第四章智能体经典范式构建(从0到1手敲三大范式)

一句话总结

这一章你要亲手敲出智能体世界的"三套祖传功法":ReAct(边想边做,走一步看一步)、Plan-and-Solve(先列计划再逐条执行)、Reflection(做完自己挑毛病再改)。后面所有花哨的智能体框架,内核都逃不出这三套的组合变形。

本章你能学到什么

三大范式各自解决什么问题、适合什么场景、有什么坑
"让 LLM 用工具"的完整技术闭环:提示词约定格式 → 正则解析 → 真实执行 → 结果回喂
手敲一个 10 文件的小项目(约 860 行),而且全部离线可跑:不需要 API Key、不需要联网、每次输出逐字一致。做法是给 LLM 写一个按内容应答的替身——这个技巧本身比三个范式更值钱,后面每一章都会用到
亲手量出三件"读文档读不出来"的事:
ReAct 反复横跳时,有熔断和没熔断分别烧掉几次调用
Reflection 那一轮反思到底改对了没有——自己跑 timeit 验,不听评审的(实测 n=20000 时快了 754 倍,数量级差距,你机器上的具体倍数会不同)
三个范式跑同一批任务各花多少次 LLM 调用
给 agent 写测试:19 个测试 0.01 秒跑完,测的是控制流和保险丝,不是模型输出
面试必考:三大范式对比、ReAct 为什么容易死循环、Reflection 的成本收益账

与原书的差异说明:原书用 SerpApi 网页搜索做工具(需注册拿密钥),本文换成计算器 + 本地迷你知识库两个零成本工具。更进一步:本文加了一个 ScriptedLLM 替身,所以连 LLM 密钥都不需要,10 个文件的每一段输出都能在你的机器上原样复现。想接真模型只要设一个环境变量,文末给了方法。

一、先看懂三大范式在解决什么问题

三个范式不是三个"框架",是三种决定"下一步干什么"的方式。看清这一点,后面所有 agent 框架的文档你都会觉得眼熟。

1.1 ReAct:边想边做(Reasoning + Acting)

第一章的旅行助手其实就是它。核心循环:Thought(我现在该干嘛)→ Action(调这个工具)→ Observation(工具返回了什么)→ 再 Thought…… 直到攒够信息输出 Finish[答案]。

强项:每一步都基于最新信息决策,遇到意外(工具报错、结果为空)能当场改道,适合探索型任务(查资料、多跳问答);
弱点:目光短浅——只看眼前一步,没有全局计划,复杂任务容易绕路、兜圈子,甚至在两个工具之间反复横跳死循环;而且每走一步都要调一次 LLM,步数多=又慢又贵。

关于"贵"有一个容易漏掉的细节:ReAct 每一步都要把完整的历史轨迹重新发一遍。 所以它的 token 消耗不是随步数线性增长,而是接近二次增长——5 步的 ReAct 比 5 次独立调用贵得多。这一点在第 3.5 节会算给你看。

1.2 Plan-and-Solve:先谋后动

针对 ReAct"走哪算哪"的毛病,Plan-and-Solve 把过程拆成两个角色:

规划器(Planner):先通读任务,一次性输出完整的分步计划(比如"1. 查A的出生年份;2. 查B的出生年份;3. 计算差值");
执行器(Executor):拿着计划逐条执行,把每一步的结果带入下一步。

强项:有全局观,步骤清晰可审计,适合结构明确的多步任务(数学应用题、流程化操作);而且成本可预测——计划一出来,你就知道这次任务要花几次调用;
弱点:计划是开局一次性定死的,中途出意外不会改计划(除非再加"重规划"机制);如果第一步计划就错了,后面全白干。

"把每一步的结果带入下一步"这句话是 Plan-and-Solve 的全部价值所在。少了它,这个范式就退化成"独立跑 N 个子任务然后拼起来"——那是另一种东西(扇出/汇合),解决的问题完全不同。第 3.3 节有一个专门守着这条的测试。

1.3 Reflection:自己给自己当审稿人

前两个范式都是"一遍过",Reflection 加了一个质检环节,三步循环:

执行:先产出一版初稿(代码、文章、方案);
反思:换一顶"评审员"帽子,让 LLM 挑初稿的毛病;
优化:带着批评意见重做一版。循环直到评审说"没问题"或达到轮数上限。

强项:对质量敏感型任务(写代码、写报告)提升显著,相当于免费多请了一个审稿人;
弱点:每一轮 = 至少两次 LLM 调用(执行+反思),成本和延迟成倍涨;而且自己查自己有盲区,反思几轮后常出现"收益递减"。原书专门算了一笔成本收益账,结论:简单任务别用,质量要求高、错误代价大的任务才值得用。

这里有一个几乎所有教程都跳过的问题:你凭什么相信那一轮反思真的把代码改好了? 你看到的只是"评审说通过了"。评审说通过,代码可能还是错的;评审说"建议改用筛法",改完可能反而慢了。第 3.4 节我们会把两版代码都真的跑起来,用 timeit 量出来——这是本章我最想让你养成的习惯。

1.4 三者对比

	ReAct	Plan-and-Solve	Reflection
决策方式	走一步看一步	先定完整计划	做完自查自改
全局观	弱	强	取决于底层范式
应对意外	强	弱	中
LLM 调用次数	步数+1,事前不可知	1+步数,计划一出就确定	轮数×2
token 消耗	随步数接近二次增长	线性	线性(但每轮两次)
主要失败模式	反复横跳、绕路	计划一错全错	收益递减、自查盲区
典型场景	查资料、多跳问答	数学题、固定流程	写代码、写文章

最后一行的"主要失败模式"比"强项/弱点"更实用:你选一个范式,就是选择接受它那种失败方式,然后为它加保险丝。 本章三个 agent 各自的保险丝是:ReAct 的重复动作熔断 + max_steps、Plan 的步数校验、Reflection 的 max_iterations。

1.5 一个必须先建立的观念:范式是可以组合的

三个范式不是三选一。生产里最常见的组合是:

Plan 出骨架,每个子步骤内部跑小型 ReAct:全局有计划(便宜、可审计),局部能应变。
Reflection 包住 ReAct:一整轮 ReAct 跑完之后反思整条轨迹,再重跑一遍——这就是学术界 Reflexion 论文的思路。
Plan + Reflection:计划生成后先让"评审"看一眼计划再执行,而不是执行完才反思。这个变体在 docs/2 第 20 篇里会变成一个真实的功能——人在环中的计划审批。

所以本章的三个文件不是三个孤立的 demo,是三块可以互相嵌套的积木。

二、"让 LLM 动手"的技术闭环

三个范式在代码上共用同一套骨架,值得单独拎出来讲,因为所有工具型 agent 都是这四步:

① 格式约定    在提示词里规定 "Action: 工具名[参数]" 这种格式
② 解析        用正则(或 JSON 解析)从模型的自然语言回复里抠出结构
③ 执行        查工具注册表, 调用真正的 Python 函数
④ 回喂        把返回值拼成 "Observation: ..." 追加到历史里, 再问一次

四步里最脆的是 ②。模型不是 API,它会不按格式输出:多一句寒暄、少一个方括号、把 Action: 写成 行动:。所以这一步的代码必须回答一个问题:解析失败了怎么办?

有三种做法,难度递增:

崩掉——最差,一次格式抖动整个任务失败。
把"格式错了"当成一条 Observation 喂回去,给模型一次改正机会。本章用这个,五行代码。
改用模型原生的 function calling / structured output,让格式由服务端保证。生产首选,但你得先理解 1 和 2 在解决什么,才知道 function calling 帮你省掉了哪一层。

名词补课——为什么本章还要教正则解析这种"过时"写法:因为你会遇到不支持 function calling 的模型(很多开源模型、很多私有化部署),也因为理解 ReAct 的格式约定是理解 function calling 的前提。function calling 不是新范式,它是把第 ①②步下沉到了服务端。

三、动手实战:从0到1手敲三大范式

3.0 项目结构与"零密钥"设计

新建项目 agent_paradigms,最终长这样(10 个文件,约 860 行,自上而下依次敲):

agent_paradigms/
├── pyproject.toml         依赖与 pytest 配置
├── llm.py                 ① LLM 底座: 真客户端 + 可断言的替身      121 行
├── tools.py               ② 工具 + 工具执行器                     105 行
├── why_not_eval.py        ③ 30 行证明"为什么不能用 eval"           30 行
├── react_agent.py         ④ 范式一: ReAct(带熔断)                86 行
├── react_loop_demo.py     ⑤ 亲手复现"反复横跳", 量出熔断省了多少     51 行
├── plan_solve_agent.py    ⑥ 范式二: Plan-and-Solve(带校验)        86 行
├── reflection_agent.py    ⑦ 范式三: Reflection                    94 行
├── verify_reflection.py   ⑧ 不听评审的, 自己 timeit 验反思有没有用   61 行
├── compare_cost.py        ⑨ 三个范式的成本账                       38 行
└── test_paradigms.py      ⑩ 19 个测试, 测控制流和保险丝           189 行

建议顺序:①②③④⑤⑥⑦⑧⑨⑩ 一路做下去。 全部离线、零密钥、零花费,每个 ✅ Checkpoint 贴的都是真实运行输出,你的输出应该和书里逐字一致。

先建项目:

mkdir agent_paradigms && cd agent_paradigms

pyproject.toml:

[project]
name = "agent-paradigms"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["openai>=1.40", "python-dotenv>=1.0"]

[dependency-groups]
dev = ["pytest>=8.0"]

[tool.pytest.ini_options]
pythonpath = ["."]

uv sync

pythonpath = ["."] 那一行是第九章排错手册里的高频坑:所有脚本 python xxx.py 都好,pytest 却 ModuleNotFoundError。能钉在配置文件里的,不要靠每次记得敲 PYTHONPATH=.。

openai 和 python-dotenv 只在你想接真模型时才用得上;跑完本章不接真模型也完全没问题。

3.1 文件一:`llm.py` —— 一个能被断言的替身(约120行)

这是本章和原书最大的一处不同,也是整章最值钱的一个文件。

原书(以及市面上绝大多数教程)的 llm.py 只有真客户端,于是每个 Checkpoint 都变成"你应该会看到类似这样的输出"——因为模型每次答的不一样。而我们要的是:每一段输出都能逐字复现,而且能写成测试。

办法是加一个替身:

"""LLM 底座: 一个真客户端 + 一个能被断言的替身。

替身(ScriptedLLM)的存在理由: 三大范式的重点是**控制流**,
而控制流不该依赖网络和模型心情。有了它, 这一章的每个 Checkpoint
都能在没有任何 API Key 的机器上跑出**逐字相同**的输出。

AP_MODE=fake(默认) 用替身; AP_MODE=real 用真模型。
"""

import os
import re
from typing import Dict, List

MODE = os.getenv("AP_MODE", "fake")


class HelloAgentsLLM:
    """极简 LLM 客户端: 发一组消息, 收一段回复。"""

    def __init__(self):
        from dotenv import load_dotenv
        from openai import OpenAI

        load_dotenv()
        key = os.getenv("LLM_API_KEY")
        if not key:
            raise RuntimeError("AP_MODE=real 需要 LLM_API_KEY, 请先配 .env")
        self.client = OpenAI(api_key=key, base_url=os.getenv("LLM_BASE_URL"))
        self.model = os.getenv("LLM_MODEL_ID")
        self.calls: List[Dict] = []

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        self.calls.append({"prompt": messages[-1]["content"], "temperature": temperature})
        resp = self.client.chat.completions.create(
            model=self.model, messages=messages, temperature=temperature
        )
        return resp.choices[0].message.content

两个细节:

