前言
用户问题 → Prompt → 调用模型 → 返回答案
模型遗漏约束；在错误方向上持续推理；声称查过并不存在的资料；工具失败后重复相同动作；当前任务修正了，下次仍犯同样的错。
于是，CoT、ReAct、Reflection、Reflexion、CoVe、ToT 等概念不断被加入系统，Agent 却未必因此更可靠。原因是：这些方法不是可以互相替代的提示词技巧，而是在解决不同层面的工程问题。本文将用 Python 搭建一个最小但完整的 Agent，把常见方法还原成五个模块：用户任务 ↓Planner：拆解目标与约束 ↓ReAct Loop：选择动作 → 调用工具 → 读取观察 ↓Verifier：检查证据、约束和矛盾 ↓Refiner：根据问题修改答案 ↓Reflection Memory：保存可复用经验
好的 Agent 不是“思考得最多”，而是能在正确位置获得反馈，并让每一步都可以被观察、验证和停止。🧠
一、先定义统一的模型接口
为了把注意力放在工作流上，代码不绑定具体模型厂商，只假设项目中存在以下函数：def call_model(system: str, user: str) -> str: ”””调用你选择的 LLM，并返回文本结果。””” raise NotImplementedError
你可以在这里接入云端 API、本地模型或公司内部网关。生产环境最好让模型直接返回受 JSON Schema 约束的结构化结果。为了保持示例通用，本文使用普通 JSON 和一个最小解析器：import jsonfrom typing import Anydef parse_json(text: str) -> dict[str, Any]: start = text.find(”{”) end = text.rfind(”}”) if start == -1 or end == -1: raise ValueError(f”模型没有返回 JSON：{text[:200]}”) return json.loads(text[start : end + 1])
让自然语言负责表达，让结构化数据负责控制流程。
不要通过搜索 Thought: 或 Action: 等字符串决定下一步。一旦模型改变格式，整个 Agent 都可能失控。🧩
二、建立单次调用基线
def direct_answer(task: str) -> str: return call_model( system=”你是一个严谨的技术助手。信息不足时明确说明。”, user=task, )
多步骤工作流会增加 Token、延迟和故障点。如果它不能在真实测试集上稳定超过单次调用，就只是更复杂，而不是更好。任务成功率；事实或证据错误率；平均响应时间；每个成功任务的平均成本。
三、用 Plan-and-Solve 拆解任务
CoT 的价值，是把复杂任务拆成中间步骤。但在工程系统中，我们通常不需要保存漫长的自由推理文本。PLANNER_SYSTEM = ”””你是任务规划器。把用户任务拆成少量、可验证的步骤。只返回 JSON：{ ”goal”: ”最终目标”, ”constraints”: [”必须满足的约束”], ”steps”: [ { ”id”: 1, ”objective”: ”该步骤要完成什么”, ”needs_external_info”: true, ”success_condition”: ”如何判断完成” } ]}不要执行任务，不要虚构已经获得的信息。”””def make_plan(task: str) -> dict: raw = call_model(PLANNER_SYSTEM, task) plan = parse_json(raw) if not plan.get(”steps”): raise ValueError(”计划中没有步骤”) if len(plan[”steps”]) > 8: raise ValueError(”计划过长，需要重新压缩”) return plan
八步不是通用标准，而是示例中的复杂度上限。计划越长，状态越容易漂移，实际项目应根据任务分布调整。CoT：生成帮助模型得到答案的中间文字Plan-and-Solve：生成系统能够执行和检查的任务步骤
对于 Agent，计划应该成为结构化状态，而不是散落在上下文中的一段话。🗂️
四、用 ReAct 连接外部工具
模型无法靠“继续思考”获得实时网页、数据库记录或代码执行结果。只要任务依赖外部状态，就需要工具。from collections.abc import Callabledef search(query: str) -> str: # 替换成你的搜索服务，并返回精简后的文本结果 raise NotImplementedErrordef calculator(expression: str) -> str: # 演示用。生产环境不要直接 eval 不可信输入。 allowed = set("0123456789+-*/().% ") if not set(expression) <= allowed: raise ValueError("表达式包含不允许的字符") return str(eval(expression, {"__builtins__": {}}, {}))TOOLS: dict[str, Callable[..., str]] = { "search": search, "calculator": calculator,}
{ ”type”: ”tool”, ”tool”: ”search”, ”args”: {”query”: ”ReAct paper arXiv”}, ”reason”: ”需要获得原始资料”}
{ ”type”: ”final”, ”answer”: ”最终答案”, ”evidence”: [”支持答案的观察编号”]}
reason 只记录简短的行动依据，不要求保存模型完整的隐藏推理过程。AGENT_SYSTEM = ”””你是一个使用工具完成任务的 Agent。每轮只能返回以下两种 JSON 之一：1. 调用工具：{ ”type”: ”tool”, ”tool”: ”search | calculator”, ”args”: {}, ”reason”: ”简短说明为什么需要这个动作”}2. 完成任务：{ ”type”: ”final”, ”answer”: ”答案”, ”evidence”: [”observation-1”]}规则：- 不得声称执行过上下文中没有记录的工具调用；- 外部事实必须引用 observation；- 工具失败时调整方案，不要原样重复；- 信息不足时继续调用工具或明确停止原因。”””def run_react(task: str, plan: dict, max_steps: int = 6) -> dict: history: list[dict] = [] for step_no in range(1, max_steps + 1): context = { ”task”: task, ”plan”: plan, ”history”: history, ”available_tools”: list(TOOLS), ”remaining_steps”: max_steps - step_no + 1, } action = parse_json( call_model(AGENT_SYSTEM, json.dumps(context, ensure_ascii=False)) ) if action.get(”type”) == ”final”: return { ”status”: ”completed”, ”answer”: action.get(”answer”, ””), ”evidence”: action.get(”evidence”, []), ”trajectory”: history, } if action.get(”type”) != ”tool”: raise ValueError(f”未知动作类型：{action}”) tool_name = action.get(”tool”) if tool_name not in TOOLS: observation = f”工具不存在：{tool_name}” else: try: observation = TOOLS[tool_name](**action.get(”args”, {})) except Exception as exc: observation = f”工具执行失败：{type(exc).__name__}: {exc}” history.append( { ”id”: f”observation-{step_no}”, ”action”: action, ”result”: observation[:6000], } ) return { ”status”: ”max_steps_exceeded”, ”answer”: ””, ”evidence”: [], ”trajectory”: history, }
Reason：决定下一步Act：调用工具Observe：读取真实结果Reason：根据结果继续决策
最大步骤数和总预算；工具超时、重试和参数校验；副作用操作的单独授权；工具结果截断与去噪；Prompt Injection 防护；重复失败检测；Evidence 与 Observation 的引用校验。
ReAct 的可靠性主要来自工具边界和状态机，而不是 Prompt 写得多聪明。🔧
五、把“反思”拆成验证和修改
Agent 生成答案后，只追加一句“请检查回答”通常不够。1. Verifier：检查约束与证据
VERIFIER_SYSTEM = ”””你是结果验证器。根据任务、计划、工具观察和候选答案进行检查。只返回 JSON：{ ”verdict”: ”pass | revise | fail”, ”issues”: [ { ”type”: ”unsupported_claim | missed_constraint | contradiction | incomplete”, ”detail”: ”具体问题”, ”required_action”: ”如何修复” } ], ”supported_evidence”: [”observation-1”]}不得用常识替代不存在的工具证据。”””def verify(task: str, plan: dict, result: dict) -> dict: payload = { ”task”: task, ”plan”: plan, ”candidate_answer”: result[”answer”], ”claimed_evidence”: result[”evidence”], ”observations”: result[”trajectory”], } return parse_json( call_model(VERIFIER_SYSTEM, json.dumps(payload, ensure_ascii=False)) )
这吸收了 CoVe 的核心思想：把验证变成独立任务，而不是让模型边维护原答案，边证明自己正确。代码任务运行测试、编译和静态检查；数据任务检查 Schema、范围和唯一性；数学任务交给计算器；事实任务回到原始资料；操作任务读取系统最终状态。
2. Refiner：按照明确问题修改
REFINER_SYSTEM = ”””你是答案修订器。根据验证器指出的问题修改候选答案。不得引入 observations 中没有依据的新事实。如果现有证据不足以修复，返回 needs_more_tools=true。只返回 JSON：{ ”needs_more_tools”: false, ”answer”: ”修订后的答案”, ”evidence”: [”observation-1”], ”changes”: [”修改了什么”]}”””def refine(task: str, result: dict, report: dict) -> dict: payload = { ”task”: task, ”candidate”: result, ”verification”: report, } return parse_json( call_model(REFINER_SYSTEM, json.dumps(payload, ensure_ascii=False)) )
如果 needs_more_tools=true，不要让 Refiner 猜答案，而应把问题送回 ReAct 循环获取新证据。真正有效的 Reflection，不是重复思考，而是引入新的约束、证据或错误信号。✅
六、用 Reflexion 保存失败经验
Self-Refine 改进当前答案，Reflexion 解决另一个问题：为什么 Agent 下次还会犯同样的错？
可以在失败或修订后，生成一条简短、可检索、可执行的经验：REFLECTION_SYSTEM = ”””你是 Agent 复盘器。根据任务轨迹和验证报告，提取一条可复用经验。只记录能改变未来行动的内容，不要复述完整过程。只返回 JSON：{ ”trigger”: ”什么情况下应该想起这条经验”, ”failure_pattern”: ”本次失败模式”, ”lesson”: ”以后应采用的具体策略”, ”avoid”: ”不要重复的动作”}”””def reflect(task: str, result: dict, report: dict) -> dict: payload = { ”task”: task, ”trajectory”: result[”trajectory”], ”verification”: report, } return parse_json( call_model(REFLECTION_SYSTEM, json.dumps(payload, ensure_ascii=False)) )
class ReflectionMemory: def __init__(self) -> None: self.items: list[dict] = [] def add(self, item: dict) -> None: self.items.append(item) def recall(self, task: str, limit: int = 3) -> list[dict]:演示版直接取最近记录。 # 生产环境可以使用关键词、Embedding 或混合检索。 return self.items[-limit:]
memory = ReflectionMemory()def make_plan_with_memory(task: str, memory: ReflectionMemory) -> dict: past_lessons = memory.recall(task) user = json.dumps( {”task”: task, ”relevant_past_lessons”: past_lessons}, ensure_ascii=False, ) return parse_json(call_model(PLANNER_SYSTEM, user))
不要保存所有轨迹和“感想”。只有符合以下条件的经验才值得进入长期记忆：对应明确的失败模式；能转化为未来动作；不依赖已经过期的状态；不与现有经验重复或冲突；后续可以评价是否有效。
Reflexion 修改的是运行时记忆，不改变模型参数；CoH 则会把反馈变成训练数据，通过训练改变模型。大多数应用团队应先做可观察、可删除的运行时记忆，再考虑微调。🧩
七、组装完整 Agent
def solve(task: str, memory: ReflectionMemory) -> dict: plan = make_plan_with_memory(task, memory) result = run_react(task, plan) if result[”status”] != ”completed”: report = { ”verdict”: ”fail”, ”issues”: [{”detail”: result[”status”]}], } memory.add(reflect(task, result, report)) return result report = verify(task, plan, result) if report[”verdict”] == ”pass”: return result revised = refine(task, result, report) memory.add(reflect(task, result, report)) if revised.get(”needs_more_tools”): return { ”status”: ”needs_more_evidence”, ”answer”: ””, ”verification”: report, } return { ”status”: ”revised”, ”answer”: revised[”answer”], ”evidence”: revised[”evidence”], ”verification”: report, }
这仍然只是教学骨架，缺少持久化、并发控制、权限系统、严格的结构化输出和生产级工具适配器，但核心控制面已经完整：Plan：决定怎么做Search：比较候选方案Act：通过工具获得真实观察Verify：检查答案是否被证据支持Learn：把失败压缩成经验
八、什么时候加入 Self-Consistency 或 ToT？
前面的实现主要沿一条路径执行。如果任务存在多个合理方案，而且早期选择会显著影响结果，可以加入受限搜索。最简单的是 Self-Consistency：运行多个候选，再由独立评估器选择。def solve_with_candidates(task: str, memory: ReflectionMemory, k: int = 3): candidates = [solve(task, memory) for _ in range(k)] judge_input = { ”task”: task, ”candidates”: candidates, ”rule”: ”优先选择证据完整、满足约束、步骤更少的结果”, } return parse_json( call_model( ”从候选中选择最佳结果，只返回 JSON。”, json.dumps(judge_input, ensure_ascii=False), ) )
ToT 会在关键节点生成多个候选动作，评分后保留部分分支，必要时回退。它适合复杂规划和搜索问题，但成本可能成倍增加。
九、上线前必须检查的六件事
1. 所有循环都有上限
ReAct、Self-Refine、ToT 和重试必须设置最大次数、最大 Token 或预算。2. 工具权限分级
搜索和读取可以自动执行；发送消息、修改数据、付款和删除文件等操作，应设置更严格的验证与人工确认。3. 工具输出不是可信指令
网页、邮件和文档可能包含 Prompt Injection。工具结果应被视为不可信数据，而不是系统指令。4. 验证尽量外部化
测试、编译器、Schema、数据库约束和原始资料，通常比模型自己的置信度可靠。5. 记忆可以过期和删除
每条经验应记录来源、时间和适用范围。过期经验需要淘汰，冲突经验需要重新验证。6. 评价完整任务
工具调用成功率；平均执行步骤；无效或重复调用比例；证据覆盖率；验证后的修订率；失败恢复成功率；每个成功任务的成本和延迟。
如果只看“答案读起来不错”，就无法判断 Agent 是可靠完成任务，还是偶然猜对。🛡️
结语
CoT、ReAct、Reflection 和 Reflexion 经常同时出现在 Agent 系统中，但它们负责不同阶段：CoT / Plan-and-Solve：拆解问题Self-Consistency / ToT：探索候选路径ReAct：从外部环境获得信息CoVe / Self-Refine：验证并修改结果Reflexion：把失败变成可复用经验CoH：把稳定反馈用于训练
先建立直接回答的基线，再加入计划、工具调用和确定性验证。只有评测证明某类错误持续出现时，再增加搜索、修订或记忆。如果你正在开发 LLM 应用，可以拿出最近十个失败案例，逐个标记它们发生在：错误属于哪一层，下一步应该补什么机制，通常就会立刻清楚。好的 Agent 不是一个无限思考的模型，而是一套能规划、行动、验证、学习，并在必要时停止的工程系统。🚀
