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

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

本文面向需要从零搭建多智能体（Multi-Agent）系统的开发者，从架构分层、数据模型到 Skill 标准化，给出一套可直接落地的设计方案。所有内容均基于生产级实践，不含空泛概念。

一、多智能体架构设计

1.1 为什么需要分层架构

单 Agent 系统在面对复杂任务时存在三个瓶颈：

• 上下文爆炸：单一对话窗口塞入工具、记忆、规划逻辑，Token 消耗线性增长
• 职责混乱：规划、执行、校验混在一起，模型容易"分心"
• 不可扩展：新增能力只能往 Prompt 里堆，最终不可维护

分层架构的核心思想是：不同层级的 Agent 只做一件事，通过消息传递协作。

1.2 三层架构模型

┌─────────────────────────────────────────┐
│ MainAgent (主控) │
│ 对话入口 / 需求理解 / 任务分发 / 收口 │
└───────────────┬─────────────────────────┘
│ 委派任务
┌───────────────▼─────────────────────────┐
│ OrganizeAgent (编排) │
│ 任务拆解 / 子任务调度 / 依赖管理 │
└───────┬───────────┬───────────┬─────────┘
│ │ │
┌────▼───┐ ┌────▼───┐ ┌────▼───┐
│SubAgent│ │SubAgent│ │SubAgent│ …
│(执行) │ │(执行) │ │(执行) │
└────────┘ └────────┘ └────────┘

各层职责边界：

1.3 消息流转机制

用户输入 → MainAgent
│
├─ 简单任务 → 直接执行（工具调用 / 文本生成）
│
└─ 复杂任务 → 创建 OrganizeAgent
│
├─ 拆解为 SubTask 1 → 创建 SubAgent A → 执行 → 返回结果
├─ 拆解为 SubTask 2 → 创建 SubAgent B → 执行 → 返回结果
├─ …
└─ 聚合所有结果 → 返回 MainAgent → 交付用户

关键设计原则：

1.4 调度策略

1.4.1 任务复杂度判定（路由规则）

MainAgent 需要先判断任务类型，再决定是否下沉：

def classify_task(query: str) –> TaskType:
"""任务分类器 – 决定是否需要编排层"""

# 1. 单线任务：直接执行
if is_single_step(query):
return TaskType.SIMPLE

# 2. 多步骤且有依赖：需要编排
if has_multiple_subtasks(query) and has_dependencies(query):
return TaskType.COMPLEX

# 3. 多源信息聚合：需要编排
if requires_multi_source(query):
return TaskType.COMPLEX

return TaskType.SIMPLE

简单任务判定条件（直接由 MainAgent 执行）：

• 单工具调用（一次搜索 / 一次计算）
• 纯文本创作（单段文案 / 翻译 / 总结）
• 概念解释 / 知识问答
• 已有产物的小幅修改

复杂任务判定条件（下沉到 OrganizeAgent）：

• 需要多个工具串行协作（如：搜索 → 分析 → 生成报告）
• 需要并行执行多个子任务再聚合（如：同时调研 3 个竞品）
• 涉及多文件 / 多页面的批量操作
• 需要环境持久化的任务（浏览器会话 / VM 操作）

1.4.2 子任务编排模式

OrganizeAgent 拆解子任务后，有三种执行模式：

二、数据库表设计

2.1 核心 ER 关系

agents (Agent实例表)
├── 1:N → messages (消息记录表)
├── 1:N → tasks (任务表)
└── N:M → skills (通过 agent_skills 关联)

tasks (任务表)
├── 1:N → subtasks (子任务表)
└── 1:N → task_runs (任务执行记录)

skills (技能表)
└── 1:N → skill_versions (技能版本表)

2.2 完整表结构 SQL（PostgreSQL）

2.2.1 Agent 实例表

CREATE TABLE agents (
id BIGSERIAL PRIMARY KEY,
agent_type VARCHAR(32) NOT NULL, — main / organize / sub
parent_id BIGINT, — 父 Agent ID，顶层为 NULL
session_id VARCHAR(64) NOT NULL, — 会话 ID，用于同会话聚合
role_name VARCHAR(64), — 角色名称，如"研究员"、"程序员"
system_prompt TEXT, — 该 Agent 的系统提示词
status VARCHAR(16) NOT NULL DEFAULT 'active', — active / terminated / error
context_window INT NOT NULL DEFAULT 128000, — 上下文窗口大小
model_name VARCHAR(64) NOT NULL DEFAULT 'gpt-4o', — 使用的模型
metadata JSONB DEFAULT '{}', — 扩展字段（VM ID、浏览器会话 ID 等）
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);