if not key: raise RuntimeError(...):没配密钥就报错,不要"没配就用一个默认值"或者静默继续。这叫 fail closed,是整套 docs 反复出现的一条纪律。
self.calls 记录每次调用:真客户端也记,不只是替身记。有了它,你能断言"这个 agent 到底调了几次模型、每次的温度是多少"——这是成本和纪律的断言,比断言输出内容有用得多。

替身本体:

class ScriptedLLM:
    """按**输入内容**决定输出的替身。

    刻意不用调用计数器(`if self.n == 0: ... elif self.n == 1: ...`):
    计数器型替身在两个 agent 共用一个实例、或者你临时多插一次调用时
    就会错位, 而且错得很隐蔽。按内容判断的替身, 你怎么调它都对。
    """

    def __init__(self):
        self.calls: List[Dict] = []

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        p = messages[-1]["content"]
        self.calls.append({"prompt": p, "temperature": temperature})
        for match, reply in self._rules():
            if match(p):
                return reply
        return "(替身没有为这个提示词准备剧本)"

"按内容而不是按次数"这条纪律值得单独记住。 计数器型替身长这样:

# 反面教材, 别这么写
def think(self, messages, temperature=0):
    self.n += 1
    if self.n == 1: return "Action: Search[乔布斯]"
    if self.n == 2: return "Action: Calculate[(2024-1955)*365]"
    return "Action: Finish[25185天]"

它在单个 agent 单次运行时完全正常,但只要你:①在两个 agent 之间共用一个替身实例,②在中间临时加一次调用调试,③或者跑并发——顺序就错位了,而且错位之后测试是随机红的,你会以为是自己的 agent 逻辑有问题。docs/2 第 20 篇里我在这个坑上真栽过一次。

剧本本体是一串"条件 → 回复":

    # ---------------- 剧本 ----------------
    def _rules(self):
        return [
            # ---- ReAct: 靠 history 里已有的 Observation 判断走到哪了 ----
            (lambda p: "Action:" in p and "25185" in p,
             "Thought: 我已经知道答案了\nAction: Finish[25185天]"),
            (lambda p: "Action:" in p and "1955年" in p,
             "Thought: 乔布斯生于1955年, 现在算天数。\nAction: Calculate[(2024-1955)*365]"),
            (lambda p: "Action:" in p,
             "Thought: 我需要先查到乔布斯的出生年份。\nAction: Search[乔布斯]"),
            # ---- Plan-and-Solve ----
            (lambda p: "规划专家" in p,
             '["计算午餐后剩下的苹果数", "加上新买的6个苹果", "给出最终数量"]'),
            (lambda p: "当前步骤:计算午餐后" in p, "23 - 20 = 3, 午餐后剩 3 个。"),
            (lambda p: "当前步骤:加上新买" in p, "3 + 6 = 9。"),
            (lambda p: "当前步骤:给出最终" in p, "现在有 9 个苹果。"),
            # ---- Reflection: 按代码内容判断该不该通过 ----
            (lambda p: "代码评审员" in p and "sieve" in p, "通过"),
            (lambda p: "代码评审员" in p,
             "1. 效率问题: 对每个数都从 2 试除到 n-1, 复杂度 O(n^2), n=10^6 时无法接受\n"
             "2. 建议改用埃拉托斯特尼筛法(sieve of Eratosthenes)\n"
             "3. 边界情况: n < 2 时应返回空列表, 当前实现未显式处理"),
            (lambda p: "Python程序员" in p and "评审意见" in p, _SIEVE_CODE),
            (lambda p: "Python程序员" in p, _BRUTE_CODE),
        ]

规则的顺序有讲究:从最具体到最宽松。 看 ReAct 那三条——第三条 "Action:" in p 会匹配所有 ReAct 提示词,所以它必须排在最后;前两条靠"历史里出现过 1955年 / 25185"判断"我已经走到第几步了"。

这个设计的巧妙之处在于:替身读的是 ReAct 自己拼出来的 history。 它不知道自己是第几次被调用,只知道"提示词里已经有了 1955 年这个事实,那我该去算天数了"。这正是真模型的行为方式,所以替身跑出来的控制流和真模型是同构的。

Reflection 那两条同理:评审看到代码里有 sieve 就说"通过",看不到就给三条意见;程序员看到提示词里有"评审意见"就给筛法版,否则给暴力版。"改进意见有没有真的进到下一轮提示词里"这件事,被替身的规则直接编码进去了——如果你的 agent 忘了把 feedback 拼进去,替身就会一直返回暴力版,测试立刻红。

两段代码常量和工厂函数:

_BRUTE_CODE = '''def find_primes(n):
    primes = []
    for x in range(2, n + 1):
        for d in range(2, x):
            if x % d == 0:
                break
        else:
            primes.append(x)
    return primes'''

_SIEVE_CODE = '''def find_primes(n):
    """埃氏筛: O(n log log n)"""
    if n < 2:
        return []
    sieve = [True] * (n + 1)
    sieve[0] = sieve[1] = False
    for x in range(2, int(n**0.5) + 1):
        if sieve[x]:
            sieve[x * x :: x] = [False] * len(sieve[x * x :: x])
    return [i for i, ok in enumerate(sieve) if ok]'''


def get_llm():
    """工厂: 让三个范式的代码不关心自己拿到的是真模型还是替身。"""
    return HelloAgentsLLM() if MODE == "real" else ScriptedLLM()


if __name__ == "__main__":
    llm = get_llm()
    print(f"MODE={MODE}  类型={type(llm).__name__}")
    if MODE == "real":
        print(llm.think([{"role": "user", "content": "用五个字打个招呼"}]))
    else:
        demo = "请尽你所能回答问题。Action: 工具名[工具输入]\n问题:乔布斯"
        print("替身回复:", re.sub(r"\n", " | ", llm.think([{"role": "user", "content": demo}])))
        print(f"已记录 {len(llm.calls)} 次调用")

get_llm() 这个工厂是整章能同时支持"离线跑通"和"接真模型"的唯一原因:三个 agent 的代码里只写 self.llm = llm or get_llm(),它们完全不知道自己拿到的是什么。这是依赖注入最朴素的形式,也是 agent 项目可测试性的地基。

✅ Checkpoint 1:python llm.py,真实输出:

MODE=fake  类型=ScriptedLLM
替身回复: Thought: 我需要先查到乔布斯的出生年份。 | Action: Search[乔布斯]
已记录 1 次调用

要接真模型:写好 .env(内容同第一章的三个变量),然后 AP_MODE=real python llm.py,会打印一句问候。没有 .env 也不影响本章后面任何一步。

3.2 文件二:`tools.py` —— 工具与工具执行器(约105行)

"""工具与工具执行器。

两个零成本工具: 安全计算器(ast 白名单) + 迷你本地知识库。
换成真搜索只要替换 `search` 一个函数, 执行器和三个范式都不用动 ——
这就是"把工具收在一个注册表后面"的回报。
"""

import ast
import operator
from typing import Callable, Dict

# ---------- 工具一:安全计算器 ----------
_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(node):
    if isinstance(node, ast.Constant):
        if not isinstance(node.value, (int, float)):
            raise ValueError("只允许数字")
        return node.value
    if isinstance(node, ast.BinOp):
        return _OPS[type(node.op)](_eval_node(node.left), _eval_node(node.right))
    if isinstance(node, ast.UnaryOp):
        return _OPS[type(node.op)](_eval_node(node.operand))
    raise ValueError(f"不支持的表达式节点: {type(node).__name__}")


def calculator(expression: str) -> str:
    """计算数学表达式, 如 (2024-1955)*365。"""
    try:
        return str(_eval_node(ast.parse(expression, mode="eval").body))
    except Exception as e:
        # 工具永远不该把异常抛给 agent 循环 —— 抛出去整个 agent 就挂了。
        # 把错误变成一句 Observation, 模型下一步才有机会改道。
        return f"计算失败:{e}"

三处值得停下来看:

raise ValueError(f"不支持的表达式节点: {type(node).__name__}") 带上了节点类型。 报错信息里说清"是哪种语法被挡了",调试时省一大截时间——后面 Checkpoint 2 里你会看到它打印出 Call,一眼就知道"有人试图调用函数"。

if not isinstance(node.value, (int, float)) 这一行是原书没有的。ast.Constant 不只装数字,字符串、True、None 都是 Constant。少了这行,calculator("'a'*10**9") 会真的构造一个 1GB 的字符串把你的进程撑爆——这是一个不需要执行任何函数就能实现的拒绝服务。 白名单要白到值的类型,不只是节点的类型。

except Exception 把错误变成字符串返回,而不是抛出去。 这是工具设计的一条硬纪律:工具的异常必须在工具内部终结。 抛给 agent 循环的后果是整个任务崩掉;返回一句"计算失败:xxx"的后果是模型看到一条 Observation,下一步有机会换个表达式。前者让一次小失误变成任务失败,后者只是多花一步。

# ---------- 工具二:迷你本地知识库(代替网页搜索) ----------
KNOWLEDGE = {
    "乔布斯": "史蒂夫·乔布斯,苹果公司联合创始人,生于1955年,逝于2011年。",
    "比尔盖茨": "比尔·盖茨,微软公司联合创始人,生于1955年。",
    "珠穆朗玛峰": "珠穆朗玛峰是世界最高峰,海拔8848.86米。",
    "长城": "长城总长度约21196千米,是世界文化遗产。",
}


def search(query: str) -> str:
    """在本地知识库中检索信息。"""
    hits = [v for k, v in KNOWLEDGE.items() if k in query]
    return " ".join(hits) if hits else f"知识库中没有关于'{query}'的信息"

search 找不到时返回一句明确的"没有",而不是空字符串。空字符串会让模型以为"工具坏了"或者"我格式写错了",于是重试同一个动作——这正是 ReAct 死循环最常见的触发方式。工具返回"我查了,没有"和返回空,对模型是完全不同的两件事。

# ---------- 工具执行器 ----------
class ToolExecutor:
    def __init__(self):
        self._tools: Dict[str, Dict] = {}
        self.log: list = []  # 记录每次调用, 让测试能断言"工具被调了几次"

    def register(self, name: str, description: str, func: Callable):
        self._tools[name] = {"description": description, "func": func}

    def run(self, name: str, arg: str) -> str:
        self.log.append((name, arg))
        if name not in self._tools:
            return f"未知工具:{name},可用工具:{list(self._tools)}"
        return self._tools[name]["func"](arg)

    def describe(self) -> str:
        return "\n".join(f"- {n}: {t['description']}" for n, t in self._tools.items())


def default_executor() -> ToolExecutor:
    ex = ToolExecutor()
    ex.register("Search", "检索人物、地点等事实信息,输入检索关键词", search)
    ex.register("Calculate", "计算数学表达式,输入如 (2024-1955)*365", calculator)
    return ex

三点:

describe() 直接生成提示词里的工具清单。 于是"注册了什么工具"和"提示词里写了什么工具"永远一致——手写两份必然有一天不同步,而不同步的症状是模型调用一个不存在的工具。
run() 遇到未知工具返回可读提示,还把可用工具列出来。 模型看到"可用工具:['Search', 'Calculate']"能自己纠正;看到 KeyError 就只能崩。
self.log:和 llm.calls 一个道理。有了它,测试能断言"这次任务一共调了 2 次工具、顺序是 Search 然后 Calculate"。给关键组件加一个调用记录,是让 agent 变得可测试的最小代价。

3.3 文件三:`why_not_eval.py` —— 30 行证明一件事(约30行)

原书这里只有一句提醒"别用 eval"。提醒不如亲眼看见:

"""为什么不能用 eval() 算 LLM 生成的表达式 —— 30 行跑一遍就明白。

这个文件只做一件事: 把同一个字符串分别喂给 eval 和我们的 calculator,
看两边分别发生什么。**不要在这个例子上加"更狠"的 payload 试**,
`echo` 已经足够说明问题了。
"""

