LangGraph 进阶实战（六）：多智能体协作 (Multi-Agent) —— 打造你的 AI 团队

1. 痛点：全能 Agent 的局限性

在前两篇中，我们构建了一个拥有工具的 ReAct Agent。如果你的工具只有 2-3 个（查天气、算数），它工作得很好。

但是，如果你想构建一个“全自动软件开发助手”，你可能需要给它几十个工具：写 Python、写 Java、运行测试、部署 AWS、查文档、写文档...

当你把这 50 个工具全塞给一个 GPT-4 时，会发生灾难：

上下文爆炸：Prompt 太长，模型记不住所有工具的用法。
能力混淆：它可能用写 Python 的工具去写文档，导致格式错误。
调试困难：一旦出错，你不知道是哪个环节的问题。

解决方案：专业分工。 不要雇佣一个全能神，而是组建一个团队：

Agent A（研究员）：只负责搜索网页，提取信息。
Agent B（程序员）：只负责写代码，运行代码。
Supervisor（经理）：负责通过对话来指挥 A 和 B，并汇总结果。

这就是 Multi-Agent（多智能体） 架构。

2. 架构设计：Supervisor 模式

我们将构建一个星形拓扑结构：

        [Supervisor (经理)]
           /        \
          /          \
[Researcher]      [Coder]

Supervisor 接收用户任务。
它判断需要谁来干活。如果是查资料，路由给 Researcher。
Researcher 干完活，把结果返回给 Supervisor（而不是直接回复用户）。
Supervisor 拿到结果，看是否还需要写代码。如果需要，路由给 Coder。
直到所有任务完成，Supervisor 输出最终回复。

3. 环境准备

我们需要安装实验性的 langchain_experimental 库，里面有一些方便构建多 Agent 的工具函数。

1	pip install langchain_experimental langgraph langchain langchain-openai

4. 实战：构建 Research-Coder 团队

第一步：定义工具

我们需要两组工具。为了演示简单，我们用模拟工具。

from langchain_core.tools import tool

# --- 研究员的工具 ---
@tool
def search_web(query: str):
    """用于搜索互联网上的技术文档或信息。"""
    print(f"--- [Researcher] 正在搜索: {query} ---")
    return f"搜索结果: 关于 {query} 的最新文档内容..."

# --- 程序员的工具 ---
@tool
def execute_python(code: str):
    """执行 Python 代码并返回结果。"""
    print(f"--- [Coder] 正在执行代码: \n{code} ---")
    return "代码执行成功，输出: Hello World"

第二步：创建一个通用的 Agent 制造工厂

因为所有的“员工”其实本质上都是一样的：LLM + 特定工具 + 特定提示词。我们可以写一个辅助函数来批量生产 Agent。

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

