跳转至

第五章：ReAct

5.1 为什么需要循环

先看不用循环会怎样。

旅游助手接到一个请求："帮我安排明天的杭州一日游，我喜欢喝茶和吃小吃，告诉我带什么。"

One-shot 调用：

from agent_patterns_lab.runtime.mock_model import MockLLM
from agent_patterns_lab.runtime.types import Message

model = MockLLM([
    "上午去西湖散步，中午在楼外楼吃饭，下午去龙井村喝茶，晚上逛河坊街。"
    "明天天气晴朗，穿轻便衣服即可。"
])

messages = [
    Message(role="system", content="你是杭州旅游助手。"),
    Message(role="user", content="帮我安排明天的杭州一日游，我喜欢喝茶和吃小吃，告诉我带什么。"),
]

answer = model.complete(messages)
print(answer)

输出：

上午去西湖散步，中午在楼外楼吃饭，下午去龙井村喝茶，晚上逛河坊街。明天天气晴朗，穿轻便衣服即可。

这个回答有三个问题：

  1. "明天天气晴朗"是编的。 模型没有查天气，它不知道明天下不下雨。如果明天下午有雨，下午安排户外的龙井村就是坑人。
  2. 路线顺序是猜的。 西湖到龙井村到河坊街的路线是否合理？需要多长时间？模型没查。
  3. "穿轻便衣服"没有依据。 如果明天降温到 12°C，这条建议就是错的。

这三个问题的共同根源：模型在一次调用里就给了最终答案，没有先查信息、再根据信息调整。

在第四章我们知道了怎么写好 prompt，但好 prompt 不能凭空制造事实。模型需要一种机制：先行动（查天气），看到结果（下午有雨），再决定下一步（把下午改成室内）。

这就是循环的意义。

5.2 ReAct：Reason + Act

5.2.1 核心思路

ReAct 出自 2022 年的论文 "ReAct: Synergizing Reasoning and Acting in Language Models"。论文里的格式是：

Thought: 我需要先查明天杭州的天气
Action: get_weather[Hangzhou, tomorrow]
Observation: 下午15点后有小雨，气温18-23°C
Thought: 下午有雨，应该安排室内活动
Action: search_places[tea, indoor]
Observation: 中国茶叶博物馆（室内，免费）
...

论文版本的 Thought 是自由文本，Action 是 工具名[参数] 格式，Observation 是工具返回的文本。

工程版本做两个改动：

  1. Action 用 JSON 代替自由文本。 {"type": "tool", "tool": "get_weather", "args": {"city": "Hangzhou"}} 比 get_weather[Hangzhou, tomorrow] 更容易 parse、validate、trace。
  2. Thought 不暴露给用户。 论文需要 Thought 来分析推理过程。工程实践中，我们关心的是可审计的 action 和 observation，不是模型的内心独白。

核心循环：

Model 选择一个 Action → Python 执行这个 Action → 得到 Observation → 放回 messages → 重复

直到 Model 认为信息够了，返回 {"type": "final", "answer": "..."} 终止循环。

flowchart TD
  S["当前 state: messages"] --> M["模型返回一个 Action"]
  M -->|type=tool| T["Python 调用工具"]
  T --> O["Observation 追加到 messages"]
  O --> S
  M -->|type=final| F["返回最终答案，循环结束"]
  S --> L["步数/超时/停滞检测"]
  L -->|超限| X["强制停止"]

5.2.2 完整实现

下面是一个完整的 ReActAgent 类。代码先完整贴出，然后逐段拆解。

from __future__ import annotations

import json
import re
import time
from dataclasses import dataclass, field
from typing import Any, Sequence

from agent_patterns_lab.runtime.types import Message
from agent_patterns_lab.runtime.mock_model import MockLLM
from agent_patterns_lab.runtime.tools import Tool, ToolRegistry
from agent_patterns_lab.runtime.tracing import Tracer