from tools import calculator

PAYLOAD = "__import__('os').system('echo 我是被-eval-执行的任意命令')"

print("=== 1. 交给 eval() ===", flush=True)  # flush: 否则子命令的输出会插到前面
try:
    result = eval(PAYLOAD)  # noqa: S307  ← 这一行是反面教材, 别抄
    print(f"eval 的返回值: {result}(0 表示那条命令执行成功)")
except Exception as e:
    print(f"eval 抛异常: {e}")

print("\n=== 2. 交给 ast 白名单版 calculator() ===")
print(f"calculator 返回值: {calculator(PAYLOAD)}")

print("\n=== 3. 正常表达式两边结果一致 ===")
for expr in ["(2024-1955)*365", "2**10", "(1+2)*(3+4)"]:
    print(f"  {expr:20s} eval={eval(expr):<8} calculator={calculator(expr)}")

print(
    "\n结论: 白名单不是'更严格的 eval', 是**完全不同的执行模型** ——\n"
    "eval 先把字符串编译成任意 Python 再跑; calculator 只遍历 AST 上\n"
    "被允许的那几种节点, 其它一律 ValueError。前者的攻击面是整个语言。"
)

✅ Checkpoint 2:python why_not_eval.py,真实输出:

=== 1. 交给 eval() ===
我是被-eval-执行的任意命令
eval 的返回值: 0(0 表示那条命令执行成功)

=== 2. 交给 ast 白名单版 calculator() ===
calculator 返回值: 计算失败:不支持的表达式节点: Call

=== 3. 正常表达式两边结果一致 ===
  (2024-1955)*365      eval=25185    calculator=25185
  2**10                eval=1024     calculator=1024
  (1+2)*(3+4)          eval=21       calculator=21

结论: 白名单不是'更严格的 eval', 是**完全不同的执行模型** ——
eval 先把字符串编译成任意 Python 再跑; calculator 只遍历 AST 上
被允许的那几种节点, 其它一律 ValueError。前者的攻击面是整个语言。

第一段的第二行 我是被-eval-执行的任意命令 是你的 shell 真的执行了那条命令打印出来的。把 echo 换成任何别的命令,它就会真的执行任何别的命令——而这个字符串是模型生成的,也就是说是提示词能影响的,也就是说是用户能间接影响的。

那个 flush=True 是我调试这个脚本时踩的坑:第一次跑,我是被-eval-执行的任意命令 打在了 === 1. 交给 eval() === 上面。原因是 Python 的 stdout 有缓冲,而 os.system 起的子进程直接写终端,不经过 Python 的缓冲区。混用 print 和子进程输出时,顺序不是你写代码的顺序。 加 flush=True 才对得上。

✅ Checkpoint 3:python tools.py,真实输出:

可用工具:
- Search: 检索人物、地点等事实信息,输入检索关键词
- Calculate: 计算数学表达式,输入如 (2024-1955)*365

正常用法:
  Calculate[(2024-1955)*365] -> 25185
  Calculate[2**10] -> 1024
  Calculate[-3+5] -> 2
  Search[乔布斯] -> 史蒂夫·乔布斯,苹果公司联合创始人,生于1955年,逝于2011年。
  Search[图灵]   -> 知识库中没有关于'图灵'的信息

恶意/错误输入(重点看它是不是老老实实返回字符串而不是崩):
  Calculate[__import__('os').system('echo 我被执行了')]
    -> 计算失败:不支持的表达式节点: Call
  Calculate[open('/etc/passwd').read()]
    -> 计算失败:不支持的表达式节点: Call
  Calculate[1/0]
    -> 计算失败:division by zero
  Calculate[2 +]
    -> 计算失败:invalid syntax (<unknown>, line 1)
  未注册的工具 -> 未知工具:Translate,可用工具:['Search', 'Calculate']

执行器共记录 10 次调用

四种失败输入(恶意调用、读文件、除零、语法错)全部返回字符串,一个都没抛异常。这就是"工具的异常在工具内部终结"长什么样。

3.4 文件四:`react_agent.py` —— 范式一:ReAct(约86行)

第二节那四步在这里第一次凑齐。比原书多了两样东西,都是踩过坑才加的:

"""范式一: ReAct —— 边想边做。

循环: Thought → Action → Observation → Thought → ... → Finish[答案]

比原书多了两样东西, 都是踩过坑才加的:
  1. **重复动作熔断**: 连续两次发出同一个 Action 就停 —— ReAct 最常见的
     失败模式不是答错, 是在两个工具之间反复横跳直到烧完 max_steps。
  2. **格式错误的容错**: 解析不到 Action 时不是崩掉, 而是把"格式错了"
     当成一条 Observation 喂回去, 给模型一次改正机会。
"""

import re

from llm import get_llm
from tools import default_executor

PROMPT = """请尽你所能回答问题。你可以使用以下工具:
{tools}

请严格按照以下格式回答(每次只输出一组 Thought/Action):

Thought: 你的思考过程
Action: 工具名[工具输入]

当你有了最终答案时,输出:
Thought: 我已经知道答案了
Action: Finish[最终答案]

问题:{question}
{history}"""

提示词里三处细节:

{tools} 是 describe() 填进来的,不是手写的。工具清单只有一个来源。
"每次只输出一组 Thought/Action":不加这句,模型会一口气把三步全编出来——包括它自己编造的 Observation。模型编造 Observation 是 ReAct 最危险的失败模式,因为它看起来完全正常,只是那个"工具返回值"根本没调过工具。
Finish 被写成一个"工具":它不在注册表里,是个假工具。这样"结束"和"调用工具"共用同一套格式和同一个解析器,少一条分支。

主体循环:

class ReActAgent:
    def __init__(self, max_steps: int = 5, llm=None):
        self.llm = llm or get_llm()
        self.tools = default_executor()
        self.max_steps = max_steps
        self.trace: list = []  # 每一步的 (action, arg), 让测试能断言路径

    def _parse(self, text: str):
        m = re.search(r"Action:\s*(\w+)\[(.*?)\]", text, re.S)
        return (m.group(1), m.group(2).strip()) if m else (None, None)

    def run(self, question: str, verbose: bool = True) -> str:
        history = ""
        last = None
        for step in range(1, self.max_steps + 1):
            prompt = PROMPT.format(
                tools=self.tools.describe(), question=question, history=history
            )
            reply = self.llm.think([{"role": "user", "content": prompt}])
            if verbose:
                print(f"\n--- 第{step}步 ---\n{reply}")

            action, arg = self._parse(reply)
            if action is None:
                # 不崩: 把格式错误当成一条观察喂回去
                if verbose:
                    print("Observation: 格式错误")
                history += "\nObservation: 格式错误,请严格按 Thought/Action 格式输出"
                continue

            self.trace.append((action, arg))
            if action == "Finish":
                return arg

            # 熔断: 连续两次同一个动作 = 模型卡住了, 再问下去只是烧钱
            if (action, arg) == last:
                if verbose:
                    print(f"⚠ 检测到重复动作 {action}[{arg}], 熔断")
                return f"熔断:模型连续两次执行 {action}[{arg}],未能推进"
            last = (action, arg)

            observation = self.tools.run(action, arg)
            if verbose:
                print(f"Observation: {observation}")
            history += f"\n{reply}\nObservation: {observation}"
        return f"达到最大步数({self.max_steps}),未能得出答案"


if __name__ == "__main__":
    agent = ReActAgent()
    answer = agent.run("乔布斯出生那一年到2024年,一共过去了多少天?按每年365天算。")
    print(f"\n最终答案:{answer}")
    print(f"工具调用轨迹:{agent.tools.log}")
    print(f"LLM 调用次数:{len(agent.llm.calls)}")

四个地方值得逐行看:

re.search(r"Action:\s*(\w+)\[(.*?)\]", text, re.S)。 (.*?) 是非贪婪的:参数里如果还有 ](比如 Calculate[max([1,2])]),贪婪版会一直吃到最后一个 ]。非贪婪在这里更安全,但它也解决不了嵌套方括号——这就是 function calling / JSON 输出存在的理由。正则解析永远有它扛不住的输入,你要知道边界在哪。

history 是纯字符串拼接,而且每一步都把它整个重发一遍。 这就是第 1.1 节说的"接近二次增长":第 1 步发 1 份提示词,第 2 步发提示词+1 条观察,第 3 步发提示词+2 条……你没写任何"发送历史"的代码,但它就在那里,而且是你账单的主要部分。

if (action, arg) == last 只比对上一次,不是比对全部历史。 为什么不用 if (action, arg) in 所有历史动作?因为合理的重复是存在的:先 Search[A]、再 Search[B]、再回头 Search[A] 补一个细节,这是正常的多跳推理。只有连续两次完全相同才是真卡住了。保险丝要挡住死循环,但不能挡住正常绕路。

self.trace 记 (action, arg),self.tools.log 记工具真实调用。 两者不一样:trace 包含 Finish(假工具,没进执行器),log 只有真调用。测试里断言 trace 看的是决策路径,断言 log 看的是副作用。分开记,断言才精确。

✅ Checkpoint 4:python react_agent.py,真实输出:

--- 第1步 ---
Thought: 我需要先查到乔布斯的出生年份。
Action: Search[乔布斯]
Observation: 史蒂夫·乔布斯,苹果公司联合创始人,生于1955年,逝于2011年。

--- 第2步 ---
Thought: 乔布斯生于1955年, 现在算天数。
Action: Calculate[(2024-1955)*365]
Observation: 25185

--- 第3步 ---
Thought: 我已经知道答案了
Action: Finish[25185天]

最终答案:25185天
工具调用轨迹:[('Search', '乔布斯'), ('Calculate', '(2024-1955)*365')]
LLM 调用次数:3

把这三步和替身的三条规则对照着看,你就看懂 ReAct 了:第 2 步之所以知道该去算天数,唯一的原因是第 1 步的 Observation 被拼进了 history。 替身的规则 "Action:" in p and "1955年" in p 判断的就是这件事。如果你的 agent 忘了拼 history,替身会永远返回 Search[乔布斯],然后熔断。这个替身其实是一个"history 有没有拼对"的检测器。

LLM 调用次数:3 也值得记住:3 步任务 = 3 次调用,通式是"步数 + 1"里的那个 +1 已经含在最后一步的 Finish 里了。这个数字事前是不可知的,取决于模型走几步——这正是 ReAct 最难做预算的地方。

3.5 文件五:`react_loop_demo.py` —— 亲手复现"反复横跳"(约51行)

上一节的熔断代码你看过了,但看代码和看它救了你多少钱是两件事。这个文件专门把 ReAct 的头号失败模式演一遍。

"""ReAct 最常见的失败模式: 反复横跳 —— 以及熔断怎么救它。

不用真模型也能演: 造一个"永远只会 Search[图灵]"的替身。
这正是知识库里没有答案时, 真模型的典型行为 —— 它不知道自己不知道,
于是把同一个动作再试一遍。
"""

from react_agent import ReActAgent


class StuckLLM:
    """永远输出同一个 Action 的模型。"""

    def __init__(self):
        self.calls = []

    def think(self, messages, temperature=0):
        self.calls.append({"prompt": messages[-1]["content"], "temperature": temperature})
        return "Thought: 我再查一次这个人。\nAction: Search[图灵]"


print("=== 有熔断(默认) ===")
a = ReActAgent(max_steps=5, llm=StuckLLM())
print("答案:", a.run("图灵是哪年生的?", verbose=False))
print(f"LLM 调用 {len(a.llm.calls)} 次, 工具调用 {len(a.tools.log)} 次")

print("\n=== 假设没有熔断(把 max_steps 当唯一保险) ===")


