如何构建多智能体网络（函数式 API）¶

前提条件

本指南假设你熟悉以下内容：

在本操作指南中，我们将演示如何实现一个多智能体网络架构，其中每个智能体都可以与其他每个智能体进行通信（多对多连接），并且可以决定接下来调用哪个智能体。我们将使用函数式 API —— 各个智能体将被定义为任务，智能体交接将在主 entrypoint() 中定义：

from langgraph.func import entrypoint
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool


# 定义一个工具，用于表明将交接给不同智能体的意图
@tool(return_direct=True)
def transfer_to_hotel_advisor():
    """向酒店顾问智能体寻求帮助。"""
    return "已成功转接至酒店顾问"


# 定义一个智能体
travel_advisor_tools = [transfer_to_hotel_advisor, ...]
travel_advisor = create_react_agent(model, travel_advisor_tools)


# 定义一个调用智能体的任务
@task
def call_travel_advisor(messages):
    response = travel_advisor.invoke({"messages": messages})
    return response["messages"]


# 定义多智能体网络工作流
@entrypoint()
def workflow(messages):
    call_active_agent = call_travel_advisor
    while True:
        agent_messages = call_active_agent(messages).result()
        messages = messages + agent_messages
        call_active_agent = get_next_agent(messages)
    return messages

API Reference: tool

安装设置¶

首先，让我们安装所需的软件包。

%%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")

ANTHROPIC_API_KEY:  ········

为 LangGraph 开发设置 LangSmith

注册 LangSmith，以便快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 — 点击此处了解更多关于如何开始使用的信息。

旅行社示例¶

在这个示例中，我们将构建一个可以相互通信的旅行助理智能体团队。

我们将创建 2 个智能体：

旅行顾问：可以提供旅行目的地推荐。可以向 酒店顾问 寻求帮助。
酒店顾问：可以提供酒店推荐。可以向 旅行顾问 寻求帮助。

这是一个全连接网络——每个智能体都可以与其他任何智能体进行交流。

首先，让我们创建一些智能体将使用的工具：

import random
from typing_extensions import Literal
from langchain_core.tools import tool


@tool
def get_travel_recommendations():
    """Get recommendation for travel destinations"""
    return random.choice(["aruba", "turks and caicos"])


@tool
def get_hotel_recommendations(location: Literal["aruba", "turks and caicos"]):
    """Get hotel recommendations for a given destination."""
    return {
        "aruba": [
            "The Ritz-Carlton, Aruba (Palm Beach)"
            "Bucuti & Tara Beach Resort (Eagle Beach)"
        ],
        "turks and caicos": ["Grace Bay Club", "COMO Parrot Cay"],
    }[location]


@tool(return_direct=True)
def transfer_to_hotel_advisor():
    """Ask hotel advisor agent for help."""
    return "Successfully transferred to hotel advisor"


@tool(return_direct=True)
def transfer_to_travel_advisor():
    """Ask travel advisor agent for help."""
    return "Successfully transferred to travel advisor"

API Reference: tool

转接工具

你可能已经注意到，我们在转接工具中使用了 @tool(return_direct=True)。这样做是为了让单个智能体（例如 travel_advisor）在调用这些工具后能够提前退出 ReAct 循环。这是我们期望的行为，因为我们希望检测到智能体调用此工具时，能立即将控制权移交给另一个智能体。

注意：这是为了与预构建的 create_react_agent 配合使用的 —— 如果你正在构建自定义智能体，请确保手动添加逻辑，以处理标记为 return_direct 的工具的提前退出情况。

现在让我们定义我们的智能体任务，并将它们组合成一个单一的多智能体网络工作流：

from langchain_core.messages import AIMessage
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task

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

# Define travel advisor ReAct agent
travel_advisor_tools = [
    get_travel_recommendations,
    transfer_to_hotel_advisor,
]
travel_advisor = create_react_agent(
    model,
    travel_advisor_tools,
    state_modifier=(
        "You are a general travel expert that can recommend travel destinations (e.g. countries, cities, etc). "
        "If you need hotel recommendations, ask 'hotel_advisor' for help. "
        "You MUST include human-readable response before transferring to another agent."
    ),
)


@task
def call_travel_advisor(messages):
    # You can also add additional logic like changing the input to the agent / output from the agent, etc.
    # NOTE: we're invoking the ReAct agent with the full history of messages in the state
    response = travel_advisor.invoke({"messages": messages})
    return response["messages"]


# Define hotel advisor ReAct agent
hotel_advisor_tools = [get_hotel_recommendations, transfer_to_travel_advisor]
hotel_advisor = create_react_agent(
    model,
    hotel_advisor_tools,
    state_modifier=(
        "You are a hotel expert that can provide hotel recommendations for a given destination. "
        "If you need help picking travel destinations, ask 'travel_advisor' for help."
        "You MUST include human-readable response before transferring to another agent."
    ),
)


@task
def call_hotel_advisor(messages):
    response = hotel_advisor.invoke({"messages": messages})
    return response["messages"]


