多智能体协作

LangGraph 非常适合构建多智能体系统，其中多个智能体协同工作以完成复杂任务。

🤖 核心概念

1. 智能体角色

• 协调器 (Orchestrator): 负责任务分解、路由和结果聚合的中心智能体。
• 专家智能体 (Expert Agents): 专注于特定任务的智能体，如研究、写作、编码等。

2. 通信机制

• 共享状态: 智能体通过读写共享的 AgentState 来进行通信。
• 消息传递: 通过在 messages 列表中添加消息来传递信息。

🏗️ 构建多智能体系统

1. 定义共享状态

from typing import TypedDict, List, Annotated
from langchain_core.messages import BaseMessage
import operator

class TeamState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    team_members: List[str]
    next: str

• Annotated[..., operator.add]: 这是一个技巧，用于告诉 LangGraph 在更新状态时，对 messages 字段使用追加（add）而不是覆盖。

2. 创建专家智能体

每个专家智能体都是一个独立的图或节点。

# 研究智能体
def research_agent_node(state: TeamState) -> dict:
    # ... 执行研究任务 ...
    return {"messages": [ToolMessage(content="研究完成", tool_call_id="...")]}

# 写作智能体
def writing_agent_node(state: TeamState) -> dict:
    # ... 执行写作任务 ...
    return {"messages": [AIMessage(content="报告初稿...")]}

3. 创建协调器

协调器是一个特殊的节点，负责将任务路由到合适的智能体。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 定义协调器的工具，代表每个专家智能体
tools = [research_tool, writing_tool] 
llm = ChatOpenAI(model="gpt-4-turbo").bind_tools(tools)

def orchestrator_node(state: TeamState) -> dict:
    # 调用 LLM 来决定下一个动作
    response = llm.invoke(state["messages"])
    
    # 如果没有工具调用，则完成
    if not response.tool_calls:
        return {"next": "end"}
    
    # 否则，路由到指定的工具（智能体）
    return {"next": response.tool_calls[0]["name"]}

4. 构建主图

workflow = StateGraph(TeamState)

# 添加专家智能体节点
workflow.add_node("researcher", research_agent_node)
workflow.add_node("writer", writing_agent_node)

# 添加协调器节点
workflow.add_node("orchestrator", orchestrator_node)

# 定义路由逻辑
def route_logic(state: TeamState) -> str:
    return state["next"]

# 设置入口点
workflow.set_entry_point("orchestrator")

# 添加条件边
workflow.add_conditional_edges(
    "orchestrator",
    route_logic,
    {
        "researcher": "researcher",
        "writer": "writer",
        "end": END,
    },
)

# 将专家智能体的输出返回给协调器
workflow.add_edge("researcher", "orchestrator")
workflow.add_edge("writer", "orchestrator")

app = workflow.compile()

🚀 运行示例

from langchain_core.messages import HumanMessage

# 初始请求
request = {"messages": [HumanMessage(content="写一篇关于 LangGraph 的报告")]}

# 流式输出
for event in app.stream(request, {"recursion_limit": 10}):
    for key, value in event.items():
        print(f"--- {key} ---")
        print(value)

执行流程

1. orchestrator: 接收到请求，决定需要先进行研究，将 next 设置为 researcher。
2. 路由: route_logic 将流程导向 researcher 节点。
3. researcher: 执行研究任务，并将结果添加到 messages。
4. 返回 orchestrator: 协调器接收到研究结果。
5. orchestrator: 根据新状态，决定下一步需要写作，将 next 设置为 writer。
6. 路由: 流程导向 writer 节点。
7. writer: 执行写作任务，生成报告初稿。
8. 返回 orchestrator: 协调器接收到初稿。
9. orchestrator: 任务完成，将 next 设置为 end。
10. 路由: 流程结束。

🎯 设计模式

1. 主从模式 (Master-Slave)

• 一个协调器（Master）管理多个专家（Slave）。
• 适用于任务可以清晰分解的场景。

2. 协作模式 (Collaborative)

• 智能体之间可以自由通信，没有中心化的协调器。
• 适用于需要动态协商和合作的复杂任务。
• 实现方式：所有智能体都连接到一个共享的 MessageGraph。

3. 层次化模式 (Hierarchical)

• 将智能体组织成树状结构。
• 高层智能体负责宏观规划，低层智能体负责具体执行。
• 适用于大型、多阶段的项目。

💡 最佳实践

• 明确职责: 每个智能体应该有单一、明确的职责。
• 标准化接口: 定义智能体之间通信的标准格式（如使用 ToolMessage）。
• 状态隔离: 除非必要，否则不要让专家智能体修改不属于其职责范围的状态。
• 循环检测: 在复杂的协作图中，要小心死循环，设置递归限制（recursion_limit）。
• 可视化调试: 经常使用 draw_mermaid_png() 来可视化和调试你的多智能体系统。

---

💡 下一步: 探索 持久化 和 流式输出，构建生产级的多智能体应用。