第9章 一个人不行？组个团

前面几章我们一直在打造一个全能Agent——脑子好、手脚快、记性好、会规划。但现实中，一个Agent再强也有天花板。就像你不会让一个人同时当销售、财务、程序员和客服——你得组团队。这一章，我们讲Multi-Agent协作。

---

9.1 一个Agent忙不过来？那就多找几个

9.1.1 一个全栈Agent的困境

假设你要做一个"智能软件开发助手"，它需要：

• 理解需求（产品经理的活）
• 写代码（程序员的活）
• 审查代码（Code Review的活）
• 写测试（测试工程师的活）
• 写文档（技术写作的活）

全塞给一个Agent？它的Prompt会变得又长又乱，工具列表爆炸，一不留神就搞混角色——明明在写代码，突然跳到审查模式。

每人只干一件事？更靠谱。五个Agent，各司其职。

9.1.2 什么时候该用Multi-Agent

简单判断标准：如果一个人能干的活，用一个Agent；如果需要不同角色配合的活，用Multi-Agent。

---

9.2 多Agent怎么配合？

9.2.1 三种经典架构

层级模式（Supervisor）：有一个"总监"Agent，负责分配任务、检查结果、做最终决策。适合任务多变、需要灵活调度的场景。

总监Agent：用户想做一个登录功能。
Agent A（前端）：你负责写登录表单的HTML和CSS。
Agent B（后端）：你负责写登录接口和数据库设计。
Agent C（测试）：等A和B完成后，你负责写测试用例。

流水线模式（Pipeline）：每个Agent处理完交给下一个。适合有明确步骤的任务。

Agent A（数据清洗）→ Agent B（数据分析）→ Agent C（报告生成）

平行模式（Parallel）：多个Agent同时干活，互不依赖。适合可并行的任务。

Agent A 查天气 → 
Agent B 查新闻 →  同时进行，结果汇总给总监
Agent C 查股价 →

9.2.2 层级模式的代码骨架

# 总监Agent的核心逻辑
class SupervisorAgent:
    def __init__(self, workers, llm):
        self.workers = workers  # {"前端": agent_a, "后端": agent_b, ...}
        self.llm = llm
    
    def handle_task(self, user_request):
        # 1. 理解任务，决定派给谁
        plan = self.llm.invoke(f"""
        任务：{user_request}
        可用的工人：{list(self.workers.keys())}
        请决定：把任务分配给哪些工人？按什么顺序？
        """)
        
        # 2. 分派任务
        results = {}
        for worker_name in plan["assignments"]:
            worker = self.workers[worker_name]
            results[worker_name] = worker.do(plan["subtasks"][worker_name])
        
        # 3. 汇总结果
        final = self.llm.invoke(f"""
        各工人的结果：{results}
        请汇总成一个完整的交付物。
        """)
        return final

---

9.3 Agent之间怎么"说话"？

9.3.1 通信方式：消息传递

多个Agent协作，就需要互相沟通。最简单的方式就是消息传递——一个Agent发消息，另一个Agent收消息。

# Agent之间的消息格式
message = {
    "from": "前端Agent",
    "to": "后端Agent", 
    "type": "request",  # request / response / notify
    "content": "需要你提供一个 /api/login 接口，输入用户名密码，返回token",
    "context": {"framework": "FastAPI", "auth_method": "JWT"}
}

9.3.2 三种通信模式

点对点通信：Agent A直接给Agent B发消息。最简单，适合固定搭档。

前端Agent → 后端Agent："接口文档发我一下"
后端Agent → 前端Agent："好了，看这里"

广播通信：一个Agent发消息，所有Agent都能收到。适合通知类的信息。

总监Agent → 所有人："需求变更了，大家注意"

共享黑板：所有Agent往同一个地方写，也读同一个地方。适合需要共享状态、追踪进度的场景。

# 共享状态
blackboard = {
    "task": "用户登录功能",
    "status": "进行中",
    "progress": {
        "前端": "表单已完成",
        "后端": "接口开发中",
        "测试": "等待中"
    }
}

---

9.4 意见不合怎么办？

9.4.1 多Agent一定会吵架

两个Agent给出矛盾的建议很正常：

Agent A（成本优先）：建议用SQLite，零配置，够用。
Agent B（性能优先）：建议用PostgreSQL，并发强，可扩展。

这不是bug，这是多Agent系统的特性。不同Agent有不同的"视角"和"偏好"。

9.4.2 解决冲突的三种方式

方式一：投票。 三个以上的Agent，少数服从多数。

opinions = [
    {"agent": "A", "choice": "SQLite", "reason": "简单够用"},
    {"agent": "B", "choice": "PostgreSQL", "reason": "性能和扩展"},
    {"agent": "C", "choice": "PostgreSQL", "reason": "团队熟悉"}
]
# PostgreSQL 2票 vs SQLite 1票 → 选PostgreSQL

方式二：总监裁决。 层级模式下，总监Agent做最终决定。

总监Agent：A说的有道理，项目初期SQLite够了。但我们预期用户量会快速增长，
          B的考虑也很重要。折中方案：先用SQLite开发，但数据层加抽象接口，
          后期可以平滑切换到PostgreSQL。

方式三：辩论后共识。 让不同意见的Agent各自陈述理由，再由一个Agent综合出最优解。

