AI AGENT 开发 · 从入门到实战 · #03

ReAct 与循环工程:让大模型学会「思考→行动→观察」

上篇回顾:上一篇解剖四大件时说过,它们全都挂在同一颗「心脏」上——那个循环。本篇就把这颗心脏放上手术台:它怎么跳(节拍)、什么时候停(停止条件)、跳错了怎么办(错误恢复),以及为什么说 Agent 的质量不取决于任何一次单跳,而取决于整个回路的设计。

一、ReAct 在讲什么:不只是「先想后做」

2022 年的 ReAct 论文,全名是《ReAct: Synergizing Reasoning and Acting in Language Models》——关键词是 Synergizing(协同),而不是简单的「先想后做」。它的核心机制是让 LLM 交替生成推理轨迹(reasoning trace)与任务动作(action):论文原话是,推理轨迹帮模型「归纳、跟踪、更新行动计划,并处理异常」,而动作让模型与知识库、环境等外部源交互获取信息 [1]。注意这四个动词——归纳、跟踪、更新、处理异常——它们几乎就是本篇的目录。

两个对照实验最能说明「交错」的分量。对照纯思维链(Chain-of-Thought, CoT):只会想的模型容易凭空幻觉——ReAct 论文实测,通过与环境(一个简单的 Wikipedia API)交互,可以克服纯思维链里常见的幻觉与错误传播 [1]。道理不玄:让模型去看一眼真实世界,比在提示词里写十遍「不要编造」都管用——外部观察是对抗幻觉的接地(grounding)手段。对照纯行动(Act-only):只会动的模型计划会漂移,而且轨迹不可解释;ReAct 的轨迹「更像人类、更可解释」[1]。至于战绩,第 1 篇已经报过:ALFWorld、WebShop 两个环境上,仅靠一两个示例的 ReAct 大幅超过模仿学习与强化学习基线——「循环 beats 训练」的最早公开证据之一。论文的官方项目页至今保留着交互式 demo 与全部提示词,值得亲手玩一遍 [2]。

有人会问:如今的推理模型(先「想」很久再输出的那一类)是不是让 ReAct 过时了?我的判断是没有——变强的只是「想」的那一拍,而 ReAct 的真正价值从来不在独白,而在行动之后必有观察、观察之后必有更新这个闭环。模型内部想一百步,不如伸手看一眼真实世界:环境的事实永远比推理的回忆可靠。这也是为什么四年来框架换了几代,这个节拍一个字都没改。

二、节拍:把思考-行动-观察写进代码

论文时代的节拍是显式的:模型按 Thought: … / Action: … / Observation: … 的格式逐轮输出,解析器抽出动作去执行,观察结果拼回提示词。今天的生产实现里,这个三拍被 API 吸收了:Action 变成结构化的工具调用(tool call),Observation 变成回填的工具结果,Thought 则或折叠进模型内部、或干脆省略。Anthropic 对 agent 的定义也随之为简:「基于环境反馈,在循环中使用工具的 LLM」[3]。范式没有死,它是被 API 标准化了。

但我建议初学者仍然把「思考」显式留下来——不是留在输出格式里,而是留在 system 提示里:要求模型每次回答先用一句话说明「我现在的判断和打算」,再决定是否调用工具。代价是几十个 token,收益是排障时你能看见模型「当时在想什么」——可解释性不是道德要求,是调试刚需。下面是一个把本篇全部要点装进去的生产级循环:

import json
from pathlib import Path
from openai import OpenAI   # pip install openai;需先设置环境变量 OPENAI_API_KEY

client = OpenAI()

def _note(note_name: str):
    """把文件名解析为安全路径:合法返回 (路径, None),否则返回 (None, 报错)。"""
    root = Path("notes").resolve()
    path = (root / note_name).resolve()
    if path.parent != root:                    # 防呆:拦截路径穿越(第 2 篇军规)
        return None, f"ERROR: 非法路径 {note_name}"
    if not path.exists():                      # 会说人话的报错:把出路指给模型看
        names = [p.name for p in root.glob("*.txt")]
        return None, f"ERROR: {note_name} 不存在,现有文件:{names}"
    return path, None

def read_note(note_name: str) -> str:
    path, err = _note(note_name)
    return err if err else path.read_text(encoding="utf-8")[:1500]

def word_count(note_name: str) -> str:
    path, err = _note(note_name)
    return err if err else str(len(path.read_text(encoding="utf-8")))

def list_notes() -> str:
    names = sorted(p.name for p in Path("notes").glob("*.txt"))
    return "\n".join(names) if names else "(空目录)"

TOOLS = {"read_note": read_note, "word_count": word_count, "list_notes": list_notes}

