如何从头开始创建一个 ReAct 代理¶
使用预构建的 ReAct 代理 create_react_agent 是一个很好的入门方式，但有时你可能需要更多的控制和定制。在这种情况下，你可以创建一个自定义的 ReAct 代理。本指南将展示如何使用 LangGraph 从头开始实现一个 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 代理¶
既然你已经安装了所需的软件包并设置了环境变量，我们就可以编写 ReAct 代理的代码了！
定义图状态¶
在这个示例中，我们将定义最基本的 ReAct 状态，它仅包含一个消息列表。
对于你的特定用例，请随意添加你需要的任何其他状态键。
from typing import (
Annotated,
Sequence,
TypedDict,
)
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""The state of the agent."""
# add_messages is a reducer
# See https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers
messages: Annotated[Sequence[BaseMessage], add_messages]
API Reference: BaseMessage
定义模型和工具¶
接下来，让我们定义将用于示例的工具和模型。
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
# Don't let the LLM know this though 😊
if any([city in location.lower() for city in ["sf", "san francisco"]]):
return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."
else:
return f"I am not sure what the weather is in {location}"
tools = [get_weather]
model = model.bind_tools(tools)
API Reference: tool
定义节点和边¶
接下来，让我们定义节点和边。在我们的基本 ReAct 代理中，只有两个节点，一个用于调用模型，另一个用于使用工具。不过，你可以修改这个基本结构，使其更适合你的用例。我们在此定义的工具节点是预构建的 ToolNode 的简化版本，后者具有一些额外的功能。
也许你想添加一个用于添加结构化输出的节点，或者一个用于执行某些外部操作（发送电子邮件、添加日历事件等）的节点。也许你只是想改变 call_model 节点的工作方式，以及 should_continue 如何决定是否调用工具 —— 可能性是无穷的，并且 LangGraph 让你可以轻松地针对特定用例自定义这个基本结构。
import json
from langchain_core.messages import ToolMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
tools_by_name = {tool.name: tool for tool in tools}
# Define our tool node
def tool_node(state: AgentState):
outputs = []
for tool_call in state["messages"][-1].tool_calls:
tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
outputs.append(
ToolMessage(
content=json.dumps(tool_result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": outputs}
# Define the node that calls the model
def call_model(
state: AgentState,
config: RunnableConfig,
):
# this is similar to customizing the create_react_agent with 'prompt' parameter, but is more flexible
system_prompt = SystemMessage(
"You are a helpful AI assistant, please respond to the users query to the best of your ability!"
)
response = model.invoke([system_prompt] + state["messages"], config)
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define the conditional edge that determines whether to continue or not
def should_continue(state: AgentState):
messages = state["messages"]
last_message = messages[-1]
# If there is no function call, then we finish
if not last_message.tool_calls:
return "end"
# Otherwise if there is, we continue
else:
return "continue"
API Reference: ToolMessage | SystemMessage | RunnableConfig
定义图¶
既然我们已经定义了所有的节点和边，就可以定义并编译我们的图了。根据你是否添加了更多节点或不同的边，你需要对其进行编辑以适应你特定的用例。
from langgraph.graph import StateGraph, END
# Define a new graph
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
# Finally we pass in a mapping.
# The keys are strings, and the values are other nodes.
# END is a special node marking that the graph should finish.
# What will happen is we will call `should_continue`, and then the output of that
# will be matched against the keys in this mapping.
# Based on which one it matches, that node will then be called.
{
# If `tools`, then we call the tool node.
"continue": "tools",
# Otherwise we finish.
"end": END,
},
)
# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("tools", "agent")
# Now we can compile and visualize our graph
graph = workflow.compile()
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
使用 ReAct 代理¶
既然我们已经创建了 ReAct 代理，那么就让我们实际对它进行测试吧！
# Helper function for formatting the stream nicely
def print_stream(stream):
for s in stream:
message = s["messages"][-1]
if isinstance(message, tuple):
print(message)
else:
message.pretty_print()
inputs = {"messages": [("user", "what is the weather in sf")]}
print_stream(graph.stream(inputs, stream_mode="values"))
================================[1m Human Message [0m=================================
what is the weather in sf
==================================[1m Ai Message [0m==================================
Tool Calls:
get_weather (call_azW0cQ4XjWWj0IAkWAxq9nLB)
Call ID: call_azW0cQ4XjWWj0IAkWAxq9nLB
Args:
location: San Francisco
=================================[1m Tool Message [0m=================================
Name: get_weather
"It's sunny in San Francisco, but you better look out if you're a Gemini \ud83d\ude08."
==================================[1m Ai Message [0m==================================
The weather in San Francisco is sunny! However, it seems there's a playful warning for Geminis. Enjoy the sunshine!
get_weather 工具，并在从该工具获取信息后对用户做出了响应。