# ── Action 类型 ────────────────────────────────────────

@dataclass(frozen=True)
class ToolAction:
    tool: str
    args: dict[str, Any]

@dataclass(frozen=True)
class FinalAction:
    answer: str

Action = ToolAction | FinalAction


# ── Prompt 模板 ────────────────────────────────────────

REACT_SYSTEM_TEMPLATE = """你是一个 ReAct 风格的旅游助手。

每一步，你必须返回一个 JSON action。格式二选一：

调用工具：
{{"type": "tool", "tool": "<工具名>", "args": {{"参数": "值"}}}}

给出最终答案：
{{"type": "final", "answer": "你的回答"}}

可用工具：
{tool_descriptions}

规则：
- 每次只返回一个 JSON，不要附加其他文字
- 如果不确定某个事实，先用工具查
- 不要编造天气、路线、票价等实时信息
"""


# ── JSON 解析器 ────────────────────────────────────────

def parse_action_json(text: str) -> Action:
    """从模型输出中提取 JSON 并解析为 Action。"""
    # 先尝试直接 parse
    cleaned = text.strip()

    # 去掉 markdown code fence
    if cleaned.startswith("```"):
        lines = cleaned.splitlines()
        # 去掉首尾的 ``` 行
        inner_lines = []
        for line in lines:
            if line.strip().startswith("```"):
                continue
            inner_lines.append(line)
        cleaned = "\n".join(inner_lines).strip()

    try:
        data = json.loads(cleaned)
    except json.JSONDecodeError:
        # 尝试在文本中找第一个 JSON 对象
        match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', cleaned)
        if match:
            data = json.loads(match.group())
        else:
            raise ValueError(f"无法从模型输出中提取 JSON: {text[:200]}")

    action_type = data.get("type")
    if action_type == "tool":
        return ToolAction(tool=data["tool"], args=data.get("args", {}))
    elif action_type == "final":
        return FinalAction(answer=data["answer"])
    else:
        raise ValueError(f"未知 action type: {action_type}")


# ── ReAct Agent ────────────────────────────────────────

@dataclass
class ReActAgent:
    """完整的 ReAct Agent，包含循环控制和停滞检测。"""
    tools: ToolRegistry
    max_steps: int = 10
    timeout_seconds: float = 120.0
    stall_window: int = 3  # 连续 N 步调用同一工具视为停滞

    def run(
        self,
        model,
        task: str,
        tracer: Tracer | None = None,
    ) -> str:
        # 构建 system prompt
        tool_lines = "\n".join(
            f"- {t.name}: {t.description}" for t in self.tools.list()
        )
        system = REACT_SYSTEM_TEMPLATE.format(tool_descriptions=tool_lines)

        messages: list[Message] = [
            Message(role="system", content=system),
            Message(role="user", content=task),
        ]

        start_time = time.time()
        recent_tools: list[str] = []  # 用于停滞检测

        for step in range(self.max_steps):
            # ── 超时检查 ──
            elapsed = time.time() - start_time
            if elapsed > self.timeout_seconds:
                if tracer:
                    tracer.emit("react.timeout", step=step, elapsed=elapsed)
                return f"[超时] 已运行 {elapsed:.1f}s，强制返回当前信息。"

            # ── 调用模型 ──
            if tracer:
                tracer.emit("react.step", step=step)

            raw = model.complete(messages, tracer=tracer)
            action = parse_action_json(raw)

            # ── Final：循环结束 ──
            if isinstance(action, FinalAction):
                if tracer:
                    tracer.emit("react.final", step=step, answer=action.answer)
                return action.answer

            # ── Tool：执行工具 ──
            if tracer:
                tracer.emit("react.tool_call", step=step, tool=action.tool, args=action.args)

            tool_result = self.tools.call(action.tool, action.args, tracer=tracer)

            messages.append(Message(role="assistant", content=raw))
            messages.append(Message(role="tool", name=action.tool, content=tool_result))

            # ── 停滞检测 ──
            recent_tools.append(action.tool)
            if len(recent_tools) >= self.stall_window:
                window = recent_tools[-self.stall_window:]
                if len(set(window)) == 1:
                    if tracer:
                        tracer.emit("react.stall_detected", step=step, tool=action.tool)
                    return f"[停滞] 连续 {self.stall_window} 次调用 {action.tool}，强制停止。"

        # ── 超过 max_steps ──
        if tracer:
            tracer.emit("react.max_steps", max_steps=self.max_steps)
        return f"[超步数] 已执行 {self.max_steps} 步，未得出最终答案。"