SCHEMAS = [
    {"type": "function", "function": {
        "name": "list_notes", "description": "列出 notes/ 下所有 .txt 笔记的文件名",
        "parameters": {"type": "object", "properties": {}, "required": []}}},
    {"type": "function", "function": {
        "name": "read_note", "description": "按文件名读取单篇笔记,返回前 1500 字",
        "parameters": {"type": "object",
                       "properties": {"note_name": {"type": "string", "description": "笔记文件名"}},
                       "required": ["note_name"]}}},
    {"type": "function", "function": {
        "name": "word_count", "description": "统计单篇笔记的字符数",
        "parameters": {"type": "object",
                       "properties": {"note_name": {"type": "string", "description": "笔记文件名"}},
                       "required": ["note_name"]}}},
]

def react(goal: str, max_steps: int = 8, max_errors: int = 2) -> str:
    messages = [
        {"role": "system", "content": "你是笔记分析员。节拍:每次先用一句话说明你当前的判断与打算(思考),再决定是否调用工具(行动);证据足够就停止调用,直接输出最终答案。"},
        {"role": "user", "content": goal},
    ]
    errors = 0                                 # 连续错误计数:熔断器的燃料
    for step in range(1, max_steps + 1):
        msg = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, tools=SCHEMAS
        ).choices[0].message
        if not msg.tool_calls:                 # 停止条件一:任务完成
            return msg.content
        messages.append(msg)
        for call in msg.tool_calls:
            print(f"[step {step}] {call.function.name}({call.function.arguments})")  # 轨迹:留给排障的你
            try:
                result = TOOLS[call.function.name](**json.loads(call.function.arguments))
                errors = 0                     # 一次成功即清零
            except Exception as e:             # 错误也是观察:原文回填
                errors += 1
                result = f"ERROR: {type(e).__name__}: {e}"
            messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
        if errors >= max_errors:               # 停止条件二:连续失败,熔断
            return f"连续 {errors} 次工具失败,提前终止,避免空转烧钱。"
    return "已到步数预算上限,任务未完成。"      # 停止条件三:预算耗尽

# 预期输出:终端逐行打印工具调用轨迹,最终返回一段汇总答案(每篇笔记的字数与最长者)
print(react("notes/ 里有哪几篇笔记?各多少字?哪一篇最长?"))

和前两篇的骨架相比,这段代码只多了三样东西:轨迹打印、错误回填、熔断器——恰好就是「玩具」与「生产」之间的全部距离。

什么时候停:三种停止条件

循环最怕的不是转得慢,是停不下来。生产级循环需要三种相互独立的停止条件,缺一不可:

• 完成信号:模型认为任务做完,给出最终答案。这是唯一体面的出口,但要小心它的判定方式——「模型不再调用工具」是一个消极信号,偶尔会出现模型「忘了自己会工具」而提前收工;要靠 system 提示里明确的收尾规则来加固;
• 预算:步数、token、费用,至少设一维,建议三维都设。Anthropic 的总原则是「找最简单的方案,只在确有收益时增加复杂度」[3]——把上限当作产品参数暴露出来,而不是埋在代码里的一个魔法数字;
• 熔断:连续 N 次同类失败就提前终止,走降级路径。预算防的是「永久空转」,熔断防的是「短期抽风」——模型陷入同一个坑里反复横跳时,预算还没烧完,你已经没心情看它跳了。

三者的关系可以这样记:完成是目标,预算是底线,熔断是保险丝。预算具体怎么定?给两个经验锚点:结构明确的任务(读几个文件、改一处配置),5-8 步就该收尾,超过就说明提示或工具有问题;开放检索类任务给到 10-15 步。把步数乘以每步的输入输出规模,就是这笔任务的成本上限——预算不是玄学,是一笔能提前算出来的账。上线后如果发现频繁触顶,先别急着加步数,九成是工具返回的信息质量太差,模型在用步数换信息。

出错了怎么办:错误是观察的一部分

新手把工具报错当异常路径,老手把它当主路径的一部分。ReAct 给我们的最重要的启示就在这里:Observation 不只有成功的结果,也包括失败的报错——把报错原文回填给模型,它大概率能自己改对。Anthropic 的工程指南对工具层的要求是:返回清晰、有信息量的错误消息,让 agent 能据此自我纠正——比如报出「参数 X 缺失、格式应为 Y」,而不是一个笼统的 error [4]。换句话说,错误消息是写给模型读的用户界面;一个会说人话的报错,等价于一次免费的重试。

但自我纠正不是无限的,失败要分层处置,每一层都有配额:

对比一下两种报错,你就明白差距在哪:ERROR: validation failed 是一句废话——模型不知道哪错了、能往哪改;而 ERROR: note_name 不存在,现有文件:[a.txt, b.txt] 是一次引导——模型下一轮就会从列表里挑一个真实文件名。写工具层时,把每条报错都想象成「模型读到的反馈」,问一句:它看完知道下一步该干什么吗?

