企业官网建设流程全解析

1. 这个问题到底在问什么：当 Agent Loop 的 while 循环开始“卡住”时，你其实已经踩进了系统设计的深水区

“Agent Loop 的 while 循环什么时候开始不够用”——这句话乍看像一句技术吐槽，但背后藏着现代智能体（Agent）开发中最常被低估、最易被忽视的底层架构分水岭。我带过十几支从零搭建 Agent 系统的团队，90% 的人最初都用一个干净利落的while True:启动主循环：接收输入 → 调用 LLM → 解析工具调用 → 执行动作 → 更新状态 → 继续下一轮。它跑得飞快，本地 demo 丝滑如德芙，连调试日志都带着自信的节奏感。直到某天，用户发来一条含糊指令：“帮我查一下上个月所有超时未处理的工单，按部门汇总，再给负责人发邮件提醒，如果附件里有 PDF 报表，顺便转成文字摘要。”——你的 while 循环突然卡在第 3 轮，CPU 占用飙到 95%，日志里反复打印waiting for tool response...，而那个“发邮件”的子任务，其实在第 2 轮就该启动，却因主循环串行阻塞，硬生生拖了 47 秒。

这就是标题所指的“不够用”的真实切口：它不是语法错误，不是性能瓶颈，而是控制流模型与现实任务复杂度之间的结构性失配。你写的不是一段 Python 代码，而是一套隐式状态机；你依赖的不是time.sleep(0.1)，而是对“下一步该做什么”的朴素直觉；你调试的不是函数返回值，而是整个执行上下文在时间维度上的坍塌。关键词里反复出现的状态机、Goose、Claude Code并非偶然——前者是解决该问题的工程范式，后者是当前最典型的实践载体：Claude Code 的编辑器插件本质就是一个轻量级 Agent Runtime，它内部早已弃用裸 while 循环，转而采用基于事件驱动的状态迁移；IEC61850 中的 GOOSE 报文机制，其核心正是用确定性状态跳转替代轮询等待；甚至单片机里“当单片机遇上状态机”的教程，讲的也是如何用有限状态摆脱死循环陷阱。

所以，这个问题真正要回答的，不是“while 循环语法能写多长”，而是：当你的 Agent 开始处理跨步骤、可中断、需回溯、带外部依赖、有超时约束、支持用户中途干预的任务时，那个曾经让你引以为豪的while True:，会在哪几个具体的技术临界点上彻底失效？失效后，你手里的代码会表现出哪些可观察、可复现、可定位的异常信号？又该如何用状态机思维，把混乱的“逻辑流”重构成清晰的“状态流”？这篇文章不讲抽象理论，只拆解我在生产环境里亲手踩过的七个坑，每个坑都对应一个 while 循环崩溃的具体时刻，每个解决方案都来自已上线半年以上的 Agent 服务真实架构。如果你还在用 while 循环写 Agent 主干，建议把手机调成勿扰模式，认真读完这五千字——它可能帮你省下三天的线上故障排查，或者避免一次关键客户的信任崩塌。

2. 为什么裸 while 循环在 Agent 场景中注定失败：从控制流到状态流的本质跃迁

2.1 控制流模型的先天缺陷：它把世界当成一台永不宕机的单线程计算器

我们先回到最基础的while True:结构：

while True: user_input = get_user_input() plan = llm_think(user_input) for step in plan.steps: action = parse_action(step) result = execute_tool(action) update_state(result) send_response(plan.final_answer)

这段代码在教科书里完美诠释了“顺序执行”：输入 → 思考 → 行动 → 输出，一气呵成。但它隐含了五个危险假设，而这些假设在真实 Agent 场景中几乎全部不成立：

1. 输入是原子性的：假设get_user_input()总能一次性拿到完整、明确、无歧义的指令。现实是用户会说“等等，刚才那个报表再加个柱状图”，或在 Agent 执行到第 3 步时追加“对了，数据要过滤掉测试部门”。while 循环无法在中间插入新输入，只能等本轮结束，导致响应延迟高达数秒甚至分钟。

2. 思考是瞬时的：假设llm_think()调用毫秒级返回。实际 LLM API 响应时间波动极大（OpenAI 通常 800ms~3s，Claude 有时达 8s），且存在重试、限流、超时等网络不确定性。while 循环在此处完全阻塞，后续所有操作（包括接收新用户消息）全部挂起。

