Multi-Agent（多智能体系统）常见架构

针对你之前关心的那7种协作模式，下面是其中3种最核心、最常用架构的图示和LangGraph代码示例。这些示例都经过简化，你可以直接复制运行（需要先替换OpenAI的API Key）。

1. 主管模式 (Supervisor Pattern)

核心逻辑：一个中央“主管”智能体负责接收任务、分析意图，然后动态地将任务委派给最合适的“专家”智能体（如研究员、数学家），并最终汇总结果。

架构图

graph TD
    User[用户] --> Supervisor[👔 主管智能体]
    Supervisor -->|路由任务| Researcher[🔍 研究员智能体]
    Supervisor -->|路由任务| Mathematician[🧮 数学家智能体]
    Researcher -->|返回结果| Supervisor
    Mathematician -->|返回结果| Supervisor
    Supervisor -->|最终答案| User

LangGraph 代码示例 (使用 langgraph-supervisor 库)

这个示例创建了一个能处理数学计算和实时信息查询的主管系统。

# 首先安装必要的库
# pip install langgraph-supervisor langchain-openai

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

# 1. 初始化模型
model = ChatOpenAI(model="gpt-4o")

# 2. 定义专家智能体会用到的工具
def add(a: float, b: float) -> float:
    """加法工具"""
    return a + b

def web_search(query: str) -> str:
    """模拟的网络搜索工具"""
    return f"已搜索到关于 '{query}' 的结果..."

# 3. 创建专家智能体
math_agent = create_react_agent(
    model=model,
    tools=[add],
    name="math_expert",
    prompt="你是一位数学专家，只处理数学计算。"
)

research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="你是一位研究员，负责搜索信息。"
)

# 4. 创建主管工作流
workflow = create_supervisor(
    agents=[research_agent, math_agent],
    model=model,
    prompt="你是一个团队主管，管理研究和数学专家。对于实时信息，使用 research_expert；对于数学，使用 math_expert。"
)

# 5. 编译并运行
app = workflow.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "25乘以4等于多少？"}]
})

# 6. 查看结果
for message in result["messages"]:
    print(f"{message.type}: {message.content}")

2. 群组/移交模式 (Swarm/Handoff Pattern)

核心逻辑：没有中心化的主管，每个智能体都是对等的。一个智能体在处理任务时，如果发现需要其他领域的专业知识，可以主动将对话“移交”(Handoff)给另一个专门的智能体。

架构图

graph TD
    User[用户] --> FlightAgent[✈️ 航班助手]
    FlightAgent -->|移交任务| HotelAgent[🏨 酒店助手]
    HotelAgent -->|移交任务| FlightAgent
    FlightAgent -->|最终答案| User
    HotelAgent -->|最终答案| User

LangGraph 代码示例 (使用 create_handoff_tool)

这个示例创建了航班和酒店两个助手，它们可以根据需要将任务移交给对方。

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# 1. 创建一个通用的移交工具生成函数
def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    @tool(name, description=description or f"将用户移交给 {agent_name}")
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"成功移交至 {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto=agent_name,
            update={"messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )
    return handoff_tool

# 2. 为每个助手创建移交工具
transfer_to_hotel = create_handoff_tool(
    agent_name="hotel_assistant",
    description="将用户移交给酒店预订助手。"
)
transfer_to_flight = create_handoff_tool(
    agent_name="flight_assistant",
    description="将用户移交给航班预订助手。"
)

# 3. 创建智能体，并赋予它们移交工具
flight_agent = create_react_agent(
    model=model,
    tools=[transfer_to_hotel],
    name="flight_assistant",
    prompt="你是一个航班预订助手。如果用户需要订酒店，请使用 transfer_to_hotel 工具移交任务。"
)

hotel_agent = create_react_agent(
    model=model,
    tools=[transfer_to_flight],
    name="hotel_assistant",
    prompt="你是一个酒店预订助手。如果用户需要订航班，请使用 transfer_to_flight 工具移交任务。"
)

# 4. 构建图：将智能体作为节点加入
builder = StateGraph(MessagesState)
builder.add_node("flight_assistant", flight_agent)
builder.add_node("hotel_assistant", hotel_agent)
builder.add_edge(START, "flight_assistant")  # 从航班助手开始
# 智能体会通过 Command(goto=...) 动态跳转，所以不需要静态边

# 5. 编译并运行
app = builder.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "帮我订一张去上海的机票，然后推荐一家附近的酒店。"}]
})

