7.多代理架构设计

当单个代理需要同时处理多个专业领域或管理大量工具时，往往会力不从心。想象一下，一个既要订机票又要订酒店的旅行助手，如果让它在一个循环里同时处理航班查询、酒店预订、价格比较、退改政策等所有任务，很快就会陷入决策混乱。工具太多会导致选择困难，上下文太复杂会让状态管理失控，专业领域跨度太大则难以保证回答质量。

这时候，把一个大而全的代理拆分成多个小而专的独立代理，再组合成多代理系统，就成了自然而然的选择。每个代理专注于自己的领域，通过精心设计的协作机制共同完成任务。这种模式不仅模块化程度更高，开发和维护更方便，还能让专业代理在特定领域深耕，最终通过明确的通信规则实现整体性能提升。

LangGraph 为多代理架构提供了完整的支持体系，核心在于代理之间的交接机制。下面我们从最基础的 Handoff 模式开始，逐步展开 Supervisor 和 Swarm 两种主流架构，最后探讨跨代理状态管理的关键问题。

Handoff 代理交接模式

Handoff 是多代理系统中最基础的交互模式。简单来说，就是一个代理把控制权交给另一个代理，同时传递必要的上下文信息。这个过程需要明确两个要素：目标代理是谁，以及要传递什么数据。

在 LangGraph 中，Handoff 通过 Command 对象实现。这个设计很巧妙，它把状态更新和节点跳转合并为一次原子操作。当代理决定移交控制权时，返回一个 Command，告诉框架下一步该去哪里，以及要更新哪些状态。

基础实现原理

让我们看看 Handoff 工具的核心实现。这里的关键是创建一个特殊的工具函数，当代理调用它时，实际上是在触发一次控制权转移。

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import MessagesState
from langgraph.types import Command

def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Transfer to {agent_name}"

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {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

这段代码值得仔细拆解。InjectedState 和 InjectedToolCallId 是 LangGraph 提供的特殊注解，框架会自动注入当前代理的状态和工具调用 ID。工具函数内部构造了一条转移成功的工具消息，然后返回 Command 对象。

Command 的三个参数各有深意：

• goto 指定目标代理的节点名称
• update 定义状态更新逻辑，这里把当前消息历史加上转移通知后传递给父图
• graph=Command.PARENT 至关重要，它告诉 LangGraph 这次跳转是在父图层面进行的，而不是在当前代理的子图内部

这个设计允许代理在不知道整体图结构的情况下，依然能正确地将控制权交还给父图，由父图调度到目标代理。代理之间保持了解耦，通信逻辑由框架统一管理。

在 ReAct 代理中使用 Handoff

创建好 Handoff 工具后，下一步是把它交给需要使用交接能力的代理。这里我们延续旅行助手的例子，创建两个专业代理：航班预订助手和酒店预订助手。

# 定义具体业务工具
def book_flight(from_airport: str, to_airport: str):
    """Book a flight"""
    return f"Successfully booked a flight from {from_airport} to {to_airport}."

def book_hotel(hotel_name: str):
    """Book a hotel"""
    return f"Successfully booked a stay at {hotel_name}."

# 创建 Handoff 工具
transfer_to_hotel = create_handoff_tool(
    agent_name="hotel_assistant",
    description="Transfer user to the hotel-booking assistant."
)
transfer_to_flight = create_handoff_tool(
    agent_name="flight_assistant",
    description="Transfer user to the flight-booking assistant."
)

# 构建专业代理
flight_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_flight, transfer_to_hotel],
    prompt="You are a flight booking assistant. If user needs hotel, transfer to hotel assistant.",
    name="flight_assistant"
)

hotel_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel, transfer_to_flight],
    prompt="You are a hotel booking assistant. If user needs flights, transfer to flight assistant.",
    name="hotel_assistant"
)

注意每个代理的工具列表都包含了对方的 Handoff 工具。这样当航班助手发现用户需要订酒店时，可以调用 transfer_to_hotel 工具，把控制权交给酒店助手。酒店助手同理。

最后，我们需要在父图中注册这些代理节点，并定义入口点：

from langgraph.graph import StateGraph, START, MessagesState

multi_agent_graph = (
    StateGraph(MessagesState)
    .add_node(flight_assistant)
    .add_node(hotel_assistant)
    .add_edge(START, "flight_assistant")
    .compile()
)

