Skip to content

如何实现在代理之间的交接操作

先决条件

本指南假设您熟悉以下内容:

在多代理架构中,代理可以表示为图节点。每个代理节点执行其步骤,并决定是否完成执行或路由到另一个代理,包括可能路由回自身(例如,循环运行)。多代理交互中的自然模式是交接,其中一个代理将控制权移交给另一个代理。交接允许您指定:

  • 目的地:要导航到的目标代理 - LangGraph 中的节点名称
  • 负载:传递给该代理的信息 - LangGraph 中的状态更新

要在 LangGraph 中实现交接,代理节点可以返回 Command 对象,以允许您结合控制流和状态更新

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # 路由/停止的条件可以是任何内容,例如 LLM 工具调用/结构化输出等
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # 指定要调用的下一个代理
        goto=goto,
        # 更新图状态
        update={"my_state_key": "my_state_value"}
    )

最常见的代理类型之一是工具调用代理。对于这些类型的代理,一种模式是将交接包装在工具调用中,例如:

@tool
def transfer_to_bob(state):
    """转移到 bob。"""
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        # 每个工具调用代理都实现为一个子图。
        # 因此,要导航到另一个代理(一个兄弟子图),我们需要指定该导航相对于父图。
        graph=Command.PARENT,
    )

本指南展示了如何:

  • 使用 Command 实现交接:代理节点根据条件(通常是基于 LLM 的)决定交接给谁,并明确返回交接。这些在您需要对代理如何路由到另一个代理进行细粒度控制时非常有用。它们非常适合实现监督架构中的监督代理。
  • 使用工具实现交接:工具调用代理可以访问工具,这些工具可以通过 Command 返回交接。代理中的工具执行节点识别工具返回的 Command 对象并相应地路由。交接工具是一种通用的原语,对于包含工具调用代理的任何多代理系统都非常有用。

设置

%%capture --no-stderr
%pip install -U langgraph langchain-anthropic
import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

为LangGraph开发设置LangSmith

注册LangSmith,可以快速发现并解决您的LangGraph项目中的问题,提高性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息,请参阅这里

使用 Command 实现交接操作

让我们实现一个包含两个代理的系统:

  • 一个加法专家(只能执行加法操作)
  • 一个乘法专家(只能执行乘法操作)

在这个例子中,代理将依赖大型语言模型(LLM)来执行数学运算。在更现实的后续示例中,我们将为代理提供执行数学运算的工具。

当加法专家需要帮助进行乘法运算时,它会将任务交给乘法专家,反之亦然。这是一个简单的多代理网络示例。

每个代理将有一个对应的节点函数,该函数可以有条件地返回一个Command对象(例如我们的交接操作)。节点函数将使用一个带有系统提示的LLM和一个工具,该工具可以让代理在需要交接给其他代理时发出信号。如果LLM的响应中包含了工具调用,我们将返回一个Command(goto=<其他代理>)

注意:虽然我们使用工具让LLM发出交接信号,但交接的条件可以是任何内容:来自LLM的特定响应文本、结构化输出、任何其他自定义逻辑等。

from typing_extensions import Literal
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.types import Command

model = ChatAnthropic(model="claude-3-5-sonnet-latest")


@tool
def transfer_to_multiplication_expert():
    """Ask multiplication agent for help."""
    # This tool is not returning anything: we're just using it
    # as a way for LLM to signal that it needs to hand off to another agent
    # (See the paragraph above)
    return


@tool
def transfer_to_addition_expert():
    """Ask addition agent for help."""
    return


