4.构建 ReAct 代理

ReAct 代理是 LangGraph 中最常用的代理模式之一。这种架构让语言模型能够像人类一样思考——观察环境、推理决策、执行动作，然后根据结果调整策略。LangGraph 提供了预构建的 ReAct 代理实现，让我们不必从零开始搭建这个循环。

预构建 ReAct 代理初始化

LangGraph 的 create_react_agent 函数封装了完整的 ReAct 循环逻辑。只需要提供三个核心组件：语言模型、工具列表和系统提示，就能得到一个功能完整的代理。

from langgraph.prebuilt import create_react_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}的天气总是阳光明媚！"

# 创建代理实例
agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
    prompt="你是一个乐于助人的助手"
)

# 运行代理
result = agent.invoke(
    {"messages": [{"role": "user", "content": "旧金山天气怎么样"}]}
)

这个简单的例子展示了 ReAct 代理的基本形态。create_react_agent 背后已经实现了完整的循环逻辑：模型生成思考、决定是否调用工具、执行工具调用、观察结果、继续推理，直到给出最终答案。

模型参数支持多种配置方式。可以直接传入模型名称字符串，LangGraph 会通过 init_chat_model 自动初始化对应的聊天模型。这种方式适合快速原型开发。对于生产环境，建议显式创建模型实例，以便精细控制温度、最大令牌数等参数。

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_react_agent

# 显式配置模型参数
model = init_chat_model(
    "anthropic:claude-3-7-sonnet-latest",
    temperature=0,  # 降低随机性
    max_tokens=2048
)

agent = create_react_agent(
    model=model,  # 传入模型实例
    tools=[get_weather],
    prompt="你是一个精确的天气查询助手"
)

系统提示的设定直接影响代理的行为模式。静态提示适合固定场景，但很多时候需要根据运行时上下文动态调整。create_react_agent 支持传入一个函数作为提示生成器，这个函数接收当前状态和配置，返回定制化的消息列表。

from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig
from langgraph.prebuilt.chat_agent_executor import AgentState

def dynamic_prompt(state: AgentState, config: RunnableConfig) -> list[AnyMessage]:
    # 从配置中获取用户信息
    user_name = config["configurable"].get("user_name", "访客")
    system_msg = f"你是专属助手，请用友好的语气称呼用户为{user_name}"
    
    # 组合系统消息和历史消息
    return [{"role": "system", "content": system_msg}] + state["messages"]

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
    prompt=dynamic_prompt  # 使用动态提示函数
)

# 调用时传入用户配置
agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    config={"configurable": {"user_name": "张三"}}
)

动态提示函数让我们能够根据用户身份、会话阶段、业务规则等条件灵活调整代理的行为。这种模式在多租户系统或需要个性化服务的场景中特别有价值。

代理工具绑定方法

工具是 ReAct 代理与外部世界交互的桥梁。LangGraph 中的工具可以是普通 Python 函数，也可以是结构化的 Tool 对象。关键在于让语言模型理解工具的用途和参数格式。

最简单的工具定义方式就是普通函数，配合清晰的文档字符串。函数名、参数类型和文档字符串都会被提取为工具的元数据，帮助模型理解何时以及如何使用这个工具。

def search_web(query: str, max_results: int = 5) -> list[dict]:
    """执行网络搜索并返回结果列表。
    
    参数:
        query: 搜索关键词
        max_results: 返回的最大结果数，默认为5
    """
    # 实际实现会调用搜索API
    return [{"title": "示例结果", "url": "https://example.com"}]

def calculator(expression: str) -> float:
    """计算数学表达式的结果。
    
    参数:
        expression: 数学表达式字符串，如"2 + 3 * 4"
    """
    return eval(expression)

# 工具列表可以混合不同类型
tools = [search_web, calculator]

对于更复杂的场景，可能需要使用 LangChain 的 tool 装饰器或函数来创建工具。这种方式提供了更精细的控制，包括自定义模式验证和异步支持。

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，如北京、上海")
    unit: str = Field(default="celsius", description="温度单位，celsius或fahrenheit")

@tool("get_weather_detailed", args_schema=WeatherInput)
def get_weather_detailed(city: str, unit: str = "celsius") -> dict:
    """获取详细的天气信息，包括温度和体感温度。"""
    # 调用天气API
    return {
        "city": city,
        "temperature": 25,
        "unit": unit,
        "feels_like": 26
    }

工具绑定到模型的过程通过 bind_tools 方法完成。这个步骤至关重要，它让模型知晓可用工具的结构和用途。在 ReAct 代理中，这一步通常在代理内部自动处理，但理解其原理有助于调试和定制。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
# 绑定工具到模型
llm_with_tools = llm.bind_tools(tools)

# 现在模型能够生成工具调用
response = llm_with_tools.invoke("计算 15 * 23 是多少")
print(response.tool_calls)  # 输出将包含工具调用信息

在预构建的 ReAct 代理中，工具执行由 ToolNode 处理。这个节点接收模型的工具调用请求，执行对应的函数，并将结果返回给状态。ToolNode 已经处理了错误捕获、结果格式化等细节，让我们专注于工具本身的逻辑。

from langgraph.prebuilt import ToolNode, tools_condition

# 创建工具节点
tool_node = ToolNode(tools=[search_web, calculator])