def create_agent(llm, tools, system_message: str):
    """创建一个只拥有特定工具和特定角色的 Agent"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_message),
        MessagesPlaceholder(variable_name="messages"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])
    
    agent = create_openai_tools_agent(llm, tools, prompt)
    
    # AgentExecutor 是 LangChain 老牌的运行器，这里用来包装单个员工
    executor = AgentExecutor(agent=agent, tools=tools)
    return executor

第三步：初始化员工节点

llm = ChatOpenAI(model="gpt-3.5-turbo")

# 1. 创建研究员节点
research_agent = create_agent(
    llm, 
    [search_web], 
    "你是一个网络研究员。你只负责搜索信息，不要写代码。"
)

def researcher_node(state):
    result = research_agent.invoke(state)
    # 这一步很关键：我们需要把 Agent 的输出包装成一种 Graph 能理解的消息格式
    # 我们使用 HumanMessage 来伪装成"员工的汇报"
    return {"messages": [HumanMessage(content=result["output"], name="Researcher")]}

# 2. 创建程序员节点
code_agent = create_agent(
    llm, 
    [execute_python], 
    "你是一个 Python 程序员。你只负责写代码和执行，如果需要搜索，去问研究员。"
)

def coder_node(state):
    result = code_agent.invoke(state)
    return {"messages": [HumanMessage(content=result["output"], name="Coder")]}

第三步：定义经理 (Supervisor)

经理也是一个 LLM，但它没有工具。它的“工具”就是路由选择权。我们需要利用 OpenAI 的 Function Calling 功能，让经理输出“下一步该找谁”。

from langchain_core.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 团队成员名单
members = ["Researcher", "Coder"]

# 定义经理的系统提示词
# 核心逻辑：Given the conversation, who should act next?
system_prompt = (
    "你是一个项目经理，负责管理以下成员: {members}。"
    "根据用户的请求，决定下一个轮到谁行动。"
    "每个成员干完活会向你汇报。"
    "如果任务已经全部完成，请回复 FINISH。"
)

# 我们把路由选择定义为一个 function
options = ["FINISH"] + members
function_def = {
    "name": "route",
    "description": "选择下一个执行者",
    "parameters": {
        "title": "routeSchema",
        "type": "object",
        "properties": {
            "next": {
                "title": "Next",
                "anyOf": [
                    {"enum": options},
                ],
            }
        },
        "required": ["next"],
    },
}

# 构建经理链
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="messages"),
    (
        "system",
        "基于以上对话，下一步该谁行动？或者是 FINISH？请返回 function call。",
    ),
])

# 绑定路由函数，并强制模型必须调用这个函数 (function_call="route")
supervisor_chain = (
    prompt 
    | llm.bind_functions(functions=[function_def], function_call="route") 
    | JsonOutputFunctionsParser() # 直接把结果解析成 JSON: {"next": "Coder"}
)

第五步：构建多智能体图 (Graph)

这是最激动人心的时刻。

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

# 定义状态：和以前一样，只是存消息历史
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # next 字段用来存储经理的决定
    next: str

workflow = StateGraph(AgentState)

# 1. 添加节点
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Coder", coder_node)
workflow.add_node("Supervisor", supervisor_chain)

# 2. 添加边
# 员工干完活，必须汇报给经理（形成闭环）
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Coder", "Supervisor")

# 3. 设置入口
workflow.set_entry_point("Supervisor")

# 4. 添加条件边 (由经理决定去向)
# 这里的逻辑是：读取 state["next"]，然后映射到对应的节点
workflow.add_conditional_edges(
    "Supervisor",
    lambda x: x["next"], 
    {
        "Researcher": "Researcher",
        "Coder": "Coder",
        "FINISH": END
    }
)

app = workflow.compile()

第六步：运行团队

让我们给这个团队一个复杂的任务。

from langchain_core.messages import HumanMessage

print("--- 开始多智能体协作任务 ---")

# 任务：既需要查资料，又需要写代码
task = "请帮我查一下 LangGraph 的最新版本号，然后写一个 Python 打印它。"

for s in app.stream({"messages": [HumanMessage(content=task)]}):
    if "__end__" not in s:
        print(s)
        print("----")

5. 深度解析：运行轨迹

当你运行这段代码时，你会观察到以下流程：

Supervisor 接收任务。它分析认为需要先查数据，输出 {"next": "Researcher"}。
Researcher 被激活。它调用 search_web 工具，得到模拟结果。然后它回复：“我查到了，版本是 0.1.0”。
Supervisor 再次接收消息。它看到 Researcher 的汇报，分析认为下一步需要写代码，输出 {"next": "Coder"}。
Coder 被激活。它接收到“版本是 0.1.0”的信息，调用 execute_python，写代码 print('LangGraph 0.1.0')。
Supervisor 再次接收消息。它认为任务已闭环，输出 {"next": "FINISH"}。
END。

6. 为什么这比单个 Agent 强？

专注度：Coder 节点的 Prompt 里只有编程相关的指令，没有搜索相关的干扰。这让它写代码更稳。
容错性：如果 Coder 写错了，Supervisor 可以看出来（比如报错信息），然后重新指派 Coder 去修，或者指派 Researcher 去查报错原因。
可扩展性：如果你明天想加一个“测试员”，只需要写一个 Tester Agent，然后在 supervisor 的 members 列表里加个名字，原来的代码几乎不用动。

7. 总结

在这一篇中，我们突破了单体架构，进入了微服务架构的 AI 版本。

Single Agent 就像全栈工程师，什么都得会，容易累死。
Multi-Agent 就像软件公司，有经理、前端、后端，各司其职。

这就是构建复杂企业级 AI 应用的终极形态。