运行这个图时，框架会自动处理代理之间的交接。当航班助手调用转移工具，LangGraph 会捕获 Command 对象，暂停当前代理的执行，将状态更新应用到父图，然后跳转到酒店助手节点继续执行。

Supervisor 监督代理架构

Supervisor 架构是多代理系统中最直观、最容易理解的模式。它引入一个中心调度代理，所有任务分配和通信协调都由这个监督者统一管理。其他代理不直接对话，而是通过监督者中转，形成星型拓扑结构。

这种模式的优势在于控制流清晰。监督者拥有全局视角，可以根据当前上下文和任务需求做出最优的代理选择决策。对于需要严格流程管控的场景，比如审批工作流、任务依赖复杂的业务流程，Supervisor 架构特别合适。

使用预构建库快速实现

LangGraph 提供了 langgraph-supervisor 库，几行代码就能搭建一个 Supervisor 系统。这是最简单的入门方式。

# 安装库
# pip install langgraph-supervisor

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

# 创建专业代理
flight_assistant = create_react_agent(
    model="openai:gpt-4o",
    tools=[book_flight],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)

hotel_assistant = create_react_agent(
    model="openai:gpt-4o",
    tools=[book_hotel],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)

# 创建监督者
supervisor = create_supervisor(
    agents=[flight_assistant, hotel_assistant],
    model=ChatOpenAI(model="gpt-4o"),
    prompt="You manage a hotel booking assistant and a flight booking assistant. Assign work to them."
).compile()

# 运行
for chunk in supervisor.stream({
    "messages": [{"role": "user", "content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"}]
}):
    print(chunk)

这个高层 API 隐藏了所有细节。监督者自动识别用户意图，决定调用哪个代理，处理代理返回的结果，并在需要时继续调度。对于快速原型开发，这种方式非常高效。

从零实现 Supervisor

预构建库虽然方便，但理解底层原理更重要。手动实现 Supervisor 能让我们更灵活地控制调度逻辑，处理复杂场景。

核心思路是：Supervisor 本身也是一个代理，它的工具就是其他专业代理。当 Supervisor 决定调用某个代理时，实际上是通过 Handoff 机制将控制权转移过去。

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END

model = ChatOpenAI()

def supervisor(state: MessagesState) -> Command[Literal["flight_assistant", "hotel_assistant", END]]:
    # 分析对话历史，决定下一步
    messages = state["messages"]
    response = model.invoke(
        messages + [{
            "role": "system",
            "content": "根据用户需求，决定调用 flight_assistant 还是 hotel_assistant。如果任务完成，返回 __end__"
        }]
    )
    
    next_agent = parse_decision(response.content)
    if next_agent == "__end__":
        return Command(goto=END)
    return Command(goto=next_agent)

def flight_assistant(state: MessagesState) -> Command[Literal["supervisor"]]:
    # 执行航班预订逻辑
    response = model.invoke(state["messages"])
    # 完成后必须返回 supervisor
    return Command(
        goto="supervisor",
        update={"messages": [response]}
    )

def hotel_assistant(state: MessagesState) -> Command[Literal["supervisor"]]:
    # 执行酒店预订逻辑
    response = model.invoke(state["messages"])
    return Command(
        goto="supervisor",
        update={"messages": [response]}
    )

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor)
builder.add_node("flight_assistant", flight_assistant)
builder.add_node("hotel_assistant", hotel_assistant)

builder.add_edge(START, "supervisor")
builder.add_edge("flight_assistant", "supervisor")
builder.add_edge("hotel_assistant", "supervisor")

supervisor_graph = builder.compile()

这个实现中，Supervisor 节点负责决策，专业代理节点负责执行。每个代理完成后必须返回 Supervisor，形成闭环。Command 的 goto 参数明确指定了下一步去向，控制流完全由代码定义，非常清晰。

Supervisor 的决策逻辑可以做得更复杂。比如使用结构化输出强制模型返回特定格式，或者维护一个任务队列，支持并行任务分配。对于需要精确控制执行顺序的场景，这种显式控制流比隐式调度更可靠。

Swarm 群体智能模式