def addition_expert(
    state: MessagesState,
) -> Command[Literal["multiplication_expert", "__end__"]]:
    system_prompt = (
        "You are an addition expert, you can ask the multiplication expert for help with multiplication. "
        "Always do your portion of calculation before the handoff."
    )
    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    ai_msg = model.bind_tools([transfer_to_multiplication_expert]).invoke(messages)
    # If there are tool calls, the LLM needs to hand off to another agent
    if len(ai_msg.tool_calls) > 0:
        tool_call_id = ai_msg.tool_calls[-1]["id"]
        # NOTE: it's important to insert a tool message here because LLM providers are expecting
        # all AI messages to be followed by a corresponding tool result message
        tool_msg = {
            "role": "tool",
            "content": "Successfully transferred",
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto="multiplication_expert", update={"messages": [ai_msg, tool_msg]}
        )

    # If the expert has an answer, return it directly to the user
    return {"messages": [ai_msg]}


def multiplication_expert(
    state: MessagesState,
) -> Command[Literal["addition_expert", "__end__"]]:
    system_prompt = (
        "You are a multiplication expert, you can ask an addition expert for help with addition. "
        "Always do your portion of calculation before the handoff."
    )
    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    ai_msg = model.bind_tools([transfer_to_addition_expert]).invoke(messages)
    if len(ai_msg.tool_calls) > 0:
        tool_call_id = ai_msg.tool_calls[-1]["id"]
        tool_msg = {
            "role": "tool",
            "content": "Successfully transferred",
            "tool_call_id": tool_call_id,
        }
        return Command(goto="addition_expert", update={"messages": [ai_msg, tool_msg]})

    return {"messages": [ai_msg]}

API Reference: ToolMessage | tool | ChatAnthropic | StateGraph | START | Command

现在让我们将这两个节点合并成一个单一的图。请注意,代理之间没有任何边!如果专家有答案,它将直接返回给用户,否则它将路由到另一个专家寻求帮助。

builder = StateGraph(MessagesState)
builder.add_node("addition_expert", addition_expert)
builder.add_node("multiplication_expert", multiplication_expert)
# we'll always start with the addition expert
builder.add_edge(START, "addition_expert")
graph = builder.compile()

最后,让我们定义一个辅助函数,以便漂亮地渲染流式输出:

from langchain_core.messages import convert_to_messages


def pretty_print_messages(update):
    if isinstance(update, tuple):
        ns, update = update
        # skip parent graph updates in the printouts
        if len(ns) == 0:
            return

        graph_id = ns[-1].split(":")[0]
        print(f"Update from subgraph {graph_id}:")
        print("\n")

    for node_name, node_update in update.items():
        print(f"Update from node {node_name}:")
        print("\n")

        for m in convert_to_messages(node_update["messages"]):
            m.pretty_print()
        print("\n")

API Reference: convert_to_messages

让我们运行一个需要同时进行加法和乘法运算的表达式的图:

for chunk in graph.stream(
    {"messages": [("user", "what's (3 + 5) * 12")]},
):
    pretty_print_messages(chunk)
Update from node addition_expert:


================================== Ai Message ==================================