3. 行动是确定性的：假设execute_tool(action)总能成功、即时返回结果。但真实工具调用充满异步性：发送邮件需等待 SMTP 响应；调用数据库查询可能触发慢 SQL；调用外部 API 可能因对方服务抖动而超时。while 循环无法优雅处理“等待中”状态，只能傻等或粗暴重试。

4. 状态更新是幂等的：假设update_state(result)每次调用都安全覆盖。但 Agent 常需维护多维度状态：当前任务 ID、各子步骤完成标记、临时缓存数据、用户对话历史、LLM token 使用计数。裸循环中状态更新散落在各处，极易因异常中断导致状态不一致（例如邮件已发但状态未标记为 success）。

5. 输出是终局性的：假设send_response()发送后即任务终结。现实是用户可能回复“不对，重新生成”，或系统需根据执行结果动态决定是否启动新子任务（如“查到 3 份 PDF，需全部转文字”）。while 循环没有“暂停-恢复-分支”的能力，只能重启整个循环，丢失所有中间状态。

提示：这五个假设，恰恰是传统命令行脚本或简单 Web API 的设计前提。Agent 不是脚本，它是与人类持续协作的“数字同事”，其核心特征是状态持久化、交互异步化、决策条件化、执行可中断化。while 循环的线性控制流，与这些特征天然冲突。

2.2 状态机：不是替代方案，而是对 Agent 本质的数学建模

当你意识到 while 循环的局限，自然会想到“用状态机重构”。但很多人误以为状态机只是“把 if-else 换成 switch-case”，这是巨大误区。真正的状态机（State Machine）是对 Agent 行为的形式化建模，它强制你回答三个根本问题：

• 状态（State）是什么？
  不是“正在思考”“正在执行”这种模糊描述，而是可精确枚举、互斥、完备的离散值。例如一个审批 Agent 的状态集可能是：IDLE（空闲）、RECEIVING_INPUT（接收指令）、PLANNING（LLM 规划中）、WAITING_FOR_TOOL（工具调用中）、TOOL_FAILED_RETRYABLE（工具失败可重试）、TOOL_FAILED_FATAL（工具失败不可恢复）、GENERATING_RESPONSE（生成回复中）、AWAITING_USER_CONFIRMATION（等待用户确认）。共 8 个状态，每个状态定义了“在此状态下，哪些事件能触发迁移，迁移后进入哪个状态”。

• 事件（Event）是什么？
  不是代码里的变量，而是外部世界施加于 Agent 的可观测、可捕获的信号。它必须独立于 Agent 内部逻辑：USER_INPUT_RECEIVED（用户发来新消息）、LLM_RESPONSE_RECEIVED（LLM 返回规划）、TOOL_EXECUTION_STARTED（工具调用发起）、TOOL_SUCCESS（工具成功返回）、TOOL_TIMEOUT（工具超时）、USER_CANCELLED（用户取消）、SYSTEM_ERROR（系统级异常）。事件是状态迁移的唯一触发器。

• 迁移（Transition）是什么？
  是一个三元组(current_state, event, next_state)，定义了在特定状态下收到特定事件时，Agent 必须执行的动作（Action）并进入的新状态。例如：(PLANNING, LLM_RESPONSE_RECEIVED, WAITING_FOR_TOOL)表示“当 Agent 处于规划状态，收到 LLM 返回结果时，应解析工具调用并进入等待工具状态”。这个迁移过程必须包含副作用：保存 LLM 返回的 JSON、初始化工具执行上下文、启动超时定时器。

这才是状态机的价值：它把原本隐藏在 while 循环深处的“隐式逻辑”，显式地、可验证地、可测试地暴露出来。你不再问“下一步该做什么”，而是问“当前处于什么状态？收到了什么事件？根据迁移规则，应该进入什么状态并执行什么动作？”——这正是 Claude Code 插件底层采用的设计哲学：它的每个编辑器操作（如Ctrl+Enter触发代码生成）都被映射为一个事件，触发状态迁移，而非简单地在一个 while 循环里顺序执行。

2.3 从 while 到状态机：不是重写，而是“解耦 + 注册 + 驱动”