• 自纠:同类错误先让模型看着报错改参数,最多 N 次(这就是熔断器的 max_errors);
• 换路:重复失败说明此路不通——换工具、拆任务、改用第 2 篇说的「模型升级即重试」;
• 升级:把上下文打包转人工。Anthropic 观察到,人类给出明确反馈后 LLM 的输出可被显著改善 [3]——转人工不是失败,是回路的又一条反馈边;
• 放弃:带着完整的失败轨迹退出,把「为什么不行」记下来。这比静默挂死体面得多,也是下一批迭代最好的燃料。

从 ReAct 到 Plan-then-Execute:谱系而非对立

学 ReAct 常见的困惑是:它和上篇说的规划(Plan-then-Execute)是什么关系?答案:它们在同一条谱系上。Anthropic 把编排方式归纳成一个谱系——prompt chaining(预拆步骤的确定性流水线)、routing(分流)、parallelization(并行)、orchestrator-workers(动态拆解委派)、evaluator-optimizer(评价者驱动迭代)——而 ReAct 式的自由循环是其中自主性最高的一档 [3]。选型标准始终是任务结构,不是新鲜度:步骤可预测,用先规划后执行,成本延迟都可控;开放探索,才上自由循环。

两者还能杂交,而且杂交体往往最好用——规划件画路线,ReAct 循环跑每一段,发现路线过时就重画:

# 混合范式示意:make_plan 见第 2 篇规划件,react 即上文循环
goal = "调研 notes/ 中三篇竞品分析笔记,写一份对比摘要"
plan = make_plan(goal)                      # 先画路线:不超过 5 步
done = []
for i, step in enumerate(plan, 1):
    result = react(f"总体目标:{goal};已完成:{done};本步只做:{step}")
    done.append(f"{step} -> {result[:50]}")
    if result.startswith(("已到", "连续")):   # 预算耗尽或熔断=计划外信号
        plan[i:] = make_plan(f"原目标:{goal};已完成:{done};请重新规划剩余步骤")  # 改计划,不硬走

注意最后那行:循环检测到「计划外信号」时,把剩余步骤整个推倒重排——这就是第 2 篇说的「计划是允许被改的」在循环层的落地。

循环工程:质量来自回路

把前面的内容收拢成一个词,就是标题里的循环工程(loop engineering)。它的核心命题是:单次生成的质量上限由模型决定,而一个任务最终的质量下限,由回路的设计决定。Anthropic 在 evaluator-optimizer 模式里的表述可以作为注脚:「迭代精炼能带来可度量的收益」,例如编码 agent 「把测试结果作为反馈来迭代解决方案」[3]。把测试、报错、用户反馈接进循环,常常比换一个更大的模型更划算。

落到工程,循环三件套缺一不可:每一圈接地(从环境拿事实,不信模型的记忆);每次失败有回音(报错回填,可自纠);每一圈有边界(完成信号 + 预算 + 熔断)。

给你一个三问自检,任何跑在生产上的循环都该能全绿:第一,它的每一圈都拿到了新事实吗,还是在反复咀嚼旧上下文?第二,失败的时候,报错文本模型看得懂、改得动吗?第三,把网线拔了、把工具全删了,它能在有限时间内体面地退出吗?三问全过,才谈得上「工程」,否则只是「能转」。

---

小结

• ReAct 的四个动词——归纳、跟踪、更新、处理异常——是推理轨迹存在的理由;纯 CoT 会幻觉,纯行动会漂移;
• 显式三拍已被 API 吸收,但节拍未变;在 system 里保留一句「先说判断和打算」,换来的是可调试性;
• 三种停止条件:完成是目标,预算是底线,熔断是保险丝;错误消息是写给模型读的 UI,失败要分层处置、层层有配额;
• 谱系选型看任务结构:可预测先规划,开放才自由循环;杂交体(规划+循环+重规划)常常最好用。

下一篇预告:循环转得越久,messages 就越滚越大——上下文窗口是整个回路里最贵的资源。下一篇《上下文工程》讲往循环里装什么、装多少、什么时候该扔:token 预算分配、System Prompt 的写作清单,以及「上下文腐化」的急救。

参考来源

1. ReAct: Synergizing Reasoning and Acting in Language Models, ICLR 2023,arxiv.org
2. ReAct 项目页(demo / 提示词 / 代码), Yao et al., 2022,react-lm.github.io
3. Building Effective Agents, Anthropic, 2024-2025,anthropic.com
4. Writing Tools for Agents, Anthropic Engineering, 2025,anthropic.com