一、引言Agent 的“自主性”正在变成事故现场过去一年几乎所有的 LLM 应用都在往 Agent 方向演进。ReAct、Plan-and-Execute、Reflexion 这些范式让模型不再只是“一问一答”而是能自己拆解目标、调用工具、观察结果、修正计划。听起来很美但真正把它放到生产环境里跑长链路任务时工程师们很快就会发现三个绕不开的问题第一任务漂移。用户让它“调研三家竞品并输出对比报告”跑到第 12 步时Agent 突然开始纠结某家公司的融资历史然后又自作主张地去查了行业政策最后生成的报告里竞品对比只占了两段。这不是模型“笨”而是每一轮决策它都在重新理解“我现在该干什么”而上下文的噪声会不断稀释最初的目标。更麻烦的是这种漂移往往不是“突然发生”而是“逐渐累积”每一轮 LLM 都会基于前一轮的输出再生成下一轮的推理误差像滚雪球一样被放大。到了第 12 步模型眼里唯一的上下文可能就是最近几条工具返回的融资新闻最初那个“对比三家竞品功能与价格”的目标早被挤出了有效注意力窗口。第二失败即归零。一个 20 步的任务在第 18 步调用外部 API 超时了。因为没有中间状态持久化唯一的选择就是从头再来——重新花钱、重新耗时而且第 1 到 17 步里那些已经产生副作用的操作发了邮件、创建了工单、扣了库存会被重复执行一遍。在真实业务里重复发一封客户通知邮件可能只是尴尬重复创建一张工单可能只是脏数据但重复扣一次库存或者重复发起一笔支付那就是实打实的资金事故了。第三不可观测、不可干预。你没法知道 Agent 现在处于哪个阶段也没法在它即将“干坏事”之前拦一下。调试基本靠打印日志 猜。当产品经理问你“这个任务现在卡在哪了”你只能把长长的 message 列表贴给他看当合规要求“涉及对外发送的动作必须人工确认”你只能在 prompt 里写一句“请谨慎操作”——然后祈祷模型听话。这些问题的根因其实是同一个我们把控制流交给了概率模型。什么时候该继续、什么时候该结束、失败了怎么办全由 LLM 每轮临场发挥。而工程系统需要的恰恰是确定性。一个很朴素的类比LLM 就像一位能力很强但注意力容易分散的顾问你让他“看着办”他就会自由发挥但如果你给他一张流程图告诉他“你现在只在‘收集资料’这一格做完这一步就交给下一格的人”他就会老老实实聚焦在眼前这一格里。流程的确定性由代码保证语义的灵活性由模型提供——这才是把 Agent 推向生产的关键分界线。这篇文章讲一个已经被验证过的解法用有限状态机FSM定义确定性骨架用任务检查点Checkpoint实现可恢复执行。LLM 依然是核心但它只负责“语义判断”这一件事不再负责流程编排。二、核心原理让确定性归确定性让概率归概率2.1 状态机把“流程”从模型脑子里搬到代码里一个有限状态机由四要素构成States状态Agent 当前处于哪个阶段比如PLANNING、SEARCHING、DRAFTING、REVIEWING、DONE、FAILED。Events事件驱动状态转移的信号来自 LLM 决策、工具返回、人工输入或定时器。Transitions转移(src, event) - dst的有向边。Guards / Actions守卫与动作守卫决定这条边能不能走比如“草稿字数 ≥ 500 才允许进入评审”动作是进入/离开状态时要执行的副作用。关键在于职责切分状态机负责“能不能走、往哪走”LLM 负责“在这条边上做语义决策”。比如在SEARCHING状态LLM 的任务是“根据当前查询决定下一个搜索关键词”而不是“决定接下来该不该搜索”。后者是状态机的事。这个思路在 LangGraph 里体现为StateGraph在 Temporal 里体现为 Workflow Definition在 OpenAI 的 Assistants 里体现为 Run Steps——本质都是一样的先声明图再让模型在节点里干活。这里有一个容易被忽视但极其重要的性质状态机是可穷举、可测试的。因为状态和事件都是有限集合你完全可以在单元测试里枚举所有(状态, 事件)组合验证“非法转移一定会被拒绝”“守卫条件一定生效”。这是概率模型永远做不到的——你没法给 LLM 写一个断言“它一定不会在第 5 步发邮件”。把流程写进状态机本质上就是把原来“无法测试”的行为转化成了“可以像普通业务逻辑一样测试”的行为。再补一个实践中的判断标准如果一个决策的答案只有有限几种就该放进状态机如果答案是一个开放式内容才交给 LLM。“下一步该搜索还是该起草”只有两种答案属于状态机“下一个搜索词是什么”是开放式的属于 LLM。很多团队把 Agent 做崩就是因为把前者也交给了 LLM于是模型每天都能发明出新的“下一步”。2.2 检查点可恢复的前提是“状态可序列化”检查点不是什么新概念数据库的 WAL、Kafka 的 offset、Flink 的 snapshot 都是它。放到 Agent 场景一个检查点至少要包含四类信息执行位置当前状态 步骤序号 下一步待处理的事件队列。业务上下文任务目标、已产出的中间结果、当前对话摘要注意不是全量 messages。副作用账本已经执行过的工具调用记录带幂等键用于恢复时去重。元信息版本号、创建时间、重试次数、trace_id。有了它“恢复”就退化成一次普通的load - 继续跑。这里有一个必须想清楚的问题恢复语义到底是 at-least-once 还是 exactly-once答案是分布式系统里真正的 exactly-once 不存在我们能做的是effectively-once——即“至少执行一次工具调用但通过幂等键让副作用只生效一次”。关于第 2 点里的“对话摘要”还有一个细节值得展开不要把全量 messages 塞进检查点。一是体积会随步数线性膨胀二是全量历史会稀释模型注意力反而加剧任务漂移。更好的做法是让检查点里保存一份**“滚动摘要”**——每完成一个状态就由 LLM 把该状态产生的关键信息压缩成一两句话追加上去同时丢掉原始消息。这样检查点体积可控恢复后模型看到的也是“精华版上下文”决策质量更稳。举一个具体的幂等键设计例子假设工具是“给客户发通知邮件”那么幂等键可以设计成hash(run_id customer_id notification_type)而不是hash(run_id step)。区别在于前者保证“同一个任务对同一个客户同类型通知只发一次”即使中途步骤号变了也不会重复发后者只要步骤漂移就会重发。幂等键要绑定业务语义而不是绑定执行时序这是很多团队恢复时踩坑最多的地方。2.3 两者结合确定性骨架 概率性肌肉用户目标 │ ▼ ┌─────────────┐ event: need_research ┌─────────────┐ │ PLANNING │ ──────────────────────▶ │ SEARCHING │ └─────────────┘ └─────────────┘ ▲ │ event: enough_info │ event: plan_invalid ▼ │ ┌─────────────┐ └────────────────────────────────│ DRAFTING │ └─────────────┘ │ event: draft_ready ┌────────────────────┼────────────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ REVIEWING │ │ NEED_HUMAN │ │ RETRY │ └─────────────┘ └─────────────┘ └─────────────┘ │ event: approved ▼ ┌─────────────┐ │ DONE │ └─────────────┘每一根箭头都是可审计、可测试、可中断的。LLM 的自由度被限制在每个节点内部而节点之间的跃迁是确定的。这种“骨架 肌肉”的分层带来三个直接收益可观测任何时刻都能从检查点里读出当前状态、可干预在任何状态后插入人工确认点或强行改写状态、可恢复任何一步崩溃都能从最近的检查点续跑。三、实战案例一个可恢复的竞品调研 Agent假设我们要做一个“竞品调研并生成报告”的 Agent。目标调研 3 家竞品每个竞品需要搜索资料 → 提取要点 → 汇总成报告草稿 → 人工审核 → 输出终稿。3.1 状态机内核 检查点存储先实现一个极简但完整的状态机内核。# fsm_core.pyimportjson,time,osfromdataclassesimportdataclass,field,asdictfromtypingimportCallabledataclassclassCheckpoint:run_id:strstate:strstep:intcontext:dict# 业务上下文目标、中间产物、摘要effects:list[str]field(default_factorylist)# 已执行副作用的幂等键version:strv1created_at:floatfield(default_factorytime.time)classCheckpointStore:生产环境请换成 PostgreSQL / Redis S3这里用文件演示def__init__(self,path./ckpt):self.pathpath os.makedirs(path,exist_okTrue)defsave(self,ckpt:Checkpoint)-None:tmpf{self.path}/{ckpt.run_id}.tmpwithopen(tmp,w)asf:json.dump(asdict(ckpt),f,ensure_asciiFalse)os.replace(tmp,f{self.path}/{ckpt.run_id}.json)# 原子写defload(self,run_id:str)-Checkpoint|None:try:withopen(f{self.path}/{run_id}.json)asf:returnCheckpoint(**json.load(f))exceptFileNotFoundError:returnNonedataclassclassTransition:src:strevent:strdst:strguard:Callable[[dict],bool]lambdactx:Trueaction:Callable[[dict],dict]lambdactx:ctx# 返回新的 contextclassWorkflow:def__init__(self,name:str,initial:str,transitions:list[Transition],store:CheckpointStore):self.name,self.initial,self.storename,initial,store self.transitions{(t.src,t.event):tfortintransitions}defstart(self,run_id:str,ctx:dict)-Checkpoint:ckptCheckpoint(run_idrun_id,stateself.initial,step0,contextctx)self.store.save(ckpt)returnckptdeffire(self,ckpt:Checkpoint,event:str)-Checkpoint:tself.transitions.get((ckpt.state,event))iftisNone:raiseValueError(f非法转移:{ckpt.state}--{event}-- ?)ifnott.guard(ckpt.context):raiseValueError(f守卫拒绝:{ckpt.state}--{event}--{t.dst})ckpt.contextt.action(ckpt.context)ckpt.state,ckpt.stept.dst,ckpt.step1self.store.save(ckpt)# 每次转移后落盘returnckpt这里有两个工程细节值得强调os.replace保证检查点写入的原子性避免进程崩溃时留下半个 JSONfire里先判守卫再执行 action确保被拒绝的转移不会污染上下文。还有一个容易被忽略但很关键的顺序问题落盘必须发生在 action 之后、且发生在“产生外部副作用”之前之后要仔细权衡。对于纯粹的上下文更新比如把摘要写进 context顺序无所谓但对于会调用外部系统的 action正确顺序应该是“先执行副作用并记录幂等键再落盘”。因为如果先落盘再执行副作用中间崩溃就会导致“以为做过了、其实没做”而先执行副作用再落盘最坏情况是恢复时重放一次正好被幂等键挡住。永远选择“可能重复但绝不丢失”的一侧这是恢复语义设计的黄金法则。3.2 用 LLM 做语义决策 幂等工具调用状态机定义好了接下来在每个状态里塞进 LLM 决策。注意这里的关键约束LLM 的输出必须是结构化 JSON且只能输出枚举值。# agents.pyimporthashlib,jsonfromopenaiimportOpenAIfromfsm_coreimportCheckpoint clientOpenAI()# 用 JSON Schema 强约束模型输出杜绝自由发挥DECISION_SCHEMA{type:json_schema,json_schema:{name:search_decision,strict:True,schema:{type:object,properties:{next_query:{type:string},is_sufficient:{type:boolean},reason:{type:string}},required:[next_query,is_sufficient,reason],additionalProperties:False}}}defdecide_next_search(ckpt:Checkpoint)-tuple[str,bool]:在 SEARCHING 状态下由 LLM 决定下一个搜索词以及信息是否足够prompt(f调研目标:{ckpt.context[goal]}\nf当前竞品:{ckpt.context[current_competitor]}\nf已收集资料摘要:{ckpt.context.get(notes,)[:2000]}\n请给出下一个搜索关键词并判断现有信息是否足以撰写该竞品的分析段落。)respclient.chat.completions.create(modelgpt-4o,temperature0,messages[{role:user,content:prompt}],response_formatDECISION_SCHEMA,)datajson.loads(resp.choices[0].message.content)returndata[next_query],data[is_sufficient]defcall_tool_idempotent(ckpt:Checkpoint,tool:str,args:dict):带幂等键的工具调用恢复重放时不会重复产生副作用rawf{ckpt.run_id}:{ckpt.step}:{tool}:{json.dumps(args,sort_keysTrue)}keyhashlib.sha256(raw.encode()).hexdigest()[:16]ifkeyinckpt.effects:# 说明这次调用在崩溃前已经执行过直接跳过避免重复副作用return{status:skipped,reason:already_executed,key:key}resultTOOLS[tool](**args)# 真正的外部调用ckpt.effects.append(key)# 记录副作用幂等键随检查点一起落盘return{status:ok,key:key,result:result}# 简单工具注册表生产环境可换成插件式注册 权限校验TOOLS{web_search:lambdaquery,k5:search_api(query,top_kk),save_section:lambdacompetitor,text:doc_store.upsert(competitor,text),}注意call_tool_idempotent里key的计算把step也带进去了这在“同一状态内多次调用同一工具”时是必要的但如果工具本身具备全局唯一语义如发通知则应改成绑定业务主键。幂等键的设计没有万能公式核心是想清楚“什么叫做重复”。3.3 主循环从检查点恢复执行真正体现价值的是主循环——它既是正常执行路径也是恢复路径。# runner.pyfromfsm_coreimportWorkflow,Transition,CheckpointStorefromagentsimportdecide_next_search,call_tool_idempotentdefbuild_workflow():声明整张图状态、事件、守卫、动作一目了然returnWorkflow(namecompetitor_research,initialPLANNING,transitions[Transition(PLANNING,need_research,SEARCHING,actionlambdac:{**c,notes:}),Transition(SEARCHING,enough_info,DRAFTING),# 守卫草稿必须有实质内容才允许进入评审Transition(DRAFTING,draft_ready,REVIEWING,guardlambdac:len(c.get(draft,))200),Transition(REVIEWING,approved,DONE),Transition(REVIEWING,rejected,DRAFTING),],storeCheckpointStore(),)defrun_or_resume(run_id:str,goal:str,competitors:list[str]):wfbuild_workflow()ckptwf.store.load(run_id)ifckptisNone:# 首次启动初始化上下文ckptwf.start(run_id,{goal:goal,competitors:competitors,current_index:0,notes:,draft:,})print(f[{run_id}] 新建任务从{ckpt.state}开始)else:# 恢复直接续跑前面已执行过的副作用会被幂等键拦截print(f[{run_id}] 检测到检查点从{ckpt.state}step{ckpt.step}恢复)whileckpt.statenotin(DONE,FAILED):handlerSTATE_HANDLERS[ckpt.state]eventhandler(ckpt)# 每个状态内部调用 LLM/工具返回一个确定的事件ckptwf.fire(ckpt,event)# 由状态机决定这条边能不能走print(f[{run_id}] -{ckpt.state}(step{ckpt.step}))print(f[{run_id}] 任务结束终态{ckpt.state})returnckpt STATE_HANDLERS{PLANNING:lambdac:need_research,SEARCHING:search_handler,DRAFTING:draft_handler,REVIEWING:review_handler,}这段代码里有几处设计取舍值得说明每个 handler 只返回“事件字符串”绝不直接改状态。状态迁移的唯一入口是wf.fire这样所有跃迁都会经过守卫校验并落盘。如果让 handler 直接改ckpt.state状态机就形同虚设了。run_or_resume对“首次启动”和“恢复”走同一套循环。这正是检查点设计的价值——恢复不是一个特殊分支而是主路径的自然延续。恢复时不用做任何“跳过已完成步骤”的特殊处理。因为已执行的副作用都记在ckpt.effects里call_tool_idempotent会自动跳过业务代码零感知。3.4 中断与人工介入状态机天然适合做人工审核节点。当REVIEWING状态需要人拍板时把事件来源从“自动”换成“外部输入”即可defreview_handler(ckpt):人审网关此处可以阻塞等待也可以直接挂起等 Webhook 唤醒ifckpt.context.get(auto_approve)andpasses_quality_gate(ckpt.context[draft]):returnapproveddecisionhuman_gate.wait_for_decision(ckpt.run_id)# 阻塞/挂起returnapprovedifdecisionokelserejectedhuman_gate的实现既可以是阻塞轮询也可以是“写入检查点后直接退出进程等人在后台点通过再触发resume”。后者更适合生产——人工审核可能要等几小时不能让服务进程一直挂着。这正好又落回检查点的价值因为状态已经落盘进程可以安全退出被唤醒时从REVIEWING原样恢复。四、常见问题 FAQQ1状态机的状态数量会不会爆炸不会。状态应该按“业务阶段”而不是“具体动作”来划分。一个典型任务 510 个状态足够。如果你发现状态超过 20 个多半是把“循环内的重复动作”也拆成了独立状态此时应该用“子流程”或“循环 索引”来建模而不是继续加状态。Q2LLM 返回了非法事件怎么办两种情况要分开处理。如果返回的是不在枚举里的事件说明结构化输出约束没生效应直接判为可重试错误并记录如果返回的是合法但守卫拒绝的事件则应触发RETRY或NEED_HUMAN而不是无限循环。永远给状态机配上最大重试次数和超时否则一个守卫永远不通过的边会让任务卡死。Q3检查点存哪里小规模用 JSON 文件或 SQLite 足矣生产环境推荐 PostgreSQL 存元信息 对象存储存大块产物如长文档、日志Redis 做热缓存。关键不是存储介质而是保证写入原子性和读取一致性。文件方案里的os.replace、数据库里的事务本质都在解决同一个问题。Q4恢复后模型会不会因为“忘了前面发生了什么”而做出错误决策这正是为什么检查点里要存“业务上下文 滚动摘要”。恢复时你不是把模型重放一遍而是把关键结论喂给它。实践表明恢复后的首轮决策质量取决于摘要质量而不是上下文长度。摘要写得好恢复后几乎与从未中断一致。Q5状态机和 LangGraph、Temporal 这类框架冲突吗不冲突。LangGraph 本身就是状态图Temporal 本身就是持久化工作流引擎。本文手写状态机是为了讲清楚原理真正落地时完全可以也应该用这些框架。理解原理后你会发现它们提供的能力——节点、边、条件转移、持久化、重放——和本文的四要素一一对应。Q6如何处理“整条流程需要回滚”的场景状态机擅长的是“向前恢复”不是“事务回滚”。如果业务要求强回滚比如一组操作要么全成功要么全撤应该用 Saga 模式每个正向动作配一个补偿动作回滚时逆序执行补偿。状态机 检查点负责“记住走到哪”Saga 负责“退回去”。两者经常一起用。五、踩坑与优化建议踩坑一守卫写成“软约束”。很多团队把守卫写成“如果条件不满足就记个日志继续走”结果状态机形同虚设。守卫必须是硬闸门不满足就抛异常或转RETRY绝不静默放行。踩坑二检查点里存了全量 messages。前面提过这会让检查点体积爆炸且加剧漂移。正确做法是存摘要 结构化字段原始消息在日志系统里留档即可用于审计不用于恢复。踩坑三幂等键设计过粗或过细。过粗比如全用run_id会导致本该重试的调用被误跳过过细比如带时间戳则完全失去去重效果。经验值是幂等键应绑定“这次调用在业务上是否算重复”通常由业务主键 动作类型构成。踩坑四恢复时不校验版本。如果代码升级后状态机多了一个状态旧的检查点可能指向一个已不存在的状态。检查点里的version字段就是为此存在的——恢复时版本不匹配要能优雅降级如从最近一个兼容状态重启而不是直接崩溃。优化建议一给每个状态加超时和重试预算。状态机不只是管“往哪走”也要管“走多久还不走就报警”。一个SEARCHING状态超时 5 分钟未产生事件就该触发告警甚至强制转人工。优化建议二把状态转移做成可观测的指标。每个fire都发一条带trace_id的埋点你就能在监控面板上看到“任务在哪个状态堆积最多”“哪条边的失败率最高”。这些是优化 Agent 最有价值的数据。优化建议三用状态机做成本控制。每个状态可以绑定自己的模型简单判断用小模型、生成草稿用大模型和 token 预算超预算就转NEED_HUMAN。这比在单一超长对话里“祈祷模型省着点用”要可靠得多。六、结语把控制流交给概率模型是当前很多 Agent 事故的根源。更多硬核网安与AI工具包请扫码获取完整源码而解法并不复杂用状态机把“流程”钉死在代码里用检查点把“进度”固化到存储里让 LLM 只专注于它真正擅长的语义判断。确定性归确定性概率归概率——这不是给 Agent 戴枷锁而是给它一条走得稳、摔得起、看得见的轨道。当你的 Agent 能在第 18 步崩掉后从第 17 步的检查点毫发无损地续跑能在发送前的最后一步稳稳停下等一个人点头你就知道它终于从一个“有趣的 Demo”变成了一个“可以托付的系统”。