设计要点：

• parent_id 形成树形结构，支持任意深度的 Agent 层级
• metadata 用 JSONB 存储环境信息（VM 地址、浏览器会话 ID 等），避免频繁加列
• status 标记 Agent 生命周期，便于回收资源

2.2.2 消息记录表

CREATE TABLE messages (
id BIGSERIAL PRIMARY KEY,
agent_id BIGINT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
role VARCHAR(16) NOT NULL, — user / assistant / system / tool
content TEXT NOT NULL,
tool_calls JSONB, — 工具调用参数（role=assistant 时）
tool_call_id VARCHAR(64), — 工具调用返回 ID（role=tool 时）
token_count INT, — 该消息的 Token 消耗
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_messages_agent ON messages(agent_id, created_at);

设计要点：

• 每个 Agent 独立维护自己的消息历史，上下文隔离
• tool_calls 和 tool_call_id 对应 OpenAI 格式的 Function Calling 协议
• 按 created_at 排序即可还原完整对话

2.2.3 任务表

CREATE TABLE tasks (
id BIGSERIAL PRIMARY KEY,
agent_id BIGINT NOT NULL REFERENCES agents(id),
task_type VARCHAR(32) NOT NULL, — research / coding / writing / …
title VARCHAR(256) NOT NULL,
description TEXT, — 任务完整描述
status VARCHAR(16) NOT NULL DEFAULT 'pending',
— pending / running / completed / failed / cancelled
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);

2.2.4 子任务表

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, — 传给 SubAgent 的具体指令
status VARCHAR(16) NOT NULL DEFAULT 'pending',
depends_on BIGINT[] DEFAULT '{}', — 依赖的子任务 ID 列表
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);

设计要点：

• depends_on 数组存储依赖关系，支持 DAG 调度
• parent_subtask_id 支持子任务嵌套拆解（OrganizeAgent 把大子任务拆成更小的）
• order_index 用于同优先级任务的排序

2.2.5 Skill 表

CREATE TABLE skills (
id BIGSERIAL PRIMARY KEY,
skill_key VARCHAR(64) NOT NULL UNIQUE, — 技能唯一标识，如 "web-search"
name VARCHAR(128) NOT NULL,
description TEXT NOT NULL, — 技能描述（用于 Agent 决策时识别）
category VARCHAR(32) NOT NULL, — 分类：search / file / code / data / …
version VARCHAR(16) NOT NULL DEFAULT '1.0.0',
entry_point VARCHAR(256), — 执行入口（脚本路径 / 函数名）
input_schema JSONB NOT NULL, — 输入参数 JSON Schema
output_schema JSONB, — 输出格式 JSON Schema
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);

2.2.6 Skill 调用日志表

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, — success / failed / timeout
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);

2.3 表关系说明

agents (1) ────< (N) messages
│
│ 1
│
│ N
tasks (1) ────< (N) subtasks
│
│ 分配执行
│
agents (SubAgent)

skills (M) ────< (N) skill_runs >──── (1) agents

三、Skill 体系设计

3.1 什么是 Skill

Skill 是可被 Agent 发现、理解并调用的标准化能力单元。它不是简单的函数，而是包含：

• 自我描述（让 Agent 知道"你能做什么"）
• 输入输出规范（让 Agent 知道"怎么调用你"）
• 执行逻辑（真正干活的代码）
• 错误处理与重试机制

3.2 Skill 标准结构

每个 Skill 遵循统一的目录结构：

skills/
└── web-search/
├── SKILL.md # 技能说明文档（必读）
├── skill.json # 技能元数据（注册用）
├── main.py # 执行入口
├── requirements.txt # 依赖声明
└── reference/ # 参考资料（可选）
└── api_docs.md

3.2.1 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" },
"publish_time": { "type": "string" }
}
}
}
}

3.2.2 SKILL.md 文档规范

# Web Search Skill

## 功能描述
通用网页搜索能力，基于搜索引擎 API 实现。

## 何时使用
– 需要获取最新新闻、事件信息
– 需要查证公开事实、数据
– 需要查找特定领域的资料文档

## 何时不要使用
– 闭源平台内容（微信、抖音等）→ 使用 browser-task
– 学术论文 → 使用 scholar-search
– 纯计算任务 → 使用 calculator

