把“rea”三个字母敲进项目目录的时候我并没有预料到它后来会撑起一整条Agent运行链。read、evaluate、act也就是读取、评估、行动当时只是某个POC里的临时模块名结果随着迭代不断长出枝条最后几乎所有任务执行入口都走这一套循环。这篇文章不是讲某种包装好的Agent框架而是从设计取舍、最小实现到线上踩坑把一个真正用得住的rea循环讲透。适合看这篇内容的人基本是这几类正在用大模型接口做应用开发但不知道该怎么让模型“干活”的工程师已经写过一版Agent但出现重复调用、循环卡死、输出乱格式等问题的实践者以及只是想搞明白各种Agent产品底层运行逻辑的学习者。我尽量用偏工程的口吻来写代码骨架可以直接改踩坑清单也可以直接保存。1. 从一次对话到一整个闭环为什么Agent离不开rea1.1 模型只负责“想”不负责“做”我最早犯的错误是把复杂任务直接丢给大模型期待它一次性输出答案。比如让模型“查询订单、核对库存、生成发货单”模型大概率会给你一段有模有样的文字甚至把查询结果也编造出来但不会真正去调用订单系统也不会真的改库存状态。原因很简单大模型的预训练目标是预测下一个token不是操作系统。它擅长的是把信息加工成看起来合理的输出而不是保证每一步动作都真实发生。想要让模型像人一样“办事”就必须在外面套一个执行循环给模型输入真实环境的状态让它输出动作决策我们再去执行这个动作并把执行结果回馈给它。这个循环就是rea。在rea里read解决“模型看到什么”的问题evaluate解决“模型决定做什么”的问题act解决“动作怎么落地”的问题。三者缺一不可。没有read模型只能靠记忆或者编造没有evaluate动作没有方向没有act决策永远停留在文本里。1.2 循环的闭环属性往往被低估很多人以为只要把工具调用加上就算Agent了实际不是。Agent和聊天机器人的分水岭在于是否存在一个封闭的反馈环。聊天机器人一次生成结束就完事Agent则需要反复完成任务执行一个动作看到结果根据结果再决定下一步。这个闭环属性带来了两个工程上的本质变化。第一状态是会累积的你在闭环里不能只依赖模型内部的潜意识必须显式维护上下文否则上一轮的工具结果会被模型遗忘。第二循环必须能被终止否则遇到失败任务就会无限重试花费和延迟都会失控。rea的每一段本质上都在解决闭环带来的某个具体问题。我后来见过不少Agent项目功能点都做了但整体跑不通拆开看就是闭环断掉了要么read阶段没有把上一次的工具结果回填要么act阶段执行完工具后没有把异常反馈给模型要么循环缺少最大步数保护。这些问题放在流程图里不显眼但运行时立刻暴露。2. 搭rea之前我给自己定的四个工程约束2.1 上下文存储归read管不要让模型记得一切read阶段的第一职责是把外部状态整理成模型可以直接消费的上下文。这句话听起来简单但具体做起来很容易跑偏。我见过有人把每一轮历史消息全部堆积在上下文里结果任务进行到第20步时光是历史就占掉了大半窗口真正有用的工具结果反而没有空间。正确做法是让read阶段具备“采买”能力只挑选对当前决策有帮助的信息。比如“最近一步结果”“当前任务目标”“可用工具清单”“用户强制约束”这些放热路径而历史原始消息可以压缩、摘要或者移出主上下文只在需要时再检索回来。比较容易被忽略的是“记忆”和“上下文”的区别。记忆指的是长期保存的领域知识比如项目规范、用户偏好、历史决策原因上下文是当前这次决策的输入。rea里的read既不是全量塞进也不是完全没有长期记忆而是设计成两级热数据进上下文冷数据留外部存储。我生产里的read阶段就是把工具结果、任务元数据、关键规则拼装成一个模块化context对象消费起来非常干净。2.2 动作空间要像函数签名一样严格evaluate阶段的输出会被act阶段真正执行。所以动作空间的定义直接决定安全边界。我给自己定的规矩是每个工具必须有严格的结构化定义包括函数名、参数类型、必填项、默认值、描述、副作用说明。大模型只能在这个结构里做选择不允许自由发挥。一个常见错误是给模型返回自然语言描述比如“你可以用search来查询数据用send_email来发送邮件”。模型确实能理解但输出往往不规范可能出现各种同义改写导致解析失败。动作空间定义成JSON Schema的方式最稳妥模型输出的决策JSON可以精确校验哪个字段缺失、哪个枚举值非法程序都能立即发现。动作空间也需要控制在一个合理粒度。工具太少模型干不了活工具太细模型决策路径过长。比如“发送订单确认邮件”可以作为一个工具而不是拆成“创建邮件”“设置收件人”“设置标题”“调用SMTP”四步。粒度越接近人类可理解的原子任务evaluate阶段越稳定。2.3 步数与预算先给上限再谈智能在任何Agent系统里安全阀都比智能重要。我不止一次看到POC阶段能跑通的任务因为没加步数上限在异常场景下死循环了几十轮等开发者意识到时token账单已经很难看了。rea循环的保护机制我至少会设置四层最大步数、最大token数、单次工具执行超时、整体执行时间上限。最大步数用来限制循环节数比如8步或12步最大token数用来限制模型消耗单次工具超时防止某个接口卡住整个循环整体时间上限则适合对耗时敏感的场景比如用户在一个聊天界面里等结果。这四层保护不是摆设。任务边界清楚时Agent通常在3到6步内完成一旦出现模糊需求或者环境异常模型可能进入“反复尝试-失败-再尝试”的状态步数保护就能兜住最坏情况。我通常在配置层暴露max_steps和max_tokens两个参数线上不同任务类型可以分别调。2.4 rea和当前流行的ReAct等模式是同源设计如果你看过现在各种Agent框架里的主流模式会发现很多名字比如ReAct、Plan-and-Execute、Tool Loop本质都落在rea这个循环框架里。区别主要在提示词模板的格式以及规划方式。我自己的理解是ReAct更强调“推理和行动交替”它在提示词里要求模型先输出思考过程再输出行动rea则是一个更松散的控制流约定read、evaluate、act各是一个函数只要把它们串起来就能跑不限制内部思考格式。对工程而言rea的好处是无论底层模型换什么提示词策略主循环代码都可以不动。所以如果你在纠结要不要引入某个开源Agent方案可以先看看那个方案里的主循环是否和rea等价。如果等价你自己维护的小循环反而更轻量可观测性和排查路径也更清晰。3. 一个能跑的最小rea引擎带骨架的Python实现3.1 先定义数据结构下面这份代码是我自己项目里精简过的骨架可以直接存成一个rea.py用。我先把完整实现放在3.1这一节后面再分段解释关键逻辑。这个版本做了取舍不绑定任何具体大模型SDK你只需要实现一个llm_fn输入是messages和参数输出是文本。工具也通过注册表统一管理。from __future__ import annotations import json import re import uuid from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable class ActionKind(str, Enum): TOOL tool FINISH finish ERROR error dataclass class ToolSpec: name: str description: str parameters_schema: dict fn: Callable[..., dict] def to_schema(self) - dict: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters_schema, }, } dataclass class Decision: action_kind: ActionKind tool_name: str | None None tool_args: dict | None None reason: str terminal_answer: str def to_json(self) - str: return json.dumps({ action: self.action_kind.value, tool_name: self.tool_name, tool_args: self.tool_args or {}, reason: self.reason, terminal_answer: self.terminal_answer, }, ensure_asciiFalse) dataclass class ReaState: task: str messages: list field(default_factorylist) step: int 0 last_result: str | None None finished: bool False final_answer: str trace_id: str field(default_factorylambda: uuid.uuid4().hex[:12]) operation_log: list field(default_factorylist) class ReaLoop: def __init__( self, llm_fn: Callable[[list, dict], str], tools: list[ToolSpec], max_steps: int 8, max_tokens: int 20_000, temperature: float 0.1, ): self.llm_fn llm_fn self.tools {t.name: t for t in tools} self.max_steps max_steps self.max_tokens max_tokens self.temperature temperature self._usage 0 def _read(self, state: ReaState) - list: system_prompt ( 你是任务执行器。请严格按JSON格式输出决策。 可选actiontool、finish、error。 如果调用工具必须提供tool_name与tool_args。 ) tool_schemas [t.to_schema() for t in self.tools.values()] context_block { role: user, content: json.dumps({ trace_id: state.trace_id, task: state.task, step: state.step, available_tools: tool_schemas, last_result: state.last_result, }, ensure_asciiFalse, indent2) } return [{role: system, content: system_prompt}, *state.messages, context_block] def _parse_decision(self, raw: str) - Decision: raw raw.strip() raw re.sub(r^(?:json)?|$, , raw, flagsre.MULTILINE).strip() payload json.loads(raw) action ActionKind(payload[action]) if action is ActionKind.TOOL: name payload.get(tool_name) if name not in self.tools: raise ValueError(funknown tool: {name}) return Decision( action_kindaction, tool_namename, tool_argspayload.get(tool_args, {}), reasonpayload.get(reason, ), ) return Decision( action_kindaction, terminal_answerpayload.get(terminal_answer, ), reasonpayload.get(reason, ), ) def _act(self, state: ReaState, decision: Decision) - None: state.operation_log.append({ step: state.step, action: decision.action_kind.value, tool_name: decision.tool_name, tool_args: decision.tool_args, }) if decision.action_kind is ActionKind.TOOL: tool self.tools[decision.tool_name] result tool.fn(**decision.tool_args) state.last_result json.dumps({ ok: True, tool: decision.tool_name, result: result, }, ensure_asciiFalse) state.messages.append({ role: user, content: f第{state.step}步执行完成结果如下\n{state.last_result}, }) elif decision.action_kind is ActionKind.FINISH: state.finished True state.final_answer decision.terminal_answer else: state.last_result json.dumps({ok: False, error: decision.reason}, ensure_asciiFalse) state.messages.append({ role: user, content: f第{state.step}步报错{decision.reason}请继续。, }) def run(self, task: str) - ReaState: state ReaState(tasktask) state.messages.append({role: user, content: f任务{task}}) while state.step self.max_steps and not state.finished: if self._usage self.max_tokens: state.final_answer token预算耗尽任务终止。 break prompt self._read(state) raw self.llm_fn(prompt, {temperature: self.temperature}) try: decision self._parse_decision(raw) except Exception as e: state.last_result json.dumps({ ok: False, format_error: str(e), }, ensure_asciiFalse) state.messages.append({ role: user, content: f输出解析失败{e}请重新输出合法JSON。, }) state.step 1 continue self._act(state, decision) state.step 1 if not state.finished: state.final_answer 达到最大步数任务终止。 return state3.2 read阶段把外部信息搬进上下文read的核心函数是_read。它做的事情并不复杂把系统提示词、历史消息、最新任务上下文拼成一个消息数组。关键是这个消息数组里的context_block它把当前步数、任务目标、工具清单、上一步结果都打包成结构化JSON放进去。为什么要结构化因为模型对格式化JSON的理解比自由文本更稳定尤其当工具描述字段很多时JSON能降低模型遗漏参数的概率。我看到的一些失败案例恰恰是因为上下文只是简单拼接了几个句子比如“你上一步的结果是xx”模型难以判断字段边界。这个阶段还承担了“控制上下文体积”的作用。如果你有更长的历史记录可以在_read里先做摘要或者截断只要保证历史摘要不丢失关键结论就行。生产环境里我通常还会在这个函数里记账统计本轮输入了多少token输出是多少token好让后面的预算控制有数据依据。3.3 evaluate阶段输出结构化决策并校验_parse_decision是整个rea循环里最需要防御的地方。我见过很多模型输出要么是“我认为我应该调用search”要么是带Markdown代码块的JSON要么JSON结构合法但值完全不匹配工具定义。所以我在解析前先剥离代码块标记然后用json.loads严格解析再检查action是否合法、工具是否存在。这里的硬性规则是解析失败不能默默吞掉异常而是要把错误信息塞回消息列表让模型自己“看到”解析失败并尝试修正。因为模型有很强的自我纠错能力只要告诉它“上一次输出不是合法JSON你能不能重试”它大概率会重来一次。但如果错误被吞掉模型会一直输出同样格式循环就会卡死。生产场景里我会加一层schema校验也就是把每个工具的parameters_schema拿到决策参数上去做校验缺字段、类型错误都在evaluate阶段拦住不要等执行工具时才报错这样能把问题定位到“模型决策层”而不是“系统执行层”。3.4 act阶段执行工具并回填结果act负责把决策落地。这个阶段最少要注意三件事第一执行超时控制。工具函数如果是不受控的外部请求必须在外面包一层超时。第二执行结果标准化。无论工具成功失败都要返回一个统一格式的dict不能有的抛异常、有的返回None。第三回填结果。成功后把结果追加到历史消息里让下一轮read自然看到。失败路径也需要回填。模型决策本身可能错误比如调用了不存在的参数工具执行也可能失败。我的做法是无论成功失败都作为一条“执行结果”写进state.messages让模型在下一次评估时看到。这样Agent才具备从错误中恢复的能力而不是一次失败直接终止。另外就是操作日志operation_log每一轮决策都要记录包括动作类型、工具名、参数、步数。这个日志在调试时价值极高尤其是多轮循环之后想回溯某个参数是从哪一步来的。没有操作日志的Agent排查问题基本靠猜。3.5 主循环的终止条件主循环run的终止条件有三个达到最大步数、收到finish动作、token预算耗尽。这个设计看似简单但每一条都对应着一个真实线上事故。没设最大步数循环可能在错误路径上无限重试没支持finish动作模型即使完成了任务也只能被迫反复调用工具没设token预算一个异常长任务能把调用费拉到几十倍。还有一个隐藏细节解析失败时我让state.step 1这意味着格式错误会消耗步数。这其实是故意的因为模型如果连续多次输出非法JSON说明当前提示词或模型能力不足以完成该任务继续给无限重试机会只会恶化问题。把格式错误算作步数消耗能促使模型在下一轮更加保守、输出更规范。4. 生产环境最容易踩的五个坑4.1 卡死上下文里只有对话没有结果我在内部评审时看到过一个Agent表现是模型反复输出同样的工具调用。排查后发现read阶段只拼上了用户消息和系统提示词完全没有把上一轮工具执行结果追加进去。模型每次看到的场景都一样自然每次决策都一样形成了一个稳如泰山的死循环。修复方式就是3.4里说的结果回填每次执行完工具立刻把结果写入消息列表。如果你在调试中看到模型“鬼打墙”第一步先查上下文里是否有上一轮的结果第二步再查结果是否被放进了模型能看到的位置。4.2 重复动作工具没有幂等保护有些工具天然有副作用比如“发送邮件”“创建订单”“扣减库存”。如果模型因为网络超时没有及时收到结果可能会重试同一个参数。如果工具本身不幂等就会造成同一笔订单创建两次、库存扣两次。我的处理方式是给带副作用的工具加上请求层面的幂等键。在工具参数里增加一个request_id由主循环生成并透传工具执行时先查这个id是否处理过处理过就直接返回上次结果。这样即使模型重复调用底层也不会重复生效。4.3 格式漂移模型输出不再是合法JSON大模型对JSON格式的遵循能力在实际应用中并没有想象中可靠。尤其是工具描述变复杂、上下文变长之后模型可能把JSON包裹进Markdown代码块或者混入多余文字甚至在JSON尾部加上一段自然语言。我建议不只依赖正则或字符串清洗而是做两层校验第一层解析第二层schema校验。解析时剥离所有非JSON包裹schema校验时逐字段验证。如果连续出现3次解析失败就把这一步标为error并让系统给出提示。与其让模型乱猜不如让它换一条路。4.4 工具异常被当场抛出我早期犯过一个错工具函数执行失败时直接抛Python异常结果主循环崩溃整个Agent停摆。正确做法是把异常捕获转成结构化错误信息返回给模型。因为很多时候模型能根据错误信息修正参数重新调用不需要人工介入。比如一个工具因为权限不足失败模型看到“权限不足”后可能会选择另一个替代工具。但如果你直接抛异常这个恢复路径就断了。所以我在act阶段永远把工具调用包在try/except里并统一返回错误结构。4.5 成本失控只设了步数没设token上限步数上限能挡住循环次数但挡不住单次输出过长。有些模型的输出可能上千token如果每轮都输出一堆推理文本总消耗会远超预期。我自己给线上任务设过每月预算但第一周就超了原因就是只限制了步数没限制模型输出长度。token控制可以从两个维度入手一是主循环里的max_tokens总额每天执行完判断累计使用量二是调用时限制模型单次最大生成长度。两者结合起来才能把成本控制在可预期范围内。下面的表是我常用的排查速查表可以直接保存现象典型根因快速检查方法修复思路模型反复调用同一工具read阶段没有回填上轮结果打印一次循环内的messages在act结束把执行结果追加进上下文同样的操作被执行多次工具缺少幂等保护检查请求参数里是否有request_id增加幂等键并透传给底层输出解析偶尔失败上下文历史太长、格式漂移抓取原始输出看是否被包裹剥离代码块加schema校验工具失败后整个Agent终止工具异常直接上抛看日志里是否出现未捕获异常包try/except并回填错误消息费用超额只限制步数没限制token看监控面板的token使用趋势增加总额度和单次长度限制5. 更贴近线上的一点工程化建议5.1 把read阶段拆成“最近状态”和“长期档案”当Agent业务复杂起来后read阶段如果还只负责拼字符串迟早碰到瓶颈。我习惯把read拆成两个子模块一个负责“当前运行上下文”比如任务目标、最近结果、当前步数另一个负责“长期档案”比如领域术语、历史决策、用户偏好这部分通过检索把少量相关片段放进上下文。长期档案不能全量放而应该做成可检索片段。比如“用户偏好是尽量用XX方式发货”这类信息只要找到顶部几条放进read阶段的context就行。如果长期档案设计得好模型从一开始就不会走偏。5.2 给每一次rea打上唯一追踪号多轮任务一旦出问题没有trace几乎没法查。我这里的ReaState.trace_id就是干这个的。从任务开始到结束所有日志、工具调用、token消耗都挂在这个trace_id下面查问题时按这个id一拉整条链路就浮现出来了。对于线上生产系统我还建议把每轮的关键字段上报到日志中心比如步数、动作类型、工具名、参数摘要、结果摘要、耗时、token数。这样不仅能排查问题还能做Agent运行质量分析比如看哪些工具经常失败、哪些任务步数异常偏高。5.3 让评估阶段同时输出成本和意图信息我在实际生产中让Decision额外带两个字段estimated_cost和intent。intent指这一步的行动意图比如“查询用户余额”或“确认订单状态”它不影响执行但能在追踪和复盘时提供很多信息。estimated_cost则让决策带上成本预估这样可以在act执行前拦住明显高成本的操作。比如一个任务累计消耗已经接近上限你可以在执行前检查estimated_cost拒绝高风险动作转而询问用户是否继续。这比事后看账单要主动得多。5.4 别让Agent自己说“完成”要定义可验证的终止标准最后是终止条件。很多Agent框架里模型输出“finish”就结束了但finish的判定可能非常随意。模型可能以为任务完成了实际结果完全不对。所以我建议对每个任务类型显式列终止标准比如“完成条件已生成文件且文件存在”或者“完成条件订单状态变为已发货”。这个标准不一定要在read阶段全部给模型看但应该在主循环结束前做一层校验。如果校验未通过即使模型想finish也可以让系统发回一条“当前结果不符合完成条件继续补充”的提示避免自欺欺人式收尾。把“rea”三个字母敲进项目目录的时候我并没有预料到它后来会撑起一整条Agent运行链。回看这个项目真正让Agent从演示变成可用的不是模型本身多聪明而是这个由read、evaluate、act组成的小闭环被磨到了严丝合缝。每当我新接手一个Agent系统第一件事就是找它的主循环在哪然后画出read、evaluate、act的边界。边界清晰系统大概率能长出好结构边界混乱后面叠加再多能力也只是在给一个不稳定的底座添砖。这是我最真实的一点体会。