ReAct 框架深度解析：AI Agent 的「思考-行动-观察」循环，从字符串解析到原生 Tool Calling 的演进
引言：为什么 AI Agent 需要「思考」？
如果你用过 ChatGPT 或 Claude，一定遇到过这样的场景：
用户：帮我查一下北京今天的天气，然后推荐几个适合户外活动的地方。
AI：北京今天晴，气温18-25℃。推荐你去颐和园、北海公园、奥林匹克森林公园。
用户：等等，你是怎么知道北京天气的？你推荐的地方为什么适合今天？
问题来了：AI 是怎么「知道」天气的？它真的「思考」过吗？
传统 LLM（大语言模型）有两个致命缺陷：
- 幻觉问题：模型可能「编造」答案，因为它没有真实的信息来源
- 被动应答：模型只能根据训练数据回答，无法主动获取最新信息
ReAct（Reason + Act）框架就是为了解决这两个问题而生的。
┌─────────────────────────────────────────────────┐
│ AI Agent 的演进 │
│ │
│ v1.0: 纯 LLM（GPT-3） │
│ 只能回答训练数据中的内容 │
│ ↓ │
│ v2.0: RAG（检索增强生成） │
│ 可以检索外部知识库 │
│ ↓ │
│ v3.0: Tool Use（工具调用） │
│ 可以调用搜索引擎、计算器等工具 │
│ ↓ │
│ v4.0: ReAct Agent ← 我们现在 │
│ 推理 + 行动 + 观察，自主解决问题 │
│ ↓ │
│ v5.0: Multi-Agent（多智能体协作） │
│ 多个 Agent 协同完成复杂任务 │
└─────────────────────────────────────────────────┘
ReAct 的核心创新：让 AI 学会「思考」再「行动」
- 在采取行动之前，先进行推理（Reason）
- 执行行动后，观察结果并更新认知（Observe）
- 循环迭代，直到完成任务
本文将从原理、演进、代码实战三个维度，深度解析 ReAct 框架的技术实现。
第一章：ReAct 的核心原理——TAO 循环
1.1 什么是 TAO 循环？
TAO = Thought（思考）+ Action（行动）+ Observation（观察）
这是 ReAct 框架的核心执行循环：
┌──────────────────────────────────────────────────────────┐
│ ReAct TAO 循环 │
│ │
│ 用户输入："北京今天天气怎么样？" │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Thought（思考） │ │
│ │ "用户想查北京天气，我需要先获取 │ │
│ │ 城市的地理位置ID，再调用天气API" │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Action（行动） │ │
│ │ 调用 geocode 工具 │ │
│ │ {"query": "北京"} │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Observation（观察） │ │
│ │ 工具返回： │ │
│ │ {"id": "CN101010100", │ │
│ │ "name": "Beijing"} │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Thought（思考） │ │
│ │ "已获取城市ID，下一步调用天气API" │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Action（行动） │ │
│ │ 调用 weather_api 工具 │ │
│ │ {"city_id": "CN101010100"} │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Observation（观察） │ │
│ │ 工具返回： │ │
│ │ {"weather": "晴", │ │
│ │ "temp": "18-25℃"} │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ Final Answer（最终答案） │ │
│ │ "北京今天晴，气温18-25℃" │ │
│ └─────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────┘
1.2 为什么需要「思考」环节？
你可能会问：为什么不直接让 LLM 调用工具，而是要先「思考」？
答案在于复杂问题的分解。
# 场景：用户问一个复杂问题
user_input = "帮我查一下马斯克最近的推特，然后分析一下他对 AI 的态度变化趋势"
# 如果没有「思考」环节，模型可能会：
# 1. 直接调用 search_twitter("马斯克 AI 态度") → 结果不准确
# 2. 直接分析 → 没有真实数据支撑，产生幻觉
# 有了「思考」环节，模型会这样分解：
"""
Thought 1: 用户想分析马斯克的 AI 态度变化，首先需要获取他最近的推特
Action 1: 调用 search_twitter 工具，查询 "elonmusk AI"
Observation 1: 返回最近 100 条推特...
Thought 2: 获取到了推特数据，但需要筛选与 AI 相关的内容
Action 2: 调用 filter_tweets 工具，筛选关键词 ["AI", "artificial intelligence", "GPT", "AGI"]
Observation 2: 筛选出 30 条相关推特...
Thought 3: 需要对这些推特进行情感分析和趋势识别
Action 3: 调用 sentiment_analysis 工具
Observation 3: 分析结果显示...
Final Answer: 根据分析，马斯克对 AI 的态度呈现以下变化趋势...
"""
「思考」环节的价值：
| 维度 | 无 Thought | 有 Thought |
|---|---|---|
| 问题分解 | 一次性调用，容易遗漏 | 逐步分解，逻辑清晰 |
| 可调试性 | 黑盒，难以追踪错误 | 每一步都有推理记录 |
| 错误恢复 | 失败后无法恢复 | 可以在 Observation 中发现错误并修正 |
| 可解释性 | 无法解释为什么这样做 | 每个行动都有理由支撑 |
1.3 TAO 循环的数学表示
ReAct 的执行过程可以形式化为：
S_0 → (T_1, A_1, O_1) → S_1 → (T_2, A_2, O_2) → S_2 → ... → S_n → Final_Answer
其中：
- S_i：第 i 步的状态（上下文）
- T_i：第 i 步的思考（Thought）
- A_i：第 i 步的行动（Action）
- O_i：第 i 步的观察（Observation）
状态转移：
S_i = S_{i-1} + T_i + A_i + O_i
终止条件：
- 模型判断已收集足够信息 → 输出 Final_Answer
- 达到最大步数限制 → 强制终止
- 遇到不可恢复错误 → 报错退出
第二章：ReAct 的演进——从字符串解析到原生 Tool Calling
2.1 第一代 ReAct：字符串解析（2022-2023）
最早的 ReAct 实现，采用的是字符串模板 + 正则解析的方式：
# 第一代 ReAct：字符串模板
REACT_PROMPT = """You have access to the following tools:
{tool_descriptions}
Use the following format:
Thought: [your reasoning about what to do next]
Action: [tool_name]
Action Input: [JSON arguments for the tool]
Observation: [tool result - this will be provided by the system]
Begin!
Question: {question}
"""
# 示例输出
llm_output = """
Thought: 用户想查北京天气，我需要先获取城市ID
Action: geocode
Action Input: {"query": "北京"}
"""
# 正则解析
import re
import json
def parse_action(llm_output):
# 提取 Action
action_match = re.search(r"Action:\s*(\w+)", llm_output)
action = action_match.group(1) if action_match else None
# 提取 Action Input
input_match = re.search(r"Action Input:\s*(\{.*?\})", llm_output, re.DOTALL)
action_input = json.loads(input_match.group(1)) if input_match else {}
return action, action_input
# 执行工具
action, action_input = parse_action(llm_output)
# action = "geocode"
# action_input = {"query": "北京"}
# 问题：
# 1. 正则表达式脆弱，容易解析失败
# 2. JSON 格式不规范会导致错误
# 3. 无法处理嵌套参数
# 4. 难以做多工具并行调用
第一代 ReAct 的痛点：
# 痛点 1：格式不规范导致解析失败
llm_output = """
Thought: 我需要调用天气API
Action: weather_api
Action Input: {"city": "北京", "date": "今天"} # 多了一个逗号
"""
# JSON 解析失败：Expecting property name enclosed in double quotes
# 痛点 2：无法处理复杂的嵌套参数
action_input = {
"filters": {
"location": {"city": "北京"},
"date_range": {"start": "2024-01-01", "end": "2024-12-31"}
}
}
# JSON 序列化后再解析，容易出错
# 痛点 3：无法做多工具并行调用
# 如果需要同时调用 3 个工具，只能串行执行
2.2 第二代 ReAct：原生 Tool Calling（2024-2026）
2024 年，OpenAI、Anthropic、Google 等厂商推出了原生 Tool Calling：
# 第二代 ReAct：原生 Tool Calling
from anthropic import Anthropic
client = Anthropic()
# 定义工具
tools = [
{
"name": "search_web",
"description": "搜索网络获取最新信息",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
},
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"input_schema": {
"type": "object",
"properties": {
"city_id": {
"type": "string",
"description": "城市ID"
}
},
"required": ["city_id"]
}
}
]
# 调用 LLM
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=tools,
messages=[
{"role": "user", "content": "北京今天天气怎么样？"}
]
)
# 响应结构化输出
# 不再是字符串，而是结构化的 tool_use block
print(response.content)
# [
# TextBlock(text="我来帮你查一下北京的天气。"),
# ToolUseBlock(
# id="toolu_01A",
# name="get_weather",
# input={"city_id": "CN101010100"}
# )
# ]
# 优势：
# 1. 结构化输出，不需要正则解析
# 2. JSON Schema 校验由 LLM 提供方负责
# 3. 支持复杂嵌套参数
# 4. 支持多工具并行调用
2.3 两代 ReAct 的对比
| 维度 | 第一代（字符串解析） | 第二代（Tool Calling） |
|---|---|---|
| 输出格式 | 纯文本 + 正则解析 | 结构化 JSON |
| 校验方式 | 开发者手动校验 | LLM 提供方自动校验 |
| 参数复杂度 | 简单参数 | 复杂嵌套参数 |
| 并行调用 | 不支持 | 支持 |
| 错误处理 | 解析失败直接报错 | 返回错误信息，LLM 可以修正 |
| 可靠性 | 低（依赖正则） | 高（Schema 保证） |
| 可调试性 | 差（字符串匹配） | 好（结构化输出） |
# 第二代 ReAct 的错误处理示例
response = client.messages.create(
model="claude-sonnet-4-20250514",
tools=tools,
messages=[
{"role": "user", "content": "帮我分析一下这个网站的流量趋势"},
{"role": "assistant", "content": [
ToolUseBlock(id="toolu_01", name="get_analytics", input={"url": "example.com"})
]},
{"role": "user", "content": [
ToolResultBlock(
tool_use_id="toolu_01",
content="Error: 需要提供 API Key 才能访问分析数据"
)
]}
]
)
# LLM 会自动修正：
# "抱歉，我需要 API Key 才能访问分析数据。请提供您的 Google Analytics API Key。"
第三章：现代 ReAct 架构——Function Calling + LangGraph
3.1 Function Calling 的底层原理
Function Calling = 结构化的工具调用协议
┌──────────────────────────────────────────────────────────┐
│ Function Calling 执行流程 │
│ │
│ 1. 用户请求 + 工具描述 │
│ ↓ │
│ 2. LLM 判断是否需要调用工具 │
│ ↓ │
│ 3. 生成结构化的 tool_use block │
│ { │
│ "name": "search_web", │
│ "arguments": {"query": "React 20 new features"} │
│ } │
│ ↓ │
│ 4. 运行时执行工具 │
│ ↓ │
│ 5. 将结果作为 ToolMessage 返回 │
│ ↓ │
│ 6. LLM 根据结果决定下一步 │
│ - 继续调用工具（回到步骤 2） │
│ - 生成最终回答 │
│ │
└──────────────────────────────────────────────────────────┘
# Function Calling 的核心代码
from anthropic import Anthropic
client = Anthropic()
def run_react_agent(user_message: str, tools: list, max_iterations: int = 10):
"""
执行 ReAct Agent
"""
messages = [{"role": "user", "content": user_message}]
for i in range(max_iterations):
# 1. 调用 LLM
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=tools,
messages=messages
)
# 2. 检查是否有 tool_use
tool_uses = [block for block in response.content if block.type == "tool_use"]
if not tool_uses:
# 没有工具调用，返回最终答案
return "".join(block.text for block in response.content if hasattr(block, "text"))
# 3. 执行工具
tool_results = []
for tool_use in tool_uses:
result = execute_tool(tool_use.name, tool_use.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(result)
})
# 4. 将工具结果加入消息历史
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
return "达到最大迭代次数，强制终止"
def execute_tool(name: str, arguments: dict):
"""
执行工具（示例）
"""
if name == "search_web":
return f"搜索结果：{arguments['query']} 的相关信息..."
elif name == "get_weather":
return f"天气：{arguments['city_id']} 晴，18-25℃"
else:
return f"未知工具：{name}"
3.2 LangGraph 状态机架构
LangGraph = 基于状态图的 ReAct 实现
┌──────────────────────────────────────────────────────────┐
│ LangGraph 状态机 │
│ │
│ ┌──────────┐ │
│ │ START │ │
│ └────┬─────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ reason │ ← Thought 环节 │
│ │ (推理节点) │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ should_act │ │
│ │ (条件判断) │ │
│ └───┬──────┬───┘ │
│ │ │ │
│ 需要工具调用 │ │ 已有足够信息 │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────┐ ┌──────────┐ │
│ │ act │ │ respond │ │
│ │(执行工具)│ │(生成回答)│ │
│ └────┬────┘ └────┬─────┘ │
│ │ │ │
│ ▼ │ │
│ ┌─────────┐ │ │
│ │ observe │ │ │
│ │(观察结果)│ │ │
│ └────┬────┘ │ │
│ │ │ │
│ └─────┬──────┘ │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ END │ │
│ └──────────┘ │
│ │
└──────────────────────────────────────────────────────────┘
# LangGraph 实现 ReAct
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
thoughts: list
actions: list
observations: list
# 定义节点
def reason_node(state: AgentState) -> AgentState:
"""推理节点：分析当前状态，决定下一步行动"""
response = client.messages.create(
model="claude-sonnet-4-20250514",
tools=tools,
messages=state["messages"]
)
# 提取思考和工具调用
thoughts = [block.text for block in response.content if hasattr(block, "text")]
tool_uses = [block for block in response.content if block.type == "tool_use"]
return {
"messages": [{"role": "assistant", "content": response.content}],
"thoughts": thoughts,
"actions": tool_uses
}
def act_node(state: AgentState) -> AgentState:
"""行动节点：执行工具"""
tool_results = []
observations = []
for tool_use in state["actions"]:
result = execute_tool(tool_use.name, tool_use.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(result)
})
observations.append({
"tool": tool_use.name,
"input": tool_use.input,
"output": result
})
return {
"messages": [{"role": "user", "content": tool_results}],
"observations": observations
}
def should_act(state: AgentState) -> str:
"""条件判断：是否需要继续执行工具"""
if not state["actions"]:
return "respond"
return "act"
# 构建状态图
workflow = StateGraph(AgentState)
workflow.add_node("reason", reason_node)
workflow.add_node("act", act_node)
workflow.add_node("respond", lambda state: {"messages": state["messages"]})
workflow.add_edge("reason", "act", condition=lambda s: should_act(s) == "act")
workflow.add_edge("reason", "respond", condition=lambda s: should_act(s) == "respond")
workflow.add_edge("act", "reason") # 观察后回到推理
workflow.add_edge("respond", END)
# 编译并运行
app = workflow.compile()
result = app.invoke({"messages": [{"role": "user", "content": "北京天气怎么样？"}]})
3.3 LangGraph vs 传统 ReAct 循环
| 维度 | 传统循环 | LangGraph |
|---|---|---|
| 状态管理 | 隐式（通过消息列表） | 显式（TypedDict 状态） |
| 流程控制 | while 循环 | 状态图 |
| 可观测性 | 难以追踪 | 每个节点都有状态快照 |
| 错误处理 | try-catch | 条件边 + 错误节点 |
| 并行执行 | 难以实现 | 原生支持 |
| 调试 | print 日志 | LangSmith 追踪 |
| 持久化 | 手动实现 | 内置 checkpointer |
第四章：ReAct 与其他 Agent 模式对比
4.1 ReAct vs Chain-of-Thought（CoT）
Chain-of-Thought = 思维链，只思考不行动
# Chain-of-Thought 示例
"""
问题：小明有 5 个苹果，给了小红 2 个，又买了 3 个，现在有几个？
思考过程：
1. 小明一开始有 5 个苹果
2. 给了小红 2 个，剩下 5 - 2 = 3 个
3. 又买了 3 个，现在有 3 + 3 = 6 个
答案：6 个
"""
ReAct = 思考 + 行动
# ReAct 示例
"""
问题：北京今天天气怎么样？
Thought: 用户想查北京天气，我需要调用天气API
Action: get_weather
Action Input: {"city": "北京"}
Observation: {"weather": "晴", "temp": "18-25℃"}
Thought: 已获取天气信息，可以回答用户了
Final Answer: 北京今天晴，气温18-25℃
"""
| 维度 | Chain-of-Thought | ReAct |
|---|---|---|
| 核心能力 | 推理 | 推理 + 行动 |
| 外部信息 | 不支持 | 支持调用工具 |
| 适用场景 | 数学、逻辑推理 | 信息检索、任务执行 |
| 可执行性 | 纯文本输出 | 可执行实际操作 |
4.2 ReAct vs Plan-and-Execute
Plan-and-Execute = 先规划，再执行
# Plan-and-Execute 示例
"""
用户：帮我策划一次北京三日游
Plan:
1. 查询北京的著名景点
2. 根据景点位置规划路线
3. 查询酒店信息
4. 预订酒店和门票
Execute:
Step 1: 搜索北京景点... → 结果：故宫、长城、颐和园...
Step 2: 规划路线... → 结果：Day1 故宫 → Day2 长城 → Day3 颐和园
Step 3: 查询酒店... → 结果：推荐酒店列表
Step 4: 预订... → 结果：预订成功
"""
ReAct = 边思考边行动
# ReAct 示例
"""
用户：帮我策划一次北京三日游
Thought: 用户想要北京三日游，我先查询著名景点
Action: search_attractions
Action Input: {"city": "北京"}
Observation: 故宫、长城、颐和园、天坛...
Thought: 景点很多，我需要根据位置和评分筛选
Action: filter_attractions
Action Input: {"criteria": {"rating": "> 4.5", "location": "市中心"}}
Observation: 故宫、天坛...
Thought: 用户可能对历史感兴趣，我推荐故宫和天坛
Final Answer: 推荐三日游路线：Day1 故宫 → Day2 天坛 → Day3 颐和园
"""
| 维度 | Plan-and-Execute | ReAct |
|---|---|---|
| 规划方式 | 一次性规划所有步骤 | 边执行边调整 |
| 灵活性 | 低（按计划执行） | 高（可中途修正） |
| 适用场景 | 任务明确、步骤固定 | 任务开放、需要探索 |
| 错误恢复 | 需要重新规划 | 可即时修正 |
4.3 选择指南
┌──────────────────────────────────────────────────────────┐
│ Agent 模式选择决策树 │
│ │
│ 任务是否需要外部信息？ │
│ ├─ 否 → Chain-of-Thought │
│ │ （数学推理、逻辑判断） │
│ │ │
│ └─ 是 → 任务是否可以预先规划所有步骤？ │
│ ├─ 是 → Plan-and-Execute │
│ │ （固定流程、明确任务） │
│ │ │
│ └─ 否 → ReAct │
│ （开放任务、需要探索） │
│ │
└──────────────────────────────────────────────────────────┘
第五章：代码实战——从零构建 ReAct Agent
5.1 基础版 ReAct Agent
"""
基础版 ReAct Agent
使用 Anthropic Claude API 实现
"""
from anthropic import Anthropic
from typing import Callable, Any
import json
class ReActAgent:
def __init__(self, model: str = "claude-sonnet-4-20250514"):
self.client = Anthropic()
self.model = model
self.tools = {}
self.tool_definitions = []
def register_tool(self, name: str, description: str,
input_schema: dict, func: Callable):
"""注册工具"""
self.tools[name] = func
self.tool_definitions.append({
"name": name,
"description": description,
"input_schema": input_schema
})
def run(self, user_message: str, max_iterations: int = 10) -> str:
"""执行 ReAct 循环"""
messages = [{"role": "user", "content": user_message}]
for iteration in range(max_iterations):
print(f"\n=== 迭代 {iteration + 1} ===")
# 1. 调用 LLM
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
tools=self.tool_definitions,
messages=messages
)
# 2. 提取思考和工具调用
thoughts = []
tool_uses = []
for block in response.content:
if hasattr(block, "text"):
thoughts.append(block.text)
print(f"Thought: {block.text[:100]}...")
elif block.type == "tool_use":
tool_uses.append(block)
print(f"Action: {block.name}")
print(f"Action Input: {json.dumps(block.input, ensure_ascii=False)}")
# 3. 如果没有工具调用，返回最终答案
if not tool_uses:
return "".join(thoughts)
# 4. 执行工具
tool_results = []
for tool_use in tool_uses:
result = self._execute_tool(tool_use.name, tool_use.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": json.dumps(result, ensure_ascii=False)
})
print(f"Observation: {str(result)[:100]}...")
# 5. 更新消息历史
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
return "达到最大迭代次数，强制终止"
def _execute_tool(self, name: str, arguments: dict) -> Any:
"""执行工具"""
if name not in self.tools:
return {"error": f"未知工具: {name}"}
try:
return self.tools[name](**arguments)
except Exception as e:
return {"error": str(e)}
# 使用示例
agent = ReActAgent()
# 注册工具
agent.register_tool(
name="search_web",
description="搜索网络获取最新信息",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
},
func=lambda query: f"搜索结果：关于 '{query}' 的最新信息..."
)
agent.register_tool(
name="get_weather",
description="获取指定城市的天气信息",
input_schema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
},
func=lambda city: {"weather": "晴", "temp": "18-25℃", "city": city}
)
# 运行
result = agent.run("北京今天天气怎么样？适合户外运动吗？")
print(f"\n最终答案: {result}")
5.2 进阶版：带状态持久化的 ReAct Agent
"""
进阶版 ReAct Agent
支持状态持久化、错误恢复、并行工具调用
"""
from anthropic import Anthropic
from typing import TypedDict, Annotated, Callable, Any
import operator
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor
class AgentState(TypedDict):
"""Agent 状态"""
messages: Annotated[list, operator.add]
thoughts: list
actions: list
observations: list
errors: list
class AdvancedReActAgent:
def __init__(self, model: str = "claude-sonnet-4-20250514"):
self.client = Anthropic()
self.model = model
self.tools = {}
self.tool_definitions = []
self.state: AgentState = {
"messages": [],
"thoughts": [],
"actions": [],
"observations": [],
"errors": []
}
def register_tool(self, name: str, description: str,
input_schema: dict, func: Callable,
timeout: float = 30.0):
"""注册工具（支持超时设置）"""
self.tools[name] = {"func": func, "timeout": timeout}
self.tool_definitions.append({
"name": name,
"description": description,
"input_schema": input_schema
})
def run(self, user_message: str, max_iterations: int = 10) -> str:
"""执行 ReAct 循环"""
self.state["messages"] = [{"role": "user", "content": user_message}]
for iteration in range(max_iterations):
print(f"\n=== 迭代 {iteration + 1} ===")
# 1. 调用 LLM
response = self._call_llm()
# 2. 提取思考和工具调用
self._extract_response(response)
# 3. 如果没有工具调用，返回最终答案
if not self.state["actions"]:
return self._get_final_answer(response)
# 4. 并行执行工具
tool_results = self._execute_tools_parallel()
# 5. 更新消息历史
self.state["messages"].append({
"role": "assistant",
"content": response.content
})
self.state["messages"].append({
"role": "user",
"content": tool_results
})
return "达到最大迭代次数，强制终止"
def _call_llm(self):
"""调用 LLM"""
return self.client.messages.create(
model=self.model,
max_tokens=4096,
tools=self.tool_definitions,
messages=self.state["messages"]
)
def _extract_response(self, response):
"""提取思考和工具调用"""
thoughts = []
actions = []
for block in response.content:
if hasattr(block, "text") and block.text:
thoughts.append(block.text)
elif block.type == "tool_use":
actions.append(block)
self.state["thoughts"].extend(thoughts)
self.state["actions"] = actions
return thoughts, actions
def _execute_tools_parallel(self) -> list:
"""并行执行工具"""
tool_results = []
with ThreadPoolExecutor() as executor:
futures = []
for action in self.state["actions"]:
future = executor.submit(
self._execute_single_tool,
action.name,
action.input,
action.id
)
futures.append(future)
for future in futures:
result = future.result()
tool_results.append(result)
self.state["observations"].append(result)
return tool_results
def _execute_single_tool(self, name: str, arguments: dict,
tool_use_id: str) -> dict:
"""执行单个工具（带错误处理）"""
if name not in self.tools:
error_result = {
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": json.dumps({"error": f"未知工具: {name}"}),
"is_error": True
}
self.state["errors"].append(error_result)
return error_result
tool = self.tools[name]
try:
result = tool["func"](**arguments)
return {
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": json.dumps(result, ensure_ascii=False)
}
except Exception as e:
error_result = {
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": json.dumps({"error": str(e)}),
"is_error": True
}
self.state["errors"].append(error_result)
return error_result
def _get_final_answer(self, response) -> str:
"""获取最终答案"""
return "".join(
block.text for block in response.content
if hasattr(block, "text")
)
def save_state(self, filepath: str):
"""保存状态"""
with open(filepath, "w") as f:
json.dump(self.state, f, ensure_ascii=False, indent=2)
def load_state(self, filepath: str):
"""加载状态"""
with open(filepath, "r") as f:
self.state = json.load(f)
# 使用示例
agent = AdvancedReActAgent()
agent.register_tool(
name="search_web",
description="搜索网络",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
},
func=lambda query: {"results": [f"关于 {query} 的信息..."]}
)
agent.register_tool(
name="calculate",
description="执行数学计算",
input_schema={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
func=lambda expression: {"result": eval(expression)}
)
result = agent.run("帮我计算一下 123 * 456，然后搜索这个数字的含义")
print(f"\n最终答案: {result}")
# 保存状态
agent.save_state("/tmp/agent_state.json")
第六章：ReAct 的最佳实践
6.1 工具设计原则
# 原则 1：工具职责单一
# ✅ 好的设计
tools = [
{"name": "search_web", "description": "搜索网络信息"},
{"name": "get_weather", "description": "获取天气信息"},
{"name": "send_email", "description": "发送邮件"}
]
# ❌ 不好的设计
tools = [
{"name": "do_everything", "description": "搜索、查天气、发邮件都行"}
]
# 原则 2：参数描述清晰
# ✅ 好的设计
{
"name": "search_web",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，多个关键词用空格分隔"
},
"num_results": {
"type": "integer",
"description": "返回结果数量，默认 10",
"default": 10
}
},
"required": ["query"]
}
}
# ❌ 不好的设计
{
"name": "search_web",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"} # 缺少描述
}
}
}
# 原则 3：返回结构化数据
# ✅ 好的设计
def search_web(query: str) -> dict:
return {
"success": True,
"results": [
{"title": "...", "url": "...", "snippet": "..."},
...
],
"total": 100
}
# ❌ 不好的设计
def search_web(query: str) -> str:
return "搜索结果：..."
6.2 错误处理策略
# 策略 1：工具执行超时
import signal
from contextlib import contextmanager
@contextmanager
def timeout(seconds: int):
def timeout_handler(signum, frame):
raise TimeoutError("工具执行超时")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
def execute_tool_with_timeout(tool_func, arguments: dict, timeout_seconds: int = 30):
try:
with timeout(timeout_seconds):
return tool_func(**arguments)
except TimeoutError:
return {"error": "工具执行超时，请稍后重试"}
# 策略 2：工具调用重试
def execute_tool_with_retry(tool_func, arguments: dict, max_retries: int = 3):
for attempt in range(max_retries):
try:
return tool_func(**arguments)
except Exception as e:
if attempt == max_retries - 1:
return {"error": f"工具执行失败（重试 {max_retries} 次）: {str(e)}"}
time.sleep(1 * (attempt + 1)) # 指数退避
# 策略 3：降级策略
def execute_tool_with_fallback(primary_tool, fallback_tool, arguments: dict):
try:
result = primary_tool(**arguments)
if result.get("error"):
return fallback_tool(**arguments)
return result
except Exception:
return fallback_tool(**arguments)
6.3 安全防护
# 安全防护 1：工具调用白名单
ALLOWED_TOOLS = ["search_web", "get_weather", "calculate"]
def execute_tool_safely(name: str, arguments: dict):
if name not in ALLOWED_TOOLS:
return {"error": f"工具 {name} 不在白名单中"}
return execute_tool(name, arguments)
# 安全防护 2：输入参数校验
def validate_tool_input(name: str, arguments: dict, schema: dict):
"""使用 JSON Schema 校验输入参数"""
from jsonschema import validate, ValidationError
try:
validate(instance=arguments, schema=schema)
return True, None
except ValidationError as e:
return False, str(e)
# 安全防护 3：敏感信息过滤
SENSITIVE_PATTERNS = [
r"api[_-]?key",
r"password",
r"secret",
r"token"
]
def filter_sensitive_info(text: str) -> str:
"""过滤敏感信息"""
import re
for pattern in SENSITIVE_PATTERNS:
text = re.sub(pattern, "[REDACTED]", text, flags=re.IGNORECASE)
return text
第七章：ReAct 的真实应用场景
7.1 智能客服机器人
"""
场景：电商智能客服
"""
customer_service_agent = ReActAgent()
customer_service_agent.register_tool(
name="query_order",
description="查询订单状态",
input_schema={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "订单号"}
},
"required": ["order_id"]
},
func=lambda order_id: {"status": "已发货", "tracking": "SF123456"}
)
customer_service_agent.register_tool(
name="query_product",
description="查询商品信息",
input_schema={
"type": "object",
"properties": {
"product_id": {"type": "string", "description": "商品ID"}
},
"required": ["product_id"]
},
func=lambda product_id: {"name": "商品A", "price": 99.9, "stock": 100}
)
customer_service_agent.register_tool(
name="apply_refund",
description="申请退款",
input_schema={
"type": "object",
"properties": {
"order_id": {"type": "string"},
"reason": {"type": "string"}
},
"required": ["order_id", "reason"]
},
func=lambda order_id, reason: {"success": True, "refund_id": "R123"}
)
# 使用
result = customer_service_agent.run(
"我的订单 12345 怎么还没到？如果发货了帮我查一下物流，如果没发货就申请退款"
)
7.2 数据分析助手
"""
场景：数据分析助手
"""
data_agent = ReActAgent()
data_agent.register_tool(
name="execute_sql",
description="执行 SQL 查询",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "SQL 查询语句"}
},
"required": ["query"]
},
func=lambda query: {"rows": [...], "count": 100}
)
data_agent.register_tool(
name="generate_chart",
description="生成图表",
input_schema={
"type": "object",
"properties": {
"chart_type": {"type": "string", "enum": ["bar", "line", "pie"]},
"data": {"type": "array"}
},
"required": ["chart_type", "data"]
},
func=lambda chart_type, data: {"chart_url": "https://..."}
)
data_agent.register_tool(
name="export_report",
description="导出报告",
input_schema={
"type": "object",
"properties": {
"format": {"type": "string", "enum": ["pdf", "excel", "csv"]}
},
"required": ["format"]
},
func=lambda format: {"file_url": "https://..."}
)
# 使用
result = data_agent.run(
"帮我分析一下最近 7 天的销售数据，生成一个趋势图，然后导出 Excel 报告"
)
7.3 代码调试助手
"""
场景：代码调试助手
"""
debug_agent = ReActAgent()
debug_agent.register_tool(
name="read_file",
description="读取文件内容",
input_schema={
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"}
},
"required": ["path"]
},
func=lambda path: {"content": "..."}
)
debug_agent.register_tool(
name="search_code",
description="搜索代码",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
},
func=lambda query: {"matches": [...]}
)
debug_agent.register_tool(
name="run_test",
description="运行测试",
input_schema={
"type": "object",
"properties": {
"test_file": {"type": "string"}
},
"required": ["test_file"]
},
func=lambda test_file: {"passed": 10, "failed": 2, "errors": [...]}
)
# 使用
result = debug_agent.run(
"这个测试失败了，帮我分析一下原因：AssertionError: expected 5 but got 3"
)
第八章：ReAct 的局限性与未来方向
8.1 当前局限性
# 局限性 1：迭代次数限制
# ReAct 需要多次 LLM 调用，成本高、延迟大
# 解决方案：限制最大迭代次数，提前终止
# 局限性 2：上下文窗口限制
# 每次迭代的 Thought/Action/Observation 都会累积到上下文中
# 长时间运行后可能超出 token 限制
# 解决方案：使用滑动窗口或压缩历史记录
# 局限性 3：工具调用失败
# 工具可能因为网络、权限等原因失败
# 解决方案：重试机制、降级策略
# 局限性 4：推理错误累积
# 如果第一步推理错误，后续步骤都会受影响
# 解决方案：引入「反思」环节，定期检查推理是否正确
8.2 未来方向
ReAct 的未来演进：
1. 反思型 Agent（Reflective Agent）
- 在 TAO 循环中加入「反思」环节
- Thought → Action → Observation → Reflection → Thought...
2. 多智能体协作（Multi-Agent）
- 多个 ReAct Agent 协同工作
- 一个 Agent 规划，另一个 Agent 执行，第三个 Agent 审核
3. 自适应工具选择（Adaptive Tool Selection）
- 根据任务类型自动选择最合适的工具
- 减少无效的工具调用
4. 流式输出（Streaming Output）
- 实时展示思考和执行过程
- 提升用户体验
5. 持久化记忆（Persistent Memory）
- 保存 Agent 的状态和学习成果
- 避免重复学习
总结：ReAct 是 AI Agent 的「操作系统」
如果把 AI Agent 比作一台计算机，那么：
┌──────────────────────────────────────────────────────────┐
│ AI Agent 架构类比 │
│ │
│ LLM（大语言模型） ←→ CPU（处理器） │
│ Tools（工具） ←→ 外设（键盘、鼠标、显示器） │
│ Memory（记忆） ←→ 硬盘（存储） │
│ ReAct 框架 ←→ 操作系统（调度中心） │
│ │
│ 没有操作系统，计算机只是一堆硬件； │
│ 没有 ReAct，LLM 只是一个聊天机器人。 │
│ │
│ ReAct 让 LLM 从「被动应答」进化为「主动执行」。 │
│ │
└──────────────────────────────────────────────────────────┘
ReAct 的核心价值：
推理 + 行动 = 智能
- 只会推理的是「聊天机器人」
- 只会行动的是「自动化脚本」
- 既会推理又会行动的，才是「智能体」
可解释 + 可调试 = 可靠
- 每一步都有 Thought 记录
- 出错时可以追溯到具体环节
- 便于人类理解和干预
通用 + 灵活 = 实用
- 不依赖特定领域知识
- 可以动态注册新工具
- 适用于各种任务场景
适用场景推荐：
- ✅ 智能客服（多步骤问题解决）
- ✅ 数据分析（查询 + 可视化 + 报告）
- ✅ 代码助手（调试 + 重构 + 测试）
- ✅ 自动化工作流（跨系统集成）
- ❌ 简单问答（用普通 LLM 即可）
- ❌ 固定流程（用传统自动化即可）
参考资源
- ReAct 论文：ReAct: Synergizing Reasoning and Acting in Language Models (Yao et al., 2022)
- LangGraph 文档：https://langchain-ai.github.io/langgraph/
- Anthropic Tool Use 文档：https://docs.anthropic.com/claude/docs/tool-use
- OpenAI Function Calling 文档：https://platform.openai.com/docs/guides/function-calling
文章字数统计：约 18,500 字
完