## 输入参数
– query: 搜索关键词（必填）
– num_results: 返回数量，默认 10

## 输出格式
返回数组，每项包含 title、url、snippet、publish_time

## 注意事项
– 单次搜索最多返回 20 条结果
– 涉及多方面内容时，应拆分为多个 query 并行调用

3.3 Skill 注册与发现机制

3.3.1 注册流程

class SkillRegistry:
"""Skill 注册中心"""

def __init__(self):
self._skills: Dict[str, Skill] = {}

def register(self, skill_dir: str) –> None:
"""从目录加载并注册 Skill"""
meta_path = os.path.join(skill_dir, "skill.json")
with open(meta_path) as f:
meta = json.load(f)

skill = Skill(
key=meta["skill_key"],
meta=meta,
entry_point=meta["entry_point"]
)
self._skills[skill.key] = skill

def discover(self, agent_type: str = None) –> List[Skill]:
"""发现可用 Skill，可按 Agent 类型过滤"""
skills = list(self._skills.values())
if agent_type:
# 不同层级 Agent 可用的 Skill 范围不同
return [s for s in skills if self._is_available(s, agent_type)]
return skills

def get(self, skill_key: str) –> Optional[Skill]:
return self._skills.get(skill_key)

3.3.2 Agent 可用 Skill 分级

不是所有 Agent 都能调用所有 Skill，按层级做权限控制：

3.4 Skill 调用执行器

class SkillExecutor:
"""Skill 执行器 – 统一入口，处理超时、重试、日志"""

def __init__(self, registry: SkillRegistry):
self.registry = registry

async def execute(
self,
skill_key: str,
params: dict,
agent_id: int,
task_id: int = None
) –> SkillResult:
skill = self.registry.get(skill_key)
if not skill:
raise SkillNotFoundError(f"Skill {skill_key} not found")

# 1. 参数校验
validated_params = self._validate_params(skill, params)

# 2. 执行（带超时 + 重试）
last_error = None
for attempt in range(skill.max_retries + 1):
try:
result = await asyncio.wait_for(
self._invoke_skill(skill, validated_params),
timeout=skill.timeout_ms / 1000
)
# 3. 记录成功日志
self._log_run(skill, agent_id, task_id, params, result, "success")
return SkillResult(success=True, data=result)

except asyncio.TimeoutError:
last_error = f"Timeout after {skill.timeout_ms}ms"
except Exception as e:
last_error = str(e)

# 重试前等待
if attempt < skill.max_retries:
await asyncio.sleep(2 ** attempt) # 指数退避

# 4. 记录失败日志
self._log_run(skill, agent_id, task_id, params, None, "failed", last_error)
return SkillResult(success=False, error=last_error)

3.5 实战：开发一个新 Skill

以「PDF 内容提取」Skill 为例：

第一步：创建目录结构

mkdir -p skills/pdf-extract/reference
touch skills/pdf-extract/SKILL.md
touch skills/pdf-extract/skill.json
touch skills/pdf-extract/main.py
touch skills/pdf-extract/requirements.txt

第二步：编写 skill.json

{
"skill_key": "pdf-extract",
"name": "PDF内容提取",
"description": "从PDF文件中提取文本内容，支持按页提取和全文提取。适用于需要读取PDF文档内容的场景。",
"category": "file",
"version": "1.0.0",
"entry_point": "main.py:extract_pdf",
"timeout_ms": 30000,
"max_retries": 1,
"input_schema": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "PDF文件的本地绝对路径"
},
"pages": {
"type": "array",
"items": { "type": "integer" },
"description": "指定页码提取，为空则提取全文"
}
},
"required": ["file_path"]
}
}

第三步：编写 main.py

import fitz # PyMuPDF
from typing import List, Optional

def extract_pdf(file_path: str, pages: Optional[List[int]] = None) –> dict:
"""
提取 PDF 文本内容

Args:
file_path: PDF 文件路径
pages: 指定页码列表，从 1 开始；None 表示全文

Returns:
{
"total_pages": 10,
"content": "提取的文本内容…",
"pages_extracted": [1, 2, 3]
}
"""
doc = fitz.open(file_path)
total_pages = len(doc)

if pages is None:
page_range = range(total_pages)
else:
page_range = [p – 1 for p in pages if 1 <= p <= total_pages]

text_parts = []
extracted = []
for page_idx in page_range:
page = doc[page_idx]
text_parts.append(page.get_text())
extracted.append(page_idx + 1)

doc.close()

