前言:一个残酷的判断
我最近在github上或是社媒上看到过很多 Agent 项目,有一个规律几乎从不例外:
Demo 靠功能,生产靠失败模式。
Demo 的评判标准是"它能做什么";生产的评判标准是"它出错的时候会怎样"。
这个区别不是程度上的差异,而是设计出发点的差异。一个围绕功能设计的 Agent,你后面每加一个生产级能力(可观测、可回放、可评测、权限控制)都要改一遍循环;而一个围绕失败模式设计的 Agent,这些能力是架构的自然推论。
下面从一起真实事故说起。
一、先看一起真实事故
Replit 的 Agent 在一次操作中删除了客户的生产数据库,并且在事后谎报了这件事。
很多人看到这个新闻的第一反应是:“prompt 没写好”。
这个判断是错的。
问题不在于它没被"告诉"不要删数据库,而在于系统里根本没有任何机制能阻止它删。
区别在这里:
| 做法 | 结果 | |
|---|---|---|
| Demo 思路 | 在 system prompt 里写"删除前请确认" | 模型今天听话,明天不听话 |
| 生产思路 | delete操作被标记为 destructive,运行时强制要求人类审批 | 模型再怎么想删也删不掉 |
这就是本文的全部主题:机制(mechanism),而不是提示(prompt)。
二、Demo Agent 与生产级 Agent 的八项对比
先给一张全景对照表:
| 维度 | Demo Agent | 生产级 Agent |
|---|---|---|
| 循环 | 隐藏的 while 循环 | 显式、可中断、有硬预算 |
| 上下文 | 当缓冲区用,塞满就截断 | 当预算管理,可出具"上下文收据" |
| 工具 | 能调通就行 | 为模型设计的 API,错误信息可教学 |
| 安全 | 靠 prompt 请求 | 运行时强制,默认拒绝 |
| 状态 | 存在内存里 | 可序列化、可恢复、可 fork |
| 非确定性 | 接受它、“反正结果随机” | 收缩到最小边界并全部记录 |
| 测试 | 手动跑几次看看 | 不调真实 LLM 就能测主循环 |
| 可观测 | print()打日志 | 结构化事件流作为唯一事实来源 |
接下来逐条展开,每条都给反面写法和生产写法。
三、机制 1:日志即本体(The Log Is the Agent)
这是我认为最重要的一条架构决策。
有一篇论文直接把这个论点写进了标题:《The Log Is the Agent: Event-Sourced Reactive Graphs for Auditable, Forkable Agentic Systems》。
它的核心主张是:Agent 的本质是日志(事件流),而不是内存里的状态对象。
一旦接受这个前提,「可审计」「可 fork」「可回放」「可评测」就不再是需要额外开发的功能,而是架构的直接推论。
❌ Demo 写法
defrun_agent(task:str):messages=[{"role":"user","content":task}]whileTrue:resp=llm.chat(messages)ifresp.tool_calls:result=call_tool(resp.tool_calls)print(f"called tool:{result}")# 日志 = 用完即弃messages.append(result)else:returnresp.content问题:print出来的东西不是数据。你无法对它做聚合、归因、回放——它只是一串给人看的文本。
✅ 生产级写法
fromdataclassesimportdataclass,field,asdictfromtypingimportLiteralimporttime,json@dataclassclassAgentEvent:seq:int# 单调递增序号type:Literal["run_start","llm_call","tool_call","tool_result","policy_denied","budget_exceeded","stagnation_detected","run_end",]ts:float# 时间戳payload:dict# 结构化载荷tokens_in:int=0tokens_out:int=0cost_usd:float=0.0classEventSink:"""事件流是 Agent 的唯一事实来源(single source of truth)。"""def__init__(self,path:str):self._f=open(path,"a",encoding="utf-8")self._seq=0defemit(self,type_:str,payload:dict,**metrics)->AgentEvent:self._seq+=1ev=AgentEvent(seq=self._seq,type=type_,ts=time.time(),payload=payload,**metrics)self._f.write(json.dumps(asdict(ev),ensure_ascii=False)+"\n")self._f.flush()returnev关键点:Agent 的运行时状态可以从事件流重建。这意味着——
- 崩溃后能从事件流恢复,而不是从内存快照
- 能从任意历史事件 fork 出一条新分支
- 能对历史事件做聚合分析(成本、延迟、失败率)
- 能拿真实事件流直接生成测试用例
四、机制 2:硬预算 + 停滞检测
失控循环是 Agent 最常见的生产事故,没有之一。
一个没有预算的 Agent 在遇到工具持续报错时,会一直重试到你的账单爆炸。
❌ Demo 写法
whileTrue:resp=llm.chat(messages)# 祈祷它能自己停下来✅ 生产级写法
fromdataclassesimportdataclassclassBudgetExceeded(Exception):...@dataclassclassBudget:max_steps:int=20max_tokens:int=200_000max_wall_clock_s:float=120.0max_cost_usd:float=1.0defcheck(self,*,steps:int,tokens:int,elapsed:float,cost:float)->None:"""必须是硬限制,不是"建议"。"""ifsteps>=self.max_steps:raiseBudgetExceeded(f"步数超限:{steps}/{self.max_steps}")iftokens>=self.max_tokens:raiseBudgetExceeded(f"token 超限:{tokens}/{self.max_tokens}")ifelapsed>=self.max_wall_clock_s:raiseBudgetExceeded(f"耗时超限:{elapsed:.1f}s")ifcost>=self.max_cost_usd:raiseBudgetExceeded(f"花费超限: ${cost:.4f}")别漏掉停滞检测。预算只能防烧钱,防止不了"原地打转 20 步然后失败":
defis_stagnant(recent:list["Action"],window:int=3)->bool:"""连续 N 步调用同一个工具且参数完全相同 -> 判定为停滞。"""iflen(recent)<window:returnFalsefingerprints={a.fingerprint()forainrecent[-window:]}returnlen(fingerprints)==1停滞检测的价值不只是省钱——它把一个"沉默烧钱 5 分钟然后失败"的体验,变成"3 秒内明确告诉你卡在哪"。
五、机制 3:上下文是「预算」,不是「缓冲区」
绝大多数 Agent 项目把上下文当垃圾桶:什么都往里塞,塞满了就静默截断。
静默截断是 bug,不是优化。它会让 Agent 神秘地丢失关键信息,而你完全不知道发生了。
✅ 生产级做法:出具「上下文收据」
fromdataclassesimportdataclass@dataclassclassSection:name:str# 例如 "system_prompt" / "tool_schemas" / "history" / "retrieved_docs"tokens:intsource:str# 这段内容从哪来(文件路径 / 事件 seq / 检索 id)truncated:bool=False@dataclassclassContextReceipt:sections:list[Section]total:intbudget:intdefexplain(self)->str:lines=[f"上下文预算{self.total}/{self.budget}tokens"]forsinself.sections:flag=" ⚠️ 已截断"ifs.truncatedelse""lines.append(f"{s.name:<20}{s.tokens:>7}tok <-{s.source}{flag}")return"\n".join(lines)配上输出:
上下文预算 48210/64000 tokens system_prompt 1204 tok <- prompts/v3/system.md tool_schemas 18320 tok <- 42 tools ⚠️ 已截断 history 24100 tok <- 事件 seq 12..87 retrieved_docs 4586 tok <- kb://policies/refund这一张表能立刻暴露大部分上下文问题。上面这个例子一眼就能看出:42 个工具的 schema 占了 38% 的预算——这正对应工具渐进披露(按需加载工具,而不是一次性全塞进去)的需求。
核心认知:上下文组装应该是一个显式、可测试、可观测的函数,而不是散落在各处的messages.append()。
六、机制 4:工具的错误信息要「教模型怎么恢复」
工具是给模型设计的 API,不是给人用的函数。它的命名、描述、错误信息都是设计对象。
而这里有一个极其普遍、又极少被修的缺陷:
❌ 大多数项目的工具错误
raiseToolError("invalid argument")模型看到这句话,只能瞎猜。于是它重试、重试、再重试——直接喂给了你的失控循环。
✅ 生产级写法
raiseToolError("参数 'start_date' 格式错误:收到 '2024/13/01'。\n""要求:ISO 8601 格式(YYYY-MM-DD),且必须是过去的日期。\n""示例:'2024-12-01'\n""请修正该参数后重试。")记住这句话:错误信息本身就是一段 prompt。
你花在打磨错误信息上的每一分钟,都会直接转化为 Agent 的自恢复能力。
另外三条工具设计原则
# 1. 返回值默认截断/分页——一个能往上下文倒 10 万 token 的工具是 bug@tool(max_output_tokens=2000)defsearch_logs(query:str,page:int=1)->str:...# 2. 写操作必须幂等,或提供 dry-run@tool(idempotent=True)defcreate_ticket(title:str,idempotency_key:str)->str:...@tool(dry_run_supported=True)defdelete_records(filter_:str,dry_run:bool=True)->str:...# 3. 显式声明能力等级——这是机制 5 的前提CAPABILITIES={"read_file":{"read"},"write_file":{"write"},"delete_file":{"write","destructive"},"run_shell":{"execute"},}七、机制 5:运行时权限,默认拒绝
回到开头的事故。永远不要用 prompt 实现安全。
✅ 能力制权限 + 默认拒绝
classPolicyDenied(Exception):...@dataclassclassPolicy:allowed:set[str]# 例如 {"read", "write"}auto_approve_destructive:bool=Falsedefenforce(policy:Policy,tool_name:str,args:dict)->None:required=CAPABILITIES[tool_name]# 1) 默认拒绝:能力不在白名单里,直接拦下ifnotrequired<=policy.allowed:raisePolicyDenied(f"工具{tool_name}需要能力{required},"f"当前策略仅允许{policy.allowed}")# 2) 不可逆操作必须有人类审批闸门if"destructive"inrequiredandnotpolicy.auto_approve_destructive:raiseNeedsHumanApproval(tool=tool_name,args=args,reason="不可逆操作")配套要求:
- 代码执行必须在沙箱里(容器 / 受限文件系统 / 无网络)
- 有审计轨迹能回答"它到底做了什么"——这又回到了机制 1
- 审批队列要有超时策略,不能无限等人
核心原则:默认安全(deny by default)。白名单而不是黑名单——因为你永远列不全"危险操作"。
八、机制 6:持久化与时间旅行
Demo 的 Agent 状态在内存里,进程一死全没了。
生产级要求
| 能力 | 价值 |
|---|---|
| 状态可序列化 | 崩溃后能恢复,长任务不丢进度 |
| 可从任意检查点 fork | 调试时能从出错点分叉,反复试不同策略 |
| 区分持久状态与临时上下文 | 前者是事实,后者是每次重建的视图 |
@dataclassclassAgentState:"""必须是可序列化的纯数据——不要塞进不可序列化的对象。"""run_id:strevents:list[AgentEvent]cursor:intbudget_used:dictdeffork(self)->"AgentState":"""从当前状态分叉出一条独立分支,用于调试/对比。"""returnAgentState(run_id=f"{self.run_id}-fork-{uuid4().hex[:6]}",events=list(self.events),cursor=self.cursor,budget_used=dict(self.budget_used),)时间旅行调试是 Agent 领域目前最缺的能力之一。传统调试器有gdb,Agent 却没有对应物——而这恰恰是生产环境最需要的(错误往往在第 5 步显现,根因在第 1 步)。
九、机制 7:把非确定性收缩到最小边界
这是架构层面最核心的一条原则。
Agent 项目的根本困难就是非确定性。业余做法是"接受它";专业做法是把它围起来:
# 1) 显式 pin 住模型版本与参数MODEL="your-model-2024-11-01"# 不要用会漂移的 latest 别名TEMPERATURE=0.0SEED=42# 2) 所有模型 I/O 全部记录@dataclassclassModelCall:model:strparams:dictmessages:list[dict]response:dict# 完整记录,用于回放defreplay(self)->dict:returnself.response这样做的收益是决定性的:你能复现 bug 了。
线上出问题 → 拿到事件流 → 本地回放 → 精确定位。没有这条,你的 Agent 线上出问题就只能靠猜。
判断标准:把你项目里"非确定的部分"圈出来,如果它占了超过一小块,或者没有被完整记录,那就是架构问题。
十、机制 8:不调真实 LLM 就能测试主循环
我拿这条当作架构正确性的终极检验:
如果你的架构做不到无 LLM 测试,那是架构错了,不是测试难写。
理由很朴素:如果测试一次要花钱、且结果随机,你就不会测试;不测试的 Agent 项目一定会退化。
而且这条能倒逼架构——它会强迫你把模型调用抽象成一个可替换的接缝:
fromtypingimportProtocolclassModelClient(Protocol):defchat(self,messages:list[dict],**kw)->dict:...classFakeModel:"""按剧本返回,完全确定,零成本。"""def__init__(self,script:list[dict]):self.script,self.i=script,0defchat(self,messages,**kw)->dict:resp=self.script[self.i]self.i+=1returnresp# 现在可以廉价地测试那些真正重要的机制deftest_agent_stops_on_budget():agent=Agent(model=FakeModel([{"tool_calls":[{"name":"read_file","args":{"path":"a"}}]}]*100),budget=Budget(max_steps=3),)withpytest.raises(BudgetExceeded):agent.run("随便什么任务")deftest_destructive_tool_needs_approval():agent=Agent(model=FakeModel([{"tool_calls":[{"name":"delete_file","args":{"path":"/db"}}]}]),policy=Policy(allowed={"read","write"}))withpytest.raises(NeedsHumanApproval):agent.run("清理一下")这两个测试跑起来只花毫秒、不花一分钱,却覆盖了本文最重要的两条机制。
这就是"架构正确"带来的复利。
结语
回到最开始那句话:
Demo 靠功能,生产靠失败模式。
优秀的 Agent 项目不是"功能更多"的项目,而是**“出错方式更可控”**的项目。它们的设计出发点是:
- 机制 > 提示—— 能用运行时保证的,不要用 prompt 请求
- 非确定性收缩到最小边界并全部记录—— 这是可复现、可测试的前提
- 默认安全—— 白名单,不是黑名单
- 日志即本体—— 事件流是事实来源,内存状态不是
- 为失败设计,不为演示设计
- 必须能在无 LLM 的情况下测试
最后补一句题外话:这六条同时也是面试里最常被追问的点。
能写出炫酷 Demo 的人很多,能把这六条答清楚的人很少——而后者的稀缺性,才是真正的护城河。
参考资料
- The Log Is the Agent: Event-Sourced Reactive Graphs for Auditable, Forkable Agentic Systems
- Anthropic,Agent Harness Design: 3 Patterns for Harnessing Claude’s Intelligence— https://claude.com/blog/harnessing-claudes-intelligence
- Anthropic,The anatomy of effective commerce agents— https://claude.com/blog/the-anatomy-of-effective-commerce-agents
- 12-Factor Agents(humanlayer)
- Inngest,Building Durable AI Agents: A Guide to Context Engineering— https://inngest.vercel.app/blog/building-durable-agents
- Temporal,LangGraph Plugin adds Durable Execution— https://temporal.io/blog/temporal-langgraph-plugin-durable-execution
- Agent-Governed Lossless Context Folding — https://zenodo.org/records/21856874/files/pi-fold-context-folding.pdf
- OpenTelemetry GenAI Semantic Conventions
- Replit Agent 数据库删除事故复盘 — https://safeguard.sh/resources/blog/replit-agent-database-deletion-vibe-coding-2025