for message in result["messages"]:
    print(f"{message.type}: {message.content}")

3. 网络/协作模式 (Network/Collaboration Pattern)

核心逻辑：多个对等的智能体在共享状态下协作，像接力一样依次处理任务，共同完成一个复杂的目标。上一个智能体的输出是下一个的输入。

架构图

graph TD
    User[用户] --> Researcher[🔍 研究员]
    Researcher -->|研究资料| ChartGenerator[📊 图表生成器]
    ChartGenerator -->|图表数据| Reviewer[✅ 审核员]
    Reviewer -->|最终报告| User

💻 LangGraph 代码示例

此示例展示了一个研究-生成-审核的协作流程。研究员搜索信息，图表生成器根据信息制作图表，最后由审核员检查并输出报告。

from langgraph.graph import StateGraph, END, MessagesState
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from typing import Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.types import Command

model = ChatOpenAI(model="gpt-4o")

# 1. 创建工具
def tavily_tool(query: str) -> str:
    """模拟网络搜索工具"""
    return f"关于 '{query}' 的搜索结果：...（模拟数据）"

# 2. 创建三个专家智能体
research_agent = create_react_agent(
    model, tools=[tavily_tool],
    prompt="你是一个研究员，负责搜索信息。"
)
chart_agent = create_react_agent(
    model, tools=[],
    prompt="你是一个图表专家，根据文本数据生成图表描述。"
)
reviewer_agent = create_react_agent(
    model, tools=[],
    prompt="你是一个审核员，负责检查并整合最终报告。"
)

# 3. 辅助函数：判断下一个节点
def get_next_node(last_message: BaseMessage, goto: str):
    if "FINAL ANSWER" in last_message.content:
        return END
    return goto

# 4. 将智能体包装为图节点
def research_node(state: MessagesState) -> Command[Literal["chart_agent", END]]:
    result = research_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "chart_agent")
    result["messages"][-1] = HumanMessage(content=result["messages"][-1].content, name="researcher")
    return Command(update={"messages": result["messages"]}, goto=goto)

def chart_node(state: MessagesState) -> Command[Literal["reviewer", END]]:
    result = chart_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "reviewer")
    result["messages"][-1] = HumanMessage(content=result["messages"][-1].content, name="chart_generator")
    return Command(update={"messages": result["messages"]}, goto=goto)

def reviewer_node(state: MessagesState) -> Command[Literal[END]]:
    result = reviewer_agent.invoke(state)
    result["messages"][-1] = HumanMessage(content=result["messages"][-1].content, name="reviewer")
    return Command(update={"messages": result["messages"]}, goto=END)

# 5. 构建图
builder = StateGraph(MessagesState)
builder.add_node("researcher", research_node)
builder.add_node("chart_agent", chart_node)
builder.add_node("reviewer", reviewer_node)

builder.set_entry_point("researcher")
# 节点内部通过 Command(goto=...) 控制流转，无需静态边

# 6. 编译并运行
app = builder.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "研究一下2026年AI的发展趋势，并生成一份带图表的报告。"}]
})

for message in result["messages"]:
    print(f"{message.type} - {message.name}: {message.content[:100]}...")

总结

这三种模式是LangGraph构建多智能体应用的基础。在实际项目中，你往往需要将它们组合使用，才能构建出强大、灵活的系统。上面的代码示例都是可以直接运行的，你可以根据实际需求调整其中的提示词（prompt）和工具（tools）。