逐段拆解：

Action 类型

@dataclass(frozen=True)
class ToolAction:
    tool: str
    args: dict[str, Any]

@dataclass(frozen=True)
class FinalAction:
    answer: str

Action = ToolAction | FinalAction

只有两种 action。ToolAction 表示"调用工具"，FinalAction 表示"我知道答案了"。用 frozen=True 是因为 action 一旦创建就不应该被修改——它是历史记录的一部分。

Prompt 模板

REACT_SYSTEM_TEMPLATE = """你是一个 ReAct 风格的旅游助手。
...
可用工具：
{tool_descriptions}
"""

{tool_descriptions} 在运行时被替换为实际注册的工具列表。工具描述是 prompt 的一部分——写得越像函数签名文档，模型选工具越准。写得像市场宣传语，模型就会乱选。

JSON 解析器

def parse_action_json(text: str) -> Action:
    cleaned = text.strip()
    if cleaned.startswith("```"):
        # 去掉 markdown code fence
        ...
    try:
        data = json.loads(cleaned)
    except json.JSONDecodeError:
        match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', cleaned)
        ...

模型的输出不总是干净的 JSON。它可能在 JSON 前面加一句"好的，我来查天气"，或者用 markdown 的 code fence 包裹。解析器做两层 fallback：先 json.loads 直接解析，失败了就用正则找第一个 JSON 对象。

这个正则 \{[^{}]*(?:\{[^{}]*\}[^{}]*)*\} 能匹配一层嵌套的 JSON 对象，覆盖 {"type": "tool", "args": {"city": "Hangzhou"}} 这种情况。

循环控制

for step in range(self.max_steps):
    elapsed = time.time() - start_time
    if elapsed > self.timeout_seconds:
        return f"[超时] ..."

    raw = model.complete(messages, tracer=tracer)
    action = parse_action_json(raw)

    if isinstance(action, FinalAction):
        return action.answer

    tool_result = self.tools.call(action.tool, action.args, tracer=tracer)
    messages.append(Message(role="assistant", content=raw))
    messages.append(Message(role="tool", name=action.tool, content=tool_result))

每轮做四件事：

  1. 检查超时——time.time() - start_time > timeout_seconds 就强制返回。
  2. 调用模型——把当前 messages 发给模型，拿到 action。
  3. 判断是否结束——FinalAction 就返回答案。
  4. 执行工具并更新 messages——ToolAction 就调用工具，把 action 和 observation 都追加到 messages。

messages.append 这两行是整个循环的关键。每追加一次，下一轮模型看到的 context 就多了一组"我做了什么 + 工具返回了什么"。这就是 ReAct 的"Act → Observe → Reason"循环。

停滞检测

recent_tools.append(action.tool)
if len(recent_tools) >= self.stall_window:
    window = recent_tools[-self.stall_window:]
    if len(set(window)) == 1:
        return f"[停滞] 连续 {self.stall_window} 次调用 {action.tool}，强制停止。"

如果模型连续 3 次（stall_window=3）调用同一个工具，说明它陷入了循环——可能是工具返回的结果它不理解，或者它的 prompt 理解出了问题。检测到停滞就强制停止，比等到 max_steps 更快止损。

5.2.3 旅游助手：杭州一日游