Swarm 架构走的是另一条路。它没有中心调度器，代理之间直接通过 Handoff 传递控制权。系统记住当前活跃的代理，下次交互时自动从这个代理继续。这种模式模仿了蜂群或蚁群的去中心化协作方式。

Swarm 的优势在于灵活性和可扩展性。新增代理只需要实现 Handoff 工具，无需修改中心调度逻辑。代理可以根据专业领域动态选择下一个处理者，形成自然的处理链条。对于对话流程不确定、需要频繁切换专业领域的场景，比如客服系统，Swarm 特别合适。

使用预构建库实现 Swarm

和 Supervisor 类似，LangGraph 提供了 langgraph-swarm 库来快速搭建 Swarm 系统。

# 安装库
# pip install langgraph-swarm

from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_swarm, create_handoff_tool

# 创建 Handoff 工具
transfer_to_hotel = create_handoff_tool(
    agent_name="hotel_assistant",
    description="Transfer user to the hotel-booking assistant."
)
transfer_to_flight = create_handoff_tool(
    agent_name="flight_assistant",
    description="Transfer user to the flight-booking assistant."
)

# 构建代理，每个代理都有能力转移到其他代理
flight_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_flight, transfer_to_hotel],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)

hotel_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel, transfer_to_flight],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)

# 创建 Swarm，指定默认启动代理
swarm = create_swarm(
    agents=[flight_assistant, hotel_assistant],
    default_active_agent="flight_assistant"
).compile()

# 运行
for chunk in swarm.stream({
    "messages": [{"role": "user", "content": "book a flight and a hotel"}]
}):
    print(chunk)

Swarm 的运行时行为很有趣。第一次调用从 default_active_agent 开始。如果航班助手决定转移给酒店助手，框架会记录这个状态。下次用户输入时，系统会直接从酒店助手继续，而不是从头开始。这种"记忆"让对话更连贯。

Swarm 的自主调度逻辑

Swarm 的核心是每个代理自主决定下一步。这种去中心化调度通过 Handoff 工具实现，但调度逻辑分布在各个代理中。

# 航班助手的系统提示
flight_prompt = """
You are a flight booking assistant. 
When user needs hotel service, call transfer_to_hotel_assistant tool.
When user needs flight service, handle it yourself.
"""

# 酒店助手的系统提示
hotel_prompt = """
You are a hotel booking assistant.
When user needs flight service, call transfer_to_flight_assistant tool.
When user needs hotel service, handle it yourself.
"""

每个代理都有自己的转移规则。运行时，代理根据用户输入和当前上下文决定是否转移、向谁转移。这种设计让系统更容易扩展——新增一个租车助手，只需要在相关代理的工具列表里加上 transfer_to_car_assistant，无需修改整体调度逻辑。

Swarm 架构下，代理之间的通信是点对点的。虽然框架提供了状态持久化，但代理需要自行维护足够的上下文信息，确保接收方能理解当前任务状态。这对代理的设计提出了更高要求。

跨代理状态共享管理

多代理系统的核心挑战是状态管理。代理之间如何通信？传递什么信息？如何保持上下文一致性？这些问题直接影响系统的可靠性和可维护性。

通信模式选择

代理间通信主要有两种方式：Handoff 和 Tool Call。

Handoff 模式通过 Command 传递整个图状态，适合需要保持完整对话历史的场景。代理可以看到之前的所有交互，做出更明智的决策。但缺点是消息列表可能变得很长，增加 token 消耗和处理延迟。

Tool Call 模式把子代理当作工具调用，只传递必要的参数。Supervisor 架构中的工具调用变体就是典型例子。这种方式更轻量，但代理无法看到完整历史，适合任务边界清晰的场景。

# Handoff 模式：传递完整状态
Command(
    goto="hotel_assistant",
    update={"messages": state["messages"] + [new_message]}
)

# Tool Call 模式：只传递必要参数
tool_call = {
    "name": "hotel_assistant",
    "arguments": {"destination": "NYC", "dates": "2024-01-01 to 2024-01-05"}
}

选择哪种模式取决于具体需求。对于需要深度上下文理解的复杂对话，Handoff 更合适。对于任务明确、参数简单的调用，Tool Call 更高效。

消息传递策略

当使用共享消息列表通信时，还有一个重要决策：代理应该分享完整的思考过程，还是只分享最终结果？