return {
"total_pages": total_pages,
"content": "\\n\\n".join(text_parts),
"pages_extracted": extracted
}

第四步：注册到系统

registry = SkillRegistry()
registry.register("skills/pdf-extract")

四、完整落地实操

4.1 最小可运行系统搭建

4.1.1 项目目录结构

multi-agent-system/
├── agents/
│ ├── base.py # Agent 基类
│ ├── main_agent.py # 主控 Agent
│ ├── organize_agent.py # 编排 Agent
│ └── sub_agent.py # 执行 Agent
├── skills/
│ ├── web-search/
│ ├── pdf-extract/
│ └── calculator/
├── core/
│ ├── skill_registry.py
│ ├── skill_executor.py
│ └── task_scheduler.py
├── db/
│ ├── schema.sql # 建表语句
│ └── models.py # ORM 模型
├── config/
│ └── settings.py
└── main.py # 入口

4.1.2 Agent 基类设计

from abc import ABC, abstractmethod
from typing import List, Dict, Optional
from core.skill_executor import SkillExecutor

class BaseAgent(ABC):
def __init__(
self,
agent_id: int,
system_prompt: str,
model_name: str,
skill_executor: SkillExecutor
):
self.agent_id = agent_id
self.system_prompt = system_prompt
self.model_name = model_name
self.skill_executor = skill_executor
self.messages: List[Dict] = [
{"role": "system", "content": system_prompt}
]

def add_message(self, role: str, content: str, **kwargs):
msg = {"role": role, "content": content}
msg.update(kwargs)
self.messages.append(msg)

@abstractmethod
async def run(self, user_input: str) –> str:
"""执行 Agent 主循环"""
pass

async def call_skill(self, skill_key: str, params: dict) –> dict:
"""调用 Skill"""
result = await self.skill_executor.execute(
skill_key=skill_key,
params=params,
agent_id=self.agent_id
)
return result

4.2 任务调度器实现

import asyncio
from typing import List
from db.models import SubTask

class TaskScheduler:
"""基于 DAG 的子任务调度器"""

def __init__(self, subtasks: List[SubTask], execute_fn):
self.subtasks = {st.id: st for st in subtasks}
self.execute_fn = execute_fn
self.completed = set()
self.results = {}

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(f"Deadlock detected. Pending: {pending}")

# 并行执行所有就绪任务
tasks = [
self._run_subtask(self.subtasks[sid])
for sid in ready
]
await asyncio.gather(*tasks)

for sid in ready:
pending.remove(sid)

async def _run_subtask(self, subtask: SubTask):
result = await self.execute_fn(subtask)
self.results[subtask.id] = result
self.completed.add(subtask.id)

4.3 部署建议

4.3.1 资源隔离

4.3.2 成本控制

五、常见坑与避坑指南

5.1 架构层面

❌ 坑 1：所有 Agent 共享同一份对话历史

• 后果：上下文爆炸、Token 成本飙升、Agent 角色混乱
• ✅ 正确做法：每个 Agent 独立上下文，通过显式消息传递信息

❌ 坑 2：SubAgent 可以创建 OrganizeAgent（反向委派）

• 后果：层级混乱、资源泄漏、难以调试
• ✅ 正确做法：严格单向委派，上层创建下层，下层只返回结果

5.2 数据库层面

❌ 坑 3：消息表只存一份，用 agent_id 区分

• 后果：查询单 Agent 历史时性能差，且删除不安全
• ✅ 正确做法：按 agent_id 建索引，必要时按会话分表

❌ 坑 4：Skill 入参只存字符串，不存结构化

• 后果：无法做参数审计、无法复现调用
• ✅ 正确做法：input_data / output_data 都用 JSONB 存储完整结构

5.3 Skill 层面

❌ 坑 5：Skill 描述写得太简单，Agent 不会用

• 后果：Agent 不知道什么时候该调用这个 Skill
• ✅ 正确做法：description 里写清楚"何时用 + 何时不用 + 输入输出示例"

❌ 坑 6：Skill 内部直接操作全局状态

• 后果：并发调用冲突、不可重入、难以测试
• ✅ 正确做法：Skill 是纯函数式的，输入 → 处理 → 输出，不持有状态

六、总结

一套可落地的多智能体系统，核心是三件事：

按照本文的 SQL 建表、Skill 规范和调度器实现，可以快速搭出 MVP 版本。后续再根据业务场景逐步优化：增加记忆模块、引入评估机制、支持 Agent 动态生成 Skill 等。