---

9.5 用 LangGraph 实现多 Agent 协作

前面 9.2.2 节用纯 Python 写了一个 SupervisorAgent 类，虽然讲清了原理，但缺少状态管理和条件路由的支持。LangGraph 提供了更专业的实现方式。

9.5.1 用 @tool 装饰器定义工具

LangGraph 中，每个 Agent 的能力以"工具"的形式暴露。@tool 装饰器自动生成 Schema，LLM 根据 Schema 决定何时调用：

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息

    Args:
        query: 搜索关键词，越具体越好
    """
    return f"搜索结果：关于 {query} 的最新信息..."

@tool
def get_weather(city: str) -> str:
    """查询城市天气

    Args:
        city: 城市名称，如"北京"、"上海"
    """
    return f"{city}今天晴，25°C"

# 绑定工具到 LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools([search_web, get_weather])

# LLM 会自动判断何时调用工具
from langchain_core.messages import HumanMessage
response = llm_with_tools.invoke([HumanMessage(content="北京今天天气怎么样？")])
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'xxx'}]

关键点：bind_tools 之后，LLM 不再直接回答，而是返回工具调用的指令。Agent 拿到这个指令，执行工具，再把结果喂回 LLM。

9.5.2 Supervisor 模式：LangGraph 版本

9.2.2 节用类封装实现了主管模式。LangGraph 的做法是把 Supervisor 也变成一个节点，用条件边来路由：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Literal

class MultiAgentState(TypedDict):
    task: str
    current_agent: Literal["researcher", "coder", "reviewer", "__end__"] | None
    research_result: str | None
    code_result: str | None
    review_result: str | None

def supervisor_node(state: MultiAgentState) -> dict:
    """Supervisor 决定下一个处理者"""
    if not state.get("research_result"):
        return {"current_agent": "researcher"}
    elif not state.get("code_result"):
        return {"current_agent": "coder"}
    elif not state.get("review_result"):
        return {"current_agent": "reviewer"}
    else:
        return {"current_agent": "__end__"}

def route_to_agent(state: MultiAgentState) -> str:
    return state["current_agent"]

# 构建图
graph = StateGraph(MultiAgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("reviewer", reviewer_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_to_agent, {
    "researcher": "researcher",
    "coder": "coder",
    "reviewer": "reviewer",
    "__end__": END
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("reviewer", "supervisor")

app = graph.compile()

和 9.2.2 节的区别：LangGraph 版本中，Supervisor、Worker 都被建模为图的节点，调度逻辑通过 add_conditional_edges 实现。每个 Worker 完成后自动回到 Supervisor，由 Supervisor 决定下一步——无需手动写调度循环。

9.5.3 Send API：让多个 Agent 并行干活

9.2.1 节讲的"平行模式"，在 LangGraph 中靠 Send API 实现。它可以让多个 Agent 同时开工，互不等待：

from langgraph.constants import Send

def parallel_research(state: MainState) -> list:
    """并行分发研究任务"""
    topic = state["topic"]
    return [
        Send("web_search", {"query": f"{topic} 最新动态"}),
        Send("academic_search", {"query": f"{topic} 学术论文"}),
        Send("social_search", {"query": f"{topic} 社区讨论"}),
    ]

graph.add_node("parallel_search", parallel_research)
graph.add_node("aggregate", aggregate_results)
graph.add_edge(START, "parallel_search")
graph.add_edge("parallel_search", "aggregate")
graph.add_edge("aggregate", END)

Send(target_node, state) 把任务分别打包发给三个搜索 Agent，它们并行执行，结果汇总到 aggregate 节点。

普通边是串行，Send 是并行。 如果你的查询任务互相独立——比如同时查天气、新闻、股价——Send 能让它们同时跑，总耗时等于最慢的那个 Agent 的耗时。

---

9.6 本章小结

这一章我们突破了"单Agent"的思维：

1. 什么时候用Multi-Agent：任务可分拆、需要不同能力时。全栈Agent往往不如专业Agent团队。
2. 三种协作模式：层级（有总监）、流水线（接力棒）、平行（各干各的）。场景决定模式。
3. Agent之间靠消息沟通：点对点、广播、共享黑板——和人类团队沟通方式一模一样。
4. 意见不合是正常的：投票、总监裁决、辩论后共识——三种方式解决分歧。

---

✅ 知识点检查

学完这一章，试试回答这几个问题：

• [ ] 什么情况下该用Multi-Agent而不是单Agent？
• [ ] 层级、流水线、平行三种模式各适合什么场景？
• [ ] Agent之间有哪三种通信方式？
• [ ] 两个Agent意见不合时，有哪三种解决方法？

---

📚 延伸阅读

• LangGraph Multi-Agent 教程：https://langchain-ai.github.io/langgraph/tutorials/multi_agent/
• CrewAI 官方文档：https://docs.crewai.com
• AutoGen（微软）Multi-Agent框架：https://microsoft.github.io/autogen/
• 本书配套源码：关注公众号「图解AI系列」免费领取

---

🎯 下一章预告

第10章，Agent框架那么多，到底该学哪个——

"巨人的肩膀就在那，挑哪个站？LangChain、LangGraph、CrewAI、AutoGen……一次讲清楚。"

---