class NoFuse(ReActAgent):
    def run(self, question, verbose=False):
        history, n = "", 0
        for _ in range(self.max_steps):
            reply = self.llm.think([{"role": "user", "content": question + history}])
            action, arg = self._parse(reply)
            obs = self.tools.run(action, arg)
            history += f"\n{reply}\nObservation: {obs}"
            n += 1
        return f"达到最大步数({self.max_steps}),未能得出答案"


b = NoFuse(max_steps=5, llm=StuckLLM())
print("答案:", b.run("图灵是哪年生的?"))
print(f"LLM 调用 {len(b.llm.calls)} 次, 工具调用 {len(b.tools.log)} 次")

print(
    f"\n差距: 熔断版花了 {len(a.llm.calls)} 次调用, 无熔断版花了 {len(b.llm.calls)} 次 ——\n"
    f"省了 {len(b.llm.calls) - len(a.llm.calls)} 次。max_steps=5 时差距不大,\n"
    "但生产里 max_steps 常设 15~30, 那时候差距就是 10 倍。\n"
    "**max_steps 是安全网, 熔断才是止损线。** 两个都要。"
)

为什么"图灵"这个例子是真实的,而不是我编的极端情况: 知识库里只有乔布斯、比尔盖茨、珠穆朗玛峰、长城。问图灵,search 返回"知识库中没有关于'图灵'的信息"。真模型看到这句话时的第一反应,往往是换个关键词再查一次,甚至用同一个关键词再查一次——因为它不知道"没有"是终局还是自己没查对。这不是模型笨,这是"模型不知道自己不知道"的必然结果。

NoFuse 那个子类是反面对照组,故意写成"只靠 max_steps 兜底"。注意它连熔断判断都删了,循环里那个 n 变量也没用上——这就是绝大多数教程里 ReAct 的样子。

✅ Checkpoint 5:python react_loop_demo.py,真实输出:

=== 有熔断(默认) ===
答案: 熔断:模型连续两次执行 Search[图灵],未能推进
LLM 调用 2 次, 工具调用 1 次

=== 假设没有熔断(把 max_steps 当唯一保险) ===
答案: 达到最大步数(5),未能得出答案
LLM 调用 5 次, 工具调用 5 次

差距: 熔断版花了 2 次调用, 无熔断版花了 5 次 ——
省了 3 次。max_steps=5 时差距不大,
但生产里 max_steps 常设 15~30, 那时候差距就是 10 倍。
**max_steps 是安全网, 熔断才是止损线。** 两个都要。

三个数字请对着看:

熔断版工具只调了 1 次,LLM 调了 2 次。 第 1 次调用产出 Search[图灵],执行;第 2 次调用又产出 Search[图灵],这次不执行工具就直接熔断了——所以工具是 1 次而 LLM 是 2 次。保险丝在花钱之前拦下了。
无熔断版 5 次 LLM + 5 次工具,并且答案还是"未能得出"。 后面 3 次调用产出了完全相同的内容,是纯浪费。
这里的 5 是我调小了的。 生产里 max_steps 常设 15~30(多跳任务确实需要),那时同样一次卡死会烧掉 15~30 次调用,而熔断永远只花 2 次。保险丝的价值和 max_steps 成正比,所以 max_steps 越大,你越需要熔断。

3.6 文件六:`plan_solve_agent.py` —— 范式二:Plan-and-Solve(约86行)

"""范式二: Plan-and-Solve —— 先谋后动。

两个角色: Planner 一次性出完整计划, Executor 逐条执行并把上一步的
结果带进下一步。和 ReAct 的手感差别就一句话:
**计划在开局定死, 执行中不再改。**

比原书多了一样: Planner 的输出要**校验**。模型偶尔会拆出 0 步或 30 步,
不校验就会一路跑偏(0 步时 Executor 直接返回空字符串, 而且不报错)。
"""

import json
import re

from llm import get_llm

PLANNER_PROMPT = """你是一个规划专家。请把下面的问题拆解成有序的、可独立执行的步骤。
只输出 JSON 数组,不要任何多余文字,例如:["步骤1", "步骤2", "步骤3"]

问题:{question}"""

EXECUTOR_PROMPT = """请完成计划中的当前步骤,只输出这一步的结果,不要多余解释。

原始问题:{question}
完整计划:{plan}
之前步骤的结果:
{context}

当前步骤:{step}"""

EXECUTOR_PROMPT 里的四个变量,每一个都在解决一个具体的失败:

question(原始问题):不给它,执行到第 3 步时模型已经忘了这题在问什么,会答一个孤立的算式。
plan(完整计划):让模型知道"我在整件事的哪个位置、后面还有什么",于是它不会在第 1 步就急着给最终答案。
context(之前步骤的结果):这是整个范式的命门,下面细说。
step(当前步骤):唯一真正的指令。

class Planner:
    def __init__(self, llm):
        self.llm = llm

    def plan(self, question: str) -> list:
        reply = self.llm.think(
            [{"role": "user", "content": PLANNER_PROMPT.format(question=question)}]
        )
        m = re.search(r"\[.*\]", reply, re.S)  # 从回复里抠出 JSON 数组
        if not m:
            raise ValueError(f"规划器没有输出 JSON 数组, 原始回复: {reply[:120]}")
        plan = json.loads(m.group(0))
        # 宁可失败也不要默默跑偏: 0 步会让 Executor 返回空字符串且不报错
        if not (1 <= len(plan) <= 10):
            raise ValueError(f"计划步数异常({len(plan)} 步): {plan}")
        return plan