分享完整过程意味着代理的所有中间消息（包括工具调用、观察结果等）都会进入全局消息列表。好处是其他代理能看到完整的推理链条，有助于做出更好的决策。缺点是消息列表会迅速膨胀，可能超出模型的上下文窗口。

# 完整分享模式
def agent_with_full_history(state):
    # 所有中间步骤都更新到 state["messages"]
    response = model.invoke(state["messages"])
    tool_result = tool.invoke(response.tool_calls[0])
    final_answer = model.invoke(state["messages"] + [response, tool_result])
    return {"messages": [response, tool_result, final_answer]}

只分享最终结果则让代理保持私有的"草稿本"，只把最终答案添加到共享状态。这种方式更节省 token，适合代理内部逻辑复杂、中间步骤繁多的场景。

# 仅分享结果模式
def agent_with_private_scratchpad(state):
    # 内部处理，不更新全局状态
    internal_messages = state["messages"].copy()
    response = model.invoke(internal_messages)
    tool_result = tool.invoke(response.tool_calls[0])
    final_answer = model.invoke(internal_messages + [response, tool_result])
    
    # 只返回最终结果
    return {"messages": [final_answer]}

实践中，混合策略往往效果最好。关键决策点使用完整分享，确保透明度；内部细节使用私有草稿，保持效率。

状态模式设计

对于复杂系统，不同代理可能需要不同的状态结构。LangGraph 支持为子图定义独立的状态模式，通过状态转换实现解耦。

class FlightState(MessagesState):
    booking_reference: str
    passenger_details: dict

class HotelState(MessagesState):
    reservation_number: str
    room_preferences: dict

# 航班代理使用 FlightState
flight_assistant = create_react_agent(
    model=model,
    tools=[book_flight],
    state_schema=FlightState
)

# 酒店代理使用 HotelState
hotel_assistant = create_react_agent(
    model=model,
    tools=[book_hotel],
    state_schema=HotelState
)

父图需要处理状态转换。当从航班代理切换到酒店代理时，可能需要提取共享字段（如消息列表），同时保留各自的专业字段。这可以通过在节点函数中显式转换实现。

def call_flight_assistant(state: MessagesState) -> Command:
    # 转换状态
    flight_state = FlightState(
        messages=state["messages"],
        booking_reference="",
        passenger_details={}
    )
    result = flight_assistant.invoke(flight_state)
    
    # 转换回父图状态
    return Command(
        goto="supervisor",
        update={"messages": result["messages"]}
    )

这种状态隔离让系统更健壮。每个代理只关心自己的状态结构，不会意外修改其他代理的数据。同时，父图保持对整体流程的控制，确保数据在代理间正确传递。

多轮对话管理

实际应用中，用户往往需要与系统多轮交互。如何记住当前活跃的代理，并在下一轮继续从这个代理开始？这需要结合检查点和 Handoff 机制。

class MultiAgentState(MessagesState):
    last_active_agent: str

def call_flight_assistant(state: MultiAgentState) -> Command:
    result = flight_assistant.invoke(state)
    return Command(
        update={**result, "last_active_agent": "flight_assistant"},
        goto="human"
    )

def human_node(state: MultiAgentState) -> Command:
    user_input = interrupt("Ready for user input")
    return Command(
        update={"messages": [{"role": "human", "content": user_input}]},
        goto=state["last_active_agent"]  # 回到上次活跃的代理
    )

interrupt 让图在等待用户输入时暂停，状态被检查点保存。下次调用时，从 human_node 恢复，根据 last_active_agent 字段决定跳转到哪个代理。这种设计让对话体验更自然，用户感觉一直在和同一个"专家"对话，即使背后是不同的代理在协作。

多代理架构设计本质上是分布式系统的微缩版。Handoff 是 RPC 调用，Supervisor 是中心调度器，Swarm 是点对点网络，状态管理是分布式一致性。理解这些类比有助于我们更好地设计和调试系统。记住，没有银弹架构，选择哪种模式取决于具体场景：需要严格控制用 Supervisor，追求灵活扩展用 Swarm，简单场景直接用 Handoff 也能解决问题。

下一章将探讨如何在多代理系统中引入人工介入，让机器智能和人类判断更好地结合。