ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

状态机驱动的可控Agent:解决大模型流程失控的工程实践

2026/9/11 5:11:45 拓冰建站 浏览量
状态机驱动的可控Agent:解决大模型流程失控的工程实践 最近在带一个智能体项目团队里最常被问的一个问题就是Agent 跑着跑着就失控了怎么办。不是模型能力不够而是执行链路太自由。让大模型在一个开放式任务里自己决定下一步干什么看起来很美但真正落到业务里尤其是涉及订单、审批、工单这类有明确流程的场景自由发挥就是灾难。它会漏步骤、会跳流程、会在不该调工具的时候调工具而且一旦出错你连“它现在到底走到哪一步了”都说不清楚。后来我们换了个思路把流程控制权从大模型手里拿回来交给状态机。大模型只负责它擅长的事——理解自然语言、抽取信息、做局部判断至于下一步该进哪个状态、哪些状态是合法的、哪些状态绝对不能返回全部由状态机说了算。这一换项目立刻从“炼丹”变成“写业务代码”可控性上了一个大台阶。这篇文章就是针对“第一个 Agent 应用”的阶段总结主题是“使用状态机开发一个可控 Agent”。我会从为什么需要状态机讲起到状态建模、核心代码实现再到实际运行中的翻车场景完整走一遍。适合正在做 Agent 开发、但被失控问题困扰的工程师也适合刚接触智能体开发、想找一种工程化落地方式的同学。1. 为什么 Agent 需要状态机先解决“失控”这个老问题1.1 让大模型自由发挥的代价很多人第一次做 Agent 都是这么起步的给大模型一个系统提示词配上几个工具函数然后循环“思考→调用工具→观察结果→再思考”直到模型自己说“任务完成”。这种 ReAct 模式确实灵活但灵活和失控往往是同一个词的两面。举个例子。我们做一个内部工单 Agent任务是收集用户反馈并创建工单。自由模式下大模型可能会在用户还没有提供完必要信息时就去调创建工单接口结果工单里缺字段或者用户想先咨询再创建模型却直接跳过了咨询环节再或者用户在对话中改变了主意模型完全可能在一段历史上下文中搞混当前意图做出一个和用户期望完全不符的操作。这类问题不是调 prompt 能解决的。因为根源在于责任边界不清。自由循环模式下大模型既当裁判又当运动员它自己决定流程走到哪、什么时候结束、能不能回溯。可大模型的概率输出决定了它做这种强约束决策时天生就有不可控的成分。流程控制需要的不是概率而是确定性。1.2 状态机能收敛什么复杂度状态机的思路恰好相反流程逻辑是写死的大模型只在规定好的“状态槽位”里做判断。状态、事件、迁移、动作四个要素把整个流程画成一张明确的图Agent 的每一步都落在某条迁移边上不会有图外行为。这里我用一个生活化的类比。自由模式的 Agent 像一个没有 SOP 的实习生你告诉他“去办一件事”他靠理解力临场发挥办好了是运气办砸了你连怎么复盘都不知道。状态机 Agent 则像一个严格执行 SOP 的员工每一步做什么、做完交给谁、什么情况下可以跳过、什么情况下必须上报都是白纸黑字的他只是在一个个节点上做选择题但选择范围已经被限死了。把流程控制交给状态机换来三个实打实的好处。第一是可观测性任何时候问“Agent 现在到哪一步了”都有精确答案还能输出状态迁移日志第二是可恢复性状态机可以定义失败状态和兜底动作中间挂了可以从上一个稳定状态重试而不是一错到底第三是可测试性迁移表是纯逻辑可以脱离大模型单独写单元测试这在 AI 工程里是巨大的优势。2. 动手之前的状态建模把一个 Agent 任务翻译成状态图2.1 选对业务场景状态机不是所有 Agent 场景的银弹它尤其适合那些“步骤固定、环节有先后约束、失败需要处理”的业务。比如下单结算、工单流转、审批流程、多轮信息收集这一类任务天然就是状态机的主场。我建议你第一个状态机 Agent 别选太复杂的挑一个“有一条清晰主链路外加一条取消/失败分支”的场景即可。下面我用一个下单 Agent 贯穿全文来讲用户描述想买什么Agent 负责查询库存、确认价格、收集收货信息、下单支付这个流程。这个场景简单直观状态不多但足够展示状态机的核心价值。2.2 从需求里提炼状态、事件和动作建模的第一步是列状态。所谓状态就是 Agent 在某个时刻所处的稳定局面它必须能回答“目前进行到哪一步了”这个问题。以下单场景为例我梳理出这么几个状态IDLE空闲状态等待用户发起需求QUERYING查询商品信息和库存CONFIRMING与用户确认商品、价格、数量COLLECTING_ADDRESS收集收货地址和联系方式PAYING处理支付环节DONE流程正常完成CANCELLED用户主动取消或流程失败终止状态列好之后列事件。事件是“触发迁移的信号”通常对应 Agent 接收到的外部输入或内部判定结果。比如“用户表达了购买意图”“库存查询完成”“用户确认订单”“用户取消”“支付成功”“支付失败”。动作则是进入某个状态后要做的事比如“调用商品库查询接口”“调用支付接口”“向用户提问”。在设计状态时可以借鉴嵌入式状态机里常说的“三段式”思路正常流程段、异常处理段、收尾段。很多新手只画了晴天路径——一切顺利一直到完成结果一旦用户说“等一下地址填错了”或者“支付超时”整个状态图就无处可去Agent 直接卡死。所以从建模第一天起把“取消”和“失败”当成一等公民它们不是异常它们是流程的一部分。2.3 用迁移表把状态图写清楚有了状态和事件接下来最有力的一步把状态图转成迁移表。迁移表是状态机的真身代码是表的解释器。以下单 Agent 的迁移表为例当前状态事件下一状态动作IDLE用户表达购买意图QUERYING解析商品关键词调查询接口QUERYING查询成功且有库存CONFIRMING组装商品信息向用户确认QUERYING查询失败或库存不足CANCELLED告知用户无货流程终止CONFIRMING用户确认COLLECTING_ADDRESS询问收货地址CONFIRMING用户修改信息QUERYING重新查询并再次确认CONFIRMING用户取消CANCELLED记录取消原因COLLECTING_ADDRESS地址信息完整PAYING生成支付订单PAYING支付成功DONE输出订单号流程完成PAYING支付失败CANCELLED提示失败支持重试这张表看起来平平无奇但它就是整个 Agent 的“宪法”。所有代码都围绕这张表展开增删一个流程只需要改表格不需要去大模型 prompt 里“碰运气”。这也是热词里“表驱动状态机”的核心思想状态机逻辑与业务动作解耦表是数据动作是函数架构清爽。2.4 不要把所有东西都塞进状态机有个容易走偏的地方我需要提醒一下状态机负责的是流程控制不是语义理解。不要试图把大模型的每一个细微动作都定义成状态那样状态数会爆炸状态机会变成一张谁都看不懂的蜘蛛网。合理的方法是粗粒度控制流程细粒度留给大模型。以下单 Agent 为例“理解用户这句话是在确认还是在修改信息”这件事不需要用状态机去穷举那是大模型该干的活但“确认之后下一步只能去收集地址不能跳去支付”这事必须由状态机锁死。简单说状态机管骨架大模型管血肉各干各擅长的部分。3. 核心代码实现用 Python 写一个可控的 Agent3.1 工程结构与目录这里我用 Python 来实现一个最小但完整的版本不依赖重型框架核心状态机引擎不到一百行加上 Agent 逻辑也不过两百行左右。先看工程结构order_agent/ ├── state_machine.py # 状态机核心引擎 ├── agent.py # Agent 执行体负责调用 LLM 和工具 ├── actions.py # 具体动作函数如查询库存、创建订单 ├── prompts.py # LLM 提示词模板 └── main.py # 入口跑一个完整示例这个拆分方式建议保留。把状态机引擎和业务动作分开最大的好处是状态机的迁移逻辑可以单独测试不需要拉起整个 Agent 服务而动作函数是纯业务代码也可以用常规单测覆盖。将来如果要换大模型供应商只动 agent.py 和 prompts.py状态机引擎完全不用碰。3.2 状态机核心引擎状态机引擎的设计要简单、通用别整那些花哨的框架。核心就一张迁移表加一个当前状态。# state_machine.py from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional dataclass class Transition: event: str next_state: str action: Optional[Callable[..., Any]] None class StateMachine: 一个极简的表驱动状态机。 核心思路所有迁移规则都放在 transitions 字典里 格式为 {当前状态: [Transition, ...]}。 触发事件后引擎查找匹配的迁移执行动作切换状态。 def __init__( self, states: List[str], transitions: Dict[str, List[Transition]], initial_state: str, ): self.states states self.transitions transitions self.state initial_state self.history: List[Dict[str, str]] [] def can_trigger(self, event: str) - bool: 判断当前状态是否允许触发某个事件。 for t in self.transitions.get(self.state, []): if t.event event: return True return False def trigger(self, event: str, context: Optional[Dict[str, Any]] None) - Any: 触发一个事件执行迁移。 如果当前状态没有匹配的迁移则抛出异常避免静默失败。 参数: event: 事件名 context: 传给动作函数的上下文通常是收集到的业务数据 if not self.can_trigger(event): raise ValueError( fIllegal transition: 状态 {self.state} 不允许事件 {event} ) for t in self.transitions.get(self.state, []): if t.event event: self.history.append({ from: self.state, event: event, to: t.next_state, }) self.state t.next_state if t.action: return t.action(**(context or {})) return None raise ValueError(f未找到迁移: {self.state} - {event})这个引擎最关键的约束在trigger里当前状态不允许的事件直接抛异常而不是默默忽略。这一点极其重要。Agent 场景里大模型经常给出意料之外的输出如果你在解析层不设防错误的事件照样往后走会让状态机名存实亡。宁可抛异常让上层兜底也不能让非法迁移发生。3.3 接入大模型的 Agent 执行体状态机引擎是骨架Agent 执行体就是给骨架供血的心脏。它的职责是接收用户的自然语言输入 → 判断当前状态下该做什么 → 调用大模型提取关键信息 → 决定触发哪个事件 → 执行迁移。# agent.py from typing import Dict, Any from state_machine import StateMachine, Transition class OrderAgent: def __init__( self, machine: StateMachine, llm_client: Any, # 传入你常用的大模型客户端 ): self.machine machine self.llm llm_client # 这里存放整个流程中收集到的业务数据 # 比如商品关键词、价格、收货地址、订单号等 self.context: Dict[str, Any] {} def _parse_with_llm(self, user_input: str) - Dict[str, str]: 用大模型解析用户输入返回结构化的意图信息。 返回示例: {intent: purchase, product: 机械键盘, quantity: 1} 注意这里只做大模型的语义抽取不涉及流程控制。 prompt f 你是订单助手的意图解析器请从用户输入中提取 - intent: 用户意图只能是 purchase(购买)/confirm(确认)/ modify(修改)/cancel(取消)/provide_address(提供地址) - product: 商品关键词如果提到 - quantity: 数量如果提到 - address: 收货地址如果提到 用户输入: {user_input} 只输出 JSON不要多余解释。 resp self.llm.chat(prompt) # 这里假设 resp 已经是解析好的 JSON 对象 return resp def handle(self, user_input: str): 处理用户一条消息驱动状态机流转。 state self.machine.state parsed self._parse_with_llm(user_input) intent parsed.get(intent, ) # 把解析结果写入共享上下文 self.context.update({k: v for k, v in parsed.items() if v}) # 根据当前状态和解析到的意图映射到状态机事件 if state IDLE and intent purchase: self.context[product] parsed.get(product) self.machine.trigger(user_intends_buy, self.context) elif state CONFIRMING and intent confirm: self.machine.trigger(user_confirmed, self.context) elif state CONFIRMING and intent modify: self.machine.trigger(user_modified, self.context) elif state CONFIRMING and intent cancel: self.machine.trigger(user_cancelled, self.context) elif state COLLECTING_ADDRESS and intent provide_address: self.context[address] parsed.get(address) self.machine.trigger(address_provided, self.context) else: # 无法映射的输入直接回复当前需要什么 return f我现在需要你提供的是{self._current_state_hint()} return self.context def _current_state_hint(self) - str: hints { QUERYING: 正在查询商品请稍等, CONFIRMING: 请确认商品和数量或告诉我你想修改的地方, COLLECTING_ADDRESS: 请提供收货地址和联系电话, PAYING: 正在创建支付订单请确认支付, DONE: 订单已完成, CANCELLED: 订单已取消, } return hints.get(self.machine.state, 处理中)注意这里有个设计细节_parse_with_llm的输出只是“事件候选”真正能不能触发是由状态机判断的。就算大模型犯糊涂把一个“确认”意图在当前状态解析出来了如果当前状态是 COLLECTING_ADDRESS这一事件根本不在迁移表里就会走 else 兜底分支提醒用户该提供地址。这是状态机 Agent 和大模型“相互制衡”的典型写法。3.4 动作函数集把状态迁移落到真实操作上光迁移状态没有意义真正的业务发生在动作函数里。以下单 Agent 为例需要的动作很简单查库存、算价格、生成订单、模拟支付。# actions.py def query_product(product: str, quantity: int 1) - dict: 模拟查询商品库存和价格。 # 真实项目里这里会调商品中心接口 product_db { 机械键盘: {price: 399, stock: 10}, 无线鼠标: {price: 89, stock: 0}, 显示器: {price: 1299, stock: 5}, } info product_db.get(product) if not info: return {ok: False, reason: 商品不存在} if info[stock] quantity: return {ok: False, reason: 库存不足} return {ok: True, price: info[price] * quantity, stock: info[stock]} def save_address(address: str, **kwargs) - dict: 保存收货地址。真实场景会写库这里只打印。 return {ok: True, saved_address: address} def create_order(product: str, quantity: int, address: str, **kwargs) - dict: 创建订单返回订单号。 # 真实场景会调订单服务生成订单并锁定库存 order_id fSO{product[:2]}{quantity}{len(address)} return {ok: True, order_id: order_id} def mock_pay(order_id: str, **kwargs) - dict: 模拟支付。真实场景会拉起收银台。 return {ok: True, paid: True}这些动作函数有一个共性它们都是纯业务、纯确定的没有大模型的参与。状态机到状态机之间该干什么早就写死了大模型插不进手。这也是可控性的来源——关键业务动作不是“生成”的而是“执行”的。3.5 把它们组装起来跑通一个完整流程现在我们把引擎、Agent、动作函数组装到一起在主程序里注册迁移表并跑一遍流程。# main.py from state_machine import StateMachine, Transition from agent import OrderAgent import actions def build_order_machine() - StateMachine: 构建下单 Agent 的状态机。 states [ IDLE, QUERYING, CONFIRMING, COLLECTING_ADDRESS, PAYING, DONE, CANCELLED, ] transitions { IDLE: [ Transition(user_intends_buy, QUERYING, actions.query_product), ], QUERYING: [ Transition(query_success, CONFIRMING), Transition(query_failed, CANCELLED), ], CONFIRMING: [ Transition(user_confirmed, COLLECTING_ADDRESS), Transition(user_modified, QUERYING, actions.query_product), Transition(user_cancelled, CANCELLED), ], COLLECTING_ADDRESS: [ Transition(address_provided, PAYING, actions.create_order), ], PAYING: [ Transition(pay_success, DONE, actions.mock_pay), Transition(pay_failed, CANCELLED), ], } return StateMachine(states, transitions, initial_stateIDLE) class MockLLM: 开发期用 Mock 大模型方便调试。 def __init__(self): self.current_intent purchase def chat(self, prompt: str): # 真实项目里这里会调用 OpenAI/Claude/本地模型 # 开发期我们可以根据状态机的阶段返回固定的意图 import json if 机械键盘 in prompt: resp {intent: purchase, product: 机械键盘, quantity: 1} elif 确认 in prompt: resp {intent: confirm} elif 地址 in prompt: resp {intent: provide_address, address: 北京市海淀区中关村大街1号} else: resp {intent: cancel} return resp if __name__ __main__: machine build_order_machine() llm MockLLM() agent OrderAgent(machine, llm) # 模拟用户一轮对话 print(状态:, agent.machine.state) print(agent.handle(我想买一个机械键盘)) print(状态:, agent.machine.state) print(agent.handle(确认)) print(状态:, agent.machine.state) print(agent.handle(地址是北京市海淀区中关村大街1号)) print(状态:, agent.machine.state) # 查看状态迁移历史 print(历史:, agent.machine.history)跑完这个示例后你会看到状态按预期一路从 IDLE 走到 DONE整个过程里没有任何一步是“超出预期”的。你可能觉得这跟传统软件没什么区别——对这正是状态机 Agent 追求的让 AI 应用在流程上像传统软件一样可靠。3.6 工程化细节日志、超时、重试与人工介入示例能跑通只是第一步上线运行还有几个绕不开的工程化问题。第一是日志。状态机的 history 一定要持续记录每次迁移都带上时间戳和上下文关键字段。我习惯把它接到日志系统里方便故障时回放“这个订单当时是怎么一步步走过来的”。第二是超时。Agent 流程里最容易出问题的是等待用户输入。用户可能聊到一半就不回了这时候你得有超时策略比如 PAYING 状态超过 15 分钟没收到支付结果自动迁移到 CANCELLED并给用户发一条提醒。状态机引擎可以加一个trigger_with_timeout但更简单的做法是在业务层定时扫描超时状态。第三是人工介入。状态机再严谨也挡不住真实世界的奇葩订单。我会在设计所有状态机 Agent 时都预留一个EScalated状态任何迁移表里没覆盖的情况最后都可以升级到人工处理。这个状态不是摆设它是你最后的兜底能让你在线上睡得着觉。4. 状态机 Agent 的常见翻车现场与排查技巧4.1 典型问题速查表我把实际踩过的坑整理成一张表对照着看能省不少排查时间。症状根本原因解决方案Agent 在某一步卡住不动事件名映射不完整大模型输出的意图没有对应的事件在解析层加“意图→事件”映射表映射不到就兜底询问而不是静默非法迁移直接抛异常迁移表漏了某个合法路径把“用户取消”“失败重试”等路径提前设计进迁移表大模型乱填字段提示词没有约束输出格式用结构化输出 / 函数调用模式让模型输出固定 JSON Schema流程重入导致上下文污染同一轮对话里手动调了多次 handle状态机只允许事件驱动迁移不要直接改 state 字段用户中途改变需求自由流程下模型搞混历史意图状态机天然支持在 CONFIRMING 状态通过 user_modified 回退到 QUERYING状态机日志丢失只打日志不落库迁移历史写入独立表订单维度可查询4.2 死锁和循环迁移怎么处理状态机 Agent 最常见的“看起来活着实际死了”的问题就是来回迁移。比如用户在 CONFIRMING 状态反复说“改一下”Agent 就不断从 CONFIRMING 迁到 QUERYING 再迁回来如果不加限制这个循环可以永远跑下去。处理办法是加“迁移次数上限”。每次迁移时带上一个计数器同一个会话里如果来回迁移超过了 N 次比如 3 次状态机自动迁移到“HUMAN_ESCALATION”状态提示用户“修改次数过多已转人工”。这既保护了系统也给了用户体验一个交代。4.3 上下文污染是隐蔽杀手我在真实项目里踩过最大的坑不是状态机本身而是状态机的共享上下文被污染了。上面代码里Agent 有一个self.context字典用来在各个状态之间传递数据。问题在于大模型解析用户输入时可能把旧信息解析到新字段里或者用户一次说了多个信息把不需要的字段也塞进来了。比如用户在 COLLECTING_ADDRESS 阶段说了一句“对了我上次说的键盘不要了改成鼠标”。这句话里既有修改意图也有新的商品信息。如果解析层把“鼠标”更新进了 context但状态机还在等待地址那么地址收集完成后创建订单时订单里可能是鼠标而不是键盘。这种问题没有银弹但有一个非常有效的习惯每个状态只消费它关心的字段下游动作函数只从 context 里读自己需要的 key并且一个 key 一旦被“消费”了比如用于创建订单了就立刻从 context 里标记为已使用防止被后续状态重复读取。4.4 状态判定和解析出现偏差时怎么办状态机的严谨性依赖一件事事件是从哪里来的。我见过不少人直接在 handle 里写一套 if-else 去判断意图结果意图解析错事件就错状态机再严谨也拦不住一个错误的事件。所以正确的做法是意图解析和大模型必须“确认过眼神”再触发。实践中我建议两层设计。第一层大模型输出必须用结构化格式不要让它自由文本输出第二层在触发事件之前设定“必备字段校验”。比如要触发user_confirmed必须保证 context 里有 product 和 price要触发address_provided必须保证 context 里有 address。校验不通过就留在原状态并提示用户补全信息。这样即使大模型解析错了状态机也不会带着残缺数据往前走。5. 状态机 Agent 的边界什么时候该用什么时候别用5.1 状态机和自由式 ReAct 的对比说到这你可能会想是不是所有 Agent 都应该套状态机不是。状态机适合有稳定流程的业务但它也有限制——它本质上是“预设路径”的你没法让它在完全没有路径的地方自己“探索”。我整理了一个对比维度状态机 Agent自由式 ReAct Agent流程确定性高路径写死低模型自由决策可观测性高随时知道状态低只能看思考过程失败恢复可从稳定状态重试难可能一错到底对未知场景的适应性低没建模的流程走不了高可以临场发挥调试难度低迁移表是明确约束高需要反复调 prompt适合场景下单、审批、工单、信息收集开放问答、研究探索、代码生成5.2 推荐的混合路线状态机管骨架大模型做局部决策如果你想要两者的优点我更推荐混合架构外层用状态机控制主流程内层在某个具体步骤里用大模型做开放式判断。举个例子状态机走到 CONFIRMING 时需要判断用户这句话到底是对订单内容的确认还是修改意见还是取消意图。这个判断适合大模型做因为自然语言千变万化但它做出的判断永远不会直接触发一个“非法路径”因为它输出的事件要经过迁移表校验才能生效。如果用户说的太模糊解析层返回“无法确定意图”状态机就停留在原状态继续追问。我之前把这个模式用在一个售前咨询 Agent 上外层状态包括“识别需求→提供方案→处理异议→完成意向确认→转销售”其中“处理异议”这一步完全交给大模型让它自由回答用户的技术疑问。因为异议处理本身没有固定路径但整条销售流程必须可控。这套架构跑下来销售团队的接受度非常高因为管理者能看清每一个线索走到了哪一步、卡在哪一步。5.3 我的几点实操体会文章最后说点个人经验算不上总结就是几个值得注意的点。状态机的第一版别贪大。我见过很多人一开始就把状态图画到十几个状态、几十条迁移结果代码写不下去。建议先画主链路 5 个状态以内跑通了再逐步加分支。状态只加不减很容易减掉一个状态往往牵一发动全身。给每个迁移动作写单元测试。状态机引擎是确定性的意味着你可以把迁移表和动作函数从 Agent 里拆出来单独测。这个测试成本极低收益率极高。我们团队现在所有状态机 Agent 的迁移测试覆盖率要求 100%这是整个项目里唯一敢说“雷打不动”的质量红线。大模型的角色一定要收敛。不要让它做流程控制不要让它决定“下一步是什么状态”。它的职责只有一个在给定的状态里理解输入、填充信息。一开始这么设计有点反直觉因为大家都觉得 Agent 就该“自主”但你一旦试过被大模型带偏流程的滋味就会回来拥抱状态机。我现在的习惯是拿到一个新业务第一件事不是写 prompt而是画迁移表。流程图一画清楚Agent 的骨架就立起来了剩下的只是填肉。方向对了事情就成了一半。