跳过内容

如何从零开始创建 ReAct Agent(函数式 API)

先决条件

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

本指南演示了如何使用 LangGraph 的 函数式 API 实现一个 ReAct Agent。

ReAct Agent 是一种 工具调用型 Agent,其工作方式如下

  1. 向聊天模型发出查询;
  2. 如果模型没有生成 工具调用,则返回模型的响应。
  3. 如果模型生成了工具调用,我们将使用可用工具执行这些工具调用,将它们作为 工具消息 添加到我们的消息列表中,然后重复此过程。

这是一个简单且多功能的设置,可以扩展记忆、人机协作能力和其他功能。有关示例,请参阅专门的操作指南

设置

首先,让我们安装所需的软件包并设置我们的 API 密钥

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 Agent

安装完所需的软件包并设置好环境变量后,我们就可以创建 Agent 了。

定义模型和工具

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

本示例将使用 OpenAI 聊天模型,但任何支持工具调用的模型都可以使用。

API 参考:ChatOpenAI | 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]

定义任务

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

  1. 调用模型:我们希望使用消息列表查询我们的聊天模型。
  2. 调用工具:如果模型生成了工具调用,我们就想执行它们。

API 参考: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 任务将为每个调用生成响应。我们将所有消息添加到单个消息列表中。

提示

请注意,由于任务返回类似 Future 的对象,以下实现会并行执行工具。

API 参考: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

使用方法

要使用我们的 Agent,我们通过消息列表来调用它。根据我们的实现,这些可以是 LangChain 的 message 对象或 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 跟踪记录。

添加线程级持久化

添加线程级持久化使我们能够支持 Agent 的对话体验:后续调用将附加到先前的消息列表,保留完整的对话上下文。

要为我们的 Agent 添加线程级持久化

  1. 选择一个检查点:这里我们将使用 MemorySaver,一个简单的内存检查点。
  2. 更新我们的入口点,使其接受先前的消息状态作为第二个参数。这里,我们只需将消息更新附加到先前的消息序列中。
  3. 使用 entrypoint.final 选择哪些值将从工作流中返回,以及哪些值将由检查点保存为 previous(可选)

API 参考: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)

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

提示

请在我们关于持久化概念页面操作指南中了解更多关于线程级持久化的信息。

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:
================================== 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.
LangSmith 跟踪记录中,我们可以看到完整的对话上下文保留在每次模型调用中。

评论