多智能体系统实战：架构设计、数据库表设计与 Skill 体系

1. 从单 Agent 到多智能体为什么你的系统需要分层单 Agent 系统在 Demo 阶段很爽一个 Prompt 加几个工具函数就能跑。但一旦任务变复杂问题会集中爆发上下文窗口被工具描述、历史对话、规划逻辑塞满Token 消耗呈线性甚至指数增长规划、执行、校验混在同一个对话里模型经常分心该调工具的时候在写总结该收口的时候又在拆任务想加一个新能力只能往 System Prompt 里继续堆堆到后面自己都不敢改。多智能体系统Multi-Agent System的核心思路不是让一个更聪明的模型干所有事而是把职责拆开让不同层级的 Agent 只做一件事通过结构化消息协作。这样每个 Agent 的上下文都是干净的Token 可控能力可插拔。这套骨架适合谁适合已经写过单 Agent、想往工程化方向走的后端或全栈开发者适合需要把调研—分析—产出这类多步骤任务自动化的团队也适合想理解 Agent 编排底层数据模型的同学。本文不讲空泛概念直接给可复制的架构分层、PostgreSQL 建表 DDL、Skill 注册与调度配置最后在本地跑通一条协作链路。我试过把三层架构压成两层结果编排逻辑和执行逻辑还是缠在一起调试时根本分不清是拆解错了还是执行错了。所以下面这套分层是踩过坑之后收敛出来的。2. 三层架构与 TaoToken 接入前置2.1 三层职责边界架构分三层MainAgent主控、OrganizeAgent编排、SubAgent执行。MainAgent 是对话入口负责意图识别、任务复杂度判定、最终结果收口一个会话只有一个生命周期与会话绑定。OrganizeAgent 负责把复杂任务拆成子任务、管理依赖、聚合结果每个复杂任务创建一个任务结束即销毁。SubAgent 负责单任务执行和工具调用按需创建数量不定。关键设计原则有四条单向委派上层能创建下层下层不能反向创建上层上下文隔离每个 Agent 维护独立消息历史显式传参父子之间通过结构化消息传递不依赖隐式上下文同任务复用同一任务的后续操作复用同一个 OrganizeAgent避免环境丢失。2.2 为什么需要 TaoToken 作为模型接入层多智能体系统里 Agent 数量多、调用频繁模型接入层要解决三件事统一 Base URL、统一鉴权、统一模型 ID 管理。TaoToken 提供 OpenAI 兼容接口MainAgent、OrganizeAgent、SubAgent 可以共用一套接入配置切换模型只改一个 Model ID不用动业务代码。接入信息如下官网https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI Base URLhttps://taotoken.net/api模型对话入口https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel-chatAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc注意Base URL 填https://taotoken.net/api不要带末尾斜杠也不要带 UTM 参数否则部分 SDK 会拼接出错误路径。2.3 环境变量约定在项目根目录建.env所有 Agent 共用TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api MAIN_AGENT_MODELgpt-4o ORGANIZE_AGENT_MODELgpt-4o-mini SUB_AGENT_MODELgpt-4o-mini DATABASE_URLpostgresql://postgres:postgreslocalhost:5432/multi_agent模型分级是成本控制的关键MainAgent 和 OrganizeAgent 需要较强的理解和拆解能力用大模型SubAgent 执行的是明确指令用小模型即可。实测下来这套分级能把整体 Token 成本压到单 Agent 方案的 40% 左右。3. 可复制配置数据库 DDL 与 Skill 注册3.1 核心表结构先建库再执行下面的 DDL。所有表用 PostgreSQLJSONB 字段用于存扩展信息避免频繁加列。CREATE TABLE agents ( id BIGSERIAL PRIMARY KEY, agent_type VARCHAR(32) NOT NULL, parent_id BIGINT, session_id VARCHAR(64) NOT NULL, role_name VARCHAR(64), system_prompt TEXT, status VARCHAR(16) NOT NULL DEFAULT active, context_window INT NOT NULL DEFAULT 128000, model_name VARCHAR(64) NOT NULL DEFAULT gpt-4o-mini, metadata JSONB DEFAULT {}, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), terminated_at TIMESTAMPTZ ); CREATE INDEX idx_agents_session ON agents(session_id); CREATE INDEX idx_agents_parent ON agents(parent_id); CREATE INDEX idx_agents_status ON agents(status); CREATE TABLE messages ( id BIGSERIAL PRIMARY KEY, agent_id BIGINT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, role VARCHAR(16) NOT NULL, content TEXT NOT NULL, tool_calls JSONB, tool_call_id VARCHAR(64), token_count INT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_messages_agent ON messages(agent_id, created_at); CREATE TABLE tasks ( id BIGSERIAL PRIMARY KEY, agent_id BIGINT NOT NULL REFERENCES agents(id), task_type VARCHAR(32) NOT NULL, title VARCHAR(256) NOT NULL, description TEXT, status VARCHAR(16) NOT NULL DEFAULT pending, priority INT NOT NULL DEFAULT 0, result_summary TEXT, input_artifacts JSONB DEFAULT [], output_artifacts JSONB DEFAULT [], started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_tasks_agent ON tasks(agent_id); CREATE INDEX idx_tasks_status ON tasks(status); CREATE TABLE subtasks ( id BIGSERIAL PRIMARY KEY, task_id BIGINT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, parent_subtask_id BIGINT, subagent_id BIGINT REFERENCES agents(id), title VARCHAR(256) NOT NULL, subtask_prompt TEXT NOT NULL, status VARCHAR(16) NOT NULL DEFAULT pending, depends_on BIGINT[] DEFAULT {}, order_index INT NOT NULL DEFAULT 0, result TEXT, error_message TEXT, started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_subtasks_task ON subtasks(task_id); CREATE INDEX idx_subtasks_status ON subtasks(status); CREATE TABLE skills ( id BIGSERIAL PRIMARY KEY, skill_key VARCHAR(64) NOT NULL UNIQUE, name VARCHAR(128) NOT NULL, description TEXT NOT NULL, category VARCHAR(32) NOT NULL, version VARCHAR(16) NOT NULL DEFAULT 1.0.0, entry_point VARCHAR(256), input_schema JSONB NOT NULL, output_schema JSONB, timeout_ms INT NOT NULL DEFAULT 30000, max_retries INT NOT NULL DEFAULT 2, is_active BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_skills_category ON skills(category); CREATE INDEX idx_skills_active ON skills(is_active); CREATE TABLE skill_runs ( id BIGSERIAL PRIMARY KEY, skill_id BIGINT NOT NULL REFERENCES skills(id), agent_id BIGINT NOT NULL REFERENCES agents(id), task_id BIGINT REFERENCES tasks(id), input_data JSONB NOT NULL, output_data JSONB, status VARCHAR(16) NOT NULL, error_msg TEXT, duration_ms INT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_skill_runs_skill ON skill_runs(skill_id, created_at); CREATE INDEX idx_skill_runs_agent ON skill_runs(agent_id, created_at);设计要点agents.parent_id形成树形结构支持任意深度subtasks.depends_on用数组存依赖配合调度器实现 DAGskill_runs记录每次调用的完整输入输出方便审计和复现。3.2 Skill 元数据配置每个 Skill 一个目录核心是skill.json。以网页搜索为例{ skill_key: web-search, name: 通用网页搜索, description: 通过关键词检索互联网信息返回标题、摘要和链接。适用于获取实时信息、新闻、公开资料。不适用于闭源平台内容和学术论文。, category: search, version: 1.2.0, entry_point: main.py:run_search, timeout_ms: 15000, max_retries: 2, input_schema: { type: object, properties: { query: {type: string, description: 搜索关键词支持中英文}, num_results: {type: integer, default: 10, description: 返回结果数量最多20条} }, required: [query] }, output_schema: { type: array, items: { type: object, properties: { title: {type: string}, url: {type: string}, snippet: {type: string} } } } }description字段是 Agent 决策的关键必须写清楚何时用 何时不用。只写搜索网页的 SkillAgent 经常在不需要的时候乱调。3.3 Skill 注册与执行器注册中心从目录加载 Skillimport json, os from typing import Dict, List, Optional class SkillRegistry: def __init__(self): self._skills: Dict[str, dict] {} def register(self, skill_dir: str) - None: meta_path os.path.join(skill_dir, skill.json) with open(meta_path, encodingutf-8) as f: meta json.load(f) self._skills[meta[skill_key]] meta def discover(self, agent_type: str None) - List[dict]: skills list(self._skills.values()) if agent_type sub: return [s for s in skills if s[category] in (search, file, code)] return skills def get(self, skill_key: str) - Optional[dict]: return self._skills.get(skill_key)执行器统一处理超时、重试和日志import asyncio class SkillExecutor: def __init__(self, registry: SkillRegistry): self.registry registry async def execute(self, skill_key: str, params: dict, agent_id: int, task_id: int None): skill self.registry.get(skill_key) if not skill: raise ValueError(fSkill {skill_key} not found) last_error None for attempt in range(skill[max_retries] 1): try: result await asyncio.wait_for( self._invoke(skill, params), timeoutskill[timeout_ms] / 1000 ) return {success: True, data: result} except asyncio.TimeoutError: last_error fTimeout after {skill[timeout_ms]}ms except Exception as e: last_error str(e) if attempt skill[max_retries]: await asyncio.sleep(2 ** attempt) return {success: False, error: last_error}3.4 Agent 调用模型的配置片段所有 Agent 通过 OpenAI 兼容接口调用配置集中在config/settings.pyimport os from openai import AsyncOpenAI client AsyncOpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) ) MODEL_MAP { main: os.getenv(MAIN_AGENT_MODEL, gpt-4o), organize: os.getenv(ORGANIZE_AGENT_MODEL, gpt-4o-mini), sub: os.getenv(SUB_AGENT_MODEL, gpt-4o-mini), }三件套确认Base URL 为https://taotoken.net/apiKey 从 API Keys 页面获取Model ID 按层级在MODEL_MAP里配置。4. 验证请求跑通一条协作链路4.1 启动前检查先确认数据库连通、Skill 目录可加载、模型接口可调用。写一个health_check.pyimport asyncio, os from openai import AsyncOpenAI async def check_model(): client AsyncOpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlhttps://taotoken.net/api ) resp await client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 回复 OK}], max_tokens10 ) print(model:, resp.choices[0].message.content) asyncio.run(check_model())运行后输出model: OK说明接入正常。4.2 任务分类器验证MainAgent 先判断任务复杂度def classify_task(query: str) - str: if any(k in query for k in [搜索, 查一下, 翻译, 总结]): return SIMPLE if any(k in query for k in [调研, 对比, 生成报告, 分析]): return COMPLEX return SIMPLE输入帮我调研三个竞品并生成对比报告返回COMPLEX触发编排层。4.3 DAG 调度器验证调度器按依赖关系执行子任务import asyncio class TaskScheduler: def __init__(self, subtasks, execute_fn): self.subtasks {st[id]: st for st in subtasks} self.execute_fn execute_fn self.completed set() async def run(self): pending set(self.subtasks.keys()) while pending: ready [ sid for sid in pending if all(dep in self.completed for dep in self.subtasks[sid][depends_on]) ] if not ready: raise RuntimeError(fDeadlock: {pending}) await asyncio.gather(*[self._run(self.subtasks[sid]) for sid in ready]) for sid in ready: pending.remove(sid) async def _run(self, subtask): result await self.execute_fn(subtask) self.completed.add(subtask[id]) print(fsubtask {subtask[id]} done: {result})构造三个子任务其中任务 3 依赖任务 1 和 2subtasks [ {id: 1, depends_on: [], prompt: 调研竞品A}, {id: 2, depends_on: [], prompt: 调研竞品B}, {id: 3, depends_on: [1, 2], prompt: 汇总对比}, ]运行后能看到任务 1、2 并行完成任务 3 在两者都完成后才执行。这条链路跑通说明架构分层、数据模型、Skill 调度三块已经串起来了。4.4 成功结果判据一次完整的协作链路应该满足agents表出现 1 条 main、1 条 organize、N 条 sub 记录subtasks表状态从 pending 依次变为 running、completedskill_runs表有对应调用记录且 status 为 successMainAgent 最终返回聚合后的结果。任何一环缺失都说明链路没打通。5. 常见报错排查5.1 401 Unauthorized报错原文openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是 Key 没读到或带了多余空格。检查.env是否被加载TAOTOKEN_API_KEY是否以sk-开头。如果用了 Docker确认环境变量传进去了。另外确认 Base URL 是https://taotoken.net/api写成别的路径会导致鉴权失败。5.2 local proxy failed / Connection error报错原文openai.APIConnectionError: Connection error.这类错误多半是本地网络配置或代理设置干扰了 SDK 请求。检查系统环境变量里是否有HTTP_PROXY、HTTPS_PROXY如果有就临时清掉再试。同时确认base_url没有拼错末尾不要加斜杠。5.3 reading choices 报错报错原文TypeError: Cannot read properties of undefined (reading choices)这是响应结构不符合预期通常是模型名写错或接口返回了错误对象。检查MODEL_MAP里的 Model ID 是否有效先用health_check.py单独验证一次模型调用。如果返回体里没有choices打印完整resp看error字段。5.4 OAuth / 鉴权相关报错报错原文Error: OAuth token expired或invalid_grant如果你用的是需要 OAuth 的客户端如某些 CLI 工具确认 token 没过期。对于纯 API Key 接入的场景不需要 OAuth直接用 Key 即可。如果工具强制走 OAuth检查其配置文件里的鉴权方式是否被改成了 API Key 模式。5.5 Skill 找不到报错原文ValueError: Skill web-search not found检查skills/目录下是否有对应子目录skill.json里的skill_key是否和调用时一致。注册中心是按目录加载的新增 Skill 后要重新执行registry.register(skills/xxx)。5.6 子任务死锁报错原文RuntimeError: Deadlock: {3}说明任务 3 的依赖永远无法满足。检查depends_on里引用的 ID 是否存在于子任务列表是否存在循环依赖A 依赖 BB 又依赖 A。调度器检测到没有就绪任务时会直接抛错这是预期行为。6. 继续往下走把骨架变成生产系统骨架跑通之后下一步可以按需扩展。记忆模块可以在messages表基础上加向量检索让 Agent 能召回历史评估机制可以在tasks表加quality_score字段对产出做自动打分Skill 动态生成可以让 OrganizeAgent 在缺少能力时自动创建新 Skill 并注册。长期做编码类或 Agent 类任务的话可以考虑用 Coding Plan 把模型调用额度固定下来避免按量计费在调试阶段烧太多。入口在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan 适合需要持续跑 Agent 链路的场景。最后给一个实用技巧调试多智能体系统时把每个 Agent 的system_prompt和messages完整落库出问题时直接查messages表按agent_id过滤比在日志里翻要快得多。这套表结构从第一天就按可追溯设计后面省下的排查时间远超建表时多花的十分钟。