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 场景中几乎全部不成立输入是原子性的假设get_user_input()总能一次性拿到完整、明确、无歧义的指令。现实是用户会说“等等刚才那个报表再加个柱状图”或在 Agent 执行到第 3 步时追加“对了数据要过滤掉测试部门”。while 循环无法在中间插入新输入只能等本轮结束导致响应延迟高达数秒甚至分钟。思考是瞬时的假设llm_think()调用毫秒级返回。实际 LLM API 响应时间波动极大OpenAI 通常 800ms~3sClaude 有时达 8s且存在重试、限流、超时等网络不确定性。while 循环在此处完全阻塞后续所有操作包括接收新用户消息全部挂起。行动是确定性的假设execute_tool(action)总能成功、即时返回结果。但真实工具调用充满异步性发送邮件需等待 SMTP 响应调用数据库查询可能触发慢 SQL调用外部 API 可能因对方服务抖动而超时。while 循环无法优雅处理“等待中”状态只能傻等或粗暴重试。状态更新是幂等的假设update_state(result)每次调用都安全覆盖。但 Agent 常需维护多维度状态当前任务 ID、各子步骤完成标记、临时缓存数据、用户对话历史、LLM token 使用计数。裸循环中状态更新散落在各处极易因异常中断导致状态不一致例如邮件已发但状态未标记为 success。输出是终局性的假设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接收指令、PLANNINGLLM 规划中、WAITING_FOR_TOOL工具调用中、TOOL_FAILED_RETRYABLE工具失败可重试、TOOL_FAILED_FATAL工具失败不可恢复、GENERATING_RESPONSE生成回复中、AWAITING_USER_CONFIRMATION等待用户确认。共 8 个状态每个状态定义了“在此状态下哪些事件能触发迁移迁移后进入哪个状态”。事件Event是什么不是代码里的变量而是外部世界施加于 Agent 的可观测、可捕获的信号。它必须独立于 Agent 内部逻辑USER_INPUT_RECEIVED用户发来新消息、LLM_RESPONSE_RECEIVEDLLM 返回规划、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 插件底层采用的设计哲学它的每个编辑器操作如CtrlEnter触发代码生成都被映射为一个事件触发状态迁移而非简单地在一个 while 循环里顺序执行。2.3 从 while 到状态机不是重写而是“解耦 注册 驱动”很多工程师看到状态机就头疼觉得要推翻重写。其实改造路径非常清晰只需三步且每步都可独立验证解耦把 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注册定义状态、事件、迁移规则表用数据结构如字典或专用类声明所有状态、事件及迁移规则。这是最核心的一步它让系统行为变得“可读、可审、可审计”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), }, # ... 其他状态迁移 }驱动用一个极简的事件循环替代 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(timeout0.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 # 无事件时继续轮询注意这个新循环的timeout0.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 官方建议 30sOpenAI 为 15sretry_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 成功生成部分答案并告知用户“邮件发送失败但数据已查到”。对比表格两种模式对工具失败的处理维度while 循环模式状态机模式失败定位模糊只知道“第 N 步失败”精确TOOL_FAILED_FATAL状态附带失败工具 ID 和错误码恢复能力无需重启整个循环强可停留在TOOL_FAILED_FATAL状态等待人工干预或自动降级用户反馈“操作失败”无上下文“邮件发送失败错误码 503但销售数据已为您提取详见附件”可观测性日志分散在各函数中所有状态迁移记录在统一审计日志可追踪完整路径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, payloadmessage) )每个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(stepssteps)策略文件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(targetself._run_loop, daemonTrue) self._thread.start() def stop(self): 停止状态机 self._running False if self._thread and self._thread.is_alive(): self._thread.join(timeout1.0) def _run_loop(self): 核心事件循环 while self._running: try: # 非阻塞获取事件超时 0.05 秒避免 CPU 空转 event self.event_queue.get(timeout0.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(fAction 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( targetself._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( targetself._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