如何在多智能体应用中添加多轮对话（函数式 API）¶
在本操作指南中，我们将构建一个应用程序，允许最终用户与一个或多个智能体进行多轮对话。我们将创建一个节点，该节点使用 interrupt 来收集用户输入并路由回活动智能体。
智能体将作为工作流程中的任务来实现，该工作流程执行智能体步骤并确定下一步操作
- 等待用户输入以继续对话，或者
- 通过 移交 路由到另一个智能体（或返回自身，例如在循环中）。
from langgraph.func import entrypoint, task
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langgraph.types import interrupt
# Define a tool to signal intent to hand off to a different agent
# Note: this is not using Command(goto) syntax for navigating to different agents:
# `workflow()` below handles the handoffs explicitly
@tool(return_direct=True)
def transfer_to_hotel_advisor():
"""Ask hotel advisor agent for help."""
return "Successfully transferred to hotel advisor"
# define an agent
travel_advisor_tools = [transfer_to_hotel_advisor, ...]
travel_advisor = create_react_agent(model, travel_advisor_tools)
# define a task that calls an agent
@task
def call_travel_advisor(messages):
response = travel_advisor.invoke({"messages": messages})
return response["messages"]
# define the multi-agent network workflow
@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 参考：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")
设置 LangSmith 以进行 LangGraph 开发
注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——在此处阅读有关如何开始的更多信息 here。
在此示例中，我们将构建一个旅行助手智能体团队，他们可以相互通信。
我们将创建 2 个智能体
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 参考：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 参考：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?