[{'text': "Let me help break this down:\n\nFirst, I'll handle the addition part since I'm the addition expert:\n3 + 5 = 8\n\nNow, for the multiplication of 8 * 12, I'll need to ask the multiplication expert for help.", 'type': 'text'}, {'id': 'toolu_015LCrsomHbeoQPtCzuff78Y', 'input': {}, 'name': 'transfer_to_multiplication_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_multiplication_expert (toolu_015LCrsomHbeoQPtCzuff78Y)
 Call ID: toolu_015LCrsomHbeoQPtCzuff78Y
  Args:
================================= Tool Message =================================

Successfully transferred


Update from node multiplication_expert:


================================== Ai Message ==================================

[{'text': 'I see there was an error in my approach. I am actually the multiplication expert, and I need to ask the addition expert for help with (3 + 5) first.', 'type': 'text'}, {'id': 'toolu_01HFcB8WesPfDyrdgxoXApZk', 'input': {}, 'name': 'transfer_to_addition_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_addition_expert (toolu_01HFcB8WesPfDyrdgxoXApZk)
 Call ID: toolu_01HFcB8WesPfDyrdgxoXApZk
  Args:
================================= Tool Message =================================

Successfully transferred


Update from node addition_expert:


================================== Ai Message ==================================

Now that I have the result of 3 + 5 = 8 from the addition expert, I can multiply 8 * 12:

8 * 12 = 96

So, (3 + 5) * 12 = 96
你可以看到加法专家首先处理了括号内的表达式,然后交由乘法专家完成计算。

现在让我们看看如何可以使用特殊的交接工具来实现同样的系统,并为我们的代理提供实际的数学工具。

使用工具实现交接工作

实现交接工具

在前面的例子中,我们显式地在每个代理节点中定义了自定义交接。另一种模式是创建专门的**交接工具**,这些工具直接返回Command对象。当一个代理调用这样的工具时,它将控制权交给了另一个代理。具体来说,代理中的工具执行节点会识别工具返回的Command对象,并相应地路由控制流。注意:与前面的例子不同,调用工具的代理不是一个单独的节点,而是一个可以作为子图节点添加到多代理图中的另一个图。

实现交接工具时有几点重要的考虑:

  • 由于每个代理都是另一个图中的一个__子图__节点,并且工具将在代理子图节点之一中被调用(例如工具执行器),因此需要在Command中指定graph=Command.PARENT,这样LangGraph就知道要在代理子图之外导航。
  • 我们可以选择指定一个状态更新,该更新将在下一个代理被调用之前应用于父图的状态。

  • 我们可以选择向工具提供以下内容(在工具函数签名中):

    这些不是必需的,但对创建传递给下一个代理的状态更新很有用。

from typing import Annotated

from langchain_core.tools import tool
from langchain_core.tools.base import InjectedToolCallId
from langgraph.prebuilt import InjectedState


def make_handoff_tool(*, agent_name: str):
    """Create a tool that can return handoff via a Command"""
    tool_name = f"transfer_to_{agent_name}"

    @tool(tool_name)
    def handoff_to_agent(
        # # optionally pass current graph state to the tool (will be ignored by the LLM)
        state: Annotated[dict, InjectedState],
        # optionally pass the current tool call ID (will be ignored by the LLM)
        tool_call_id: Annotated[str, InjectedToolCallId],
    ):
        """Ask another agent for help."""
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": tool_name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            # navigate to another agent node in the PARENT graph
            goto=agent_name,
            graph=Command.PARENT,
            # This is the state update that the agent `agent_name` will see when it is invoked.
            # We're passing agent's FULL internal message history AND adding a tool message to make sure
            # the resulting chat history is valid. See the paragraph above for more information.
            update={"messages": state["messages"] + [tool_message]},
        )

    return handoff_to_agent

API Reference: tool | InjectedToolCallId | InjectedState

使用自定义代理

为了演示如何使用交接工具,我们首先实现一个简单的预构建create_react_agent版本。这在你希望拥有自定义工具调用代理实现并利用交接工具时非常有用。

from typing_extensions import Literal
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.types import Command


def make_agent(model, tools, system_prompt=None):
    model_with_tools = model.bind_tools(tools)
    tools_by_name = {tool.name: tool for tool in tools}

    def call_model(state: MessagesState) -> Command[Literal["call_tools", "__end__"]]:
        messages = state["messages"]
        if system_prompt:
            messages = [{"role": "system", "content": system_prompt}] + messages

        response = model_with_tools.invoke(messages)
        if len(response.tool_calls) > 0:
            return Command(goto="call_tools", update={"messages": [response]})

        return {"messages": [response]}

    # NOTE: this is a simplified version of the prebuilt ToolNode
    # If you want to have a tool node that has full feature parity, please refer to the source code
    def call_tools(state: MessagesState) -> Command[Literal["call_model"]]:
        tool_calls = state["messages"][-1].tool_calls
        results = []
        for tool_call in tool_calls:
            tool_ = tools_by_name[tool_call["name"]]
            tool_input_fields = tool_.get_input_schema().model_json_schema()[
                "properties"
            ]

            # this is simplified for demonstration purposes and
            # is different from the ToolNode implementation
            if "state" in tool_input_fields:
                # inject state
                tool_call = {**tool_call, "args": {**tool_call["args"], "state": state}}

            tool_response = tool_.invoke(tool_call)
            if isinstance(tool_response, ToolMessage):
                results.append(Command(update={"messages": [tool_response]}))

            # handle tools that return Command directly
            elif isinstance(tool_response, Command):
                results.append(tool_response)

        # NOTE: nodes in LangGraph allow you to return list of updates, including Command objects
        return results

    graph = StateGraph(MessagesState)
    graph.add_node(call_model)
    graph.add_node(call_tools)
    graph.add_edge(START, "call_model")
    graph.add_edge("call_tools", "call_model")

    return graph.compile()

API Reference: ToolMessage | tool | StateGraph | START | Command

让我们也为代理定义一些数学工具:

@tool
def add(a: int, b: int) -> int:
    """Adds two numbers."""
    return a + b


@tool
def multiply(a: int, b: int) -> int:
    """Multiplies two numbers."""
    return a * b

让我们测试一下代理实现,确保它按预期工作:

agent = make_agent(model, [add, multiply])

for chunk in agent.stream({"messages": [("user", "what's (3 + 5) * 12")]}):
    pretty_print_messages(chunk)
Update from node call_model:


================================== Ai Message ==================================

[{'text': "I'll help break this down into two steps:\n1. First calculate 3 + 5\n2. Then multiply that result by 12\n\nLet me make these calculations:\n\n1. Adding 3 and 5:", 'type': 'text'}, {'id': 'toolu_01DUAzgWFqq6XZtj1hzHTka9', 'input': {'a': 3, 'b': 5}, 'name': 'add', 'type': 'tool_use'}]
Tool Calls:
  add (toolu_01DUAzgWFqq6XZtj1hzHTka9)
 Call ID: toolu_01DUAzgWFqq6XZtj1hzHTka9
  Args:
    a: 3
    b: 5


Update from node call_tools:


================================= Tool Message =================================
Name: add

8


Update from node call_model:


================================== Ai Message ==================================

[{'text': '2. Multiplying the result (8) by 12:', 'type': 'text'}, {'id': 'toolu_01QXi1prSN4etgJ1QCuFJsgN', 'input': {'a': 8, 'b': 12}, 'name': 'multiply', 'type': 'tool_use'}]
Tool Calls:
  multiply (toolu_01QXi1prSN4etgJ1QCuFJsgN)
 Call ID: toolu_01QXi1prSN4etgJ1QCuFJsgN
  Args:
    a: 8
    b: 12


Update from node call_tools:


================================= Tool Message =================================
Name: multiply

96


Update from node call_model:


================================== Ai Message ==================================

The result of (3 + 5) * 12 = 96
现在,我们可以使用乘法和加法专家代理来实现我们的多智能体系统。这一次我们将为它们提供进行数学运算的工具,以及我们特殊的交接工具:

addition_expert = make_agent(
    model,
    [add, make_handoff_tool(agent_name="multiplication_expert")],
    system_prompt="You are an addition expert, you can ask the multiplication expert for help with multiplication.",
)
multiplication_expert = make_agent(
    model,
    [multiply, make_handoff_tool(agent_name="addition_expert")],
    system_prompt="You are a multiplication expert, you can ask an addition expert for help with addition.",
)

builder = StateGraph(MessagesState)
builder.add_node("addition_expert", addition_expert)
builder.add_node("multiplication_expert", multiplication_expert)
builder.add_edge(START, "addition_expert")
graph = builder.compile()

让我们使用与之前相同的多步计算输入来运行图:

for chunk in graph.stream(
    {"messages": [("user", "what's (3 + 5) * 12")]}, subgraphs=True
):
    pretty_print_messages(chunk)
Update from subgraph addition_expert:


Update from node call_model:


================================== Ai Message ==================================

[{'text': "I can help with the addition part (3 + 5), but I'll need to ask the multiplication expert for help with multiplying the result by 12. Let me break this down:\n\n1. First, let me calculate 3 + 5:", 'type': 'text'}, {'id': 'toolu_01McaW4XWczLGKaetg88fxQ5', 'input': {'a': 3, 'b': 5}, 'name': 'add', 'type': 'tool_use'}]
Tool Calls:
  add (toolu_01McaW4XWczLGKaetg88fxQ5)
 Call ID: toolu_01McaW4XWczLGKaetg88fxQ5
  Args:
    a: 3
    b: 5


Update from subgraph addition_expert:


Update from node call_tools:


================================= Tool Message =================================
Name: add

8


Update from subgraph addition_expert:


Update from node call_model:


================================== Ai Message ==================================

[{'text': "Now that we have 8, we need to multiply it by 12. I'll ask the multiplication expert for help with this:", 'type': 'text'}, {'id': 'toolu_01KpdUhHuyrmha62z5SduKRc', 'input': {}, 'name': 'transfer_to_multiplication_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_multiplication_expert (toolu_01KpdUhHuyrmha62z5SduKRc)
 Call ID: toolu_01KpdUhHuyrmha62z5SduKRc
  Args:


Update from subgraph multiplication_expert:


Update from node call_model:


================================== Ai Message ==================================

[{'text': 'Now that we have 8 as the result of the addition, I can help with the multiplication by 12:', 'type': 'text'}, {'id': 'toolu_01Vnp4k3TE87siad3BNJgRKb', 'input': {'a': 8, 'b': 12}, 'name': 'multiply', 'type': 'tool_use'}]
Tool Calls:
  multiply (toolu_01Vnp4k3TE87siad3BNJgRKb)
 Call ID: toolu_01Vnp4k3TE87siad3BNJgRKb
  Args:
    a: 8
    b: 12


Update from subgraph multiplication_expert:


Update from node call_tools:


================================= Tool Message =================================
Name: multiply

96


Update from subgraph multiplication_expert:


Update from node call_model:


================================== Ai Message ==================================

The final result is 96.

To break down the steps:
1. 3 + 5 = 8
2. 8 * 12 = 96
我们可以看到,在添加专家完成计算的第一部分(调用add工具之后),它决定将任务交给乘法专家,由后者计算最终结果。

与预构建的ReAct代理一起使用

如果你不需要额外的自定义功能,可以使用预构建的create_react_agent,它通过ToolNode内置了交接工具的支持。

from langgraph.prebuilt import create_react_agent

addition_expert = create_react_agent(
    model,
    [add, make_handoff_tool(agent_name="multiplication_expert")],
    prompt="You are an addition expert, you can ask the multiplication expert for help with multiplication.",
)

multiplication_expert = create_react_agent(
    model,
    [multiply, make_handoff_tool(agent_name="addition_expert")],
    prompt="You are a multiplication expert, you can ask an addition expert for help with addition.",
)

builder = StateGraph(MessagesState)
builder.add_node("addition_expert", addition_expert)
builder.add_node("multiplication_expert", multiplication_expert)
builder.add_edge(START, "addition_expert")
graph = builder.compile()

API Reference: create_react_agent

我们现在可以验证预构建的ReAct代理与上面的自定义代理工作方式完全相同:

for chunk in graph.stream(
    {"messages": [("user", "what's (3 + 5) * 12")]}, subgraphs=True
):
    pretty_print_messages(chunk)
Update from subgraph addition_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': "I can help with the addition part of this calculation (3 + 5), and then I'll need to ask the multiplication expert for help with multiplying the result by 12.\n\nLet me first calculate 3 + 5:", 'type': 'text'}, {'id': 'toolu_01GUasumGGJVXDV7TJEqEfmY', 'input': {'a': 3, 'b': 5}, 'name': 'add', 'type': 'tool_use'}]
Tool Calls:
  add (toolu_01GUasumGGJVXDV7TJEqEfmY)
 Call ID: toolu_01GUasumGGJVXDV7TJEqEfmY
  Args:
    a: 3
    b: 5


Update from subgraph addition_expert:


Update from node tools:


================================= Tool Message =================================
Name: add

8


Update from subgraph addition_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': "Now that we have 8, we need to multiply it by 12. Since I'm an addition expert, I'll transfer this to the multiplication expert to complete the calculation:", 'type': 'text'}, {'id': 'toolu_014HEbwiH2jVno8r1Pc6t9Qh', 'input': {}, 'name': 'transfer_to_multiplication_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_multiplication_expert (toolu_014HEbwiH2jVno8r1Pc6t9Qh)
 Call ID: toolu_014HEbwiH2jVno8r1Pc6t9Qh
  Args:


Update from subgraph multiplication_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': 'I notice I made a mistake - I actually don\'t have access to the "add" function or "transfer_to_multiplication_expert". Instead, I am the multiplication expert and I should ask the addition expert for help with the first part. Let me correct this:', 'type': 'text'}, {'id': 'toolu_01VAGpmr4ysHjvvuZp3q5Dzj', 'input': {}, 'name': 'transfer_to_addition_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_addition_expert (toolu_01VAGpmr4ysHjvvuZp3q5Dzj)
 Call ID: toolu_01VAGpmr4ysHjvvuZp3q5Dzj
  Args:


Update from subgraph addition_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': "I'll help you with the addition part of (3 + 5) * 12. First, let me calculate 3 + 5:", 'type': 'text'}, {'id': 'toolu_01RE16cRGVo4CC4wwHFB6gaE', 'input': {'a': 3, 'b': 5}, 'name': 'add', 'type': 'tool_use'}]
Tool Calls:
  add (toolu_01RE16cRGVo4CC4wwHFB6gaE)
 Call ID: toolu_01RE16cRGVo4CC4wwHFB6gaE
  Args:
    a: 3
    b: 5


Update from subgraph addition_expert:


Update from node tools:


================================= Tool Message =================================
Name: add

8


Update from subgraph addition_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': "Now that we have 8, we need to multiply it by 12. Since I'm an addition expert, I'll need to transfer this to the multiplication expert to complete the calculation:", 'type': 'text'}, {'id': 'toolu_01HBDRh64SzGcCp7EX1u3MFa', 'input': {}, 'name': 'transfer_to_multiplication_expert', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_multiplication_expert (toolu_01HBDRh64SzGcCp7EX1u3MFa)
 Call ID: toolu_01HBDRh64SzGcCp7EX1u3MFa
  Args:


Update from subgraph multiplication_expert:


Update from node agent:


================================== Ai Message ==================================

[{'text': 'Now that I have the result of 3 + 5 = 8, I can help with multiplying by 12:', 'type': 'text'}, {'id': 'toolu_014Ay95rsKvvbWWJV4CcZSPY', 'input': {'a': 8, 'b': 12}, 'name': 'multiply', 'type': 'tool_use'}]
Tool Calls:
  multiply (toolu_014Ay95rsKvvbWWJV4CcZSPY)
 Call ID: toolu_014Ay95rsKvvbWWJV4CcZSPY
  Args:
    a: 8
    b: 12


Update from subgraph multiplication_expert:


Update from node tools:


================================= Tool Message =================================
Name: multiply

96


Update from subgraph multiplication_expert:


Update from node agent:


================================== Ai Message ==================================

The final result is 96. Here's the complete calculation:
(3 + 5) * 12 = 8 * 12 = 96

Comments