很多工程师看到状态机就头疼，觉得要推翻重写。其实改造路径非常清晰，只需三步，且每步都可独立验证：

1. 解耦：把 while 循环的“大块头”拆成原子动作
   将原循环中的每个逻辑块（如llm_think、parse_action、execute_tool）封装为纯函数或方法，它们只做一件事、返回明确结果、不修改全局状态。例如：

   def plan_with_llm(user_input: str, context: dict) -> Plan: # 仅调用 LLM，返回结构化 Plan 对象，不更新任何状态 pass def execute_tool(tool_call: ToolCall, context: dict) -> ToolResult: # 仅执行工具，返回结果或异常，不更新任何状态 pass

2. 注册：定义状态、事件、迁移规则表
   用数据结构（如字典或专用类）声明所有状态、事件及迁移规则。这是最核心的一步，它让系统行为变得“可读、可审、可审计”：

   TRANSITIONS = { State.IDLE: { Event.USER_INPUT_RECEIVED: (State.RECEIVING_INPUT, handle_user_input), }, State.RECEIVING_INPUT: { Event.INPUT_PARSED: (State.PLANNING, start_llm_plan), }, State.PLANNING: { Event.LLM_RESPONSE_RECEIVED: (State.WAITING_FOR_TOOL, parse_and_launch_tools), Event.LLM_TIMEOUT: (State.IDLE, log_timeout_and_reset), }, # ... 其他状态迁移 }

3. 驱动：用一个极简的事件循环替代 while True
   这个新循环只做三件事：监听事件队列、查找当前状态对应的迁移规则、执行迁移动作。它本身不包含业务逻辑，只是一个“状态路由器”：

   class AgentStateMachine: def __init__(self): self.state = State.IDLE self.event_queue = queue.Queue() def run(self): while not self.should_stop: try: event = self.event_queue.get(timeout=0.1) # 非阻塞获取事件 if (self.state, event) in TRANSITIONS: next_state, action = TRANSITIONS[(self.state, event)] action(self, event.payload) # 执行动作，可能向队列投递新事件 self.state = next_state else: self.handle_unexpected_event(event) except queue.Empty: continue # 无事件时继续轮询

   注意：这个新循环的timeout=0.1是为了不阻塞，它本质是“事件驱动”的轻量级轮询，与原 while 循环的“业务逻辑驱动”有本质区别。

实操心得：我在重构一个金融风控 Agent 时，就是按这三步走。第一步解耦花了 2 天，把 800 行 while 循环拆成 12 个职责单一的函数；第二步注册用了半天，和产品、算法同学一起白板画出所有状态和事件，发现原来漏掉了“用户修改原始请求”这个关键事件；第三步驱动循环写了不到 100 行。上线后，平均任务响应时间从 4.2s 降到 1.7s，更重要的是，当出现“工具调用超时”时，系统能自动降级到备用方案（如用缓存数据生成近似结果），而不是卡死。这证明：状态机不是增加复杂度，而是把复杂度从“隐藏在代码里”转移到“显式在设计中”，从而让系统真正可控。

3. 七个 while 循环崩溃的临界点：从现象到根因的精准定位

3.1 临界点一：用户中途修改指令（交互异步性爆发）

现象：Agent 正在执行“查询销售数据”步骤，用户突然发来新消息：“不用查销售了，改成查库存”。while 循环仍在原计划里执行，直到当前步骤完成才处理新消息，导致用户等待 8 秒后收到“已查询销售数据”的过期回复。

根因分析：while 循环是同步阻塞式的。get_user_input()通常设计为阻塞调用（如input()或 WebSocketrecv()），它必须等到本次输入完成才能进入下一轮。用户新消息被缓冲在队列末尾，而循环正卡在execute_tool()的网络等待中，无法及时响应。

状态机解法：引入USER_INPUT_RECEIVED事件，并允许在任意状态（除IDLE外）接收该事件。迁移规则定义为：

• (PLANNING, USER_INPUT_RECEIVED) → (IDLE, cancel_current_plan_and_reset)
• (WAITING_FOR_TOOL, USER_INPUT_RECEIVED) → (IDLE, cancel_pending_tool_and_reset)
• (GENERATING_RESPONSE, USER_INPUT_RECEIVED) → (IDLE, discard_pending_response_and_reset)

关键在于：事件监听是独立的，不依赖于主循环进度。只要新消息到达，立即入队，状态机在下一个轮询周期就能处理。

实操细节：在 Claude Code 插件中，这个机制体现为“中断生成”。当你在代码生成过程中按下Esc键，编辑器会立即向 Agent Runtime 发送USER_CANCELLED事件，Runtime 收到后立刻终止当前 LLM 请求（通过 HTTP Cancel），并将状态切回IDLE。这背后没有复杂的线程管理，只有事件队列和状态迁移表的精准匹配。

3.2 临界点二：LLM 响应超时（网络不确定性失控）

现象：Agent 向 Claude 发送规划请求，30 秒后仍未返回。while 循环卡在llm_think()函数内，CPU 占用低但进程无响应，用户界面显示“思考中...”长达半分钟，期间无法接收任何新指令。

根因分析：llm_think()函数内部做了同步 HTTP 请求，且未设置合理超时。while 循环在此处完全挂起，既不能降级（如用本地小模型快速生成草稿），也无法通知用户“正在努力，请稍候”，更无法启动备用通道。

状态机解法：将 LLM 调用拆分为两个事件：LLM_EXECUTION_STARTED和LLM_RESPONSE_RECEIVED（或LLM_TIMEOUT）。在PLANNING状态下，收到USER_INPUT_RECEIVED后，触发LLM_EXECUTION_STARTED事件，同时启动一个独立的超时定时器（如threading.Timer或asyncio.create_task）。若定时器到期，自动向事件队列投递LLM_TIMEOUT事件。

参数计算：超时阈值不能拍脑袋定。我推荐公式：timeout = base_delay * (1 + retry_count)，其中base_delay根据 LLM 服务商 SLA 设定（Claude 官方建议 30s，OpenAI 为 15s），retry_count是当前重试次数。首次调用设为 30s，重试一次后升至 45s，避免雪崩。

实操记录：我们在对接 IEC61850 GOOSE 通讯时遇到类似问题。GOOSE 报文要求严格实时性（< 4ms），但网络抖动会导致偶尔超时。解决方案不是加长超时，而是定义GOOSE_SEND_TIMEOUT事件，触发状态迁移到GOOSE_RETRY，并启用备用 VLAN 通道。这与 Agent 的 LLM 超时处理逻辑完全同源——都是用状态迁移应对不确定性。

3.3 临界点三：工具调用链式依赖（执行流程不可控）

现象：Agent 计划执行三步：“1. 查数据库 → 2. 用结果调用邮件 API → 3. 生成总结”。第一步成功，第二步邮件 API 返回 503 服务不可用。while 循环尝试重试三次后放弃，但第三步“生成总结”从未执行，因为代码逻辑是for step in plan.steps:顺序执行，第二步失败直接跳出循环。

根因分析：while 循环隐含了“全有或全无”的强一致性假设。它把工具调用视为线性流水线，无法表达“步骤 2 失败时，步骤 3 是否仍可执行？”这类条件逻辑。状态不明确，导致恢复点模糊。

状态机解法：为每个工具调用定义独立状态和事件。例如：

• (WAITING_FOR_TOOL, TOOL_SUCCESS) → (WAITING_FOR_NEXT_TOOL, launch_next_tool)
• (WAITING_FOR_TOOL, TOOL_FAILED_RETRYABLE) → (TOOL_FAILED_RETRYABLE, schedule_retry)
• (WAITING_FOR_TOOL, TOOL_FAILED_FATAL) → (GENERATING_FALLBACK_RESPONSE, use_cached_data)

关键创新是TOOL_FAILED_FATAL状态：它不终止整个 Agent，而是触发一个专门的“降级响应生成”动作，利用已获取的数据库结果（步骤 1 成功）生成部分答案，并告知用户“邮件发送失败，但数据已查到”。

对比表格：两种模式对工具失败的处理

3.4 临界点四：多任务并发竞争（资源争用显性化）

现象：同一 Agent 实例被两个用户同时调用。User A 的“生成报告”任务刚启动，User B 的“查询状态”请求到达。while 循环无法区分上下文，user_input变量被覆盖，plan对象被 User B 的数据污染，最终 User A 收到 User B 的查询结果。

根因分析：while 循环共享全局状态（如user_input,plan,state变量）。它本质上是单实例、单会话模型，无法天然支持多租户。工程师常通过加锁（threading.Lock）强行解决，但这带来死锁风险，且无法解决长任务（如生成 PDF）占用资源的问题。

状态机解法：每个会话（Session）拥有独立的状态机实例。Agent Runtime 不是一个 while 循环，而是一个状态机工厂：

class AgentRuntime: def __init__(self): self.sessions = {} # {session_id: AgentStateMachine()} def handle_message(self, session_id: str, message: dict): if session_id not in self.sessions: self.sessions[session_id] = AgentStateMachine() self.sessions[session_id].event_queue.put( Event(USER_INPUT_RECEIVED, payload=message) )

每个AgentStateMachine维护自己的state、context和event_queue，完全隔离。资源争用问题转化为会话 ID 的路由问题，由上层负载均衡器（如 Nginx）或消息队列（如 Kafka）解决。

实操心得：我们在部署一个支持 500+ 并发客服 Agent 的系统时，最初用单实例 while 循环，加了各种锁，QPS 卡在 80。改用会话级状态机后，QPS 直接飙升到 1200，且 CPU 利用率从 95% 降到 65%。因为不再有锁竞争，每个会话的事件处理是真正并行的。Claude Code 桌面版也采用此设计：每个打开的编辑器标签页对应一个独立 Agent 实例，互不干扰。

3.5 临界点五：长时任务阻塞（时间维度失控）

现象：Agent 启动一个需 5 分钟完成的“批量数据清洗”任务。while 循环卡在此处，期间所有其他用户请求排队等待，系统响应时间从 200ms 暴涨到 30s。

根因分析：while 循环将“任务执行”与“状态更新”耦合。它假设所有动作都是瞬时的，无法表达“任务已启动，正在后台运行，预计 5 分钟后完成”这样的中间态。

状态机解法：引入BACKGROUND_TASK_STARTED和BACKGROUND_TASK_COMPLETED事件。当 Agent 进入EXECUTING_LONG_TASK状态时，它不阻塞主循环，而是：

• 启动一个后台线程/进程执行耗时操作；
• 立即返回BACKGROUND_TASK_STARTED事件，状态迁移到AWAITING_BACKGROUND_RESULT；
• 在后台任务完成时，向对应会话的事件队列投递BACKGROUND_TASK_COMPLETED事件。

关键技术点：后台任务与状态机的通信必须安全。我们采用multiprocessing.Queue或 Redis Pub/Sub，确保事件可靠投递。状态机本身始终保持轻量，只负责状态流转，不执行重负载。

生活类比：这就像餐厅点餐系统。while 循环是“服务员全程盯着厨房，直到菜做好才去服务下一位客人”。状态机是“服务员下单后，厨房给个取餐号，服务员立即去服务新客人，等厨房广播‘XX号菜好了’，再凭号取餐”。前者效率低下，后者并发高效。

3.6 临界点六：状态持久化缺失（崩溃后无法恢复）

现象：Agent 正在执行“审批流程”，已走到第 3 步（财务审核），服务器意外宕机。重启后，while 循环从头开始，用户需重复提交申请，已做的两步审核工作全部丢失。

根因分析：while 循环的状态（如current_step,approval_history）全在内存中。它没有“检查点（Checkpoint）”概念，无法在崩溃后恢复到最近的稳定状态。

状态机解法：状态迁移即持久化。每次状态变更（如(APPROVAL_STEP_2, APPROVAL_APPROVED) → (APPROVAL_STEP_3, notify_finance)）发生时，立即将新状态和必要上下文（如审批人 ID、时间戳）写入数据库或 Redis。Agent 启动时，首先从持久化存储加载最新状态，直接进入对应状态，而非IDLE。

参数选择：持久化频率需权衡性能与可靠性。我们采用“状态变更必存”策略，但对高频事件（如HEARTBEAT）做合并写入。数据库选型上，Redis 因其高吞吐和原子性操作（SET key value EX 3600 NX）成为首选，用于存储会话状态；PostgreSQL 用于存储审计日志和长期任务记录。

实操记录：某政务 OA 系统的审批 Agent 曾因机房断电宕机。得益于状态机的自动恢复，327 个进行中的审批流程在服务重启后 2 秒内全部恢复到宕机前的精确步骤，用户无感知。而旧版 while 循环系统，每次宕机后需人工导出日志、手动重建状态，平均耗时 47 分钟。

3.7 临界点七：动态策略注入（业务逻辑僵化）

现象：公司新出台政策：“所有金额 > 100 万的合同，需额外增加法务审核步骤”。工程师需修改 while 循环代码，在plan.steps中插入新步骤，然后重新部署。期间所有 Agent 服务暂停，用户无法提交合同。

根因分析：while 循环将业务规则（如“何时加法务审核”）硬编码在控制流中。它缺乏“策略即配置”的灵活性，无法在运行时动态调整。

状态机解法：将业务规则外置为可热加载的策略文件。状态机在PLANNING状态下，不直接生成固定步骤，而是调用策略引擎：

def generate_plan(user_input: str, context: dict) -> Plan: # 从 Redis 加载最新策略 policy = load_policy_from_redis("contract_approval_policy") # 根据策略和上下文动态生成步骤 steps = policy.apply(context) return Plan(steps=steps)

策略文件（JSON/YAML）可由产品经理在管理后台编辑，实时生效。状态机只负责执行策略引擎返回的步骤，不关心规则本身。

对比优势：

• 发布零停机：策略更新无需重启 Agent 服务。
• 灰度发布：可为特定用户群（如tenant_id == "finance"）加载不同策略。
• 可追溯：每次状态迁移记录使用的策略版本号，便于审计。

我们在一个跨国电商 Agent 中应用此方案。不同国家的税务规则差异巨大（如欧盟 VAT、美国州税），通过动态策略加载，实现了 23 个国家税务计算逻辑的独立管理，上线新国家规则仅需 15 分钟配置，而非 3 天代码开发。

4. 从理论到落地：一个可直接复用的轻量级 Agent 状态机框架

4.1 核心设计原则：极简、可嵌入、零依赖

我见过太多“企业级状态机框架”，动辄数百个文件、依赖七八个包、需要 YAML 配置。对于大多数 Agent 项目，你需要的只是一个50 行核心代码 + 清晰约定的轻量方案。以下是我在线上服务中稳定运行两年的SimpleAgentSM框架：

# simple_agent_sm.py from enum import Enum from typing import Dict, Callable, Any, Optional import threading import queue import time class State(Enum): IDLE = "idle" RECEIVING_INPUT = "receiving_input" PLANNING = "planning" WAITING_FOR_TOOL = "waiting_for_tool" TOOL_FAILED_RETRYABLE = "tool_failed_retryable" TOOL_FAILED_FATAL = "tool_failed_fatal" GENERATING_RESPONSE = "generating_response" AWAITING_USER_CONFIRMATION = "awaiting_user_confirmation" class Event: def __init__(self, name: str, payload: Any = None): self.name = name self.payload = payload class SimpleAgentSM: def __init__(self, session_id: str): self.session_id = session_id self.state = State.IDLE self.context = {} # 存储会话上下文 self.event_queue = queue.Queue() self._running = False self._thread = None def start(self): """启动状态机线程""" self._running = True self._thread = threading.Thread(target=self._run_loop, daemon=True) self._thread.start() def stop(self): """停止状态机""" self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=1.0) def _run_loop(self): """核心事件循环""" while self._running: try: # 非阻塞获取事件，超时 0.05 秒避免 CPU 空转 event = self.event_queue.get(timeout=0.05) self._handle_event(event) except queue.Empty: continue # 无事件，继续轮询 def _handle_event(self, event: Event): """处理事件：查找迁移规则，执行动作""" # 迁移规则：(current_state, event_name) -> (next_state, action_func) transition = self._get_transition(self.state, event.name) if transition: next_state, action = transition try: # 执行动作，传入 self 和 event.payload action(self, event.payload) self.state = next_state except Exception as e: # 动作执行失败，进入错误状态 self.state = State.TOOL_FAILED_FATAL self.context["error"] = str(e) self._log_error(f"Action failed: {e}") else: self._handle_unexpected_event(event) def _get_transition(self, state: State, event_name: str) -> Optional[tuple]: """获取迁移规则，此处可替换为外部配置""" # 示例：硬编码规则，实际项目中可从 Redis 或 DB 加载 rules = { (State.IDLE, "USER_INPUT_RECEIVED"): ( State.RECEIVING_INPUT, self._on_user_input_received ), (State.RECEIVING_INPUT, "INPUT_PARSED"): ( State.PLANNING, self._on_input_parsed ), (State.PLANNING, "LLM_RESPONSE_RECEIVED"): ( State.WAITING_FOR_TOOL, self._on_llm_response_received ), # ... 更多规则 } return rules.get((state, event_name)) # 以下是具体的动作函数，按需实现 def _on_user_input_received(self, payload: dict): self.context["user_input"] = payload.get("text", "") # 解析输入，可能触发 INPUT_PARSED 事件 self.event_queue.put(Event("INPUT_PARSED", {"parsed": True})) def _on_input_parsed(self, payload: dict): # 启动 LLM 调用，异步执行 threading.Thread( target=self._call_llm_async, args=(self.context["user_input"],) ).start() def _call_llm_async(self, user_input: str): # 模拟 LLM 调用 time.sleep(1.0) # 真实调用需用 requests 或 aiohttp self.event_queue.put(Event("LLM_RESPONSE_RECEIVED", {"plan": "..."})) def _on_llm_response_received(self, payload: dict): # 解析 plan，启动工具调用 self.context["plan"] = payload["plan"] # 模拟工具调用 threading.Thread( target=self._execute_tool_async, args=("database_query",) ).start() def _execute_tool_async(self, tool_name: str): time.sleep(0.5) self.event_queue.put(Event("TOOL_SUCCESS", {"result": "data"})) def _log_error(self, msg: str): print(f"[{self.session_id}] ERROR: {msg}") def _handle_unexpected_event(self, event: Event): print(f"[{self.session_id}] Unexpected event: {event.name}") # 公共方法：供外部调用 def post_event(self, event: Event): """外部投递事件的入口""" self.event_queue.put(event)

为什么这个框架足够用？

• 50 行核心：去掉注释和空行，核心状态机逻辑不足 50 行，易于理解、审计和定制。
• 零外部依赖：只用 Python 标准库threading和queue，无 pip install。
• 可嵌入性：SimpleAgentSM是一个普通类，可轻松集成到 FastAPI、Flask 或任何 Web 框架中。
• 可扩展性：_get_transition方法是钩子，可无缝替换为从 Redis 加载规则，或接入 GraphQL 查询。

4.2 快速集成指南：3 分钟接入现有项目

假设你有一个基于 Flask 的 Agent 服务，原 while 循环如下：

# old_app.py from flask import Flask, request, jsonify import time app = Flask(__name__) @app.route("/chat", methods=["POST"]) def chat(): user_input = request.json.get("message") # while 循环逻辑... time.sleep(2) # 模拟 LLM 调用 return jsonify({"response": "Hello from while loop!"})

步骤 1：创建状态机管理器

# sm_manager.py from simple_agent_sm import SimpleAgentSM from collections import defaultdict import threading class SMManager: _instances = defaultdict(dict) _lock = threading.Lock() @classmethod def get_instance(cls, session_id: str) -> SimpleAgentSM: with cls._lock: if session_id not in cls._instances: sm = SimpleAgentSM(session_id) sm.start() cls._instances[session_id] = sm return cls._instances[session_id] @classmethod def cleanup_session(cls, session_id: str): with cls._lock: if session_id in cls._instances: cls._instances[session_id].stop() del cls._instances[session_id]

步骤 2：改造 Flask 路由

# new_app.py from flask import Flask, request, jsonify from sm_manager import SMManager app = Flask(__name__) @app.route("/chat", methods=["POST"]) def chat(): session_id = request.json.get("session_id", "default") user_input = request.json.get("message", "") # 获取或创建状态机实例 sm = SMManager.get_instance(session_id) # 投递用户输入事件 sm.post_event(Event("USER_INPUT_RECEIVED", {"text": user_input})) # 等待响应（生产环境建议用 WebSocket 或 SSE） # 此处简化：轮询检查 context 中的 response for _ in range(20): # 最多等待 2 秒 if "response" in sm.context