多智能体系统（Multi-Agent Systems）

问题 1: 上下文过载
  一个 Agent 需要同时处理代码编写、测试、文档、部署——
  上下文窗口装不下所有信息，导致遗漏和错误。

问题 2: 专业能力不足
  一个 Agent 同时扮演产品经理、架构师、开发者、测试员——
  每个角色都做不精，输出质量参差不齐。

问题 3: 串行效率低
  一个 Agent 做完 A 才能做 B，做完 B 才能做 C——
  无法并行处理独立子任务，效率低下。

问题 4: 缺乏对抗与制衡
  一个 Agent 自我评审容易出现"确认偏差"——
  没有独立视角的交叉验证，错误难以发现。

# 模式 1: 直接通信（Agent-to-Agent）
agent_a.send_message(agent_b, "请帮我检查这段代码的逻辑")

# 模式 2: 广播通信（Agent-to-All）
orchestrator.broadcast("任务已完成 80%，请各 Agent 汇报进度")

# 模式 3: 共享黑板（Blackboard）
blackboard.write("research_results", agent_a.findings)
data = blackboard.read("research_results")  # 其他 Agent 可读

# 模式 4: 消息队列（Message Queue）
queue.publish("task.completed", {"agent": "A", "result": result})
queue.subscribe("task.completed", handler)  # 订阅特定事件

@dataclass
class AgentMessage:
    """Agent 之间的通信消息"""
    sender: str           # 发送方 Agent ID
    receiver: str         # 接收方 Agent ID（"all" 表示广播）
    msg_type: str         # 消息类型: task / result / query / feedback
    content: Any          # 消息内容
    priority: int = 0     # 优先级
    timestamp: float = 0  # 时间戳
    reply_to: str = ""    # 回复哪条消息
    metadata: Dict = field(default_factory=dict)

# 消息类型
MSG_TYPES = {
    "task_assign":   "管理者分配任务",
    "task_result":   "工作者返回结果",
    "query":         "Agent 之间的询问",
    "response":      "对询问的回答",
    "feedback":      "评审反馈",
    "status_update": "状态更新",
    "terminate":     "终止信号"
}

任务: "评估某公司的技术架构方案"

Round 1:
  Agent A (架构师): "应该用微服务架构，便于独立部署和扩展"
  Agent B (运维专家): "微服务会增加运维复杂度，建议单体 + 模块化"
  Agent C (CTO 视角): "取决于团队规模，小团队用单体，大团队用微服务"

Round 2:
  Agent A: "同意 C 的观点，补充：可以先单体后拆分"
  Agent B: "同意渐进式演进，但要提前设计好模块边界"
  Agent C: "最终建议：模块化单体起步，核心模块预留拆分接口"

Round 3:
  共识达成：采用模块化单体 + 渐进式拆分策略

def division_of_labor(task: str):
    # 1. 任务分解
    subtasks = planner.decompose(task)
    
    # 2. 角色分配
    assignments = {}
    for subtask in subtasks:
        best_agent = select_agent(subtask)  # 根据能力匹配
        assignments[subtask.id] = best_agent
    
    # 3. 并行执行
    results = {}
    with ThreadPoolExecutor() as executor:
        futures = {
            executor.run(agent.execute, subtask): subtask
            for subtask, agent in assignments.items()
        }
        for future in as_completed(futures):
            subtask = futures[future]
            results[subtask.id] = future.result()
    
    # 4. 结果合并
    final = integrator.merge(task, results)
    return final

from crewai import Agent, Task, Crew

# 定义 Agent
researcher = Agent(
    role="资深行业研究员",
    goal="深入调研指定主题，提供详尽的数据和分析",
    backstory="你是一位有 10 年经验的行业分析师，擅长数据挖掘和趋势预测",
    tools=[search_tool, web_scraper],
    llm="gpt-4"
)

writer = Agent(
    role="专业内容作者",
    goal="将研究结果整理成高质量的文章",
    backstory="你是一位资深科技作者，擅长将复杂概念用通俗语言表达",
    llm="gpt-4"
)

reviewer = Agent(
    role="内容审核专家",
    goal="审核文章的准确性、逻辑性和可读性",
    backstory="你是一位严谨的编辑，对事实核查和逻辑一致性要求极高",
    llm="gpt-4"
)

# 定义任务
research_task = Task(
    description="调研 2024 年 AI Agent 技术的发展现状和趋势",
    expected_output="详细的调研报告，包含数据、案例和趋势分析",
    agent=researcher
)

writing_task = Task(
    description="基于调研报告，撰写一篇 3000 字的深度分析文章",
    expected_output="结构清晰、数据充分的分析文章",
    agent=writer
)

review_task = Task(
    description="审核文章，确保事实准确、逻辑自洽、表述清晰",
    expected_output="审核报告和修改后的最终版本",
    agent=reviewer
)

# 组建团队
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process="sequential",  # 顺序执行
    verbose=True
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 创建 Agent
coder = AssistantAgent(
    name="Coder",
    system_message="你是一位资深 Python 开发者，负责编写代码",
    llm_config=llm_config
)

reviewer = AssistantAgent(
    name="Reviewer",
    system_message="你是一位代码审查专家，负责审查代码质量和安全性",
    llm_config=llm_config
)

user = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "coding"}
)

# 创建群聊
groupchat = GroupChat(
    agents=[user, coder, reviewer],
    messages=[],
    max_round=10,
    speaker_selection_method="round_robin"  # 轮流发言
)

manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# 启动对话
user.initiate_chat(manager, message="帮我写一个 REST API 的用户管理系统")

1. 明确角色边界
   每个 Agent 有清晰且互不重叠的职责

2. 最小化通信
   Agent 之间只传递必要的结构化信息

3. 设置超时和上限
   每个 Agent 有执行时间上限和最大重试次数

4. 统一错误处理
   任何 Agent 失败都有明确的降级和恢复策略

5. 可观测性
   记录所有 Agent 的决策和交互，便于调试和优化

6. 渐进式构建
   先用单 Agent 验证可行性，再拆分为多 Agent