re.search(r"\[.*\]", reply, re.S) 而不是直接 json.loads(reply)。 因为模型很爱在 JSON 前面加一句"好的!我来帮你规划一下",或者用 ```json 把它裹起来。先抠再解析,能扛住这两种最常见的抖动。注意这里用的是贪婪 .*——和 ReAct 的解析相反,因为我们要的是从第一个 [ 到最后一个 ] 的完整数组,里面允许有嵌套。

if not (1 <= len(plan) <= 10): raise 是原书没有的,也是这个文件最重要的五行。下限 1 挡的是最阴的一种失败:计划 0 步。 0 步的时候 Executor 的 for 循环一次都不进,result 保持初始值 "",函数返回空字符串,不报错、不警告、看起来就像"任务完成了但答案是空的"。上限 10 挡的是模型把简单问题拆成 30 步——那不是规划,那是烧钱。

为什么选 raise 而不是"返回一个默认计划"? 因为规划错了,后面每一步都是错的,而且执行完你才发现——那时你已经花了 30 次调用。fail closed:错在第一步,就在第一步停。

class Executor:
    def __init__(self, llm):
        self.llm = llm

    def execute(self, question: str, plan: list, verbose: bool = True) -> str:
        context = ""
        result = ""
        for i, step in enumerate(plan, 1):
            prompt = EXECUTOR_PROMPT.format(
                question=question, plan=plan, context=context or "(无)", step=step
            )
            result = self.llm.think([{"role": "user", "content": prompt}])
            if verbose:
                print(f"\n--- 执行步骤{i}:{step} ---\n{result}")
            context += f"步骤{i}({step})的结果:{result}\n"
        return result

context += f"步骤{i}({step})的结果:{result}\n" 这一行就是 Plan-and-Solve 的全部价值。 把它删掉,代码照样跑、照样不报错、照样输出一个答案——但这个范式已经死了,退化成"独立跑 N 个不相干的子任务,取最后一个的输出"。第 2 步会不知道午餐后剩 3 个,只能瞎猜。

这种"删掉不报错、只是悄悄答错"的代码,是 agent 项目里最难发现的 bug。 所以 3.10 节有一个测试专门守着这一行:

assert "3 + 6 = 9" in b.llm.calls[-1]["prompt"], "第 2 步的结果没有进到第 3 步的上下文里"

context or "(无)":第 1 步时 context 是空字符串,填进提示词会变成"之前步骤的结果:\n\n当前步骤:…"。模型看到一片空白容易犯迷糊(它可能以为是自己没收到),写明"(无)"是明确的信息。这和 search 找不到时返回"知识库中没有"是同一条纪律:告诉模型"空",不要给它空。

class PlanAndSolveAgent:
    def __init__(self, llm=None):
        llm = llm or get_llm()
        self.llm = llm
        self.planner = Planner(llm)
        self.executor = Executor(llm)
        self.plan: list = []

    def run(self, question: str, verbose: bool = True) -> str:
        self.plan = self.planner.plan(question)
        if verbose:
            print(f"生成的计划:{json.dumps(self.plan, ensure_ascii=False, indent=2)}")
        return self.executor.execute(question, self.plan, verbose=verbose)


if __name__ == "__main__":
    agent = PlanAndSolveAgent()
    answer = agent.run("食堂有23个苹果,午餐用掉20个,又买来6个,现在有多少个苹果?请一步步推理。")
    print(f"\n最终答案:{answer}")
    print(f"LLM 调用次数:{len(agent.llm.calls)}(1 次规划 + {len(agent.plan)} 次执行)")

Planner 和 Executor 共用同一个 llm 实例,这样 agent.llm.calls 里能看到规划和执行的全部调用。如果各建一个实例,你就得加起来数——而且很容易漏。"一个 agent 一本账"比"每个组件一本账"好用。

✅ Checkpoint 6:python plan_solve_agent.py,真实输出:

生成的计划:[
  "计算午餐后剩下的苹果数",
  "加上新买的6个苹果",
  "给出最终数量"
]

--- 执行步骤1:计算午餐后剩下的苹果数 ---
23 - 20 = 3, 午餐后剩 3 个。

--- 执行步骤2:加上新买的6个苹果 ---
3 + 6 = 9。

--- 执行步骤3:给出最终数量 ---
现在有 9 个苹果。

最终答案:现在有 9 个苹果。
LLM 调用次数:4(1 次规划 + 3 次执行)

和 Checkpoint 4 的 ReAct 对比着看,两个范式的性格差别就出来了:

	ReAct(Checkpoint 4)	Plan-and-Solve(Checkpoint 6)
第一次调用产出什么	一个动作 `Search[乔布斯]`	一整份计划(3 步)
你什么时候知道要花几次调用	永远不知道,跑完才知道	第 1 次调用之后就知道
中途工具报错	下一步就能改道	计划不变,硬着头皮往下走
调用次数	3(步数,末步是 Finish)	4(1 + 3)

注意 Plan-and-Solve 这次没调用任何工具。 它不是不能用工具(把 Executor 的每一步换成一个小 ReAct 就行,这就是第 1.5 节说的组合),而是这个范式的原始形态只负责"分步骤思考"。所以它天生适合数学题、固定流程,天生不适合"我不知道答案在哪、得边找边看"的探索任务。

3.7 文件七:`reflection_agent.py` —— 范式三:Reflection(约94行)

"""范式三: Reflection —— 做完自己挑毛病再改。

三步循环: 执行 → 反思 → 带着意见重做。出口有两个:
评审说"通过", 或者达到 max_iterations。**第二个出口是硬要求** ——
一个刻薄的评审能让你无限重写, 每一轮都是真金白银。
"""

from llm import get_llm

EXECUTE_PROMPT = """你是一位Python程序员,请完成以下任务,只输出代码(可带注释):
任务:{task}
{feedback}"""

REFLECT_PROMPT = """你是一位严格的代码评审员。请审查下面的代码:效率、边界情况、可读性。
如果代码已经足够好,只回答两个字:通过
否则,列出需要改进的具体问题。

任务:{task}
代码:
{code}"""

两个提示词的设计各有一处讲究:

EXECUTE_PROMPT 里的 {feedback} 第一轮是空字符串。 于是同一个提示词模板既是"从零写"又是"带着意见改",不需要两套模板。少一套模板就少一个会不同步的地方。

REFLECT_PROMPT 规定"足够好就只回答两个字:通过"。 这是给程序留一个可判定的信号。如果让评审自由发挥,你就得去理解"这段代码写得不错,不过如果非要说的话……"到底算通过还是不通过——那是没有尽头的。要让 LLM 的输出参与控制流,就必须给它一个窄到能用 in 判断的出口。

class Memory:
    """记录执行与反思的轨迹。

    留着它不只是为了打印: 达到轮数上限时要**交出最后一版代码**,
    而不是交出一句"失败了"。半成品比空手强。
    """

    def __init__(self):
        self.records: list = []

    def add(self, kind: str, content: str):
        self.records.append({"kind": kind, "content": content})

    def last_code(self):
        for r in reversed(self.records):
            if r["kind"] == "code":
                return r["content"]
        return None

    def rounds(self) -> int:
        return sum(1 for r in self.records if r["kind"] == "code")

Memory 不是为了"记忆"存在的,是为了那个失败出口存在的。 想清楚这件事:评审连着三轮都不满意,max_iterations 到了,你交什么?

交 None 或抛异常:用户等了 6 次 LLM 调用,拿到一句"失败"。你已经把钱花完了,却把成果扔了。
交最后一版代码 + 一句"未通过评审":用户至少拿到了改了 3 轮的代码,可以自己判断。

rounds() 数的是 code 记录的条数,不是 records 总数。因为一轮 = 一条 code + 一条 review,直接数总数会翻倍。这个方法存在的意义是让测试能断言"到底改了几版"——assert c.memory.rounds() == 2 比 assert "sieve" in code 更能说明"第一版真的被打回了"。

class ReflectionAgent:
    def __init__(self, max_iterations: int = 3, llm=None):
        self.llm = llm or get_llm()
        self.memory = Memory()
        self.max_iterations = max_iterations
        self.passed = False

    def run(self, task: str, verbose: bool = True) -> str:
        feedback = ""
        for i in range(1, self.max_iterations + 1):
            if verbose:
                print(f"\n===== 第{i}轮:执行 =====")
            code = self.llm.think(
                [{"role": "user", "content": EXECUTE_PROMPT.format(task=task, feedback=feedback)}]
            )
            self.memory.add("code", code)
            if verbose:
                print(code)

            if verbose:
                print(f"\n===== 第{i}轮:反思 =====")
            review = self.llm.think(
                [{"role": "user", "content": REFLECT_PROMPT.format(task=task, code=code)}]
            )
            self.memory.add("review", review)
            if verbose:
                print(review)

            if "通过" in review[:10]:
                self.passed = True
                if verbose:
                    print(f"\n✅ 第{i}轮评审通过,任务完成")
                return code
            feedback = f"上一版代码的评审意见如下,请改进:\n{review}"

        if verbose:
            print(f"\n⚠ 达到轮数上限({self.max_iterations}),交出最后一版")
        return self.memory.last_code()

三处要停下来看:

if "通过" in review[:10] —— 为什么切前 10 个字符? 因为不切的话,评审说"这段代码通过不了效率审查"也会被判成通过。用子串判断 LLM 输出时,一定要把判断范围收窄到你在提示词里约定的那个位置。 这个 [:10] 是个补丁,不优雅,但它挡住的是一整类"关键词出现在否定句里"的误判。更正规的做法是让模型输出 JSON {"passed": true},那是第二节说的第 3 条路。

self.passed = False 是一个独立的标志位,不是从返回值推出来的。 因为两个出口都返回代码字符串,光看返回值分不清"通过了"还是"改到上限了"。调用方(和测试)需要这个区别:assert not c.passed 是 3.10 节那个"永不满意的评审"测试的核心断言。当一个函数有多个出口而返回类型相同时,你需要一个额外的状态位来区分它们。

feedback = f"上一版代码的评审意见如下,请改进:\n{review}" 是这个范式的命门,和 Plan-and-Solve 的 context += 一模一样的地位。删了它,循环照样跑三轮,每轮都从零重写同一份暴力版代码,评审每轮都提同样的意见,你付了 6 次调用的钱得到 1 次调用的结果。而且不报错。

注意这里是 = 不是 +=:只带上最新一轮的意见,不累积历史意见。 累积的话提示词会越来越长,而且旧意见(已经改好的那些)会干扰模型。这是一个取舍——累积能防止"改了 A 又把 B 改坏"的来回震荡,不累积则更便宜更聚焦。本章选后者,因为轮数上限只有 3。

if __name__ == "__main__":
    agent = ReflectionAgent()
    final = agent.run("写一个函数 find_primes(n),找出1到n之间的所有素数,要求高效。")
    print(f"\n最终代码:\n{final}")
    print(
        f"\n统计:改了 {agent.memory.rounds()} 版, "
        f"LLM 调用 {len(agent.llm.calls)} 次(每轮 2 次), "
        f"评审通过={agent.passed}"
    )

✅ Checkpoint 7:python reflection_agent.py,真实输出:

===== 第1轮:执行 =====
def find_primes(n):
    primes = []
    for x in range(2, n + 1):
        for d in range(2, x):
            if x % d == 0:
                break
        else:
            primes.append(x)
    return primes

===== 第1轮:反思 =====
1. 效率问题: 对每个数都从 2 试除到 n-1, 复杂度 O(n^2), n=10^6 时无法接受
2. 建议改用埃拉托斯特尼筛法(sieve of Eratosthenes)
3. 边界情况: n < 2 时应返回空列表, 当前实现未显式处理

===== 第2轮:执行 =====
def find_primes(n):
    """埃氏筛: O(n log log n)"""
    if n < 2:
        return []
    sieve = [True] * (n + 1)
    sieve[0] = sieve[1] = False
    for x in range(2, int(n**0.5) + 1):
        if sieve[x]:
            sieve[x * x :: x] = [False] * len(sieve[x * x :: x])
    return [i for i, ok in enumerate(sieve) if ok]

===== 第2轮:反思 =====
通过

✅ 第2轮评审通过,任务完成

最终代码:
def find_primes(n):
    """埃氏筛: O(n log log n)"""
    if n < 2:
        return []
    sieve = [True] * (n + 1)
    sieve[0] = sieve[1] = False
    for x in range(2, int(n**0.5) + 1):
        if sieve[x]:
            sieve[x * x :: x] = [False] * len(sieve[x * x :: x])
    return [i for i, ok in enumerate(sieve) if ok]

统计:改了 2 版, LLM 调用 4 次(每轮 2 次), 评审通过=True

这个输出看起来非常漂亮:暴力版 → 三条具体意见 → 筛法版 → 通过。漂亮得让人不想再问一句。

但你必须再问一句:筛法版真的对吗?真的更快吗? 你现在掌握的全部证据只有一句"通过"——那是同一个模型说的。让写代码的人自己评审自己的代码,这在人类团队里叫什么,你心里清楚。 下一节就把这两版代码都真的跑起来。

3.8 文件八:`verify_reflection.py` —— 不听评审的,自己验(约61行)

这一节是整章我最想让你带走的习惯。 前面七个文件教的是怎么搭 agent,这一个教的是怎么不被自己的 agent 骗。

"""反思到底改对了没有? —— 别信评审的话, 自己跑一遍。

Reflection 最大的风险是**你只看到"评审说通过了", 没看到代码变好了**。
这个脚本把两版代码都真的执行一遍: 比对结果是否一致, 再比耗时。
"""

import timeit

BRUTE = """
def find_primes(n):
    primes = []
    for x in range(2, n + 1):
        for d in range(2, x):
            if x % d == 0:
                break
        else:
            primes.append(x)
    return primes
"""

SIEVE = """
def find_primes(n):
    if n < 2:
        return []
    sieve = [True] * (n + 1)
    sieve[0] = sieve[1] = False
    for x in range(2, int(n**0.5) + 1):
        if sieve[x]:
            sieve[x * x :: x] = [False] * len(sieve[x * x :: x])
    return [i for i, ok in enumerate(sieve) if ok]
"""

ns = {}
exec(BRUTE, ns)  # noqa: S102  ← 这里 exec 的是我们自己写的常量, 不是模型输出
brute = ns["find_primes"]
ns2 = {}
exec(SIEVE, ns2)  # noqa: S102
sieve = ns2["find_primes"]

等一下——3.3 节刚花 30 行证明"不能执行模型生成的字符串",这里怎么就 exec 了?

这个矛盾必须说清楚,因为它是安全判断的核心:危险的不是 exec 这个函数,是"字符串从哪来"。 这里 exec 的是我写在文件里的两个常量,和直接把函数定义写进文件完全等价。而 3.3 节那个 payload 是模型生成的、提示词能影响的、用户能间接影响的。同一个函数,一个安全一个是远程代码执行,区别只在数据来源。

那如果我真想跑模型生成的代码怎么办? 唯一的正解是换执行环境,不是换函数:容器、gVisor、Firecracker、或者 RestrictedPython 这类受限解释器。在同一个进程里"小心地"执行不可信代码,是做不到的——这一点在 docs/2 里讲代码执行沙箱时会展开。

print("=== 1. 正确性: 两版结果必须完全一致 ===")
for n in [0, 1, 2, 3, 10, 100, 1000]:
    a, b = brute(n), sieve(n)
    print(f"  n={n:<5} 结果一致={a == b}  素数个数={len(b)}")
assert all(brute(n) == sieve(n) for n in range(0, 200)), "两版结果不一致, 反思改错了!"
print("  n=0..199 全部一致 ✅")

print("\n=== 2. 边界: 评审专门提到的 n<2 ===")
for n in [-5, 0, 1]:
    print(f"  find_primes({n}) -> 暴力版 {brute(n)} / 筛法版 {sieve(n)}")

print("\n=== 3. 效率: 评审说 O(n^2) 不能接受, 验一下 ===")
for n in [1_000, 5_000, 20_000]:
    t1 = timeit.timeit(lambda: brute(n), number=3) / 3
    t2 = timeit.timeit(lambda: sieve(n), number=3) / 3
    print(f"  n={n:<7} 暴力 {t1 * 1000:8.2f} ms   筛法 {t2 * 1000:6.2f} ms   快 {t1 / t2:6.1f} 倍")

验证的顺序是有讲究的:先验对不对,再验快不快。 一个错的快版本毫无价值,而"改快了但改错了"是优化里最常见的事故。所以正确性那一段用了 assert(不一致就直接崩),效率那一段只是打印数字给你看。

assert all(... for n in range(0, 200)) 覆盖 200 个输入,而不是只看上面打印的那 7 个。 打印是给人看的,断言是给机器看的。给人看的样本要少而典型,给机器看的要多而无聊。

for n in [-5, 0, 1] 专门测评审提到的那条意见。 评审说"n < 2 时应返回空列表",那就去验这句话到底有没有落实。把评审的每一条具体意见都变成一个检查项,这是防止"改了个样子"的最直接办法。

✅ Checkpoint 8:python verify_reflection.py,真实输出(这一节的时间数字取决于你的机器,倍数会不同,但数量级会一致):

=== 1. 正确性: 两版结果必须完全一致 ===
  n=0     结果一致=True  素数个数=0
  n=1     结果一致=True  素数个数=0
  n=2     结果一致=True  素数个数=1
  n=3     结果一致=True  素数个数=2
  n=10    结果一致=True  素数个数=4
  n=100   结果一致=True  素数个数=25
  n=1000  结果一致=True  素数个数=168
  n=0..199 全部一致 ✅

=== 2. 边界: 评审专门提到的 n<2 ===
  find_primes(-5) -> 暴力版 [] / 筛法版 []
  find_primes(0) -> 暴力版 [] / 筛法版 []
  find_primes(1) -> 暴力版 [] / 筛法版 []

=== 3. 效率: 评审说 O(n^2) 不能接受, 验一下 ===
  n=1000    暴力     1.74 ms   筛法   0.04 ms   快   47.4 倍
  n=5000    暴力    39.60 ms   筛法   0.18 ms   快  218.0 倍
  n=20000   暴力   552.37 ms   筛法   0.73 ms   快  754.5 倍

结论: 这一轮反思是**真的有收益**的 —— 结果一致、边界补上了、n=20000 时快了两个量级。
但请注意这件事是**我们自己验的**, 不是评审说的。
评审只会说'建议改用筛法', 它不会告诉你改完到底快了几倍、结果还对不对。

现在你才有资格说"这一轮反思是有收益的"。 三条证据都拿到了:结果一致(200 个输入)、边界补上了(-5/0/1 都返回空列表)、快了两到三个数量级。

注意那一列倍数怎么涨的:47 → 218 → 754。 n 翻 20 倍,倍数涨了 16 倍——这就是 O(n²) 和 O(n log log n) 之间差距的形状:它不是一个固定倍数,是一个随规模张开的口子。 所以"快 754 倍"这个数字本身不重要(换台机器就变),重要的是它随 n 增长。评审说的"n=10^6 时无法接受"由此得到了印证:按这个趋势,n 到一百万时暴力版要跑几分钟。

同时请注意一件反面的事:如果评审这一轮建议错了,上面的代码会原样把它抓出来。 假设它建议了一个"更聪明"但其实更慢的写法,第 3 段的倍数就会小于 1;假设它把 n=2 的边界改错了,第 1 段的 assert 直接崩。这个文件的价值不在于它这次报了好消息,在于它有能力报坏消息。 只会报好消息的验证不是验证。

3.9 文件九:`compare_cost.py` —— 三个范式的成本账(约38行)

"""三个范式跑同一个任务, 把账算清楚。

面试里"Reflection 一定更好吗"这种问题, 答案不该是感觉, 该是这张表。
用替身跑, 所以 LLM 调用次数是**确定的**, 可以直接当成本代理。
"""

from plan_solve_agent import PlanAndSolveAgent
from react_agent import ReActAgent
from reflection_agent import ReflectionAgent

rows = []

a = ReActAgent()
ans = a.run("乔布斯出生那一年到2024年,一共过去了多少天?按每年365天算。", verbose=False)
rows.append(("ReAct", len(a.llm.calls), len(a.tools.log), ans[:24]))

b = PlanAndSolveAgent()
ans = b.run("食堂有23个苹果,午餐用掉20个,又买来6个,现在有多少个苹果?请一步步推理。", verbose=False)
rows.append(("Plan-and-Solve", len(b.llm.calls), 0, ans[:24]))

c = ReflectionAgent()
ans = c.run("写一个函数 find_primes(n),找出1到n之间的所有素数,要求高效。", verbose=False)
rows.append(("Reflection", len(c.llm.calls), 0, f"{c.memory.rounds()} 版代码, 通过={c.passed}"))

print(f"{'范式':<16}{'LLM调用':>8}{'工具调用':>9}   结果")
print("-" * 68)
for name, n_llm, n_tool, res in rows:
    print(f"{name:<16}{n_llm:>8}{n_tool:>9}   {res}")

这个文件短,但它能存在完全靠 3.1 节那个替身。想想:如果用真模型,三个范式每次跑出来的调用次数都不一样(模型有时 3 步有时 5 步),这张表就没有意义了。替身把"次数"变成了确定值,于是成本可以被断言、可以被写进文档、可以被 CI 监控回归。

len(a.llm.calls) 和 len(a.tools.log) 分别来自 3.1 和 3.2 节加的那两行记录。当初加的时候只是"顺手记一下",现在它变成了一张成本表。 这就是给关键组件加调用记录的复利。

✅ Checkpoint 9:python compare_cost.py,真实输出:

范式                 LLM调用     工具调用   结果
--------------------------------------------------------------------
ReAct                  3        2   25185天
Plan-and-Solve         4        0   现在有 9 个苹果。
Reflection             4        0   2 版代码, 通过=True

读法:
  ReAct 的调用次数 = 步数 + 1, **由任务难度决定, 事前不可知** —— 这是它最难做预算的地方
  Plan-and-Solve = 1(规划) + 步数, 计划一出来预算就确定了
  Reflection = 轮数 × 2, 一轮就是两次调用, 所以它是三者里单位任务最贵的

注意这张表只算了**次数**, 没算 token。ReAct 每一步都要把完整 history
重新发一遍, 所以它的 token 消耗是**随步数二次增长**的 —— 5 步的 ReAct
比 5 次独立调用贵得多。真要控成本, 得看 token 不是看次数。

三个数字看着差不多(3、4、4),但这张表最重要的信息不在数字里,在数字的"可预测性"里:

ReAct 的 3 是跑完才知道的。 换一个需要五跳的问题,它就是 6;遇到卡住,它是熔断的 2 或者 max_steps 的 15。你没法给这类任务报价。
Plan-and-Solve 的 4 在第 1 次调用之后就确定了。 计划出来是 3 步,那就是 1+3。你可以在规划完成之后决定"这个计划太贵,不执行了"——这是一个真实可用的成本闸门,ReAct 给不了你这个机会。
Reflection 的 4 = 2 轮 × 2 次。 它的上限是 max_iterations × 2,写死在代码里,是三者里唯一一个上限完全由你决定的。

还有一件这张表故意不告诉你的事:token。 次数只是成本的粗代理。ReAct 第 3 步发出去的提示词包含了前两步的全部内容,所以三步的总 token 大约是 1+2+3=6 个"单位",而三次独立调用是 3 个单位。步数到 10 时,是 55 对 10。面试里能说出"ReAct 的 token 是二次增长,因为历史每步重发"这句话的人,和只会说"ReAct 比较贵"的人,是两个层次。

3.10 文件十:`test_paradigms.py` —— 19 个测试(约189行)

给 agent 写测试,难点从来不是技术,是想清楚"测什么"。 模型的输出不确定,你不能断言"它答得对";但控制流是确定的,你完全可以断言"它按什么顺序走、走了几步、保险丝有没有响"。

"""给三个范式写测试。

测的**不是模型答得对不对**, 是三件事:
  1. 控制流走对了没有(ReAct 的动作顺序、Plan 的步数、Reflection 的轮数)
  2. 保险丝有没有用(熔断、max_steps、max_iterations)
  3. 工具的安全边界(白名单挡住了什么)

因为有 ScriptedLLM, 这些测试不需要网络、不花钱、每次结果一样。
运行: python -m pytest test_paradigms.py -q   (或 python test_paradigms.py)
"""

import pytest

from llm import ScriptedLLM
from plan_solve_agent import PlanAndSolveAgent
from react_agent import ReActAgent
from reflection_agent import ReflectionAgent
from tools import calculator, default_executor


# ---------------- 工具 ----------------
def test_计算器只认数学表达式():
    assert calculator("(2024-1955)*365") == "25185"
    assert calculator("2**10") == "1024"


@pytest.mark.parametrize(
    "payload",
    [
        "__import__('os').system('echo x')",
        "open('/etc/passwd').read()",
        "[].__class__.__base__",
        "'a'*10**9",
    ],
)
def test_计算器挡住非数学输入且不抛异常(payload):
    """工具必须返回字符串, 绝不能把异常抛进 agent 循环。"""
    out = calculator(payload)
    assert out.startswith("计算失败"), out


def test_未注册的工具返回可读提示():
    ex = default_executor()
    out = ex.run("Translate", "hello")
    assert "未知工具" in out and "Search" in out

那四个 payload 是四类不同的攻击,不是四个变体: 执行命令、读文件、爬类继承链(Python 沙箱逃逸的经典起手式)、以及不调用任何函数的资源耗尽('a'*10**9)。最后一个最容易被漏掉——很多人的白名单只挡 Call 节点,于是这个 payload 顺利通过,然后进程被一个 1GB 的字符串撑爆。

断言写的是 out.startswith("计算失败") 而不是 pytest.raises。 这个区别就是 3.2 节那条纪律的测试化表达:工具挡住恶意输入的正确方式是"返回一句失败",不是"抛异常"。 如果哪天有人把 calculator 里的 except 删了,这个测试会红——它守的不是"安全",是"安全的实现方式"。

# ---------------- ReAct ----------------
def test_react_按_搜索_计算_收尾_的顺序走():
    a = ReActAgent(llm=ScriptedLLM())
    ans = a.run("乔布斯出生那一年到2024年,一共过去了多少天?按每年365天算。", verbose=False)
    assert ans == "25185天"
    assert [x[0] for x in a.trace] == ["Search", "Calculate", "Finish"]
    assert len(a.llm.calls) == 3, "3 步 = 3 次 LLM 调用"


def test_react_重复动作会熔断而不是烧到上限():
    class Stuck:
        def __init__(self):
            self.calls = []

        def think(self, messages, temperature=0):
            self.calls.append(1)
            return "Thought: 再查一次\nAction: Search[图灵]"

    a = ReActAgent(max_steps=20, llm=Stuck())
    ans = a.run("图灵是哪年生的?", verbose=False)
    assert "熔断" in ans
    assert len(a.llm.calls) == 2, f"熔断应在第 2 次调用后触发, 实际 {len(a.llm.calls)}"

assert [x[0] for x in a.trace] == ["Search", "Calculate", "Finish"] 是本章最典型的一条 agent 断言。 它不关心模型说了什么,只关心走了哪条路。这种断言的好处:换个模型、换个措辞、换个提示词,只要控制流对,测试就绿;控制流一错,立刻红。

max_steps=20 配合 assert len(a.llm.calls) == 2 是这个测试的精髓。 把上限设到 20,然后断言"只花了 2 次"——如果熔断失效,这个数字会变成 20,测试用成本把 bug 抓出来。断言调用次数,是 agent 测试里最便宜也最有效的一种断言:它同时守住了正确性和账单。

def test_react_格式错误不崩_而是喂回去再给一次机会():
    class Messy:
        def __init__(self):
            self.calls = []

        def think(self, messages, temperature=0):
            self.calls.append(1)
            # 第一次胡说, 之后正常收尾
            if len(self.calls) == 1:
                return "我觉得应该是 25185 天吧。"
            return "Thought: 我已经知道答案了\nAction: Finish[25185天]"

    a = ReActAgent(llm=Messy())
    assert a.run("随便问问", verbose=False) == "25185天"
    assert len(a.llm.calls) == 2


def test_react_达到上限会明确说没答出来():
    class Wanderer:
        def __init__(self):
            self.calls = []
            self.i = 0

        def think(self, messages, temperature=0):
            self.calls.append(1)
            self.i += 1
            return f"Thought: 试试\nAction: Search[目标{self.i}]"  # 每次都不一样, 绕过熔断

    a = ReActAgent(max_steps=3, llm=Wanderer())
    ans = a.run("一个查不到的问题", verbose=False)
    assert "达到最大步数(3)" in ans
    assert len(a.llm.calls) == 3

这两个类是"专用假模型",和 ScriptedLLM 分工不同。 ScriptedLLM 演的是"正常模型",这两个演的是具体的病:Messy 得的是格式病,Wanderer 得的是绕路病。给每一种失败模式造一个专门的假模型,是测 agent 保险丝的标准手法。

Wanderer 每次返回不同的目标,是为了故意绕过熔断。 因为这个测试要验的是另一条保险丝(max_steps)。两条保险丝要分别测,不能让一条挡住另一条的测试路径——否则你永远不知道 max_steps 那条到底通不通。这也解释了为什么两条都需要:熔断挡"卡在同一步",max_steps 挡"一直换着步走却不收敛"。

# ---------------- Plan-and-Solve ----------------
def test_plan_solve_调用次数等于_1加步数():
    b = PlanAndSolveAgent(llm=ScriptedLLM())
    ans = b.run("食堂有23个苹果,午餐用掉20个,又买来6个,现在有多少个苹果?请一步步推理。", verbose=False)
    assert "9" in ans
    assert len(b.plan) == 3
    assert len(b.llm.calls) == 1 + len(b.plan)


def test_plan_solve_把上一步结果带进下一步():
    """这是 Plan-and-Solve 和"独立跑 N 个子任务"的唯一区别, 必须有测试守着。"""
    b = PlanAndSolveAgent(llm=ScriptedLLM())
    b.run("食堂有23个苹果,午餐用掉20个,又买来6个,现在有多少个苹果?请一步步推理。", verbose=False)
    第三次执行的提示词 = b.llm.calls[-1]["prompt"]
    assert "3 + 6 = 9" in 第三次执行的提示词, "第 2 步的结果没有进到第 3 步的上下文里"

assert len(b.llm.calls) == 1 + len(b.plan) 把范式的成本公式写成了断言。 这一条会在两种情况下红:有人在循环里多加了一次调用(成本回归),或者有人漏执行了某一步(逻辑 bug)。把"这个范式应该花几次调用"写成测试,你的成本模型就永远不会和代码脱节。

test_plan_solve_把上一步结果带进下一步 是这一组里最不能删的测试。 它守的是 3.6 节那一行 context +=。回想一下:删掉那行,程序不报错、有输出、看起来正常。只有这个测试会红。 这就是"给 agent 写测试"真正的价值——它抓的不是崩溃,是静默降级。

@pytest.mark.parametrize("bad", ["[]", '["1","2","3","4","5","6","7","8","9","10","11"]'])
def test_plan_步数异常直接失败而不是默默跑偏(bad):
    class BadPlanner(ScriptedLLM):
        def think(self, messages, temperature=0):
            if "规划专家" in messages[-1]["content"]:
                return bad
            return super().think(messages, temperature)

    with pytest.raises(ValueError, match="计划步数异常"):
        PlanAndSolveAgent(llm=BadPlanner()).run("随便", verbose=False)


def test_plan_不是_json_也会明确报错():
    class Chatty(ScriptedLLM):
        def think(self, messages, temperature=0):
            if "规划专家" in messages[-1]["content"]:
                return "好的!我来帮你规划一下这个问题。"
            return super().think(messages, temperature)

    with pytest.raises(ValueError, match="没有输出 JSON 数组"):
        PlanAndSolveAgent(llm=Chatty()).run("随便", verbose=False)

注意这两个假模型是 ScriptedLLM 的子类,只改一条规则,其余 super() 兜着。 这个写法很值得学:造"病模型"时只改你要测的那一处,别整个重写。 重写的话,你测的就不是"规划器坏了会怎样",而是"一个全新的模型会怎样"——那里面藏着别的变量。

Chatty 那个回复"好的!我来帮你规划一下这个问题。"是真模型每天都在干的事。 它不是恶意输入,是最正常的模型行为。这就是为什么 Planner.plan 里要先 re.search 再 json.loads,以及为什么找不到数组时要给一条说清原始回复的错误(原始回复: {reply[:120]})——排查这类问题时,你第一个想看的就是模型到底说了什么。

# ---------------- Reflection ----------------
def test_reflection_评审不过会真的改一版():
    c = ReflectionAgent(llm=ScriptedLLM())
    code = c.run("写一个函数 find_primes(n),找出1到n之间的所有素数,要求高效。", verbose=False)
    assert c.memory.rounds() == 2, "第一版被打回, 应该有第二版"
    assert c.passed
    assert "sieve" in code, "第二版应该真的换成筛法, 而不是原地重跑"
    assert len(c.llm.calls) == 4, "每轮 2 次调用"


def test_reflection_改进意见真的进了下一轮的提示词():
    c = ReflectionAgent(llm=ScriptedLLM())
    c.run("写一个函数 find_primes(n),找出1到n之间的所有素数,要求高效。", verbose=False)
    第二轮执行提示词 = c.llm.calls[2]["prompt"]
    assert "评审意见" in 第二轮执行提示词
    assert "埃拉托斯特尼" in 第二轮执行提示词, "评审的具体意见必须带进去, 否则等于重跑"

c.llm.calls[2] 为什么是第二轮的执行? 数一下:calls[0] 第一轮执行、calls[1] 第一轮反思、calls[2] 第二轮执行。这种"按下标定位某次调用"的断言,只有在你记录了每次调用的提示词时才写得出来。 又是 self.calls 的复利。

两条断言分工明确:"评审意见" in ... 验的是"拼接的动作发生了","埃拉托斯特尼" in ... 验的是"拼进去的是评审的具体内容"。 少了第二条,一个偷懒的实现(feedback = "上一版代码的评审意见如下,请改进:\n(略)")也能通过。断言要盯住信息本身,而不只是包装信息的那个壳。

def test_reflection_循环有硬性出口():
    """造一个永远不满意的评审 —— 出口写错的话这个测试会挂住, 不是失败。"""

    class NeverHappy(ScriptedLLM):
        def think(self, messages, temperature=0):
            if "代码评审员" in messages[-1]["content"]:
                self.calls.append({"prompt": "review", "temperature": temperature})
                return "1. 还是不够好\n2. 再改"
            return super().think(messages, temperature)

    c = ReflectionAgent(max_iterations=3, llm=NeverHappy())
    code = c.run("写一个函数 find_primes(n),找出1到n之间的所有素数,要求高效。", verbose=False)
    assert not c.passed
    assert c.memory.rounds() == 3
    assert code, "达到上限也要交出最后一版代码, 不能返回 None"


def test_替身按内容而不是按次数应答():
    """同一个提示词问两次, 答案必须一样 —— 计数器型替身在这里会露馅。"""
    llm = ScriptedLLM()
    p = [{"role": "user", "content": "你是一位Python程序员,请完成以下任务:任务:素数"}]
    assert llm.think(p) == llm.think(p)


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))

NeverHappy 这个测试的文档字符串请读两遍:"出口写错的话这个测试会挂住,不是失败。" 这是一类特殊的测试——它验的是程序会终止。如果 max_iterations 那个出口写错了(比如 while True 忘了计数),这个测试不会红,它会永远跑不完,pytest 挂在那里。"挂住"也是一种失败信号,只是它长得不像失败。 遇到 CI 卡死,先去看有没有这类测试。

assert code 那一条守的是 3.7 节 Memory 存在的理由: 达到上限也要交出最后一版,不能返回 None。空手而归是最差的失败方式。

最后一个测试 test_替身按内容而不是按次数应答 测的不是 agent,是测试本身的基础设施。 同一个提示词问两次,答案必须一样。计数器型替身在这里必然露馅(第一次返回暴力版,第二次返回筛法版)。给你的测试替身写一个测试,听起来像洁癖,但替身一旦有状态错位,它会让所有其他测试变成随机红——那种 bug 你会先怀疑自己的 agent,查几个小时才想到替身。

✅ Checkpoint 10:python -m pytest test_paradigms.py -q,真实输出:

...................                                                      [100%]
19 passed in 0.01s

19 个测试 0.01 秒。 这个数字是本节的重点:没有网络、没有 API Key、没有花钱、没有等待。 你可以把它挂进 CI,每次改 agent 代码都跑一遍。

对比一下"用真模型跑测试"是什么体验:19 个测试至少 50 次 LLM 调用、几十秒到几分钟、每次几分钱、而且随机红(模型这次多说了一句话,断言就挂了)。随机红的测试比没有测试更糟,因为团队很快会学会忽略它。

这一节最该带走的一句话:agent 的测试要测控制流,不要测模型输出。 模型输出的质量用评测集(eval)去管,那是另一套东西;控制流、保险丝、成本公式、安全边界,这四样是普通单元测试就能完全覆盖的,而且它们才是线上事故的主要来源。

3.11 想接真模型 / 继续往下走

整章跑完你手里有 10 个文件、19 个绿灯,全程零密钥。四个升级方向,难度递增:

① 换成真模型(5 分钟)。 在 agent_paradigms/ 下写一个 .env(三个变量和第一章一样:LLM_API_KEY、LLM_BASE_URL、LLM_MODEL_ID),然后:

AP_MODE=real python react_agent.py

三个 agent 一行代码都不用改——这是 3.1 节 get_llm() 工厂的回报。跑之前先预期一件事:真模型的 ReAct 轨迹和 Checkpoint 4 大概率不完全一样(它可能多想一步、可能一次就查对、可能把 Search[乔布斯] 写成 Search[史蒂夫·乔布斯的出生年份])。不一样是正常的,你现在有 Checkpoint 4 作为对照,能一眼看出差在哪。 这就是先离线跑通的好处。

② 换成真搜索。 tools.py 里只有 search 那一个函数需要改,执行器和三个范式完全不动。这就是"把工具收在注册表后面"的回报。 真正的检索工具怎么建(向量库、混合检索、重排),docs/2 第 13 篇会从零搭一个。

③ 给 Plan-and-Solve 加"重规划"。 现在的计划一旦生成就定死了,第 1.2 节说这是它的主要失败模式。加法:执行每一步之后,让 LLM 判断"计划还成立吗",不成立就重新规划剩余步骤。代价是每步多一次调用,成本从"1+步数"变回"不可预知"——所以更常见的做法是只在某一步失败时才触发重规划。

④ 用 Reflection 包住 ReAct(Reflexion)。 让一整轮 ReAct 跑完,把整条轨迹交给评审,评审给出"你第 2 步不该查这个"之类的意见,然后带着意见重跑整轮。这是学术界 Reflexion 论文的思路,也是第 1.5 节说的组合。实现上你会立刻遇到一个问题:反思的对象从"一段代码"变成了"一条轨迹",提示词要怎么写? 这个问题的答案就是 agent 工程的真正难度所在。

四、踩坑与专家提示

这 10 条全部来自把上面 10 个文件真敲一遍的过程,不是从文档里抄的。

坑 1:pytest 报 ModuleNotFoundError: No module named llm,但 python test_paradigms.py 好的。 因为 python xxx.py 会把脚本所在目录加进 sys.path,pytest 不会。解法就是 pyproject.toml 里那一行 pythonpath = ["."]。能钉在配置文件里的,别靠每次记得敲 PYTHONPATH=.。

坑 2:子进程的输出跑到了 print 前面。 why_not_eval.py 第一次跑,我是被-eval-执行的任意命令 打在了标题上面。原因:Python 的 stdout 有缓冲,os.system 起的子进程直接写终端。加 flush=True 才对得上。混用 print 和子进程输出时,顺序不是你写代码的顺序。

坑 3:payload 里带 shell 元字符会让实验结论假。 我最初的 payload 是 echo >>> 我是被 eval 执行的任意命令,shell 在 >>> 上报 syntax error,eval 返回 512 而不是 0——看起来像"eval 被挡住了",其实是命令自己写错了。 做安全演示时,payload 要尽量简单,否则你会得到一个假的安心结论。

坑 4:计数器型测试替身,会让测试随机红。 if self.n == 1: ... elif self.n == 2: ... 这种写法在单个 agent 单次运行时完全正常,但只要共用实例、临时加一次调试调用、或者跑并发,顺序就错位,而且错位之后你会先怀疑 agent 逻辑,查很久才想到替身。改成按输入内容匹配,怎么调都对。

坑 5:if "通过" in review 会把"通过不了效率审查"判成通过。 用子串判断 LLM 输出参与控制流时,一定要把范围收窄到提示词里约定的位置(本章用 review[:10])。正规做法是让模型输出 JSON。

坑 6:白名单只挡节点类型,挡不住 'a'*10**9。 ast.Constant 不只装数字,字符串也是 Constant。少了 isinstance(node.value, (int, float)) 那一行,一个不调用任何函数的表达式就能把你的进程撑爆。白名单要白到值的类型,不只是节点的类型。

坑 7:工具抛异常 = 整个 agent 崩。 除零、语法错、找不到 key,全都要在工具内部变成一句"计算失败:xxx"返回。抛出去的后果是一次小失误变成任务失败,返回字符串的后果只是多花一步。

坑 8:工具找不到结果时返回空字符串,是 ReAct 死循环的头号触发器。 模型看到空,会以为"工具坏了"或"我格式写错了",于是重试同一个动作。返回"知识库中没有关于'图灵'的信息"和返回 "",对模型是完全不同的两件事。 同理,Plan-and-Solve 第一步的 context 要写 "(无)"。

坑 9:Plan-and-Solve 的计划 0 步时,程序不报错、返回空字符串。 for 循环一次都不进,result 保持初始的 "",看起来就像"任务完成了但答案是空的"。所以 Planner 必须校验步数。这类"静默降级"是 agent 里最难发现的 bug,普通异常处理抓不到它,只有主动校验能。

坑 10:max_iterations 写错了,测试不会红,会挂住。 验"程序会终止"的测试失败时长得不像失败——CI 卡在那里不动。遇到 CI 卡死,先去看有没有这类"硬性出口"测试。

专家提示 · 三条纪律,值钱程度超过三个范式本身:

给每个关键组件加一行调用记录(llm.calls、executor.log)。当初只是顺手记一下,后来它变成了成本表(3.9)、控制流断言(3.10)、以及"改进意见有没有真的拼进去"的证据(calls[2]["prompt"])。这是让 agent 变得可测试的最小代价。
不确定的东西全部收到一个工厂后面(get_llm())。三个 agent 完全不知道自己拿到的是真模型还是替身,于是"离线可复现"和"接真模型"同时成立。依赖注入最朴素的形式,也是 agent 项目可测试性的地基。
别信评审,自己跑。 3.8 节那个 timeit 脚本是本章唯一一个"验证 LLM 说的是不是真的"的文件。只会报好消息的验证不是验证。

五、面试视角

Q1:讲一下 ReAct 的循环。

Thought → Action → Observation 三段循环,直到模型输出 Finish[答案]。工程上是四步闭环:提示词里约定格式 → 正则或 JSON 解析出 工具名[参数] → 查注册表调真函数 → 把返回值拼成 Observation: 追加进历史,再问一次。关键点是"历史每一步都要整个重发",这既是它能连续推理的原因,也是它贵的原因。

Q2:ReAct 为什么容易死循环?怎么防?

根本原因是模型不知道自己不知道:工具返回"查不到"时,它倾向于换个措辞再查一次,甚至用同样的参数再查一次。三层防线:①工具返回明确的"没有"而不是空字符串(消除"是不是我格式错了"的歧义);②连续两次相同 Action 就熔断;③max_steps 兜底。熔断是止损线,max_steps 是安全网,两个都要——熔断挡"卡在同一步",max_steps 挡"一直换着步走却不收敛"。本章实测:卡死时熔断版花 2 次调用,只有 max_steps 的版本花满 5 次(生产里 max_steps 常设 15~30,差距就是 10 倍)。

Q3:为什么熔断只比对上一次动作,不比对全部历史?

因为合理的重复是存在的:Search[A] → Search[B] → 回头 Search[A] 补细节,这是正常的多跳推理。只有连续两次完全相同才是真卡住。保险丝要挡死循环,不能挡正常绕路。

Q4:ReAct 的 token 消耗是怎么增长的?

接近二次。 因为每一步都把完整历史轨迹重新发一遍:第 1 步 1 份、第 2 步 2 份、第 3 步 3 份……n 步的总量是 O(n²)。所以"5 步的 ReAct"远比"5 次独立调用"贵。只数调用次数会严重低估 ReAct 的成本。 缓解手段:历史摘要、只保留最近 k 步观察、提示词缓存。

Q5:Plan-and-Solve 和"把任务拆成 N 个子任务分别跑"有什么区别?

区别只有一件事,但是全部价值所在:上一步的结果要带进下一步的上下文。 少了这一行,它就退化成扇出/汇合(fan-out/fan-in),解决的是完全不同的问题。而且删掉它程序不报错、有输出、看起来正常——所以必须有测试守着(断言"第 2 步的结果出现在第 3 步的提示词里")。

Q6:Plan-and-Solve 的成本为什么比 ReAct 好管?

因为 1 + 步数,而计划在第 1 次调用之后就确定了。你可以在规划完成、执行开始之前决定"这个计划要 30 步,太贵,不执行"。ReAct 给不了你这个闸门——它的次数由任务难度决定,事前不可知。

Q7:规划器输出不是合法 JSON 怎么办?

先用正则从回复里抠出 \[.*\](贪婪,要完整数组)再 json.loads,能扛住模型加寒暄和加 ```json 这两种最常见抖动。抠不到就抛异常,并且把原始回复带进错误信息。另外要校验步数在合理区间——0 步会让执行器返回空字符串且不报错,是最阴的一种失败。生产首选是 structured output / function calling,让格式由服务端保证。

