如何在多代理应用程序中添加多轮对话（功能API）¶
在本指南中，我们将构建一个应用程序，允许最终用户与一个或多个代理进行*多轮对话*。我们将创建一个节点，使用一个中断来收集用户输入，并返回到**活动**代理。
代理将作为工作流中的任务来实现，执行代理步骤并决定下一步行动：
- **等待用户输入**以继续对话，或
- 转交到另一个代理（或返回自身，如在循环中）通过一个转交。
from langgraph.func import entrypoint, task
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langgraph.types import interrupt
# 定义一个工具，用于指示转交到不同代理的意图
# 注意：这不使用Command(goto)语法来导航到不同的代理：
# 下面的`workflow()`显式处理转交
@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(checkpointer)
def workflow(messages):
call_active_agent = call_travel_advisor
while True:
agent_messages = call_active_agent(messages).result()
ai_msg = get_last_ai_msg(agent_messages)
if not ai_msg.tool_calls:
user_input = interrupt(value="Ready for user input.")
messages = messages + [{"role": "user", "content": user_input}]
continue
messages = messages + agent_messages
call_active_agent = get_next_agent(messages)
return entrypoint.final(value=agent_messages[-1], save=messages)
API Reference: entrypoint | task | create_react_agent | tool | interrupt
设置环境¶
首先，让我们安装所需的包
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应用程序——更多关于如何开始的信息，请参阅这里。
在这个例子中，我们将构建一个旅行助手代理团队，这些代理可以相互交流。
我们将创建两个代理：
travel_advisor：可以提供旅行目的地推荐。可以向hotel_advisor寻求帮助。hotel_advisor：可以提供酒店推荐。可以向travel_advisor寻求帮助。
这是一个完全连接的网络——每个代理都可以与其他任何代理通信。
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的工具。
现在让我们使用预构建的create_react_agent和我们的多代理工作流来创建代理。请注意，每次从每个代理获得最终响应后，我们都会调用interrupt。
import uuid
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
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
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"]
checkpointer = MemorySaver()
def string_to_uuid(input_string):
return str(uuid.uuid5(uuid.NAMESPACE_URL, input_string))
@entrypoint(checkpointer=checkpointer)
def multi_turn_graph(messages, previous):
previous = previous or []
messages = add_messages(previous, messages)
call_active_agent = call_travel_advisor
while True:
agent_messages = call_active_agent(messages).result()
messages = add_messages(messages, agent_messages)
# Find the last AI message
# If one of the handoff tools is called, the last message returned
# by the agent will be a ToolMessage because we set them to have
# "return_direct=True". This means that the last AIMessage will
# have tool calls.
# Otherwise, the last returned message will be an AIMessage with
# no tool calls, which means we are ready for new input.
ai_msg = next(m for m in reversed(agent_messages) if isinstance(m, AIMessage))
if not ai_msg.tool_calls:
user_input = interrupt(value="Ready for user input.")
# Add user input as a human message
# NOTE: we generate unique ID for the human message based on its content
# it's important, since on subsequent invocations previous user input (interrupt) values
# will be looked up again and we will attempt to add them again here
# `add_messages` deduplicates messages based on the ID, ensuring correct message history
human_message = {
"role": "user",
"content": user_input,
"id": string_to_uuid(user_input),
}
messages = add_messages(messages, [human_message])
continue
tool_call = ai_msg.tool_calls[-1]
if tool_call["name"] == "transfer_to_hotel_advisor":
call_active_agent = call_hotel_advisor
elif tool_call["name"] == "transfer_to_travel_advisor":
call_active_agent = call_travel_advisor
else:
raise ValueError(f"Expected transfer tool, got '{tool_call['name']}'")
return entrypoint.final(value=agent_messages[-1], save=messages)
API Reference: AIMessage | ChatAnthropic | create_react_agent | add_messages | entrypoint | task | MemorySaver | interrupt | Command
测试多轮对话¶
让我们用这个应用程序测试一个多轮对话。
thread_config = {"configurable": {"thread_id": uuid.uuid4()}}
inputs = [
# 1st round of conversation,
{
"role": "user",
"content": "i wanna go somewhere warm in the caribbean",
"id": str(uuid.uuid4()),
},
# Since we're using `interrupt`, we'll need to resume using the Command primitive.
# 2nd round of conversation,
Command(
resume="could you recommend a nice hotel in one of the areas and tell me which area it is."
),
# 3rd round of conversation,
Command(
resume="i like the first one. could you recommend something to do near the hotel?"
),
]
for idx, user_input in enumerate(inputs):
print()
print(f"--- Conversation Turn {idx + 1} ---")
print()
print(f"User: {user_input}")
print()
for update in multi_turn_graph.stream(
user_input,
config=thread_config,
stream_mode="updates",
):
for node_id, value in update.items():
if isinstance(value, list) and value:
last_message = value[-1]
if isinstance(last_message, dict) or last_message.type != "ai":
continue
print(f"{node_id}: {last_message.content}")
--- Conversation Turn 1 ---
User: {'role': 'user', 'content': 'i wanna go somewhere warm in the caribbean', 'id': 'f48d82a7-7efa-43f5-ad4c-541758c95f61'}
call_travel_advisor: Based on the recommendations, Aruba would be an excellent choice for your Caribbean getaway! Known as "One Happy Island," Aruba offers:
- Year-round warm weather with consistent temperatures around 82°F (28°C)
- Beautiful white sand beaches like Eagle Beach and Palm Beach
- Crystal clear waters perfect for swimming and snorkeling
- Minimal rainfall and location outside the hurricane belt
- Rich culture blending Dutch and Caribbean influences
- Various activities from water sports to desert-like landscape exploration
- Excellent dining and shopping options
Would you like me to help you find suitable accommodations in Aruba? I can transfer you to our hotel advisor who can recommend specific hotels based on your preferences.
--- Conversation Turn 2 ---
User: Command(resume='could you recommend a nice hotel in one of the areas and tell me which area it is.')
call_hotel_advisor: I can recommend two excellent options in different areas:
1. The Ritz-Carlton, Aruba - Located in Palm Beach
- Luxury beachfront resort
- Located in the vibrant Palm Beach area, known for its lively atmosphere
- Close to restaurants, shopping, and nightlife
- Perfect for those who want a more active vacation with plenty of amenities nearby
2. Bucuti & Tara Beach Resort - Located in Eagle Beach
- Adults-only boutique resort
- Situated on the quieter Eagle Beach
- Known for its romantic atmosphere and excellent service
- Ideal for couples seeking a more peaceful, intimate setting
Would you like more specific information about either of these properties or their locations?
--- Conversation Turn 3 ---
User: Command(resume='i like the first one. could you recommend something to do near the hotel?')
call_travel_advisor: Near The Ritz-Carlton in Palm Beach, here are some popular activities you can enjoy:
1. Palm Beach Strip - Take a walk along this bustling strip filled with restaurants, shops, and bars
2. Visit the Bubali Bird Sanctuary - Just a short distance away
3. Try your luck at the Stellaris Casino - Located right in The Ritz-Carlton
4. Water Sports at Palm Beach - Right in front of the hotel you can:
- Go parasailing
- Try jet skiing
- Take a sunset sailing cruise
5. Visit the Palm Beach Plaza Mall - High-end shopping just a short walk away
6. Enjoy dinner at Madame Janette's - One of Aruba's most famous restaurants nearby
Would you like more specific information about any of these activities or other suggestions in the area?