如何从零开始创建一个ReAct代理（功能API）¶

先决条件

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

本指南演示了如何使用LangGraph 功能API实现ReAct代理。

ReAct代理是一个工具调用代理，其操作如下：

向聊天模型发出查询；
如果模型生成了没有工具调用的响应，我们返回模型的响应。
如果模型生成了工具调用，我们使用可用工具执行这些调用，将它们作为工具消息附加到我们的消息列表中，并重复此过程。

这是一个简单且多功能的设置，可以扩展为具有记忆功能、人机交互能力和其他功能。请参阅专门的操作指南以获取示例。

设置¶

首先，让我们安装所需的包并设置我们的API密钥：

%%capture --no-stderr
%pip install -U langgraph langchain-openai

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 构建的 LLM 应用程序——更多关于如何开始的信息，请参阅文档。

创建ReAct代理¶

现在你已经安装了所需的包并设置了环境变量，我们可以创建我们的代理了。

定义模型和工具¶

首先，我们定义用于示例的工具和模型。这里我们将使用一个占位工具，该工具会获取某个位置的天气描述。

我们将使用一个OpenAI聊天模型来进行示例，但任何支持工具调用的模型支持工具调用都可以。

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: ChatOpenAI | tool

定义任务¶

接下来，我们定义将要执行的任务。这里有两个不同的任务：

调用模型：我们希望使用消息列表查询聊天模型。
调用工具：如果模型生成工具调用，我们希望执行它们。

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

API Reference: ToolMessage | entrypoint | task

定义入口点¶

我们的入口点将负责这两个任务的编排。如上所述，当call_model任务生成工具调用时，call_tool任务将为每个调用生成响应。我们将所有消息附加到一个单一的消息列表中。

Tip

请注意，由于任务返回类似于未来的对象，因此下面的实现将并行执行工具。

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

API Reference: add_messages

使用方法¶

要使用我们的代理，我们需要用一个消息列表来调用它。根据我们的实现，这些可以是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:
==================================[1m Ai Message [0m==================================
Tool Calls:
  get_weather (call_tNnkrjnoz6MNfCHJpwfuEQ0v)
 Call ID: call_tNnkrjnoz6MNfCHJpwfuEQ0v
  Args:
    location: san francisco

call_tool:
=================================[1m Tool Message [0m=================================

It's sunny!

call_model:
==================================[1m Ai Message [0m==================================

The weather in San Francisco is sunny!

完美！该图正确地调用了get_weather工具，并在接收到来自工具的信息后对用户进行了响应。查看LangSmith跟踪此处。

添加线程级持久化¶

添加线程级持久化让我们能够支持与代理的对话体验：后续的调用将会追加到之前的对话消息列表中，保留完整的对话上下文。

要将线程级持久化添加到我们的代理中：

选择一个检查点器：这里我们将使用MemorySaver，一个简单的内存检查点器。
更新我们的入口点以接受前一个消息状态作为第二个参数。在这里，我们只是将消息更新追加到之前的对话消息序列中。
选择哪些值将从工作流中返回，哪些值将由检查点器保存为previous，使用entrypoint.final（可选）

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)

API Reference: MemorySaver

我们现在需要在运行应用程序时传入一个配置文件。该配置将指定会话线程的标识符。

提示

有关线程级持久性的更多信息，请参阅我们的概念页面和操作指南。

config = {"configurable": {"thread_id": "1"}}

我们以同样的方式启动一个线程，这一次传入了配置：

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:
==================================[1m Ai Message [0m==================================
Tool Calls:
  get_weather (call_lubbUSdDofmOhFunPEZLBz3g)
 Call ID: call_lubbUSdDofmOhFunPEZLBz3g
  Args:
    location: San Francisco

call_tool:
=================================[1m Tool Message [0m=================================

It's sunny!

call_model:
==================================[1m Ai Message [0m==================================

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:
==================================[1m Ai Message [0m==================================
Tool Calls:
  get_weather (call_8sTKYAhSIHOdjLD5d6gaswuV)
 Call ID: call_8sTKYAhSIHOdjLD5d6gaswuV
  Args:
    location: Boston, MA

call_tool:
=================================[1m Tool Message [0m=================================

It's rainy!

call_model:
==================================[1m Ai Message [0m==================================

Compared to San Francisco, which is sunny, Boston, MA is experiencing rainy weather.

在LangSmith跟踪中，我们可以看到每次模型调用时都保留了完整的对话上下文。