@entrypoint()
def workflow(messages):
    messages = add_messages([], messages)

    call_active_agent = call_travel_advisor
    while True:
        agent_messages = call_active_agent(messages).result()
        messages = add_messages(messages, agent_messages)
        ai_msg = next(m for m in reversed(agent_messages) if isinstance(m, AIMessage))
        if not ai_msg.tool_calls:
            break

        tool_call = ai_msg.tool_calls[-1]
        if tool_call["name"] == "transfer_to_travel_advisor":
            call_active_agent = call_travel_advisor
        elif tool_call["name"] == "transfer_to_hotel_advisor":
            call_active_agent = call_hotel_advisor
        else:
            raise ValueError(f"Expected transfer tool, got '{tool_call['name']}'")

    return messages

API Reference: AIMessage

最后，让我们定义一个辅助函数来渲染智能体的输出：

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 workflow.stream(
    [
        {
            "role": "user",
            "content": "i wanna go somewhere warm in the caribbean. pick one destination and give me hotel recommendations",
        }
    ],
    subgraphs=True,
):
    pretty_print_messages(chunk)

Update from subgraph call_travel_advisor:


Update from node agent:


==================================[1m Ai Message [0m==================================

[{'text': "I'll help you find a warm Caribbean destination and then get some hotel recommendations for you.\n\nLet me first get some destination recommendations for the Caribbean region.", 'type': 'text'}, {'id': 'toolu_015vT8PkPq1VXvjrDvSpWUwJ', 'input': {}, 'name': 'get_travel_recommendations', 'type': 'tool_use'}]
Tool Calls:
  get_travel_recommendations (toolu_015vT8PkPq1VXvjrDvSpWUwJ)
 Call ID: toolu_015vT8PkPq1VXvjrDvSpWUwJ
  Args:


Update from subgraph call_travel_advisor:


Update from node tools:


=================================[1m Tool Message [0m=================================
Name: get_travel_recommendations

turks and caicos


Update from subgraph call_travel_advisor:


Update from node agent:


==================================[1m Ai Message [0m==================================

[{'text': "Based on the recommendation, I suggest Turks and Caicos! This beautiful British Overseas Territory is known for its stunning white-sand beaches, crystal-clear turquoise waters, and year-round warm weather. Grace Bay Beach in Providenciales is consistently ranked among the world's best beaches. The islands offer excellent snorkeling, diving, and water sports opportunities, plus a relaxed Caribbean atmosphere.\n\nNow, let me connect you with our hotel advisor to get some specific hotel recommendations for Turks and Caicos.", 'type': 'text'}, {'id': 'toolu_01JY7pNNWFuaWoe9ymxFYiPV', 'input': {}, 'name': 'transfer_to_hotel_advisor', 'type': 'tool_use'}]
Tool Calls:
  transfer_to_hotel_advisor (toolu_01JY7pNNWFuaWoe9ymxFYiPV)
 Call ID: toolu_01JY7pNNWFuaWoe9ymxFYiPV
  Args:


Update from subgraph call_travel_advisor:


Update from node tools:


=================================[1m Tool Message [0m=================================
Name: transfer_to_hotel_advisor

Successfully transferred to hotel advisor


Update from subgraph call_hotel_advisor:


Update from node agent:


==================================[1m Ai Message [0m==================================

[{'text': 'Let me get some hotel recommendations for Turks and Caicos:', 'type': 'text'}, {'id': 'toolu_0129ELa7jFocn16bowaGNapg', 'input': {'location': 'turks and caicos'}, 'name': 'get_hotel_recommendations', 'type': 'tool_use'}]
Tool Calls:
  get_hotel_recommendations (toolu_0129ELa7jFocn16bowaGNapg)
 Call ID: toolu_0129ELa7jFocn16bowaGNapg
  Args:
    location: turks and caicos


Update from subgraph call_hotel_advisor:


Update from node tools:


=================================[1m Tool Message [0m=================================
Name: get_hotel_recommendations

["Grace Bay Club", "COMO Parrot Cay"]


Update from subgraph call_hotel_advisor:


Update from node agent:


==================================[1m Ai Message [0m==================================

Here are two excellent hotel options in Turks and Caicos:

1. Grace Bay Club: This luxury resort is located on the world-famous Grace Bay Beach. It offers all-oceanfront suites, exceptional dining options, and personalized service. The resort features adult-only and family-friendly sections, making it perfect for any type of traveler.

2. COMO Parrot Cay: This exclusive private island resort offers the ultimate luxury escape. It's known for its pristine beach, world-class spa, and holistic wellness programs. The resort provides an intimate, secluded experience with top-notch amenities and service.

Would you like more specific information about either of these properties or would you like to explore hotels in another destination?

瞧！travel_advisor 选定了一个目的地，然后决定调用 hotel_advisor 以获取更多信息！

如何构建多智能体网络（函数式 API）¶

安装设置¶

旅行社示例¶

Comments