Q8:Reflection 的成本收益怎么算?值不值得用?

一轮 = 至少 2 次调用(执行 + 反思),上限 max_iterations × 2。判断标准是"错误代价": 错了没什么损失的任务(闲聊、草稿)别用;错误代价高、且质量可被评审的任务(代码、报告、合同要点提取)才值得。另外要预期收益递减:第一轮反思收益最大,第三轮往后常常只是措辞变化。

Q9:Reflection 有什么结构性缺陷?

自己查自己有盲区。 写代码的和评审的是同一个模型、同一套先验,它看不见的地方,换顶帽子照样看不见。缓解:①换一个更强/不同家族的模型当评审;②给评审具体的检查清单而不是"你看看有没有问题";③最硬的一招——别只让 LLM 评审,把产物真的跑起来(单测、编译、benchmark)。能自动化验证的地方,不要用 LLM 评审代替。

Q10:你怎么知道那一轮反思真的把代码改好了?

这是我最想在面试里被问到的一题。"评审说通过"不是证据。 本章的做法是把两版代码都真的执行:先验正确性(200 个输入结果完全一致,不一致就直接 assert 崩)、再验评审提到的边界(它说 n<2 要返回空列表,那就去测 -5/0/1)、最后验效率(timeit,实测倍数随 n 从 47 涨到 754——说明差距是 O(n²) 那种随规模张开的口子,不是固定倍数)。注意验证的顺序:先对不对,再快不快。一个错的快版本毫无价值。

