企业官网建设流程全解析

把“rea”三个字母敲进项目目录的时候，我并没有预料到它后来会撑起一整条Agent运行链。read、evaluate、act，也就是读取、评估、行动，当时只是某个POC里的临时模块名，结果随着迭代不断长出枝条，最后几乎所有任务执行入口都走这一套循环。这篇文章不是讲某种包装好的Agent框架，而是从设计取舍、最小实现到线上踩坑，把一个真正用得住的rea循环讲透。

适合看这篇内容的人，基本是这几类：正在用大模型接口做应用开发，但不知道该怎么让模型“干活”的工程师；已经写过一版Agent，但出现重复调用、循环卡死、输出乱格式等问题的实践者；以及只是想搞明白各种Agent产品底层运行逻辑的学习者。我尽量用偏工程的口吻来写，代码骨架可以直接改，踩坑清单也可以直接保存。

1. 从一次对话到一整个闭环：为什么Agent离不开rea

1.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_ascii=False) @dataclass class ReaState: task: str messages: list = field(default_factory=list) step: int = 0 last_result: str | None = None finished: bool = False final_answer: str = "" trace_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) operation_log: list = field(default_factory=list) 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格式输出决策。" "可选action：tool、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_ascii=False, indent=2) } 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, flags=re.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(f"unknown tool: {name}") return Decision( action_kind=action, tool_name=name, tool_args=payload.get("tool_args", {}), reason=payload.get("reason", ""), ) return Decision( action_kind=action, terminal_answer=payload.get("terminal_answer", ""), reason=payload.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_ascii=False) 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_ascii=False) state.messages.append({ "role": "user", "content": f"第{state.step}步报错：{decision.reason}，请继续。", }) def run(self, task: str) -> ReaState: state = ReaState(task=task) 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_ascii=False) 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 state

3.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总额，每天执行完判断累计使用量；二是调用时限制模型单次最大生成长度。两者结合起来，才能把成本控制在可预期范围内。

下面的表是我常用的排查速查表，可以直接保存：

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的边界。边界清晰，系统大概率能长出好结构；边界混乱，后面叠加再多能力也只是在给一个不稳定的底座添砖。这是我最真实的一点体会。