# tools_condition 是预置的条件函数
# 它检查最新消息是否包含工具调用，决定路由到工具节点还是结束
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
    {"tools": "tool_node", END: END}
)

工具执行的错误处理也很关键。默认情况下，ToolNode 会捕获异常并将错误信息返回给模型，让模型有机会调整策略。可以通过配置自定义错误处理逻辑。

# 自定义错误处理
def handle_tool_error(error: Exception, tool_call: dict) -> str:
    return f"工具执行失败: {str(error)}。请尝试其他方法。"

tool_node = ToolNode(
    tools=[search_web, calculator],
    handle_tool_errors=handle_tool_error
)

代理循环执行逻辑

ReAct 代理的核心是一个循环：模型思考、行动、观察，然后再次思考。LangGraph 通过图结构优雅地实现了这个循环。理解这个执行流程对调试和优化代理行为至关重要。

预构建 ReAct 代理的图结构包含三个关键节点：调用语言模型的节点、执行工具的节点，以及一个条件路由节点。边定义了循环的流向。

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict
from typing import Annotated

class State(TypedDict):
    messages: Annotated[list, add_messages]

# 构建图
graph_builder = StateGraph(State)

# 添加聊天机器人节点
graph_builder.add_node("chatbot", call_llm)

# 添加工具节点
graph_builder.add_node("tools", tool_node)

# 设置条件边：根据模型输出决定下一步
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,  # 如果有工具调用则转到tools，否则结束
    {"tools": "tools", END: END}
)

# 工具执行后总是返回聊天机器人
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile()

tools_condition 是 LangGraph 预置的条件函数，它检查最新消息是否包含工具调用。如果有工具调用，路由到工具节点；否则，流程结束。这个简单的条件构成了 ReAct 循环的分支逻辑。

循环的执行过程可以通过可视化工具清晰展现。LangGraph 提供了 get_graph().draw_mermaid_png() 方法生成流程图，帮助理解代理的决策路径。

# 生成并保存流程图
png_data = graph.get_graph().draw_mermaid_png()
with open("react_agent_flow.png", "wb") as f:
    f.write(png_data)

在循环执行过程中，每次模型调用后，LangGraph 会检查响应内容。如果包含 tool_calls，就将状态路由到 ToolNode。工具执行完成后，结果以 ToolMessage 形式添加到消息列表，然后状态再次回到模型节点，形成闭环。

循环的终止条件由模型自身决定。当模型认为已经收集足够信息，不再生成工具调用时，tools_condition 将路由到 END，整个流程结束。这种设计让模型拥有自主决策权，能够根据任务复杂度灵活调整推理深度。

对于需要限制资源使用的场景，可以通过配置设置最大循环次数。LangGraph 的递归限制机制可以防止代理陷入无限循环。

# 设置递归深度限制
graph = graph_builder.compile(checkpointer=memory, recursion_limit=50)

超过限制时，LangGraph 会抛出 GraphRecursionError 异常，我们可以在应用层捕获并优雅处理。

添加对话记忆能力

没有记忆的代理每次调用都是独立的，无法维持对话上下文。LangGraph 的检查点机制为代理提供了强大的记忆能力，让代理能够跨轮次保持状态。

记忆功能通过 checkpointer 参数启用。LangGraph 提供了多种检查点实现，从内存中的 InMemorySaver 到持久化的 SqliteSaver 或 PostgresSaver。

from langgraph.checkpoint.memory import InMemorySaver

# 创建内存检查点
memory = InMemorySaver()

# 编译图时传入检查点
graph = graph_builder.compile(checkpointer=memory)

启用检查点后，每次调用 graph.invoke 都需要提供 thread_id 作为配置参数。这个标识符将同一会话的多次调用关联起来，实现状态持久化。

# 第一次调用
config = {"configurable": {"thread_id": "user_123"}}
result1 = graph.invoke(
    {"messages": [{"role": "user", "content": "我的名字是张三"}]},
    config=config
)

# 第二次调用，自动携带历史状态
result2 = graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字"}]},
    config=config
)

检查点不仅存储消息历史，还保存了图的执行状态。这意味着即使在中断后，也能从上次离开的地方继续执行。这种能力对于长时间运行的任务或需要人工介入的场景尤为重要。

对于生产环境，建议使用数据库存储检查点。SqliteSaver 提供了简单的文件型数据库支持，而 PostgresSaver 则适合高并发的分布式部署。

from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite 持久化存储
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = graph_builder.compile(checkpointer=checkpointer)
    
    # 后续调用会自动持久化到数据库
    result = graph.invoke(input_data, config={"configurable": {"thread_id": "1"}})

检查点机制还支持时间旅行功能，可以查看和恢复到历史状态。这在调试复杂代理行为时非常有用，能够回溯到特定决策点分析问题。

# 获取状态历史
for state in graph.get_state_history(config):
    print(f"消息数: {len(state.values['messages'])}, 下一步: {state.next}")
    
# 恢复到特定状态
graph.update_state(config, {"messages": [new_message]}, as_node="chatbot")

记忆能力让 ReAct 代理真正具备了会话智能。代理可以引用之前的对话内容，维持长期上下文，提供连贯的交互体验。结合工具使用能力，代理能够完成需要多步骤、跨会话的复杂任务。

在下一章中，我们将深入探讨如何开发自定义工具，包括处理工具错误、动态工具选择和高级工具模式。掌握这些技能后，就能构建出更加强大和可靠的代理系统。