Agent AtlasAI Agent 知识库
03.02架构模式 · Architecture

ReAct 与工具循环

Reason + Act 的工作机制、终止条件、循环控制,以及一个不到 200 行、可运行的 Python 实现。

ReAct 不是某个框架的 API,而是一种很朴素的循环:让模型在「想」和「做」之间交替,并把每次行动的结果喂回去。

核心思想

传统的一次调用是:

text
问题 → 模型 → 答案

ReAct 是:

text
问题 → 思考 → 行动 → 观察 → 思考 → 行动 → …… → 答案

它解决了一个关键问题:模型在推理时无法获得新信息。通过工具,它可以在中途查资料、算数、读文件、问用户,然后基于真实结果继续。

图 1 · ReAct 循环
THOUGHT推理我现在知道什么?还缺什么?
ACTION行动选择一个工具与参数
OBSERVATION观察工具返回结果或错误
REPEAT继续或结束信息够了就回答,否则再来一轮

用原生工具调用,而不是解析文本

早期实现会让模型输出固定格式:

text
Thought: ...
Action: search
Action Input: {"query": "..."}

然后正则解析。这种方式脆弱:模型可能少写一行、加个前缀、在 JSON 里换行。现代模型都支持原生工具调用(Function Calling),直接返回结构化的 tool_calls,不需要解析自由文本。

两者对比如下:

方式稳定性并行调用推荐度
文本协议 + 正则低难仅用于不支持工具调用的模型
原生 Function Calling高支持默认选择
结构化输出 + 自建协议中高需自行设计特殊场景

一个可运行的最小实现

下面这段代码不到 200 行,包含两个工具、完整循环、错误处理和 trace。把 base_url、api_key、model 换成你的即可运行。

python
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))

运行轨迹大致是:

text
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. 终止条件

至少三层:

text
模型不再请求工具                → 正常结束
达到 max_steps / 预算 / 超时    → 强制结束
检测到无进展(重复动作)        → 提前结束

「模型不再请求工具」并不可靠:它可能一直调用工具,或者一直说「我还需要更多信息」。所以必须有硬上限。

2. 无进展检测

一个简单有效的方法是对每次动作做指纹:

python
fingerprint = (tool_name, json.dumps(args, sort_keys=True))

如果连续 2–3 次指纹相同、且观察结果也相同,就判定卡住:

  • 注入一条系统消息:「你已经用相同参数调用过该工具,结果没有变化。请换一种方法或停止。」
  • 或直接结束并转人工。

这比单纯增加步数上限更早发现问题。

3. 观察的裁剪

工具返回可能是 100KB。直接塞回上下文会:

  • 撑爆窗口;
  • 稀释关键信息;
  • 增加成本。

在工具层或循环层做:

  • 只保留决策需要的字段;
  • 长文本摘要;
  • 列表分页;
  • HTML 转正文;
  • 错误只留 code + message。

4. 并行工具调用

如果模型一次返回多个 tool_calls,且它们互不依赖,可以并发执行:

python
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 适合探索型任务;长任务需要规划、验证和上下文隔离。

下一篇讲「先想清楚再做」:规划与任务分解。

读完了?进度会保存在本地浏览器。
内容会持续修订。发现错误或想补充,欢迎反馈。 遇到问题?查排查手册 →