ReAct 不是某个框架的 API,而是一种很朴素的循环:让模型在「想」和「做」之间交替,并把每次行动的结果喂回去。
核心思想
传统的一次调用是:
问题 → 模型 → 答案ReAct 是:
问题 → 思考 → 行动 → 观察 → 思考 → 行动 → …… → 答案它解决了一个关键问题:模型在推理时无法获得新信息。通过工具,它可以在中途查资料、算数、读文件、问用户,然后基于真实结果继续。
用原生工具调用,而不是解析文本
早期实现会让模型输出固定格式:
Thought: ...
Action: search
Action Input: {"query": "..."}然后正则解析。这种方式脆弱:模型可能少写一行、加个前缀、在 JSON 里换行。现代模型都支持原生工具调用(Function Calling),直接返回结构化的 tool_calls,不需要解析自由文本。
两者对比如下:
| 方式 | 稳定性 | 并行调用 | 推荐度 |
|---|---|---|---|
| 文本协议 + 正则 | 低 | 难 | 仅用于不支持工具调用的模型 |
| 原生 Function Calling | 高 | 支持 | 默认选择 |
| 结构化输出 + 自建协议 | 中高 | 需自行设计 | 特殊场景 |
一个可运行的最小实现
下面这段代码不到 200 行,包含两个工具、完整循环、错误处理和 trace。把 base_url、api_key、model 换成你的即可运行。
import ast
import json
import operator
import os
from dataclasses import dataclass, field
from openai import OpenAI
client = OpenAI(
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY"),
)
MODEL = os.getenv("MODEL", "gpt-4.1-mini")
# ---------------------------------------------------------------- tools
_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def _eval(node):
if isinstance(node, ast.Expression):
return _eval(node.body)
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
return node.value
if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
return _OPS[type(node.op)](_eval(node.operand))
raise ValueError("只支持数字与 + - * / ** 运算")
def calculator(expression: str) -> dict:
"""安全计算四则运算,不使用 eval。"""
try:
value = _eval(ast.parse(expression, mode="eval"))
return {"ok": True, "expression": expression, "value": value}
except Exception as e:
return {"ok": False, "error": {"code": "BAD_EXPRESSION", "message": str(e), "retryable": True}}
DOCS = {
"react": "ReAct 让模型交替推理与行动,并用工具结果更新上下文(Yao et al., 2022)。",
"rag": "RAG 在生成前检索外部文档,把相关内容放入上下文,降低幻觉。",
"mcp": "MCP 用统一协议连接模型应用与外部工具、资源、提示模板。",
}
def search_docs(query: str) -> dict:
"""模拟检索;真实项目替换为向量库或搜索引擎。"""
q = query.lower().strip()
hits = [{"id": k, "text": v} for k, v in DOCS.items() if q in k or q in v.lower()]
if not hits:
return {"ok": True, "summary": "没有找到相关资料。", "items": []}
return {"ok": True, "summary": f"找到 {len(hits)} 条资料。", "items": hits}
TOOL_IMPL = {"calculator": calculator, "search_docs": search_docs}
TOOLS = [
{
"type": "function",
"function": {
"name": "calculator",
"description": "计算数学表达式。涉及精确算术时使用,不要心算。",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "如 (38.2-42.4)/42.4*100"}
},
"required": ["expression"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "search_docs",
"description": "检索 AI Agent 主题的资料。问题涉及概念、定义、出处时使用。",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string", "description": "检索关键词,尽量简短"}},
"required": ["query"],
"additionalProperties": False,
},
},
},
]
SYSTEM = """你是一个严谨的研究助理。
规则:
1. 需要外部事实时先调用 search_docs;需要精确计算时调用 calculator。
2. 不要编造资料中没有的内容;资料不足时明确说明。
3. 工具报错时,根据错误信息修正一次;仍失败则说明原因。
4. 信息足够后直接给出最终答案,并标注引用的资料 id。
5. 回答使用中文,先结论后依据,不超过 300 字。"""
@dataclass
class Trace:
steps: list = field(default_factory=list)
tokens: int = 0
def run(goal: str, max_steps: int = 6) -> tuple[str, Trace]:
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": goal},
]
trace = Trace()
for step in range(1, max_steps + 1):
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
temperature=0,
max_tokens=800,
)
msg = resp.choices[0].message
trace.tokens += resp.usage.total_tokens
assistant_msg = {"role": "assistant", "content": msg.content}
if msg.tool_calls:
assistant_msg["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments},
}
for tc in msg.tool_calls
]
messages.append(assistant_msg)
if not msg.tool_calls:
trace.steps.append({"step": step, "type": "final", "text": msg.content})
return msg.content or "", trace
for call in msg.tool_calls:
name = call.function.name
args = {}
try:
args = json.loads(call.function.arguments or "{}")
if name not in TOOL_IMPL:
raise ValueError(f"未知工具:{name}")
result = TOOL_IMPL[name](**args)
except Exception as e:
result = {
"ok": False,
"error": {"code": "TOOL_ERROR", "message": str(e), "retryable": False},
}
trace.steps.append(
{"step": step, "type": "tool", "name": name, "args": args, "result": result}
)
messages.append(
{
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False)[:4000],
}
)
trace.steps.append({"type": "max_steps"})
return "达到最大步数,任务未完成。请缩小问题范围或转人工。", trace
if __name__ == "__main__":
answer, trace = run("ACME 的毛利率从 42.4% 降到 38.2%,下降了多少个百分点?")
print(answer)
print(json.dumps(trace.steps, ensure_ascii=False, indent=2))运行轨迹大致是:
step 1 tool calculator {"expression": "(42.4-38.2)"} → 4.2
step 2 final "毛利率下降了 4.2 个百分点。"注意代码里的几个决定:
- 温度 0:工具调用需要稳定;
- max_steps:硬性上限;
- 错误结构化:
ok、code、message、retryable; - 结果截断:
[:4000]防止工具返回撑爆上下文; - trace:记录每一步,便于调试;
- token 统计:成本可追踪。
循环的四个关键设计
1. 终止条件
至少三层:
模型不再请求工具 → 正常结束
达到 max_steps / 预算 / 超时 → 强制结束
检测到无进展(重复动作) → 提前结束「模型不再请求工具」并不可靠:它可能一直调用工具,或者一直说「我还需要更多信息」。所以必须有硬上限。
2. 无进展检测
一个简单有效的方法是对每次动作做指纹:
fingerprint = (tool_name, json.dumps(args, sort_keys=True))如果连续 2–3 次指纹相同、且观察结果也相同,就判定卡住:
- 注入一条系统消息:「你已经用相同参数调用过该工具,结果没有变化。请换一种方法或停止。」
- 或直接结束并转人工。
这比单纯增加步数上限更早发现问题。
3. 观察的裁剪
工具返回可能是 100KB。直接塞回上下文会:
- 撑爆窗口;
- 稀释关键信息;
- 增加成本。
在工具层或循环层做:
- 只保留决策需要的字段;
- 长文本摘要;
- 列表分页;
- HTML 转正文;
- 错误只留 code + message。
4. 并行工具调用
如果模型一次返回多个 tool_calls,且它们互不依赖,可以并发执行:
import asyncio
async def execute_all(calls):
return await asyncio.gather(*[execute(c) for c in calls], return_exceptions=True)注意:
- 每个结果都要按
tool_call_id对应; - 部分失败时,把每个调用的状态分别返回;
- 有副作用的写操作不要并行,除非工具本身支持幂等与并发控制。
ReAct 的局限
ReAct 是「走一步看一步」,它不擅长:
- 需要全局规划的长任务:容易走弯路、重复劳动;
- 强依赖顺序的流程:应该用状态机;
- 需要多方案比较的决策:应该用规划 + 评估;
- 超长任务:上下文会膨胀,需要压缩或分层。
对应的改进:
- 先规划再执行(Plan-and-Execute);
- 用待办列表维护进度;
- 把子任务交给子 Agent 隔离上下文;
- 对关键步骤加验证器。
这些会在后面的〈规划〉〈反思〉〈多智能体〉中展开。
常见失败
在系统提示中明确「涉及 X 必须调用 Y」,并给工具描述写清触发条件。评估时专门测「该调用却没调用」的案例。
通常是错误信息没有提供新信息,或工具返回空结果但描述没说「空结果意味着什么」。改进错误与空结果的语义。
在工具层裁剪,并在结果里加一个 summary 字段,把最重要的结论放在最前面。
每次模型调用和工具调用都要有超时。超时后返回可重试错误,而不是无限等待。
检查清单
- 使用原生工具调用,不解析自由文本
- 有最大步数、总超时、token 预算
- 有重复动作/无进展检测
- 工具错误结构化,带 retryable
- 工具结果裁剪并带摘要
- 并行调用按 tool_call_id 正确合并
- 全链路 trace,能回放每一步
- 失败时返回可理解的说明,而不是空字符串
小结
- ReAct = 推理与行动交替,观察回填上下文。
- 用原生工具调用,不要用正则解析文本协议。
- 循环的可靠性来自硬上限、无进展检测、错误结构和结果裁剪。
- ReAct 适合探索型任务;长任务需要规划、验证和上下文隔离。
下一篇讲「先想清楚再做」:规划与任务分解。