5.4 多智能体系统实现

5.4 多智能体系统实现 — Agent智能体开发实战

本节导读：前几节分别完成了通信（5.1）、协议（5.2）、冲突处理（5.3）三大组件，本节把它们组装成一套可运行的多智能体系统，并对比自研与框架（LangGraph/AutoGen/CrewAI）两条路线的取舍。

学习目标

• 能从零组装一个包含监督者、专家、黑板、冲突仲裁的完整系统
• 理解主流多智能体框架的能力边界与选型逻辑
• 掌握多智能体系统的调试与评估方法
• 了解从原型到生产的关键工程化步骤

核心概念

系统总装蓝图

用户任务
   │
   ▼
监督者 Agent ──拆解──► 任务图
   │                        │
   │        ┌───────────────┼────────────────┐
   ▼        ▼               ▼                ▼
检索Agent  写作Agent     审校Agent      数据Agent     ← 专家层
   │        │               │                │
   └────────┴───────┬───────┴────────────────┘
                    ▼
            共享黑板（版本化状态）
                    ▼
            冲突检测 → 仲裁 → 汇总
                    ▼
              监督者合成最终答案
​

四个组件即前三节成果的直接复用：MessageBus（5.1）、ContractNet+Handoff（5.2）、Blackboard+Resolver（5.3）。

自研 vs 框架

选型建议：学习阶段自研（理解每个齿轮），原型阶段CrewAI（最快见效），生产阶段LangGraph（可控可观测）。框架不会替你解决5.2/5.3的协议与冲突设计，这些认知无论用什么都缺不了。

分步实战

步骤 1：智能体基类与LLM封装

import json, time
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")

class BaseAgent:
    def __init__(self, agent_id: str, role: str, system_prompt: str):
        self.id = agent_id
        self.role = role
        self.system_prompt = system_prompt

    def llm(self, messages: list, json_mode: bool = False) -> str:
        resp = client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=[{"role": "system", "content": self.system_prompt}]
                     + messages,
            temperature=0.2,
            response_format={"type": "json_object"} if json_mode else None,
        )
        return resp.choices[0].message.content

    def run(self, task: dict, blackboard) -> dict:
        raise NotImplementedError
​

temperature=0.2 是协作系统的常用起点：执行类智能体要稳定，发散留给创作类角色单独调高。

步骤 2：实现监督者

PLANNER_PROMPT = """你是任务监督者，职责是拆解任务并分派。
规则：
1. 只做拆解与汇总，绝不自己执行子任务
2. 每个子任务给出：id、description、role（retriever/writer/reviewer/data）、
   depends_on（依赖的子任务id）、acceptance（验收标准）
3. 子任务数量不超过5个，依赖深度不超过2层
4. 输出严格JSON。"""

class Supervisor(BaseAgent):
    def __init__(self):
        super().__init__("supervisor", "planner", PLANNER_PROMPT)

    def decompose(self, user_task: str) -> list[dict]:
        out = self.llm([{"role": "user", "content": user_task}],
                       json_mode=True)
        return json.loads(out)["subtasks"]

    def synthesize(self, results: dict) -> str:
        prompt = "根据以下子任务结果生成最终答案：\n" + \
                 json.dumps(results, ensure_ascii=False, indent=1)
        return self.llm([{"role": "user", "content": prompt}])
​

步骤 3：实现专家智能体

class RetrieverAgent(BaseAgent):
    def __init__(self):
        super().__init__("retriever-1", "retriever",
                         "你是资料检索员。只依据检索到的内容回答，"
                         "每条结论标注来源。无法确认就写UNKNOWN。")

    def run(self, task, blackboard):
        docs = search(task["description"])          # 接入检索工具
        answer = self.llm([
            {"role": "user",
             "content": f"任务：{task['description']}\n"
                        f"验收：{task['acceptance']}\n"
                        f"资料：{docs}"}])
        return {"task": task["id"], "output": answer, "sources": docs}

class ReviewerAgent(BaseAgent):
    def __init__(self):
        super().__init__("reviewer-1", "reviewer",
                         "你是审校员。逐条核对事实与逻辑，输出："
                         "通过项、存疑项（附理由）、修改建议。")

    def run(self, task, blackboard):
        draft = blackboard.read("draft")[0]
        return self.llm([{"role": "user",
                          "content": f"审校以下草稿：\n{draft}"}])
​

步骤 4：组装执行引擎

class MultiAgentSystem:
    def __init__(self):
        self.supervisor = Supervisor()
        self.specialists = {"retriever": [RetrieverAgent()],
                            "reviewer": [ReviewerAgent()],
                            "writer": [WriterAgent()],
                            "data": [DataAgent()]}
        self.blackboard = SharedBlackboard()       # 5.3节实现
        self.detector = ConflictDetector()
        self.resolver = ConflictResolver(ArbiterAgent(), LockManager())
        self.trace = []                             # 全链路留痕

    def execute(self, user_task: str) -> str:
        t0 = time.time()
        plan = self.supervisor.decompose(user_task)
        done, results = set(), {}

        while len(done) < len(plan):                # 按依赖调度
            ready = [t for t in plan
                     if t["id"] not in done
                     and set(t.get("depends_on", [])) <= done]
            for task in ready:                      # 简化：串行执行
                agent = self.specialists[task["role"]][0]
                results[task["id"]] = agent.run(task, self.blackboard)
                done.add(task["id"])
                self.trace.append({"task": task["id"], "agent": agent.id,
                                   "cost_ms": time.time() - t0})

        # 冲突处理（多检索者时生效）
        conflicts = self.detector.check_facts(
            {k: r["output"] for k, r in results.items()
             if isinstance(r, dict) and "output" in r})
        for c in conflicts:
            verdict = self.resolver.resolve(c)
            self.trace.append({"conflict": c.cid, "verdict": str(verdict)})

        return self.supervisor.synthesize(results)
