AI 多智能体协作系统完全教程
1. 概述与架构模式
1.1 什么是多智能体系统
多智能体系统（Multi-Agent System, MAS）是一种由多个自主智能体（Agent）组成的分布式人工智能系统。每个智能体具备独立的感知、推理、决策和行动能力，它们通过协作、竞争或协商来完成单个智能体无法独立完成的复杂任务。
在大语言模型（LLM）时代，多智能体系统已经从理论研究走向了工程实践。每个 Agent 由 LLM 驱动，配备特定的工具集、记忆系统和行为规则，通过结构化的通信协议协同工作。
1.2 为什么需要多智能体系统
复杂任务的分解：一个复杂的软件开发任务，涉及需求分析、架构设计、编码、测试、部署等多个环节。单个 Agent 难以在所有环节都保持高质量输出，而多智能体系统可以将任务分配给专业化的 Agent。
专业化与分工：就像人类团队一样，每个 Agent 可以专注于特定领域。一个 Agent 专精于 Python 后端开发，另一个专精于前端 React，还有一个专精于数据库优化。专业化带来更高的输出质量。
容错与鲁棒性：单点故障是单 Agent 系统的致命弱点。多智能体系统中，一个 Agent 的失败可以被其他 Agent 检测和补偿。
思维多样性：不同 Agent 可以采用不同的推理策略和视角，通过讨论和辩论产生更全面、更准确的结论。
1.3 架构模式
中心化架构（Hub-and-Spoke）
┌─────────┐
│ 编排器 │
│Orchestrator│
└────┬────┘
┌────────┼────────┐
▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐
│Agent A│ │Agent B│ │Agent C│
│(设计) │ │(编码) │ │(测试) │
└──────┘ └──────┘ └──────┘
特点：
- 一个中心编排器负责任务分配和协调
- Agent 之间不直接通信，都通过编排器中转
- 实现简单，易于控制和监控
- 编排器是单点故障
适用场景：任务流程明确、Agent 数量较少（<10）、需要严格控制执行顺序。
去中心化架构（Peer-to-Peer）
┌──────┐ ┌──────┐
│Agent A│◄───►│Agent B│
└──┬───┘ └───┬──┘
│ │
▼ ▼
┌──────┐ ┌──────┐
│Agent C│◄───►│Agent D│
└──────┘ └──────┘
特点：
- Agent 之间直接通信
- 无单点故障，容错性好
- 协调复杂度高，可能出现死锁
- 适合需要频繁交互的任务
适用场景：需要大量讨论和协商的任务、Agent 数量适中、对容错性要求高。
层级化架构（Hierarchical）
┌──────────┐
│ 项目经理 │
└────┬─────┘
┌─────────┼─────────┐
▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐
│组长 A │ │组长 B │ │组长 C │
└──┬───┘ └──┬───┘ └──┬───┘
┌─┴─┐ ┌─┴─┐ ┌─┴─┐
▼ ▼ ▼ ▼ ▼ ▼
A1 A2 B1 B2 C1 C2
特点：
- 多层级的管理结构
- 任务逐级分解和传递
- 适合大规模复杂项目
- 层级过深会增加通信开销
适用场景：大型项目、需要多级协调、团队规模较大。
流水线架构（Pipeline）
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│Agent A│──►│Agent B│──►│Agent C│──►│Agent D│
│(需求) │ │(设计) │ │(编码) │ │(测试) │
└──────┘ └──────┘ └──────┘ └──────┘
特点：
- 任务按顺序在 Agent 之间传递
- 每个 Agent 完成后将结果传给下一个
- 流程清晰，易于理解和调试
- 灵活性较差，难以处理分支和回退
适用场景：任务有明确的先后顺序、每个阶段的输出是下一阶段的输入。
2. 主流多智能体框架对比
2.1 AutoGen
AutoGen 是微软开源的多智能体对话框架，以对话驱动为核心设计理念。
核心概念：
AssistantAgent：由 LLM 驱动的智能体UserProxyAgent：代表人类用户的代理，可以执行代码GroupChat：多智能体群聊机制
# AutoGen 基础示例
import autogen
# 配置 LLM
config_list = [
{
"model": "gpt-4",
"api_key": "your-api-key"
}
]
# 创建智能体
assistant = autogen.AssistantAgent(
name="assistant",
llm_config={"config_list": config_list},
system_message="你是一个有帮助的AI助手。"
)
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
code_execution_config={"work_dir": "workspace"}
)
# 启动对话
user_proxy.initiate_chat(
assistant,
message="请用 Python 编写一个快速排序算法，并写单元测试。"
)
优势：微软背书、社区活跃、支持代码执行、对话管理成熟。
局限：配置较复杂、对自定义工作流支持有限。
2.2 CrewAI
CrewAI 以"团队协作"为隐喻，强调角色定义和任务编排。
核心概念：
Agent：具有角色、目标和背景的智能体Task：具体的任务定义Crew：智能体团队Process：执行流程（顺序/层级）
# CrewAI 基础示例
from crewai import Agent, Task, Crew, Process
# 定义智能体（角色）
researcher = Agent(
role="技术研究员",
goal="深入调研最新技术趋势并提供详细分析",
backstory="""你是一位资深技术研究员，拥有计算机科学博士学位，
专注于 AI 和分布式系统领域。你善于从大量信息中提取关键洞察。""",
verbose=True,
allow_delegation=False,
tools=[search_tool, web_scraper_tool]
)
writer = Agent(
role="技术作家",
goal="将技术研究成果转化为通俗易懂的文章",
backstory="""你是一位经验丰富的技术作家，擅长将复杂的技术概念
用简单的语言解释清楚。你的文章总是结构清晰、逻辑严密。""",
verbose=True,
allow_delegation=False
)
reviewer = Agent(
role="编辑审核员",
goal="确保文章质量，检查事实准确性",
backstory="""你是一位严谨的编辑，对细节有极高的要求。
你会仔细核查每一个技术细节和数据引用。""",
verbose=True,
allow_delegation=False
)
# 定义任务
research_task = Task(
description="""调研 2024 年大语言模型领域的最新进展，重点关注：
1. 新模型架构的突破
2. 训练效率的改进
3. 推理优化技术
4. 多模态能力的提升
输出一份详细的技术调研报告。""",
expected_output="一份包含至少 2000 字的技术调研报告",
agent=researcher
)
writing_task = Task(
description="""基于调研报告，撰写一篇面向开发者的技术博客文章。要求：
1. 标题吸引人
2. 用通俗的语言解释技术概念
3. 包含代码示例
4. 总结关键要点""",
expected_output="一篇 3000 字左右的技术博客文章",
agent=writer,
context=[research_task]
)
review_task = Task(
description="""审核文章质量：
1. 检查技术准确性
2. 优化语言表达
3. 确保结构合理
4. 标注需要修改的地方""",
expected_output="审核意见和修改建议",
agent=reviewer,
context=[writing_task]
)
# 组建团队
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, writing_task, review_task],
process=Process.sequential,
verbose=True
)
# 执行
result = crew.kickoff()
print(result)
优势：概念直观、角色定义清晰、支持多种执行流程、文档完善。
局限：对复杂工作流的表达能力有限、调试工具有待完善。
2.3 MetaGPT
MetaGPT 模拟软件公司的组织结构，将 SOP（标准操作流程）融入多智能体协作。
核心概念：
Role：角色，具有专业知识和行为模式Action：具体动作Message：角色间传递的消息Environment：共享环境
# MetaGPT 基础示例
from metagpt.software_company import SoftwareCompany
from metagpt.roles import ProjectManager, Architect, Engineer, QaEngineer
# 创建软件公司
company = SoftwareCompany()
# 配置团队
company.invest(
investment=20.0, # 预算
max_rounds=5 # 最大轮次
)
# 定义需求
requirement = """
开发一个命令行的待办事项管理工具，支持以下功能：
1. 添加待办事项（标题、描述、优先级、截止日期）
2. 查看待办列表（支持按优先级、日期排序）
3. 标记完成/未完成
4. 删除待办事项
5. 数据持久化（使用 JSON 文件）
6. 支持搜索功能
技术要求：
- 使用 Python 3.10+
- 使用 argparse 处理命令行参数
- 包含完整的单元测试
- 提供 README 文档
"""
# 启动项目
company.start(requirement=requirement)
优势：面向软件工程场景深度优化、SOP 驱动、输出质量高。
局限：主要针对软件开发场景、学习曲线较陡。
2.4 框架选择指南
| 维度 | AutoGen | CrewAI | MetaGPT |
|---|---|---|---|
| 学习曲线 | 中等 | 低 | 较高 |
| 灵活性 | 高 | 中 | 中 |
| 代码执行 | 内置支持 | 需配置 | 内置支持 |
| 适合场景 | 通用对话 | 任务协作 | 软件开发 |
| 社区生态 | 活跃 | 快速增长 | 稳定 |
| 生产就绪 | 是 | 接近 | 部分 |
建议：
- 快速原型 → CrewAI
- 通用多智能体应用 → AutoGen
- 软件开发自动化 → MetaGPT
- 自定义需求强烈 → 基于 LLM API 自建
3. 智能体角色定义与分工
3.1 角色定义的核心要素
一个完整的 Agent 角色定义应包含以下要素：
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum
class AgentCapability(Enum):
"""Agent 能力枚举"""
CODE_GENERATION = "code_generation"
CODE_REVIEW = "code_review"
RESEARCH = "research"
WRITING = "writing"
DATA_ANALYSIS = "data_analysis"
TESTING = "testing"
ARCHITECTURE_DESIGN = "architecture_design"
PROJECT_MANAGEMENT = "project_management"
@dataclass
class AgentProfile:
"""Agent 角色档案"""
name: str
role: str
goal: str
backstory: str
capabilities: List[AgentCapability]
tools: List[str] = field(default_factory=list)
constraints: List[str] = field(default_factory=list)
communication_style: str = "professional"
max_tokens_per_response: int = 4000
temperature: float = 0.7
def to_system_prompt(self) -> str:
"""将角色档案转换为系统提示词"""
capabilities_str = "\n".join(
f"- {cap.value}" for cap in self.capabilities
)
tools_str = "\n".join(f"- {tool}" for tool in self.tools)
constraints_str = "\n".join(
f"- {c}" for c in self.constraints
)
return f"""
## 角色：{self.name} ({self.role})
### 目标
{self.goal}
### 背景
{self.backstory}
### 能力
{capabilities_str}
### 可用工具
{tools_str}
### 行为约束
{constraints_str}
### 沟通风格
{self.communication_style}
### 规则
1. 始终保持角色一致性
2. 基于专业能力给出建议
3. 对于超出能力范围的任务，明确说明
4. 与其他 Agent 协作时保持礼貌和专业
5. 输出必须是有结构的、可执行的内容
"""
3.2 常见角色设计模式
class AgentFactory:
"""Agent 工厂 - 预定义常用角色"""
@staticmethod
def create_architect() -> AgentProfile:
return AgentProfile(
name="架构师",
role="系统架构师",
goal="设计可扩展、可维护、高性能的系统架构",
backstory="""你是一位拥有 15 年经验的系统架构师，
曾主导过多个日活千万级系统的架构设计。你精通微服务、
分布式系统、云原生架构等技术栈。""",
capabilities=[
AgentCapability.ARCHITECTURE_DESIGN,
AgentCapability.CODE_REVIEW
],
tools=["draw_diagram", "evaluate_architecture"],
constraints=[
"架构设计必须考虑可扩展性",
"优先选择成熟稳定的技术方案",
"每个架构决策必须说明理由"
],
temperature=0.5
)
@staticmethod
def create_senior_engineer(language: str = "Python") -> AgentProfile:
return AgentProfile(
name="高级工程师",
role=f"高级{language}工程师",
goal="编写高质量、可测试、符合最佳实践的代码",
backstory=f"""你是一位资深的{language}开发者，
对{language}的生态系统有深入理解。你注重代码质量，
遵循 SOLID 原则，编写清晰的文档。""",
capabilities=[
AgentCapability.CODE_GENERATION,
AgentCapability.CODE_REVIEW,
AgentCapability.TESTING
],
tools=["code_executor", "linter", "test_runner"],
constraints=[
"所有代码必须包含类型注解",
"公共 API 必须有 docstring",
"不使用已弃用的特性",
"遵循项目既定的代码风格"
],
temperature=0.3
)
@staticmethod
def create_qa_engineer() -> AgentProfile:
return AgentProfile(
name="QA工程师",
role="质量保证工程师",
goal="确保软件质量，发现并推动修复缺陷",
backstory="""你是一位严谨的QA工程师，
擅长发现边界条件和潜在缺陷。你有丰富的测试经验，
熟悉单元测试、集成测试、端到端测试等多种测试方法。""",
capabilities=[
AgentCapability.TESTING,
AgentCapability.CODE_REVIEW
],
tools=["test_runner", "coverage_analyzer", "bug_tracker"],
constraints=[
"测试用例必须覆盖正常流程和异常流程",
"每个 bug 报告必须包含复现步骤",
"测试代码和产品代码同等重要"
],
temperature=0.4
)
@staticmethod
def create_tech_writer() -> AgentProfile:
return AgentProfile(
name="技术文档工程师",
role="技术文档工程师",
goal="创建清晰、准确、易于理解的技术文档",
backstory="""你是一位经验丰富的技术文档工程师，
擅长将复杂的技术概念转化为开发者友好的文档。
你了解文档即代码的理念，使用 Markdown 和自动化工具。""",
capabilities=[
AgentCapability.WRITING,
AgentCapability.RESEARCH
],
tools=["markdown_editor", "diagram_tool"],
constraints=[
"文档必须包含代码示例",
"术语使用必须一致",
"每个章节应有明确的学习目标"
],
temperature=0.6
)
3.3 动态角色分配
class DynamicRoleAssigner:
"""根据任务动态分配角色"""
def __init__(self):
self.factory = AgentFactory()
self.role_map = {
"architecture": self.factory.create_architect,
"coding": self.factory.create_senior_engineer,
"testing": self.factory.create_qa_engineer,
"documentation": self.factory.create_tech_writer,
}
def analyze_task(self, task_description: str) -> List[str]:
"""分析任务需要哪些角色"""
required_roles = []
keywords_map = {
"architecture": ["架构", "设计", "系统设计", "微服务", "扩展"],
"coding": ["实现", "编码", "开发", "功能", "接口"],
"testing": ["测试", "质量", "bug", "缺陷", "验证"],
"documentation": ["文档", "说明", "README", "教程", "API文档"],
}
for role, keywords in keywords_map.items():
if any(kw in task_description for kw in keywords):
required_roles.append(role)
return required_roles if required_roles else ["coding"]
def assign_roles(self, task_description: str) -> List[AgentProfile]:
"""为任务分配角色"""
required_roles = self.analyze_task(task_description)
return [self.role_map[role]() for role in required_roles]
4. 通信协议设计
4.1 消息传递机制
消息传递是多智能体系统最基本的通信方式。
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional
import json
import uuid
class MessageType(Enum):
"""消息类型"""
TASK = "task" # 任务分配
RESULT = "result" # 任务结果
QUERY = "query" # 查询请求
RESPONSE = "response" # 查询响应
FEEDBACK = "feedback" # 反馈意见
BROADCAST = "broadcast" # 广播消息
HEARTBEAT = "heartbeat" # 心跳检测
ERROR = "error" # 错误通知
class MessagePriority(Enum):
"""消息优先级"""
LOW = 0
NORMAL = 1
HIGH = 2
URGENT = 3
@dataclass
class Message:
"""消息定义"""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
sender: str = ""
receiver: str = ""
msg_type: MessageType = MessageType.TASK
priority: MessagePriority = MessagePriority.NORMAL
content: Any = None
metadata: Dict = field(default_factory=dict)
timestamp: str = field(
default_factory=lambda: datetime.now().isoformat()
)
reply_to: Optional[str] = None
def to_dict(self) -> Dict:
return {
"id": self.id,
"sender": self.sender,
"receiver": self.receiver,
"type": self.msg_type.value,
"priority": self.priority.value,
"content": self.content,
"metadata": self.metadata,
"timestamp": self.timestamp,
"reply_to": self.reply_to
}
def to_prompt_format(self) -> str:
"""转换为适合 LLM 理解的格式"""
return f"""
[{self.msg_type.value.upper()}] from {self.sender} to {self.receiver}
Priority: {self.priority.name}
Time: {self.timestamp}
---
{self.content}
"""
class MessageBus:
"""消息总线 - 管理 Agent 间通信"""
def __init__(self):
self.subscribers: Dict[str, List[Callable]] = {}
self.message_queue: Dict[str, List[Message]] = {}
self.message_history: List[Message] = []
self.dead_letter_queue: List[Message] = []
def subscribe(self, agent_name: str, callback: Callable):
"""Agent 订阅消息"""
if agent_name not in self.subscribers:
self.subscribers[agent_name] = []
self.subscribers[agent_name].append(callback)
def publish(self, message: Message) -> bool:
"""发布消息"""
self.message_history.append(message)
# 直接消息
if message.receiver and message.receiver in self.subscribers:
if message.receiver not in self.message_queue:
self.message_queue[message.receiver] = []
self.message_queue[message.receiver].append(message)
# 触发回调
for callback in self.subscribers.get(message.receiver, []):
try:
callback(message)
except Exception as e:
print(f"消息处理失败: {e}")
return True
# 广播消息
if message.msg_type == MessageType.BROADCAST:
for agent_name in self.subscribers:
if agent_name != message.sender:
if agent_name not in self.message_queue:
self.message_queue[agent_name] = []
self.message_queue[agent_name].append(message)
return True
# 无法投递
self.dead_letter_queue.append(message)
return False
def get_messages(self, agent_name: str,
msg_type: MessageType = None) -> List[Message]:
"""获取 Agent 的消息"""
messages = self.message_queue.get(agent_name, [])
if msg_type:
messages = [m for m in messages if m.msg_type == msg_type]
return messages
def clear_messages(self, agent_name: str):
"""清除已处理的消息"""
self.message_queue[agent_name] = []
4.2 共享内存机制
共享内存提供了一种更高效的数据共享方式，适合大数据量的场景。
import threading
import time
from typing import Any, Dict, Optional
class SharedMemory:
"""共享内存空间"""
def __init__(self):
self._store: Dict[str, Any] = {}
self._locks: Dict[str, threading.Lock] = {}
self._global_lock = threading.Lock()
self._access_log: List[Dict] = []
self._watchers: Dict[str, List[Callable]] = {}
def put(self, key: str, value: Any, owner: str = "system"):
"""写入数据"""
with self._global_lock:
if key not in self._locks:
self._locks[key] = threading.Lock()
with self._locks[key]:
old_value = self._store.get(key)
self._store[key] = value
self._access_log.append({
"action": "put",
"key": key,
"owner": owner,
"timestamp": datetime.now().isoformat()
})
# 通知观察者
if key in self._watchers:
for callback in self._watchers[key]:
try:
callback(key, old_value, value)
except Exception as e:
print(f"Watcher 回调失败: {e}")
def get(self, key: str, default: Any = None,
reader: str = "system") -> Any:
"""读取数据"""
with self._global_lock:
if key not in self._locks:
self._locks[key] = threading.Lock()
with self._locks[key]:
value = self._store.get(key, default)
self._access_log.append({
"action": "get",
"key": key,
"reader": reader,
"timestamp": datetime.now().isoformat()
})
return value
def watch(self, key: str, callback: Callable):
"""监视某个键的变化"""
if key not in self._watchers:
self._watchers[key] = []
self._watchers[key].append(callback)
def list_keys(self, prefix: str = "") -> List[str]:
"""列出所有键"""
if prefix:
return [k for k in self._store if k.startswith(prefix)]
return list(self._store.keys())
def delete(self, key: str, owner: str = "system") -> bool:
"""删除数据"""
with self._locks.get(key, threading.Lock()):
if key in self._store:
del self._store[key]
self._access_log.append({
"action": "delete",
"key": key,
"owner": owner,
"timestamp": datetime.now().isoformat()
})
return True
return False
def snapshot(self) -> Dict:
"""获取当前状态快照"""
return {
"store": dict(self._store),
"key_count": len(self._store),
"access_count": len(self._access_log)
}
class SharedContext:
"""基于共享内存的上下文管理"""
def __init__(self):
self.memory = SharedMemory()
def set_task_context(self, task_id: str, context: Dict):
"""设置任务上下文"""
self.memory.put(f"task:{task_id}:context", context, owner="system")
def get_task_context(self, task_id: str) -> Dict:
"""获取任务上下文"""
return self.memory.get(f"task:{task_id}:context", {})
def set_agent_output(self, agent_name: str, task_id: str, output: Any):
"""保存 Agent 输出"""
key = f"output:{agent_name}:{task_id}"
self.memory.put(key, output, owner=agent_name)
def get_agent_output(self, agent_name: str, task_id: str) -> Any:
"""获取 Agent 输出"""
return self.memory.get(f"output:{agent_name}:{task_id}")
def get_all_outputs(self, task_id: str) -> Dict[str, Any]:
"""获取某个任务的所有 Agent 输出"""
outputs = {}
for key in self.memory.list_keys(f"output:"):
parts = key.split(":")
if len(parts) >= 3 and parts[2] == task_id:
agent_name = parts[1]
outputs[agent_name] = self.memory.get(key)
return outputs
4.3 事件驱动架构
from collections import defaultdict
import asyncio
from typing import Callable, Dict, List, Any
class EventBus:
"""事件总线"""
def __init__(self):
self._handlers: Dict[str, List[Callable]] = defaultdict(list)
self._event_history: List[Dict] = []
def on(self, event_name: str, handler: Callable):
"""注册事件处理器"""
self._handlers[event_name].append(handler)
def emit(self, event_name: str, data: Any = None,
source: str = "system"):
"""触发事件"""
event = {
"name": event_name,
"data": data,
"source": source,
"timestamp": datetime.now().isoformat()
}
self._event_history.append(event)
for handler in self._handlers.get(event_name, []):
try:
handler(event)
except Exception as e:
self.emit("error", {
"handler": handler.__name__,
"error": str(e),
"original_event": event_name
})
async def emit_async(self, event_name: str, data: Any = None,
source: str = "system"):
"""异步触发事件"""
event = {
"name": event_name,
"data": data,
"source": source,
"timestamp": datetime.now().isoformat()
}
self._event_history.append(event)
tasks = []
for handler in self._handlers.get(event_name, []):
if asyncio.iscoroutinefunction(handler):
tasks.append(handler(event))
else:
handler(event)
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
def get_history(self, event_name: str = None) -> List[Dict]:
"""获取事件历史"""
if event_name:
return [e for e in self._event_history
if e["name"] == event_name]
return self._event_history
# 使用示例
event_bus = EventBus()
# 注册事件处理器
def on_task_completed(event):
print(f"任务完成: {event['data']['task_id']}")
def on_error(event):
print(f"错误发生: {event['data']}")
event_bus.on("task_completed", on_task_completed)
event_bus.on("error", on_error)
# 触发事件
event_bus.emit("task_completed", {
"task_id": "task-001",
"result": "成功",
"agent": "engineer"
}, source="orchestrator")
5. 任务分解与调度策略
5.1 任务分解
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from enum import Enum
import uuid
class TaskStatus(Enum):
PENDING = "pending"
ASSIGNED = "assigned"
IN_PROGRESS = "in_progress"
REVIEW = "review"
COMPLETED = "completed"
FAILED = "failed"
BLOCKED = "blocked"
class TaskPriority(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
CRITICAL = 4
@dataclass
class Task:
"""任务定义"""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
title: str = ""
description: str = ""
status: TaskStatus = TaskStatus.PENDING
priority: TaskPriority = TaskPriority.MEDIUM
assigned_to: Optional[str] = None
dependencies: List[str] = field(default_factory=list)
subtasks: List[str] = field(default_factory=list)
parent: Optional[str] = None
estimated_tokens: int = 0
actual_tokens: int = 0
result: Any = None
created_at: str = field(
default_factory=lambda: datetime.now().isoformat()
)
completed_at: Optional[str] = None
class TaskDecomposer:
"""任务分解器"""
def __init__(self, llm_func: Callable):
self.llm_func = llm_func
def decompose(self, task_description: str,
max_depth: int = 3) -> List[Task]:
"""将复杂任务分解为子任务"""
decomposition_prompt = f"""
请将以下任务分解为可独立执行的子任务。
任务描述：
{task_description}
要求：
1. 每个子任务应该是原子的，可以由一个 Agent 独立完成
2. 明确子任务之间的依赖关系
3. 每个子任务应有明确的完成标准
4. 子任务数量控制在 3-8 个
请以 JSON 格式输出：
[
{{
"title": "子任务标题",
"description": "详细描述",
"priority": "high/medium/low",
"dependencies": ["依赖的子任务标题"],
"estimated_tokens": 预估Token数
}}
]
"""
result = self.llm_func(decomposition_prompt)
subtask_dicts = json.loads(result)
tasks = []
for st in subtask_dicts:
task = Task(
title=st["title"],
description=st["description"],
priority=TaskPriority[st["priority"].upper()],
estimated_tokens=st.get("estimated_tokens", 1000)
)
tasks.append(task)
# 建立依赖关系
title_to_id = {t.title: t.id for t in tasks}
for i, st in enumerate(subtask_dicts):
for dep_title in st.get("dependencies", []):
if dep_title in title_to_id:
tasks[i].dependencies.append(title_to_id[dep_title])
return tasks
5.2 任务调度器
import heapq
from collections import defaultdict
class TaskScheduler:
"""任务调度器"""
def __init__(self):
self.tasks: Dict[str, Task] = {}
self.agent_assignments: Dict[str, List[str]] = defaultdict(list)
self.execution_order: List[str] = []
def add_task(self, task: Task):
"""添加任务"""
self.tasks[task.id] = task
def get_ready_tasks(self) -> List[Task]:
"""获取可以执行的任务（所有依赖已完成）"""
ready = []
for task in self.tasks.values():
if task.status != TaskStatus.PENDING:
continue
deps_met = all(
self.tasks[dep_id].status == TaskStatus.COMPLETED
for dep_id in task.dependencies
if dep_id in self.tasks
)
if deps_met:
ready.append(task)
# 按优先级排序
ready.sort(key=lambda t: t.priority.value, reverse=True)
return ready
def assign_task(self, task: Task, agent_name: str) -> bool:
"""分配任务给 Agent"""
if task.status != TaskStatus.PENDING:
return False
task.status = TaskStatus.ASSIGNED
task.assigned_to = agent_name
self.agent_assignments[agent_name].append(task.id)
return True
def complete_task(self, task_id: str, result: Any = None):
"""标记任务完成"""
task = self.tasks[task_id]
task.status = TaskStatus.COMPLETED
task.result = result
task.completed_at = datetime.now().isoformat()
def fail_task(self, task_id: str, reason: str):
"""标记任务失败"""
task = self.tasks[task_id]
task.status = TaskStatus.FAILED
task.result = {"error": reason}
def get_execution_plan(self) -> List[List[str]]:
"""获取执行计划（拓扑排序）"""
in_degree = defaultdict(int)
graph = defaultdict(list)
for task in self.tasks.values():
if task.id not in in_degree:
in_degree[task.id] = 0
for dep in task.dependencies:
graph[dep].append(task.id)
in_degree[task.id] += 1
# BFS 拓扑排序
levels = []
queue = [tid for tid, deg in in_degree.items() if deg == 0]
while queue:
levels.append(queue)
next_queue = []
for tid in queue:
for neighbor in graph[tid]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
next_queue.append(neighbor)
queue = next_queue
return levels
def get_progress(self) -> Dict:
"""获取进度信息"""
total = len(self.tasks)
completed = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.COMPLETED)
failed = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.FAILED)
in_progress = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.IN_PROGRESS)
return {
"total": total,
"completed": completed,
"failed": failed,
"in_progress": in_progress,
"pending": total - completed - failed - in_progress,
"completion_rate": completed / total if total > 0 else 0
}
5.3 自适应调度策略
class AdaptiveScheduler(TaskScheduler):
"""自适应调度器 - 根据 Agent 表现动态调整"""
def __init__(self):
super().__init__()
self.agent_performance: Dict[str, Dict] = defaultdict(lambda: {
"completed": 0,
"failed": 0,
"avg_quality": 0.0,
"avg_tokens": 0,
"total_tokens": 0
})
def record_performance(self, agent_name: str, task_id: str,
quality_score: float, tokens_used: int):
"""记录 Agent 表现"""
perf = self.agent_performance[agent_name]
total = perf["completed"] + perf["failed"]
perf["avg_quality"] = (
(perf["avg_quality"] * total + quality_score) / (total + 1)
)
perf["total_tokens"] += tokens_used
perf["avg_tokens"] = perf["total_tokens"] // (total + 1)
if quality_score >= 0.7:
perf["completed"] += 1
else:
perf["failed"] += 1
def select_best_agent(self, task: Task,
available_agents: List[str]) -> str:
"""为任务选择最佳 Agent"""
best_agent = None
best_score = -1
for agent in available_agents:
perf = self.agent_performance[agent]
# 计算综合得分
score = (
perf["avg_quality"] * 0.5 +
(perf["completed"] / max(perf["completed"] + perf["failed"], 1)) * 0.3 +
(1 - min(perf["avg_tokens"] / 10000, 1)) * 0.2 # Token 效率
)
if score > best_score:
best_score = score
best_agent = agent
return best_agent or available_agents[0]
def should_reassign(self, task: Task) -> bool:
"""判断是否需要重新分配任务"""
if task.status != TaskStatus.IN_PROGRESS:
return False
# 检查是否超时（基于预估 Token）
if task.actual_tokens > task.estimated_tokens * 2:
return True
return False
6. 冲突解决与共识机制
6.1 冲突类型与检测
class ConflictType(Enum):
"""冲突类型"""
OPINION = "opinion" # 意见分歧
RESOURCE = "resource" # 资源竞争
PRIORITY = "priority" # 优先级冲突
APPROACH = "approach" # 方案分歧
class ConflictDetector:
"""冲突检测器"""
def detect_opinion_conflict(self, responses: Dict[str, str]) -> bool:
"""检测意见冲突"""
# 简单实现：通过 LLM 判断是否存在分歧
if len(responses) < 2:
return False
conflict_prompt = f"""
请判断以下多个 Agent 的回答是否存在实质性分歧：
{json.dumps(responses, ensure_ascii=False, indent=2)}
如果存在分歧，回答 "YES"，否则回答 "NO"。
"""
result = self.llm_func(conflict_prompt)
return "YES" in result.upper()
def detect_approach_conflict(self,
proposals: Dict[str, Dict]) -> bool:
"""检测方案冲突"""
if len(proposals) < 2:
return False
# 比较方案的关键决策点
key_decisions = set()
for proposal in proposals.values():
key_decisions.update(proposal.get("key_decisions", []))
# 如果关键决策点差异较大，可能存在冲突
return len(key_decisions) > len(proposals) * 2
6.2 共识机制
class ConsensusMechanism:
"""共识机制"""
def __init__(self, llm_func: Callable):
self.llm_func = llm_func
def majority_vote(self, options: Dict[str, str]) -> str:
"""多数投票"""
from collections import Counter
votes = Counter(options.values())
return votes.most_common(1)[0][0]
def weighted_vote(self, options: Dict[str, str],
weights: Dict[str, float]) -> str:
"""加权投票"""
scores = defaultdict(float)
for agent, option in options.items():
scores[option] += weights.get(agent, 1.0)
return max(scores, key=scores.get)
def debate_and_synthesize(self, topic: str,
positions: Dict[str, str]) -> str:
"""辩论与综合"""
debate_prompt = f"""
我们正在讨论以下话题：{topic}
以下是不同专家的观点：
"""
for agent, position in positions.items():
debate_prompt += f"**{agent}** 的观点：\n{position}\n\n"
debate_prompt += """
请综合以上所有观点，给出一个平衡、全面的结论。要求：
1. 承认各方观点的合理性
2. 指出共识点和分歧点
3. 给出综合建议
4. 如果无法达成共识，说明原因并给出多个可行方案
"""
return self.llm_func(debate_prompt)
def iterative_refinement(self, initial_proposal: str,
reviewers: List[str],
max_rounds: int = 3) -> str:
"""迭代精炼"""
current = initial_proposal
for round_num in range(max_rounds):
feedbacks = {}
for reviewer in reviewers:
review_prompt = f"""
请审查以下方案，给出具体的改进建议：
{current}
要求：
1. 指出优点
2. 指出不足
3. 给出具体的改进建议
4. 对方案整体评分（1-10）
"""
feedbacks[reviewer] = self.llm_func(review_prompt)
# 检查是否已达成共识
scores = []
for feedback in feedbacks.values():
import re
score_match = re.search(r'(\d+)(?:/10|分)', feedback)
if score_match:
scores.append(int(score_match.group(1)))
if scores and min(scores) >= 8:
break
# 综合反馈改进方案
refine_prompt = f"""
当前方案：
{current}
收到的反馈：
{json.dumps(feedbacks, ensure_ascii=False, indent=2)}
请根据反馈改进方案，输出改进后的完整方案。
"""
current = self.llm_func(refine_prompt)
return current
7. 人机协作模式
7.1 人类参与节点
from enum import Enum
from typing import Callable, Any
class HumanInterventionType(Enum):
"""人类干预类型"""
APPROVAL = "approval" # 审批
REVIEW = "review" # 审查
DECISION = "decision" # 决策
GUIDANCE = "guidance" # 指导
ESCALATION = "escalation" # 升级
class HumanInTheLoop:
"""人机协作管理器"""
def __init__(self, human_callback: Callable):
"""
Args:
human_callback: 获取人类输入的回调函数
签名: (prompt: str, options: list) -> str
"""
self.human_callback = human_callback
self.intervention_points: List[Dict] = []
def request_approval(self, action: str, context: str,
auto_approve_threshold: float = 0.9,
confidence: float = 0.0) -> bool:
"""请求审批"""
if confidence >= auto_approve_threshold:
self.intervention_points.append({
"type": "auto_approval",
"action": action,
"confidence": confidence
})
return True
prompt = f"""
## 需要您的审批
**操作**：{action}
**上下文**：{context}
**AI 置信度**：{confidence:.0%}
请决定：
1. 批准执行
2. 拒绝
3. 修改后执行
"""
response = self.human_callback(prompt,
["批准", "拒绝", "修改"])
return response == "批准"
def request_decision(self, question: str,
options: List[Dict[str, str]],
context: str = "") -> str:
"""请求人类决策"""
prompt = f"""
## 需要您的决策
**问题**：{question}
**上下文**：{context}
**选项**：
"""
option_texts = []
for i, opt in enumerate(options, 1):
prompt += f"{i}. {opt['label']}: {opt['description']}\n"
option_texts.append(opt['label'])
return self.human_callback(prompt, option_texts)
def request_review(self, content: str,
review_criteria: List[str]) -> Dict:
"""请求人类审查"""
prompt = f"""
## 需要您的审查
**内容**：
{content}
**审查标准**：
"""
for criteria in review_criteria:
prompt += f"- {criteria}\n"
prompt += """
请提供：
1. 是否通过（通过/不通过）
2. 具体意见
3. 修改建议（如有）
"""
response = self.human_callback(prompt, ["通过", "不通过"])
return {
"approved": response == "通过",
"response": response
}
7.2 混合智能工作流
class HybridWorkflow:
"""混合智能工作流 - AI 与人类协作"""
def __init__(self, human_callback: Callable,
llm_func: Callable):
self.hitl = HumanInTheLoop(human_callback)
self.llm_func = llm_func
self.workflow_log: List[Dict] = []
def execute_with_guardrails(self, task: str,
agent_func: Callable,
review_required: bool = True) -> Any:
"""带护栏的执行流程"""
# 第一步：AI 执行
self._log("ai_execution_start", {"task": task})
result = agent_func(task)
self._log("ai_execution_complete", {"result": result})
# 第二步：AI 自评
self_assessment = self._self_evaluate(task, result)
self._log("self_assessment", self_assessment)
# 第三步：根据置信度决定是否需要人类审查
if review_required or self_assessment["confidence"] < 0.8:
review = self.hitl.request_review(
content=result,
review_criteria=[
"技术准确性",
"完整性",
"可执行性",
"安全性"
]
)
self._log("human_review", review)
if not review["approved"]:
# 第四步：根据反馈修改
revised = self._revise_based_on_feedback(
task, result, review.get("response", "")
)
self._log("revision", {"revised": revised})
return revised
return result
def _self_evaluate(self, task: str, result: Any) -> Dict:
"""AI 自评"""
eval_prompt = f"""
请评估以下任务执行结果的质量。
任务：{task}
结果：{result}
评估维度：
1. 准确性（1-10）
2. 完整性（1-10）
3. 可用性（1-10）
请以 JSON 格式输出：
{{"accuracy": 8, "completeness": 7, "usability": 9, "confidence": 0.8}}
"""
response = self.llm_func(eval_prompt)
return json.loads(response)
def _revise_based_on_feedback(self, task: str,
original: Any,
feedback: str) -> Any:
"""根据反馈修改"""
revise_prompt = f"""
原始任务：{task}
原始结果：
{original}
人类反馈：
{feedback}
请根据反馈修改结果，输出修改后的完整内容。
"""
return self.llm_func(revise_prompt)
def _log(self, event: str, data: Any):
"""记录工作流日志"""
self.workflow_log.append({
"event": event,
"data": data,
"timestamp": datetime.now().isoformat()
})
8. 多智能体调试与监控
8.1 系统监控
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class AgentMetrics:
"""Agent 性能指标"""
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
total_tokens: int = 0
total_latency: float = 0.0
avg_quality_score: float = 0.0
@property
def success_rate(self) -> float:
return (self.successful_calls / self.total_calls
if self.total_calls > 0 else 0)
@property
def avg_latency(self) -> float:
return (self.total_latency / self.total_calls
if self.total_calls > 0 else 0)
class SystemMonitor:
"""系统监控器"""
def __init__(self):
self.agent_metrics: Dict[str, AgentMetrics] = defaultdict(AgentMetrics)
self.system_events: List[Dict] = []
self.alerts: List[Dict] = []
self.alert_thresholds = {
"failure_rate": 0.3,
"avg_latency": 30.0,
"token_budget": 100000
}
def record_call(self, agent_name: str, success: bool,
tokens: int, latency: float, quality: float = None):
"""记录 Agent 调用"""
metrics = self.agent_metrics[agent_name]
metrics.total_calls += 1
if success:
metrics.successful_calls += 1
else:
metrics.failed_calls += 1
metrics.total_tokens += tokens
metrics.total_latency += latency
if quality is not None:
n = metrics.successful_calls
metrics.avg_quality_score = (
(metrics.avg_quality_score * (n - 1) + quality) / n
)
# 检查告警条件
self._check_alerts(agent_name, metrics)
def _check_alerts(self, agent_name: str, metrics: AgentMetrics):
"""检查是否需要告警"""
if (metrics.total_calls >= 10 and
metrics.success_rate < (1 - self.alert_thresholds["failure_rate"])):
self._create_alert(
"HIGH",
f"Agent '{agent_name}' 失败率过高: "
f"{(1-metrics.success_rate):.0%}"
)
if metrics.avg_latency > self.alert_thresholds["avg_latency"]:
self._create_alert(
"MEDIUM",
f"Agent '{agent_name}' 响应延迟过高: "
f"{metrics.avg_latency:.1f}s"
)
def _create_alert(self, level: str, message: str):
"""创建告警"""
alert = {
"level": level,
"message": message,
"timestamp": datetime.now().isoformat()
}
self.alerts.append(alert)
print(f"⚠️ [{level}] {message}")
def get_dashboard(self) -> Dict:
"""获取监控仪表盘数据"""
return {
"agents": {
name: {
"total_calls": m.total_calls,
"success_rate": f"{m.success_rate:.1%}",
"avg_latency": f"{m.avg_latency:.2f}s",
"total_tokens": m.total_tokens,
"avg_quality": f"{m.avg_quality_score:.2f}"
}
for name, m in self.agent_metrics.items()
},
"system": {
"total_events": len(self.system_events),
"active_alerts": len([a for a in self.alerts
if a["level"] == "HIGH"]),
"total_agents": len(self.agent_metrics)
},
"recent_alerts": self.alerts[-5:] if self.alerts else []
}
def print_dashboard(self):
"""打印监控仪表盘"""
dashboard = self.get_dashboard()
print("\n" + "=" * 60)
print("📊 多智能体系统监控仪表盘")
print("=" * 60)
for name, stats in dashboard["agents"].items():
print(f"\n🤖 {name}")
print(f" 调用次数: {stats['total_calls']}")
print(f" 成功率: {stats['success_rate']}")
print(f" 平均延迟: {stats['avg_latency']}")
print(f" Token 消耗: {stats['total_tokens']}")
if dashboard["recent_alerts"]:
print(f"\n⚠️ 最近告警:")
for alert in dashboard["recent_alerts"]:
print(f" [{alert['level']}] {alert['message']}")
print("=" * 60)
8.2 调试工具
class AgentDebugger:
"""Agent 调试器"""
def __init__(self):
self.debug_log: List[Dict] = []
self.breakpoints: Set[str] = set()
self.step_mode: bool = False
def log_interaction(self, sender: str, receiver: str,
message: str, response: str,
tokens_used: int, latency: float):
"""记录交互日志"""
entry = {
"sender": sender,
"receiver": receiver,
"message_preview": message[:200] + "..." if len(message) > 200 else message,
"response_preview": response[:200] + "..." if len(response) > 200 else response,
"tokens": tokens_used,
"latency": latency,
"timestamp": datetime.now().isoformat()
}
self.debug_log.append(entry)
def trace_conversation(self, conversation_id: str) -> List[Dict]:
"""追踪完整对话链"""
# 根据 conversation_id 过滤相关日志
return [
entry for entry in self.debug_log
if conversation_id in str(entry.get("metadata", {}))
]
def analyze_token_distribution(self) -> Dict:
"""分析 Token 分布"""
by_agent = defaultdict(int)
for entry in self.debug_log:
by_agent[entry["receiver"]] += entry["tokens"]
total = sum(by_agent.values())
return {
agent: {
"tokens": tokens,
"percentage": f"{tokens/total:.1%}" if total > 0 else "0%"
}
for agent, tokens in by_agent.items()
}
def generate_debug_report(self) -> str:
"""生成调试报告"""
token_dist = self.analyze_token_distribution()
report = "# 调试报告\n\n"
report += f"## 总览\n"
report += f"- 总交互次数: {len(self.debug_log)}\n"
report += f"- 总 Token 消耗: {sum(e['tokens'] for e in self.debug_log)}\n"
report += "\n## Token 分布\n"
for agent, stats in token_dist.items():
report += f"- {agent}: {stats['tokens']} ({stats['percentage']})\n"
report += "\n## 最近交互\n"
for entry in self.debug_log[-10:]:
report += f"\n### {entry['sender']} → {entry['receiver']}\n"
report += f"- Token: {entry['tokens']}\n"
report += f"- 延迟: {entry['latency']:.2f}s\n"
return report
9. 成本控制与 Token 优化
9.1 Token 预算管理
class TokenBudgetManager:
"""Token 预算管理器"""
def __init__(self, total_budget: int):
self.total_budget = total_budget
self.allocated: Dict[str, int] = {}
self.consumed: Dict[str, int] = {}
self.history: List[Dict] = []
def allocate(self, agent_name: str, budget: int) -> bool:
"""为 Agent 分配 Token 预算"""
total_allocated = sum(self.allocated.values())
if total_allocated + budget > self.total_budget:
return False
self.allocated[agent_name] = budget
self.consumed[agent_name] = 0
return True
def consume(self, agent_name: str, tokens: int) -> bool:
"""记录 Token 消耗"""
if agent_name not in self.allocated:
return False
remaining = self.allocated[agent_name] - self.consumed[agent_name]
if tokens > remaining:
return False
self.consumed[agent_name] += tokens
self.history.append({
"agent": agent_name,
"tokens": tokens,
"timestamp": datetime.now().isoformat()
})
return True
def get_status(self) -> Dict:
"""获取预算状态"""
total_consumed = sum(self.consumed.values())
return {
"total_budget": self.total_budget,
"total_consumed": total_consumed,
"remaining": self.total_budget - total_consumed,
"utilization": f"{total_consumed/self.total_budget:.1%}",
"by_agent": {
agent: {
"allocated": self.allocated[agent],
"consumed": self.consumed[agent],
"remaining": self.allocated[agent] - self.consumed[agent],
"utilization": f"{self.consumed[agent]/self.allocated[agent]:.1%}"
}
for agent in self.allocated
}
}
def should_throttle(self, agent_name: str) -> bool:
"""判断是否需要节流"""
if agent_name not in self.allocated:
return True
utilization = (self.consumed[agent_name] /
self.allocated[agent_name])
return utilization > 0.9 # 超过 90% 开始节流
9.2 Token 优化策略
class TokenOptimizer:
"""Token 优化器"""
def __init__(self):
self.cache: Dict[str, str] = {}
self.compression_stats = {
"original_tokens": 0,
"compressed_tokens": 0
}
def compress_context(self, context: str,
max_tokens: int = 2000) -> str:
"""压缩上下文"""
# 简单的截断策略（实际应用中可使用更复杂的摘要方法）
words = context.split()
if len(words) <= max_tokens // 1.3: # 粗略估计
return context
# 保留开头和结尾，中间用摘要替代
head = " ".join(words[:200])
tail = " ".join(words[-200:])
return f"{head}\n\n[... 中间内容已省略 ...]\n\n{tail}"
def cache_response(self, prompt_hash: str, response: str):
"""缓存响应"""
self.cache[prompt_hash] = response
def get_cached(self, prompt_hash: str) -> Optional[str]:
"""获取缓存"""
return self.cache.get(prompt_hash)
def optimize_messages(self, messages: List[Dict],
max_history: int = 10) -> List[Dict]:
"""优化消息历史"""
if len(messages) <= max_history:
return messages
# 保留系统消息
system_msgs = [m for m in messages if m["role"] == "system"]
other_msgs = [m for m in messages if m["role"] != "system"]
# 保留最近的消息
recent = other_msgs[-max_history:]
# 如果有被截断的消息，添加摘要
if len(other_msgs) > max_history:
omitted = other_msgs[:-max_history]
summary = f"[已省略 {len( omitted)} 条历史消息]"
recent = [{"role": "system", "content": summary}] + recent
return system_msgs + recent
def select_optimal_model(self, task_complexity: str,
budget_remaining: int) -> str:
"""根据任务复杂度和预算选择最优模型"""
model_tiers = {
"simple": {"model": "gpt-3.5-turbo", "cost_per_1k": 0.002},
"medium": {"model": "gpt-4-turbo", "cost_per_1k": 0.01},
"complex": {"model": "gpt-4", "cost_per_1k": 0.03}
}
tier = model_tiers.get(task_complexity, model_tiers["medium"])
# 检查预算是否足够
if budget_remaining < tier["cost_per_1k"] * 10:
return model_tiers["simple"]["model"]
return tier["model"]
10. 实战案例
10.1 案例一：AI 软件开发团队
"""
AI 软件开发团队 - 完整实现
团队成员：
1. 产品经理 - 需求分析和 PRD 编写
2. 架构师 - 技术方案设计
3. 高级工程师 - 代码实现
4. QA 工程师 - 测试和质量保证
5. 技术文档工程师 - 文档编写
工作流程：
需求分析 → 架构设计 → 代码实现 → 测试 → 文档
"""
class AIDevTeam:
"""AI 软件开发团队"""
def __init__(self, llm_func: Callable):
self.llm = llm_func
self.message_bus = MessageBus()
self.shared_memory = SharedMemory()
self.monitor = SystemMonitor()
self.token_budget = TokenBudgetManager(total_budget=500000)
# 初始化团队成员
self.agents = {
"product_manager": self._create_product_manager(),
"architect": self._create_architect(),
"engineer": self._create_engineer(),
"qa_engineer": self._create_qa_engineer(),
"tech_writer": self._create_tech_writer()
}
# 分配 Token 预算
budgets = {
"product_manager": 50000,
"architect": 80000,
"engineer": 200000,
"qa_engineer": 100000,
"tech_writer": 70000
}
for name, budget in budgets.items():
self.token_budget.allocate(name, budget)
def _create_product_manager(self) -> AgentProfile:
return AgentProfile(
name="产品经理",
role="Product Manager",
goal="将用户需求转化为清晰的产品需求文档（PRD）",
backstory="你是一位经验丰富的产品经理，擅长需求分析和用户故事编写。",
capabilities=[AgentCapability.RESEARCH, AgentCapability.WRITING],
temperature=0.6
)
def _create_architect(self) -> AgentProfile:
return AgentProfile(
name="架构师",
role="System Architect",
goal="设计可扩展的系统架构",
backstory="你是一位资深系统架构师，精通微服务和分布式系统。",
capabilities=[AgentCapability.ARCHITECTURE_DESIGN],
temperature=0.5
)
def _create_engineer(self) -> AgentProfile:
return AgentProfile(
name="高级工程师",
role="Senior Engineer",
goal="编写高质量的生产级代码",
backstory="你是一位全栈工程师，擅长 Python、TypeScript。",
capabilities=[AgentCapability.CODE_GENERATION, AgentCapability.CODE_REVIEW],
temperature=0.3
)
def _create_qa_engineer(self) -> AgentProfile:
return AgentProfile(
name="QA工程师",
role="QA Engineer",
goal="确保代码质量和功能正确性",
backstory="你是一位严谨的QA工程师，擅长自动化测试。",
capabilities=[AgentCapability.TESTING, AgentCapability.CODE_REVIEW],
temperature=0.4
)
def _create_tech_writer(self) -> AgentProfile:
return AgentProfile(
name="技术文档工程师",
role="Technical Writer",
goal="编写清晰的技术文档",
backstory="你是一位技术写作专家，擅长将复杂技术转化为易懂文档。",
capabilities=[AgentCapability.WRITING],
temperature=0.6
)
def develop(self, requirement: str) -> Dict:
"""执行完整的开发流程"""
results = {}
# 阶段 1：需求分析
print("\n📋 阶段 1：需求分析")
prd = self._phase_requirements(requirement)
results["prd"] = prd
# 阶段 2：架构设计
print("\n🏗️ 阶段 2：架构设计")
architecture = self._phase_architecture(prd)
results["architecture"] = architecture
# 阶段 3：代码实现
print("\n💻 阶段 3：代码实现")
code = self._phase_implementation(architecture)
results["code"] = code
# 阶段 4：测试
print("\n🧪 阶段 4：测试")
test_results = self._phase_testing(code)
results["tests"] = test_results
# 阶段 5：文档
print("\n📝 阶段 5：文档编写")
docs = self._phase_documentation(prd, architecture, code)
results["docs"] = docs
# 输出报告
self.monitor.print_dashboard()
return results
def _call_agent(self, agent_name: str, prompt: str) -> str:
"""调用 Agent"""
start_time = time.time()
profile = self.agents[agent_name]
system_prompt = profile.to_system_prompt()
response = self.llm(system_prompt, prompt)
latency = time.time() - start_time
tokens = len(response.split()) * 2 # 粗略估计
self.token_budget.consume(agent_name, tokens)
self.monitor.record_call(agent_name, True, tokens, latency)
return response
def _phase_requirements(self, requirement: str) -> str:
prompt = f"""
请分析以下需求，并输出一份完整的 PRD（产品需求文档）。
原始需求：
{requirement}
PRD 应包含：
1. 产品概述
2. 目标用户
3. 核心功能列表
4. 用户故事（至少 5 个）
5. 非功能需求
6. 验收标准
"""
return self._call_agent("product_manager", prompt)
def _phase_architecture(self, prd: str) -> str:
prompt = f"""
基于以下 PRD，设计系统架构。
PRD：
{prd}
输出：
1. 系统架构图（用文字描述）
2. 技术栈选择及理由
3. 模块划分
4. API 设计（至少 5 个核心 API）
5. 数据模型设计
6. 部署方案
"""
return self._call_agent("architect", prompt)
def _phase_implementation(self, architecture: str) -> str:
prompt = f"""
基于以下架构设计，实现核心功能代码。
架构设计：
{architecture}
要求：
1. 使用 Python 实现
2. 包含完整的类型注解
3. 包含 docstring
4. 遵循 PEP 8 规范
5. 包含错误处理
6. 输出完整的、可运行的代码
"""
return self._call_agent("engineer", prompt)
def _phase_testing(self, code: str) -> str:
prompt = f"""
为以下代码编写完整的测试用例。
代码：
{code}
要求：
1. 使用 pytest 框架
2. 包含单元测试和集成测试
3. 覆盖正常流程和异常流程
4. 包含边界条件测试
5. 测试覆盖率目标 > 80%
"""
return self._call_agent("qa_engineer", prompt)
def _phase_documentation(self, prd: str, arch: str, code: str) -> str:
prompt = f"""
为项目编写技术文档。
PRD 摘要：{prd[:500]}
架构摘要：{arch[:500]}
代码摘要：{code[:500]}
文档应包含：
1. README.md
2. 快速开始指南
3. API 文档
4. 架构说明
5. 开发指南
"""
return self._call_agent("tech_writer", prompt)
# 使用示例
def simple_llm(system: str, user: str) -> str:
"""简单的 LLM 模拟"""
# 实际使用时替换为真实的 LLM API 调用
return f"[模拟响应] 基于系统提示和用户输入生成内容..."
team = AIDevTeam(llm_func=simple_llm)
results = team.develop("""
开发一个在线代码评审工具，支持：
1. 代码提交和差异查看
2. 行级评论
3. 评审状态管理
4. 通知提醒
""")
10.2 案例二：AI 研究团队
"""
AI 研究团队 - 文献综述与研究分析
团队成员：
1. 文献搜索员 - 搜索和筛选相关论文
2. 论文分析员 - 深度分析论文内容
3. 研究综合员 - 综合分析研究趋势
4. 报告撰写员 - 撰写研究报告
"""
class AIResearchTeam:
"""AI 研究团队"""
def __init__(self, llm_func: Callable):
self.llm = llm_func
self.shared_context = SharedContext()
def conduct_research(self, topic: str,
depth: str = "comprehensive") -> str:
"""执行研究流程"""
# 阶段 1：文献搜索
print(f"\n🔍 搜索关于 '{topic}' 的文献...")
papers = self._search_papers(topic)
# 阶段 2：论文分析
print(f"\n📖 分析 {len(papers)} 篇论文...")
analyses = self._analyze_papers(papers)
# 阶段 3：综合研究
print(f"\n🔬 综合研究趋势...")
synthesis = self._synthesize_research(topic, analyses)
# 阶段 4：撰写报告
print(f"\n📝 撰写研究报告...")
report = self._write_report(topic, papers, analyses, synthesis)
return report
def _search_papers(self, topic: str) -> List[Dict]:
search_prompt = f"""
请搜索关于 "{topic}" 的最新研究论文（2023-2024年）。
要求：
1. 列出至少 10 篇相关论文
2. 每篇包含：标题、作者、年份、会议/期刊、摘要
3. 按相关性排序
4. 标注影响力较高的论文
以 JSON 格式输出。
"""
result = self.llm("你是文献搜索专家。", search_prompt)
return json.loads(result)
def _analyze_papers(self, papers: List[Dict]) -> List[Dict]:
analyses = []
for paper in papers[:5]: # 分析前 5 篇
analysis_prompt = f"""
请深度分析以下论文：
标题：{paper.get('title', '')}
摘要：{paper.get('abstract', '')}
分析维度：
1. 研究问题
2. 方法论
3. 主要贡献
4. 实验结果
5. 局限性
6. 未来方向
"""
result = self.llm("你是论文分析专家。", analysis_prompt)
analyses.append({
"paper": paper,
"analysis": result
})
return analyses
def _synthesize_research(self, topic: str,
analyses: List[Dict]) -> str:
synthesis_prompt = f"""
基于以下论文分析，综合研究 "{topic}" 的发展趋势：
{json.dumps(analyses, ensure_ascii=False)[:3000]}
请分析：
1. 主要研究方向
2. 技术演进路线
3. 共识与争议
4. 研究空白
5. 未来展望
"""
return self.llm("你是研究综合专家。", synthesis_prompt)
def _write_report(self, topic: str, papers: List[Dict],
analyses: List[Dict], synthesis: str) -> str:
report_prompt = f"""
请撰写关于 "{topic}" 的研究报告。
论文数量：{len(papers)} 篇
深度分析：{len(analyses)} 篇
研究综合：{synthesis[:500]}
报告结构：
1. 引言
2. 研究背景
3. 文献综述
4. 研究趋势分析
5. 关键发现
6. 结论与展望
7. 参考文献
要求：
1. 学术风格
2. 逻辑清晰
3. 包含引用
4. 5000 字以上
"""
return self.llm("你是学术报告撰写专家。", report_prompt)
# 使用示例
research_team = AIResearchTeam(llm_func=simple_llm)
report = research_team.conduct_research(
"大语言模型的推理优化技术",
depth="comprehensive"
)
11. 最佳实践总结
11.1 设计原则
- 单一职责：每个 Agent 专注于一个明确的职责领域
- 松耦合：Agent 之间通过标准化的消息接口通信
- 高内聚：相关功能集中在同一个 Agent 内
- 可观测：所有交互和决策都应该可追踪
- 容错性：单个 Agent 的失败不应导致整个系统崩溃
11.2 性能优化建议
- Token 预算管理：为每个 Agent 设置 Token 上限
- 上下文压缩：定期压缩对话历史，只保留关键信息
- 结果缓存：对重复查询的结果进行缓存
- 模型分级：简单任务使用小模型，复杂任务使用大模型
- 并行执行：无依赖的任务并行执行
11.3 常见陷阱
| 陷阱 | 表现 | 解决方案 |
|---|---|---|
| 角色模糊 | Agent 输出不稳定 | 明确定义角色边界和能力范围 |
| 过度通信 | Token 消耗过高 | 设计合理的通信频率和内容 |
| 缺乏监控 | 问题难以定位 | 建立完善的日志和监控体系 |
| 单点故障 | 系统不可靠 | 引入重试和降级机制 |
| 权限失控 | 安全风险 | 实施最小权限原则 |
12. 常见问题解答
Q1: 多智能体系统适合所有场景吗？ A: 不是。对于简单的任务，单个 Agent 可能更高效。多智能体系统适合复杂、多步骤、需要专业分工的任务。
Q2: 如何决定使用多少个 Agent？ A: 根据任务的复杂度和专业分工需求来决定。通常 3-7 个 Agent 是比较合理的范围。过多会增加协调成本。
Q3: Agent 之间的通信如何优化？ A: 使用结构化的消息格式、压缩冗余信息、设置合理的通信频率、使用共享内存传递大数据。
Q4: 如何处理 Agent 输出质量不一致的问题？ A: 建立质量评估机制、使用自适应调度策略、引入人类审查节点、持续优化 Prompt。
Q5: 多智能体系统的成本如何控制？ A: Token 预算管理、模型分级选择、结果缓存、上下文压缩、合理的任务分解粒度。
13. 总结
本教程系统性地介绍了 AI 多智能体协作系统的完整知识体系，从基础概念到工程实践，涵盖了架构设计、框架选择、通信协议、任务调度、冲突解决、人机协作、监控调试和成本控制等核心主题。
核心要点回顾：
- 架构选择：根据任务特点选择合适的架构模式（中心化/去中心化/层级化/流水线）
- 框架选型：AutoGen 适合通用场景，CrewAI 易于上手，MetaGPT 专精软件开发
- 角色设计：明确的角色定义是多智能体系统成功的基础
- 通信设计：消息传递、共享内存、事件驱动各有适用场景
- 任务调度：拓扑排序 + 自适应调度是高效执行的关键
- 人机协作：在关键节点引入人类审查，平衡效率与质量
- 成本控制：Token 预算管理是生产环境的必备能力
多智能体系统代表了 AI 应用的前沿方向，掌握这项技术将帮助你构建更加智能、可靠的 AI 系统。
本教程内容基于当前主流框架和最佳实践编写。随着技术的快速发展，建议持续关注各框架的更新和社区的最新实践。