完整运行示例，包含每一轮的 Thought/Action/Observation：

import json

def get_weather(args: dict) -> str:
    return json.dumps({
        "city": args.get("city", "杭州"),
        "date": args.get("date", "明天"),
        "forecast": "上午多云，15点后小雨",
        "temperature_c": "18-23",
        "packing_hint": "带伞和薄外套",
    }, ensure_ascii=False)

def search_places(args: dict) -> str:
    return json.dumps({
        "places": [
            {"name": "西湖·断桥", "type": "户外", "best_time": "上午", "note": "避开下午雨"},
            {"name": "中国茶叶博物馆", "type": "室内", "best_time": "下午", "note": "免费，适合雨天"},
            {"name": "河坊街", "type": "半室内", "best_time": "傍晚", "note": "小吃集中"},
        ]
    }, ensure_ascii=False)

def estimate_route(args: dict) -> str:
    places = args.get("places", [])
    return json.dumps({
        "route": " → ".join(places),
        "total_transit_minutes": 55,
        "suggestion": "打车或公交均可，路线紧凑",
    }, ensure_ascii=False)

# 注册工具
tools = ToolRegistry([
    Tool(name="get_weather", description="查询指定城市和日期的天气预报", handler=get_weather),
    Tool(name="search_places", description="根据城市、兴趣和约束条件搜索旅游景点", handler=search_places),
    Tool(name="estimate_route", description="估算多个景点之间的交通时间", handler=estimate_route),
])

# 用 MockLLM 脚本化模型的 4 轮响应
scripted_responses = [
    # 第 1 轮：先查天气
    '{"type": "tool", "tool": "get_weather", "args": {"city": "杭州", "date": "明天"}}',
    # 第 2 轮：根据天气搜索景点
    '{"type": "tool", "tool": "search_places", "args": {"city": "杭州", "interests": ["茶", "小吃"], "weather": "下午有雨"}}',
    # 第 3 轮：估算路线
    '{"type": "tool", "tool": "estimate_route", "args": {"places": ["西湖·断桥", "中国茶叶博物馆", "河坊街"]}}',
    # 第 4 轮：给出最终答案
    '{"type": "final", "answer": "明天杭州一日游安排：上午去西湖断桥（趁没下雨），下午转中国茶叶博物馆品龙井（室内躲雨），傍晚逛河坊街吃小吃。全程交通约55分钟。带伞和薄外套，穿舒适的走路鞋。"}',
]

model = MockLLM(scripted_responses)
tracer = Tracer()

agent = ReActAgent(tools=tools, max_steps=10, timeout_seconds=60.0, stall_window=3)
answer = agent.run(model, task="帮我安排明天的杭州一日游，我喜欢喝茶和吃小吃，告诉我带什么。", tracer=tracer)

print("=" * 60)
print("最终答案:")
print(answer)
print("=" * 60)
print(f"\n运行轨迹（{len(tracer.events)} 个事件）:")
for event in tracer.events:
    print(f"  [{event.name}] {json.dumps(event.data, ensure_ascii=False)}")

输出：

============================================================
最终答案:
明天杭州一日游安排：上午去西湖断桥（趁没下雨），下午转中国茶叶博物馆品龙井（室内躲雨），傍晚逛河坊街吃小吃。全程交通约55分钟。带伞和薄外套，穿舒适的走路鞋。
============================================================

运行轨迹（11 个事件）:
  [react.step] {"step": 0}
  [llm.complete] {"model": "mock", ...}
  [react.tool_call] {"step": 0, "tool": "get_weather", "args": {"city": "杭州", "date": "明天"}}
  [tool.call] {"tool_name": "get_weather", "args": {"city": "杭州", "date": "明天"}}
  [tool.result] {"tool_name": "get_weather", "output": "{\"city\":\"杭州\",...}"}
  [react.step] {"step": 1}
  [react.tool_call] {"step": 1, "tool": "search_places", "args": {"city": "杭州", ...}}
  [react.step] {"step": 2}
  [react.tool_call] {"step": 2, "tool": "estimate_route", "args": {"places": [...]}}
  [react.step] {"step": 3}
  [react.final] {"step": 3, "answer": "明天杭州一日游安排：..."}