​

步骤 5：评估与调试

if __name__ == "__main__":
    system = MultiAgentSystem()
    answer = system.execute("分析2024年新能源车市场格局并给出投资建议")

    # 三项体检
    print(system.trace)                    # 1) 过程留痕：谁在何时做了什么
    assert "UNKNOWN" not in answer or True # 2) 事实边界：未知是否被显式标注
    for item in eval_set:                  # 3) 端到端指标：与基线对比
        assert quality(system.execute(item.q)) >= baseline
​

调试多智能体系统的核心手段：逐层单测——先固定监督者的拆解结果人工校对，再单独跑各专家验证输入输出，最后才联调。直接端到端调试时，错误在层间传播会让你分不清是拆错了还是执行错了。

完整示例：主流框架等价实现（LangGraph）

同样的系统在LangGraph中的形态：

from langgraph.graph import StateGraph, END

builder = StateGraph(AgentState)
builder.add_node("plan", supervisor_node)      # 拆解
builder.add_node("retrieve", retriever_node)
builder.add_node("write", writer_node)
builder.add_node("review", reviewer_node)
builder.add_node("synthesize", synthesize_node)

builder.set_entry_point("plan")
builder.add_conditional_edges("plan", route_by_dependency)  # 依赖路由
builder.add_edge("write", "review")
builder.add_edge("review", "synthesize")
builder.add_edge("synthesize", END)

graph = builder.compile(checkpointer=MemorySaver())  # 状态可恢复
​

框架的价值在最后两行：条件路由与checkpoint。自研系统要把这两块写到同等可靠，通常需要一个迭代周期——这正是"生产选框架"的理由。

常见问题 FAQ

Q1：总成本失控怎么优化？
三个手段按序尝试：①削减智能体数量（多数系统有冗余角色）；②缩短各智能体的系统提示词并复用前缀缓存；③给检索/数据类智能体换小模型，只保留监督者与写作用大模型。混合模型通常能省40%以上成本。

Q2：如何防止智能体间"礼貌性附和"？
结构上隔离：各专家只看到任务描述与自己可见的资料，看不到彼此的结论（结论进黑板，裁决阶段才比对）；提示词上要求"独立判断、标注证据"。串行传递上下文的链式结构最容易产生附和，并行独立执行+汇总的结构天然抗附和。

Q3：系统跑一半挂了，如何恢复？
黑板每次状态变更落盘（快照+版本号），重启后从最后一个一致版本重放未完成任务。用LangGraph则直接用checkpointer恢复。自研系统至少要做到"任务级幂等"：同一子任务重跑不产生副作用（写操作带版本号即天然幂等）。

Q4：怎么评估多智能体系统好坏？
分三层：①任务层——端到端准确率/质量分；②协作层——冲突率、交接成功率、冗余工作占比；③成本层——每任务token数与延迟。只看任务层会掩盖协作浪费，三层联看才能定位"该优化模型还是该优化架构"。

Q5：什么时候该从单体Agent升级到多智能体？
出现明确信号再升级：①单个提示词塞进所有职责后质量明显下降；②不同步骤需要不同模型（如检索用小模型、写作用大模型）；③流程需要人工在中间环节介入。仅为"看起来更强"而拆分，只会得到更贵更慢的系统。

最佳实践与避坑

• 避坑一：角色划分按"专业边界"而非"流程步骤"。三个都什么都会点的"通用Agent"互相协作，效果远不如"检索专精+写作专精"两个角色——专业边界清晰的分工才配得上多智能体的成本；
• 避坑二：没有留痕就上线。多智能体系统的失败是复合型的，缺少逐任务trace时，排查成本随智能体数量指数增长；
• 实践：先用最便宜的模型把架构跑通，确认协作结构有效后再升级模型——架构缺陷在便宜模型上暴露得更快；
• 实践：把协作系统的评估集与单体基线放在一起持续回归，多智能体的复杂度必须始终用可度量的收益来偿还。

本节小结

本节完成了全书的最后一块拼图：把通信、协议、冲突处理组装为完整系统，自研路线吃透原理，框架路线（LangGraph）快速落地。三条铁律贯穿始终——留痕可审计、评估分三层、复杂度须有收益。至此，你已具备从ReAct单体到多智能体协作的全栈设计能力。

延伸阅读

• 相关章节：5.1 多智能体通信机制、5.2 协作协议与策略、5.3 冲突解决与一致性
• LangGraph官方教程
• AutoGen文档