Q11:三个范式怎么选?

看任务的不确定性和质量要求:路径不确定、要边找边看 → ReAct;结构清楚、步骤可预先列出、要成本可控 → Plan-and-Solve;产物质量敏感、且错误代价高 → Reflection 包在外面。而且三者可以组合:Plan 出骨架 + 每个子步骤内跑小 ReAct(全局便宜可审计,局部能应变)、Reflection 包住整轮 ReAct(即 Reflexion)、Plan + 计划审批(计划生成后先给人或给评审看一眼再执行)。

Q12:选范式时最该关心的是什么?

不是"强项",是主要失败模式:ReAct 会反复横跳、Plan 会一错全错、Reflection 会收益递减加自查盲区。你选一个范式,就是选择接受它那种失败方式,然后为它加保险丝。 说得出这三种失败模式各自的保险丝(熔断+max_steps / 步数校验+重规划 / max_iterations+外部验证),这题就满分了。

Q13:agent 怎么写测试?模型输出不确定啊。

不测模型输出,测控制流。 四类可以用普通单测完全覆盖的东西:①路径(动作顺序、计划步数、反思轮数);②保险丝(熔断、max_steps、max_iterations,而且要分别测,一条不能挡住另一条的测试路径);③成本公式(assert len(llm.calls) == 1 + len(plan));④安全边界(白名单挡住了什么,而且要断言"返回失败字符串"而非 pytest.raises)。做法是给 LLM 写一个按输入内容应答的替身,再为每种失败模式造一个专门的"病模型"。本章 19 个测试 0.01 秒跑完,零成本零网络。模型输出的质量交给 eval 体系管,那是另一套东西。