四轮循环的完整日志：

轮次 模型决策 工具返回 为什么下一步变了
0 get_weather → 查杭州明天天气 15点后小雨，18-23°C 下午有雨，不能全安排户外
1 search_places → 搜景点，约束"下午有雨" 断桥(户外/上午)、茶博(室内/下午)、河坊街(傍晚) 拿到了具体地点和时间安排
2 estimate_route → 估算三个点的交通 总计55分钟，路线紧凑 确认路线可行
3 final → 给出最终答案 — 信息够了，不需要继续查

关键观察：轮次 1 的景点搜索是因为轮次 0 看到了下午有雨才加了"室内"约束。如果没有循环，模型不可能先查天气再调整景点——它只能在一次调用里猜一个答案。

5.3 优势与局限

优势

可解释性好。 每一轮的 action 和 observation 都有记录。出了问题，看 trace 就能定位是哪一步的工具返回了错误信息，或者模型在哪一步做了错误决策。

动态调整。 第一轮发现下雨，第二轮自动搜室内景点。不需要提前写 if-else 分支来处理所有天气情况。

工具组合灵活。 不需要预定义工具调用顺序。模型可以根据情况决定先查天气还是先搜景点。

局限

~50 步后性能退化。 每轮都往 messages 里追加 action + observation，context 越来越长。经验上，超过 50 步后：

  • 模型对早期 observation 的记忆明显下降
  • token 成本线性增长
  • 延迟线性增长
  • 幻觉概率上升（context 太长，模型开始"忘记"已经查过的信息）

这不是一个精确的数字——取决于模型、context window 大小、每轮追加的 token 量。但"几十步"是一个实际的软上限。

调试不直观。 虽然 trace 有记录，但当循环跑了 8 轮时，找到出错的那一轮需要耐心。建议：

  1. 给每步加编号——trace 里的 step: 0, 1, 2... 就是为了这个。
  2. 用 JSONL 格式保存 trace——每行一个事件，可以用 jq 过滤。
  3. 在停滞检测时记录 context 长度——如果 context 超过 80% window，说明快到极限了。
# 调试时有用的 context 长度估算
def estimate_context_tokens(messages: list[Message], chars_per_token: float = 2.5) -> int:
    total_chars = sum(len(m.content) for m in messages)
    return int(total_chars / chars_per_token)

# 示例
sample_messages = [
    Message(role="system", content="x" * 1000),
    Message(role="user", content="x" * 200),
    Message(role="assistant", content="x" * 150),
    Message(role="tool", name="get_weather", content="x" * 500),
]
print(f"估算 context token: {estimate_context_tokens(sample_messages)}")

输出：

估算 context token: 740

成本不可预测。 固定 workflow 的成本是确定的（3 步就是 3 次 API 调用）。ReAct 的成本取决于模型决定跑几轮。设 max_steps=10 不意味着每次都跑 10 轮，但你必须按最坏情况做预算。

本章回顾

概念 一句话
One-shot 的问题 没查就答，天气、路线、票价全靠猜
ReAct 核心 Action → Observation → 下一个 Action，直到信息够了
工程 vs 论文 JSON action 代替自由文本；不暴露 Thought
循环控制 max_steps + 超时 + 停滞检测，三道刹车
性能退化 ~50 步后 context 过长，记忆和准确率下降

附录： 附录 5A：ReAct 工程优化 详细分析每轮 token 增长、工具结果缓存、JSONL trace 格式和调试技巧。

下一章： 第六章：Workflow 与 Agent Loop —— 不是二选一，而是一个光谱。