第 51 章
ReAct 架构精讲与代码实现
第51章：ReAct 架构精讲与代码实现
导语
ReAct（Reasoning + Acting）是现代 AI Agent 框架的基石。2022年由 Shunyu Yao 等人提出的这篇论文，改变了人们对 LLM 能力边界的认知：通过将"思考"和"行动"交织在一起，LLM 可以解决原本需要专门训练才能应对的复杂推理任务。Hermes Agent 的核心推理引擎正是基于 ReAct 范式。本章从论文原理出发，逐步剖析实现细节，并带你从零用 Python 实现一个完整的 ReAct Agent。
51.1 ReAct 论文核心思想
背景：单纯推理与单纯行动的局限
在 ReAct 出现之前，LLM 有两种主要使用方式：
Chain-of-Thought（CoT）：让 LLM 逐步推理，但所有信息必须在提示词中预先提供，无法访问外部信息。
工具调用（早期 Agent）：让 LLM 直接生成行动，但缺乏中间推理步骤，行动选择缺乏依据，容易陷入错误路径。
ReAct 的关键洞察是：思考和行动应该交替进行，相互增益。
flowchart LR
subgraph CoT["仅 CoT（推理）"]
direction TB
T1[思考1] --> T2[思考2] --> T3[思考3] --> A1[最终答案]
end
subgraph Act["仅行动（早期 Agent）"]
direction TB
AC1[行动1] --> AC2[行动2] --> AC3[行动3] --> A2[答案]
end
subgraph ReAct["ReAct（推理+行动）"]
direction TB
RT1[思考1] --> RA1[行动1] --> RO1[观察1]
RO1 --> RT2[思考2] --> RA2[行动2] --> RO2[观察2]
RO2 --> RT3[思考3] --> A3[最终答案]
end
ReAct 的优势
原论文在 HotpotQA、FEVER、AlfWorld、WebShop 等多个 Benchmark 上验证了 ReAct 的优势：
| 对比维度 | 仅 CoT | 仅行动 | ReAct |
|---|---|---|---|
| 可解释性 | 高 | 低 | 高 |
| 信息获取 | 无（依赖训练数据） | 有 | 有 |
| 错误纠正 | 有限 | 无 | 强（观察反馈驱动） |
| 幻觉风险 | 高 | 中 | 低 |
| 任务成功率 | 基准 | 接近 | 最高 |
51.2 Thought → Action → Observation 循环详解
ReAct 的核心循环由三个阶段构成：
Thought（思考）
LLM 生成内部推理，分析当前情况，制定下一步计划。Thought 不直接产生输出，而是为 Action 提供依据。
Thought: 用户问的是 2024 年诺贝尔物理学奖得主。我知道 2024 年的信息可能
不在我的训练数据中，需要搜索最新信息。我应该搜索"2024 Nobel Prize Physics winner"。
Action（行动）
LLM 生成具体的工具调用指令。Action 格式通常是结构化的（JSON 或特定语法）。
Action: search["2024 Nobel Prize Physics winner"]
Observation（观察）
工具执行结果被注入回上下文，成为 LLM 下一轮推理的输入。
Observation: The 2024 Nobel Prize in Physics was awarded to John Hopfield and
Geoffrey Hinton for foundational discoveries and inventions that enable machine
learning with artificial neural networks.
然后循环继续：LLM 基于 Observation 再次 Thought，决定是继续行动还是给出最终答案。
stateDiagram-v2
[*] --> Thought: 用户输入
Thought --> Action: 需要工具
Thought --> FinalAnswer: 信息充足
Action --> Observation: 工具执行
Observation --> Thought: 继续推理
FinalAnswer --> [*]: 返回答案
note right of Thought: LLM 内部推理
note right of Action: 工具调用
note right of Observation: 外部信息注入
51.3 Hermes 的 ReAct 实现特点
Hermes Agent 在标准 ReAct 基础上做了以下增强：
1. 函数调用格式（Function Calling）
Hermes 使用结构化的 JSON 函数调用，而非原论文的文本格式，更可靠且易于解析：
{
"thought": "需要查询当前比特币价格",
"tool_calls": [
{
"id": "call_001",
"type": "function",
"function": {
"name": "get_price",
"arguments": {"symbol": "BTC", "currency": "USD"}
}
}
]
}
2. 并行工具调用
标准 ReAct 每轮只能调用一个工具。Hermes 支持在同一轮 Action 中并行调用多个工具，显著提升效率。
3. 自定义停止条件
Hermes 支持配置 max_steps、timeout、stop_tokens 等多种停止条件，防止无限循环。
4. 上下文窗口管理
当对话历史超过模型最大上下文长度时，Hermes 自动进行摘要压缩，保持核心信息。
51.4 从零实现 ReAct Agent（完整代码）
# react_agent.py
"""
从零实现的 ReAct Agent
约 120 行核心代码，支持工具调用、循环推理、错误处理
"""
import json
import re
import asyncio
from typing import Any, Callable, Optional
from dataclasses import dataclass, field
from openai import AsyncOpenAI # 兼容 Hermes 的 OpenAI 格式 API
# ─── 数据结构 ──────────────────────────────────────────────────
@dataclass
class Tool:
"""工具定义"""
name: str
description: str
parameters: dict # JSON Schema 格式
func: Callable # 实际执行函数
def to_openai_format(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
@dataclass
class ReActStep:
"""单步 ReAct 执行记录"""
step_number: int
thought: str
action: Optional[dict] = None # 工具调用请求
observation: Optional[str] = None # 工具执行结果
final_answer: Optional[str] = None
def __str__(self) -> str:
parts = [f"=== Step {self.step_number} ==="]
parts.append(f"Thought: {self.thought}")
if self.action:
parts.append(f"Action: {json.dumps(self.action, ensure_ascii=False)}")
if self.observation:
parts.append(f"Observation: {self.observation}")
if self.final_answer:
parts.append(f"Final Answer: {self.final_answer}")
return "\n".join(parts)
@dataclass
class ReActResult:
"""ReAct 执行结果"""
success: bool
answer: str
steps: list[ReActStep]
total_steps: int
error: Optional[str] = None
# ─── ReAct Agent 核心实现 ──────────────────────────────────────
class ReActAgent:
"""
ReAct Agent 核心实现
支持：
- 多工具注册
- 最大步骤限制
- 工具调用错误处理
- 详细执行日志
"""
SYSTEM_PROMPT = """你是一个 ReAct Agent，通过交替进行"思考"和"行动"来解决问题。
工作方式：
1. 分析问题，思考需要哪些信息或操作
2. 调用工具获取信息或执行操作
3. 根据工具结果继续思考
4. 当信息足够时，给出最终答案
重要：
- 每次只专注当前最重要的一个子问题
- 观察工具结果后，更新你对问题的理解
- 如果工具调用失败，思考替代方案
- 信息足够时，直接给出最终答案，不要过度使用工具"""
def __init__(
self,
model: str = "NousResearch/Hermes-3-Llama-3.1-8B",
base_url: str = "http://localhost:8000/v1",
api_key: str = "not-needed",
max_steps: int = 15,
verbose: bool = True
):
self.model = model
self.max_steps = max_steps
self.verbose = verbose
self.tools: dict[str, Tool] = {}
self.client = AsyncOpenAI(
base_url=base_url,
api_key=api_key
)
def register_tool(self, tool: Tool) -> 'ReActAgent':
"""注册工具，支持链式调用"""
self.tools[tool.name] = tool
return self
def _log(self, message: str) -> None:
if self.verbose:
print(message)
async def _execute_tool(self, tool_name: str, arguments: dict) -> str:
"""执行工具调用，处理错误"""
if tool_name not in self.tools:
return f"错误：工具 '{tool_name}' 不存在。可用工具：{list(self.tools.keys())}"
tool = self.tools[tool_name]
try:
result = await asyncio.coroutine(tool.func)(**arguments) \
if asyncio.iscoroutinefunction(tool.func) \
else tool.func(**arguments)
# 限制观察结果长度，防止上下文溢出
result_str = str(result)
if len(result_str) > 2000:
result_str = result_str[:2000] + "\n... [结果已截断，显示前2000字符]"
return result_str
except Exception as e:
return f"工具执行错误：{type(e).__name__}: {str(e)}"
async def run(self, task: str) -> ReActResult:
"""
执行 ReAct 推理循环
Args:
task: 用户任务描述
Returns:
ReActResult: 包含答案、执行步骤、成功状态
"""
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": task}
]
tools_schema = [t.to_openai_format() for t in self.tools.values()]
steps = []
self._log(f"\n{'='*50}")
self._log(f"任务: {task}")
self._log(f"{'='*50}\n")
for step_num in range(1, self.max_steps + 1):
# ── 调用 LLM ──────────────────────────────────────
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools_schema if tools_schema else None,
tool_choice="auto",
temperature=0.1,
max_tokens=1024
)
msg = response.choices[0].message
# 提取思考内容（从 content 字段或工具调用前的推理）
thought = msg.content or "（分析中）"
# ── 判断是否有工具调用 ────────────────────────────
if msg.tool_calls:
# ReAct 的 Action 阶段
tool_call = msg.tool_calls[0] # 处理第一个工具调用
tool_name = tool_call.function.name
try:
arguments = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
arguments = {"raw": tool_call.function.arguments}
step = ReActStep(
step_number=step_num,
thought=thought,
action={"tool": tool_name, "arguments": arguments}
)
self._log(f"\n--- Step {step_num} ---")
self._log(f"Thought: {thought}")
self._log(f"Action: {tool_name}({json.dumps(arguments, ensure_ascii=False)})")
# ── 执行工具（Observation 阶段）──────────────
observation = await self._execute_tool(tool_name, arguments)
step.observation = observation
self._log(f"Observation: {observation[:200]}{'...' if len(observation) > 200 else ''}")
steps.append(step)
# 将工具调用和结果追加到消息历史
messages.append({
"role": "assistant",
"content": thought,
"tool_calls": [{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": tool_call.function.arguments
}
}]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": observation
})
else:
# ReAct 循环结束，LLM 给出最终答案
final_answer = msg.content or "任务完成，但无明确输出。"
step = ReActStep(
step_number=step_num,
thought=thought,
final_answer=final_answer
)
steps.append(step)
self._log(f"\n--- Step {step_num} (Final) ---")
self._log(f"Thought: {thought}")
self._log(f"Final Answer: {final_answer}")
self._log(f"\n{'='*50}")
self._log(f"完成，共 {step_num} 步")
self._log(f"{'='*50}")
return ReActResult(
success=True,
answer=final_answer,
steps=steps,
total_steps=step_num
)
# 超过最大步骤数
return ReActResult(
success=False,
answer="已达到最大步骤数限制，任务未完成。",
steps=steps,
total_steps=self.max_steps,
error=f"Exceeded max_steps={self.max_steps}"
)
# ─── 工具实现示例 ──────────────────────────────────────────────
import httpx
def make_search_tool() -> Tool:
"""创建网络搜索工具"""
async def web_search(query: str, num_results: int = 3) -> str:
# 使用 DuckDuckGo 或其他搜索 API
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.duckduckgo.com/",
params={"q": query, "format": "json", "no_html": 1},
timeout=10
)
data = resp.json()
results = []
if data.get("AbstractText"):
results.append(f"摘要：{data['AbstractText']}")
for r in data.get("RelatedTopics", [])[:num_results]:
if isinstance(r, dict) and r.get("Text"):
results.append(f"- {r['Text']}")
return "\n".join(results) if results else "未找到相关结果"
return Tool(
name="web_search",
description="搜索互联网获取最新信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num_results": {"type": "integer", "description": "返回结果数量", "default": 3}
},
"required": ["query"]
},
func=web_search
)
def make_calculator_tool() -> Tool:
"""创建计算器工具"""
def calculate(expression: str) -> str:
try:
# 安全计算：只允许数字和基本运算符
allowed = set('0123456789+-*/()., ')
if not all(c in allowed for c in expression):
return "错误：表达式包含不允许的字符"
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误：{str(e)}"
return Tool(
name="calculator",
description="执行数学计算",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式，如 '2 + 3 * 4'"}
},
"required": ["expression"]
},
func=calculate
)
def make_datetime_tool() -> Tool:
"""创建日期时间工具"""
from datetime import datetime, timezone
def get_datetime(timezone_name: str = "UTC") -> str:
now = datetime.now(timezone.utc)
return f"当前 UTC 时间：{now.strftime('%Y-%m-%d %H:%M:%S')} UTC"
return Tool(
name="get_datetime",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {
"timezone_name": {"type": "string", "description": "时区名称", "default": "UTC"}
},
"required": []
},
func=get_datetime
)
# ─── 主程序 ────────────────────────────────────────────────────
async def main():
# 构建 ReAct Agent
agent = (
ReActAgent(
model="NousResearch/Hermes-3-Llama-3.1-8B",
base_url="http://localhost:8000/v1",
max_steps=10,
verbose=True
)
.register_tool(make_search_tool())
.register_tool(make_calculator_tool())
.register_tool(make_datetime_tool())
)
# 运行任务
tasks = [
"今天是什么日期？距离 2025 年元旦还有多少天？",
"谁是 2024 年诺贝尔物理学奖得主？他们做了什么贡献？",
"计算：如果每月存 5000 元，年利率 3%，10 年后本息合计多少？"
]
for task in tasks:
print(f"\n{'#'*60}")
result = await agent.run(task)
print(f"\n最终结果（{result.total_steps} 步）：")
print(result.answer)
if not result.success:
print(f"警告：{result.error}")
if __name__ == "__main__":
asyncio.run(main())
51.5 调试 ReAct 循环的常见问题
问题一：无限循环
症状：Agent 不断重复相似的工具调用，永远不给出最终答案。
根本原因：
- 工具返回的信息不足以让 LLM 进展
- 系统提示词缺少明确的"信息足够时停止"指令
- LLM 过于保守，总想"再确认一次"
解决方案：
# 在系统提示词中明确停止条件
SYSTEM_PROMPT += """
**停止规则**：
- 当你已经有足够信息回答问题时，直接给出答案，不要继续调用工具
- 如果同一工具调用失败两次，换一种方法或直接说明无法获取该信息
- 最多进行 {max_steps} 步推理
"""
# 检测重复行动
class LoopDetector:
def __init__(self, window: int = 3):
self.recent_actions = []
self.window = window
def is_looping(self, action: dict) -> bool:
action_str = json.dumps(action, sort_keys=True)
self.recent_actions.append(action_str)
if len(self.recent_actions) > self.window:
self.recent_actions.pop(0)
# 如果最近 3 个行动完全相同
return len(set(self.recent_actions)) == 1 and len(self.recent_actions) == self.window
问题二：工具调用参数解析失败
症状：json.JSONDecodeError，LLM 生成了非法 JSON。
解决方案：
def safe_parse_arguments(raw: str) -> dict:
"""容错的参数解析"""
# 尝试直接解析
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
# 尝试提取 JSON 子串
json_pattern = re.search(r'\{.*\}', raw, re.DOTALL)
if json_pattern:
try:
return json.loads(json_pattern.group())
except json.JSONDecodeError:
pass
# 最后退路：作为字符串参数
return {"input": raw}
问题三：上下文溢出
症状：长任务执行到后期，LLM 开始"遗忘"早期信息，或 API 返回 context_length_exceeded 错误。
解决方案：
async def compress_messages(messages: list, max_tokens: int = 4096) -> list:
"""当消息历史过长时，摘要压缩早期内容"""
total_estimated_tokens = sum(len(m.get('content', '') or '') // 4 for m in messages)
if total_estimated_tokens <= max_tokens:
return messages
# 保留系统提示词和最近 6 条消息
system_msgs = [m for m in messages if m['role'] == 'system']
recent_msgs = messages[-6:]
middle_msgs = messages[len(system_msgs):-6]
if not middle_msgs:
return messages
# 对中间消息进行摘要
summary_prompt = "请用3-5句话总结以下对话的关键信息和已知事实：\n\n"
for m in middle_msgs:
if isinstance(m.get('content'), str):
summary_prompt += f"{m['role']}: {m['content'][:500]}\n"
# 调用 LLM 生成摘要（简化示例）
summary = f"[早期对话摘要] {summary_prompt[:200]}..."
return system_msgs + [{"role": "user", "content": summary}] + recent_msgs
常见问题速查表
| 问题 | 诊断方法 | 解决方案 |
|---|---|---|
| Agent 循环不停 | 打印每步 action，检查重复 | 添加循环检测器，强化停止指令 |
| 工具参数错误 | 打印 raw arguments | 实现容错解析，使用 JSON Schema 验证 |
| 上下文溢出 | 监控 token_count | 实现摘要压缩，限制观察结果长度 |
| 最终答案质量差 | 检查 observation 质量 | 优化工具输出格式，过滤噪音 |
| 步骤数过多 | 统计平均步骤数 | 调整系统提示词，鼓励高效推理 |
小结
本章深入讲解了 ReAct 架构的理论与实现：
- 论文原理：ReAct 通过交织 Thought（推理）和 Action（行动）克服了 CoT 与纯 Agent 各自的局限。
- 三阶段循环：Thought → Action → Observation 形成闭合反馈，每步观察都更新 LLM 的世界模型。
- Hermes 增强：结构化函数调用、并行工具、上下文压缩是 Hermes 对标准 ReAct 的主要改进。
- 完整代码：120 行实现了一个生产可用的 ReAct Agent，支持工具注册、错误处理、详细日志。
- 常见陷阱：无限循环、参数解析失败、上下文溢出是最常见的三类问题，均有对应解决方案。
思考题
- ReAct 中的 Thought 是给 LLM 自己看的，还是给用户看的？如果想在生产中隐藏 Thought，应该如何处理？
- 当工具调用失败时，Agent 应该立即重试、换工具还是向用户报告失败？如何设计这个策略？
- 如果要支持并行工具调用（多个工具同时执行），ReAct 的消息历史格式需要如何调整？
- ReAct 与普通 CoT 的本质区别是什么？对于不需要外部信息的纯推理任务，ReAct 是否仍然有优势？