Q14:eval() 和 ast 白名单的区别,只是"白名单更严格"吗?

不是,是完全不同的执行模型。eval 把字符串编译成任意 Python 再跑,攻击面是整个语言;白名单只遍历 AST 上被允许的那几种节点,其余一律报错。而且白名单要白到值的类型——'a'*10**9 不调用任何函数,却能耗尽内存。真要执行模型生成的代码,唯一正解是换执行环境(容器 / gVisor / 受限解释器),不是"小心地 eval"。

划重点

三个范式不是三个框架,是三种"决定下一步干什么"的方式。 ReAct 走一步看一步、Plan-and-Solve 先定完整计划、Reflection 做完自查自改。而且它们可以互相嵌套——生产里最常见的是 Plan 出骨架、每个子步骤内跑小 ReAct。
所有工具型 agent 都是同一个四步闭环: 格式约定 → 解析 → 执行 → 回喂。最脆的是解析,处理格式错误有三条路(崩 / 把错误当 Observation 喂回去 / 用 function calling 让服务端保证格式),function calling 不是新范式,它是把前两步下沉到了服务端。
选范式 = 选择接受它的失败方式,然后加保险丝。 ReAct 反复横跳(熔断 + max_steps)、Plan 一错全错(步数校验 + 重规划)、Reflection 收益递减和自查盲区(max_iterations + 外部验证)。
ReAct 的成本坑:token 接近二次增长,因为每步重发完整历史。只数调用次数会严重低估它。Plan-and-Solve 的成本在计划出来那一刻就确定了,这给了你一个真实可用的成本闸门。
两条命门代码,删掉都不报错、只是悄悄答错: Plan-and-Solve 的 context += 上一步结果、Reflection 的 feedback = 评审意见。这类静默降级是 agent 里最难发现的 bug,只有主动写测试守着才抓得住。
工具的异常必须在工具内部终结。 除零、语法错、恶意 payload,全部变成一句"计算失败:xxx"返回。抛出去的后果是整个任务崩;返回字符串的后果只是模型多花一步改道。
工具找不到结果时要返回明确的"没有",不要返回空。 空字符串会让模型以为"工具坏了"或"我格式写错了",于是重试同一个动作——这是 ReAct 死循环的头号触发器。
eval 和白名单是两个执行模型,不是严格程度之差。 白名单还要白到值的类型:'a'*10**9 不调用任何函数就能打爆进程。要跑模型生成的代码,换环境,别"小心地 eval"。
别信评审,自己跑。 本章实测那一轮反思确实有收益:200 个输入结果一致、n<2 边界补上了、n=20000 快了 754 倍(倍数随 n 从 47 涨到 754,这才是 O(n²) 的形状)。但这是我们自己 timeit 验的,评审只会说"建议改用筛法"。能自动化验证的地方,不要用 LLM 评审代替。
给 agent 写测试就是测控制流、保险丝、成本公式、安全边界,四样都能用普通单测覆盖。关键工具是一个按输入内容(不是按调用次数)应答的替身——计数器型替身会让测试随机红,而且你会先怀疑自己的 agent。本章 19 个测试 0.01 秒跑完,零网络零成本。
两条让整个项目可测试的地基: 给关键组件加一行调用记录(llm.calls / executor.log),把不确定的依赖收到一个工厂后面(get_llm())。当初只是顺手,后来它们变成了成本表、控制流断言和"意见有没有拼进去"的证据。

下一章预告:三大范式手敲完,你已经知道 agent 的内核长什么样了。第五章换一个视角——不写代码怎么搭智能体:Coze、Dify、FastGPT、n8n 四个低代码平台各做一个真实案例,重点不是学会拖拽,而是搞清楚什么时候该拖拽、什么时候必须回来写代码。

第四章 智能体经典范式构建(从0到1手敲三大范式)