如何从头开始创建一个ReAct代理（功能API）¶
本指南演示了如何使用LangGraph 功能API 实现ReAct代理。
ReAct代理是一个工具调用代理，其工作流程如下：
这是一个简单且灵活的设置，可以扩展为包含记忆、人机交互能力和其他功能。有关示例，请参阅专门的操作指南。
设置¶
首先，让我们安装所需的包并设置我们的API密钥：
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
使用LangSmith进行更好的调试
注册 LangSmith 可以快速发现并解决项目中的问题，提高您的 LangGraph 项目的性能。LangSmith 让您可以利用跟踪数据来调试、测试和监控使用 LangGraph 构建的大型语言模型应用——更多关于如何开始的信息，请参阅 文档。
创建ReAct代理¶
现在您已经安装了所需的包并设置了环境变量，我们可以创建我们的代理。
定义模型和工具¶
首先，让我们定义我们将用于示例的工具和模型。在这里，我们将使用一个占位工具来获取某个地点的天气描述。
我们将使用OpenAI聊天模型来进行此示例，但任何支持工具调用的模型参见聊天集成都可以。
API Reference: tool
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
model = ChatOpenAI(model="gpt-4o-mini")
@tool
def get_weather(location: str):
"""Call to get the weather from a specific location."""
# This is a placeholder for the actual implementation
if any([city in location.lower() for city in ["sf", "san francisco"]]):
return "It's sunny!"
elif "boston" in location.lower():
return "It's rainy!"
else:
return f"I am not sure what the weather is in {location}"
tools = [get_weather]
定义任务¶
接下来我们定义要执行的任务。这里有两个不同的任务：
- 调用模型：我们希望使用消息列表查询聊天模型。
- 调用工具：如果我们的模型生成了工具调用，我们希望执行它们。
API Reference: ToolMessage | entrypoint | task
from langchain_core.messages import ToolMessage
from langgraph.func import entrypoint, task
tools_by_name = {tool.name: tool for tool in tools}
@task
def call_model(messages):
"""Call model with a sequence of messages."""
response = model.bind_tools(tools).invoke(messages)
return response
@task
def call_tool(tool_call):
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
return ToolMessage(content=observation, tool_call_id=tool_call["id"])
定义入口点¶
我们的入口点将负责这两个任务的编排。如上所述，当call_model任务生成工具调用时，call_tool任务将为每个调用生成响应。我们将所有消息附加到一个单一的消息列表中。
Tip
请注意，由于任务返回类似于未来的对象，因此以下实现会并行执行工具。
API Reference: add_messages
from langgraph.graph.message import add_messages
@entrypoint()
def agent(messages):
llm_response = call_model(messages).result()
while True:
if not llm_response.tool_calls:
break
# Execute tools
tool_result_futures = [
call_tool(tool_call) for tool_call in llm_response.tool_calls
]
tool_results = [fut.result() for fut in tool_result_futures]
# Append to message list
messages = add_messages(messages, [llm_response, *tool_results])
# Call model again
llm_response = call_model(messages).result()
return llm_response
使用方法¶
要使用我们的代理，我们需要用一个消息列表来调用它。根据我们的实现，这些可以是LangChain 消息 对象或OpenAI风格的字典：
user_message = {"role": "user", "content": "What's the weather in san francisco?"}
print(user_message)
for step in agent.stream([user_message]):
for task_name, message in step.items():
if task_name == "agent":
continue # Just print task updates
print(f"\n{task_name}:")
message.pretty_print()
{'role': 'user', 'content': "What's the weather in san francisco?"}
call_model:
================================== Ai Message ==================================
Tool Calls:
get_weather (call_tNnkrjnoz6MNfCHJpwfuEQ0v)
Call ID: call_tNnkrjnoz6MNfCHJpwfuEQ0v
Args:
location: san francisco
call_tool:
================================= Tool Message =================================
It's sunny!
call_model:
================================== Ai Message ==================================
The weather in San Francisco is sunny!
get_weather工具，并在接收到来自工具的信息后回应用户。查看LangSmith跟踪此处。
添加线程级持久化¶
添加线程级持久化可以让我们的代理支持对话体验：后续调用会追加到之前的对话消息列表中，保留完整的对话上下文。
要为我们的代理添加线程级持久化：
- 选择一个检查点器：这里我们将使用MemorySaver，这是一个简单的内存检查点器。
- 更新入口点以接受前一消息状态作为第二个参数。在这里，我们只是将消息更新追加到先前的消息序列中。
- 选择哪些值将从工作流返回，哪些值将由检查点器保存为
previous，使用entrypoint.final（可选）
API Reference: MemorySaver
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
@entrypoint(checkpointer=checkpointer)
def agent(messages, previous):
if previous is not None:
messages = add_messages(previous, messages)
llm_response = call_model(messages).result()
while True:
if not llm_response.tool_calls:
break
# Execute tools
tool_result_futures = [
call_tool(tool_call) for tool_call in llm_response.tool_calls
]
tool_results = [fut.result() for fut in tool_result_futures]
# Append to message list
messages = add_messages(messages, [llm_response, *tool_results])
# Call model again
llm_response = call_model(messages).result()
# Generate final response
messages = add_messages(messages, llm_response)
return entrypoint.final(value=llm_response, save=messages)
现在我们需要在运行应用程序时传入一个配置文件。该配置将指定会话线程的标识符。
我们以同样的方式启动一个线程，这一次传入了配置：
user_message = {"role": "user", "content": "What's the weather in san francisco?"}
print(user_message)
for step in agent.stream([user_message], config):
for task_name, message in step.items():
if task_name == "agent":
continue # Just print task updates
print(f"\n{task_name}:")
message.pretty_print()
{'role': 'user', 'content': "What's the weather in san francisco?"}
call_model:
================================== Ai Message ==================================
Tool Calls:
get_weather (call_lubbUSdDofmOhFunPEZLBz3g)
Call ID: call_lubbUSdDofmOhFunPEZLBz3g
Args:
location: San Francisco
call_tool:
================================= Tool Message =================================
It's sunny!
call_model:
================================== Ai Message ==================================
The weather in San Francisco is sunny!
user_message = {"role": "user", "content": "How does it compare to Boston, MA?"}
print(user_message)
for step in agent.stream([user_message], config):
for task_name, message in step.items():
if task_name == "agent":
continue # Just print task updates
print(f"\n{task_name}:")
message.pretty_print()
{'role': 'user', 'content': 'How does it compare to Boston, MA?'}
call_model:
================================== Ai Message ==================================
Tool Calls:
get_weather (call_8sTKYAhSIHOdjLD5d6gaswuV)
Call ID: call_8sTKYAhSIHOdjLD5d6gaswuV
Args:
location: Boston, MA
call_tool:
================================= Tool Message =================================
It's rainy!
call_model:
================================== Ai Message ==================================
Compared to San Francisco, which is sunny, Boston, MA is experiencing rainy weather.