28. Agent 执行到一半想暂停？用 interrupt 给它设个“关卡“！-北京尧图网络

28. Agent 执行到一半想暂停用 interrupt 给它设个“关卡“在构建复杂的 Agent 系统时我们经常会遇到这样的场景Agent 正在执行一个多步骤的任务比如“下单购买商品”但执行到一半时我们需要它停下来等待用户确认或输入额外信息。如果没有一种机制让 Agent 在中间暂停整个过程就会失控——要么一次性执行完所有步骤可能出错要么不得不强行打断整个流程。今天我们就来深入探讨如何用interrupt中断机制给 Agent 设置“关卡”让它在执行到关键步骤时暂停、等待、再恢复。### 为什么需要 interrupt想象一个电商场景Agent 被要求“帮用户买一部手机”。正常流程是搜索手机 → 选择型号 → 加入购物车 → 确认订单 → 支付。但如果没有 interruptAgent 可能会直接执行支付步骤而用户可能还没确认价格或选择颜色。interrupt 允许我们在“确认订单”这个节点插入一个暂停点让用户检查信息然后决定继续或修改。在实际开发中interrupt 的典型用途包括-用户确认在关键操作前等待用户批准。-动态输入需要用户提供额外参数如地址、优惠码。-条件分支根据暂停点的状态决定后续流程。### 核心概念如何实现 interrupt在 Python 中我们可以利用asyncio的事件循环或自定义状态机来实现 interrupt。这里我们使用一个轻量级的模式用asyncio.Event或asyncio.Condition来阻塞 Agent 的执行直到外部信号触发恢复。下面我将用两个完整的代码示例来演示。—## 示例 1基础 interrupt —— 让 Agent 在步骤中暂停等待用户确认这个示例模拟一个简单的购物 Agent。它执行三步搜索商品、加入购物车、支付。我们在“支付”前插入一个 interrupt等待用户输入yes或no。pythonimport asyncioimport randomclass ShoppingAgent: def __init__(self): self.interrupt_event asyncio.Event() self.user_decision None async def search_item(self, item_name): print(f[Agent] 正在搜索商品{item_name}...) await asyncio.sleep(1) # 模拟搜索延迟 found_items [iPhone 15 Pro, Samsung Galaxy S24, Google Pixel 8] print(f[Agent] 找到商品{random.choice(found_items)}) return found_items[0] async def add_to_cart(self, item): print(f[Agent] 正在将 {item} 加入购物车...) await asyncio.sleep(0.5) print(f[Agent] {item} 已加入购物车当前价格$999) return True async def confirm_payment(self): print([Agent] 准备执行支付...) # --- 设置 interrupt 关卡 --- print([Agent] ⏸️ 暂停等待用户确认支付...) await self.interrupt_event.wait() # 阻塞在这里直到事件被设置 # --- 用户做出决定后恢复 --- if self.user_decision yes: print([Agent] 用户确认支付正在处理...) await asyncio.sleep(1) print([Agent] ✅ 支付成功订单完成。) return True else: print([Agent] ❌ 用户取消支付流程终止。) return False async def run(self, item_name): item await self.search_item(item_name) await self.add_to_cart(item) result await self.confirm_payment() return result def resume_with_decision(self, decision): 外部调用此方法来恢复 Agent self.user_decision decision self.interrupt_event.set() # 设置事件解除 await 阻塞async def main(): agent ShoppingAgent() # 启动 Agent 任务不等待完成 task asyncio.create_task(agent.run(手机)) # 模拟外部用户输入延迟 2 秒后 await asyncio.sleep(2) print(\n[用户] 正在查看订单信息...) # 假设用户输入 yes user_input yes # 在实际应用中这里可以是 input() 或 API 请求 print(f[用户] 输入决定{user_input}) agent.resume_with_decision(user_input) # 等待 Agent 完成 result await task print(f\n最终结果{成功 if result else 失败})if __name__ __main__: asyncio.run(main())运行结果示例[Agent] 正在搜索商品手机...[Agent] 找到商品iPhone 15 Pro[Agent] 正在将 iPhone 15 Pro 加入购物车...[Agent] iPhone 15 Pro 已加入购物车当前价格$999[Agent] 准备执行支付...[Agent] ⏸️ 暂停等待用户确认支付...[用户] 正在查看订单信息...[用户] 输入决定yes[Agent] 用户确认支付正在处理...[Agent] ✅ 支付成功订单完成。最终结果成功关键点解释-asyncio.Event是一个简单的同步原语await event.wait()会阻塞当前协程直到另一个协程调用event.set()。-resume_with_decision()是外部接口模拟用户或另一个系统向 Agent 发送恢复信号。- 这种模式非常适合需要人工介入的场景比如审批流程、动态确认。—## 示例 2高级 interrupt —— 带超时和多次中断的 Agent现实中的 Agent 可能需要多个中断点例如先确认商品再确认支付并且要处理超时情况。这个示例实现了一个更复杂的 Agent它有两个中断点并且每个中断点都有超时机制。pythonimport asyncioimport timeclass AdvancedAgent: def __init__(self): self.interrupts {} # 存储多个中断事件 def create_interrupt(self, name, timeout10): 创建一个新的中断点 event asyncio.Event() self.interrupts[name] { event: event, data: None, timeout: timeout, created_at: time.time() } return event def resolve_interrupt(self, name, data): 外部调用解决中断传入数据 if name in self.interrupts: self.interrupts[name][data] data self.interrupts[name][event].set() else: raise ValueError(f中断 {name} 不存在) async def wait_with_timeout(self, name): 等待中断带超时处理 interrupt self.interrupts[name] try: await asyncio.wait_for(interrupt[event].wait(), timeoutinterrupt[timeout]) return interrupt[data] except asyncio.TimeoutError: print(f[Agent] ⏰ 中断 {name} 超时) return None # 返回 None 表示超时 async def step1_search(self): print([Agent] 步骤1搜索商品...) await asyncio.sleep(1) products [Laptop, Tablet, Smartwatch] print(f[Agent] 找到商品{products}) # 第一个中断让用户选择商品 print([Agent] ⏸️ 中断点1请选择商品编号0-2) choice await self.wait_with_timeout(choose_product) if choice is None: return None selected products[int(choice)] print(f[Agent] 用户选择了{selected}) return selected async def step2_confirm(self, product): print(f[Agent] 步骤2确认购买 {product}...) await asyncio.sleep(0.5) # 第二个中断让用户确认 print([Agent] ⏸️ 中断点2请确认购买yes/no) decision await self.wait_with_timeout(confirm_purchase) if decision yes: print([Agent] ✅ 购买确认继续执行...) return True else: print([Agent] ❌ 购买取消) return False async def run(self): # 创建中断点 self.create_interrupt(choose_product, timeout5) self.create_interrupt(confirm_purchase, timeout5) product await self.step1_search() if product is None: print([Agent] 流程因超时而终止) return False result await self.step2_confirm(product) return resultasync def main(): agent AdvancedAgent() # 启动 Agent task asyncio.create_task(agent.run()) # 模拟外部输入延迟 1 秒后提供商品选择 await asyncio.sleep(1) print(\n[外部] 用户正在选择商品...) agent.resolve_interrupt(choose_product, 1) # 选择 Tablet # 模拟第二个中断输入延迟 2 秒后 await asyncio.sleep(2) print(\n[外部] 用户正在确认购买...) agent.resolve_interrupt(confirm_purchase, yes) # 等待结果 result await task print(f\n最终结果{成功 if result else 失败})if __name__ __main__: asyncio.run(main())运行结果示例[Agent] 步骤1搜索商品...[Agent] 找到商品[Laptop, Tablet, Smartwatch][Agent] ⏸️ 中断点1请选择商品编号0-2[外部] 用户正在选择商品...[Agent] 用户选择了Tablet[Agent] 步骤2确认购买 Tablet...[Agent] ⏸️ 中断点2请确认购买yes/no[外部] 用户正在确认购买...[Agent] ✅ 购买确认继续执行...最终结果成功关键点解释- 使用create_interrupt()动态创建多个中断点每个都有自己的事件和超时。-wait_with_timeout()利用asyncio.wait_for实现超时超时后返回NoneAgent 可以据此做降级处理。- 外部通过resolve_interrupt()传入数据实现了灵活的双向通信。—## 总结通过以上两个示例我们看到了 interrupt 机制在 Agent 系统中的强大作用。它不仅仅是暂停执行更是一种流程控制和人机协作的桥梁。**核心要点**1.中断是状态机的一部分每个中断点代表 Agent 的一个状态等待外部事件触发状态转换。2.异步事件是基础asyncio.Event提供了简单可靠的阻塞/唤醒机制适合大多数场景。3.超时处理必不可少在真实系统中用户可能无响应超时机制能防止 Agent 永久挂起。4.多中断管理复杂 Agent 需要多个中断点建议用字典或类属性统一管理。5.生产环境扩展在实际项目中可以将中断点对外暴露为 REST API 或消息队列实现分布式协作。最佳实践建议- 每个中断点赋予唯一 ID方便追踪。- 中断数据尽量序列化如 JSON以便在分布式系统中传递。- 考虑中断的幂等性防止重复恢复导致错误。有了 interrupt 这个“关卡”你的 Agent 就不再是“闭眼狂奔”的机器而是一个可以随时沟通、灵活调整的智能助手。下次构建复杂流程时记得给它加几个关卡

28. Agent 执行到一半想暂停？用 interrupt 给它设个“关卡“！

相关推荐

轨道交通PIS系统工控主板选型指南(宽温高可靠工业级嵌入式硬件解决方案)

KMS智能激活工具：一站式解决Windows和Office激活难题

免费开源RPA工具OpenRPA：企业级自动化流程的终极解决方案

更多资讯

ArcGIS Server 数据存储注册：文件夹与数据库配置及排错

Fluent UDF中DEFINE_PROPERTY物性动态定义详解

基于MCP与Docker构建LLM Agent的hindsight记忆系统实战

Power SI AC阻抗仿真全攻略：从PDN建模到目标阻抗曲线解读

RK3576 I3C实战：从I2C迁移到12.5MHz高速总线的DTS配置与避坑指南

数字电路30秒倒计时器设计原理与Proteus仿真实战

今日更新

开源模型端侧落地实战：量化、推理加速与Agent上下文管理

AI Evals实战指南：从零搭建LLM应用评估体系与CI/CD集成

Java采购管理系统实战：从数据库设计到事务一致性

本周精选

如何彻底离线使用 Laya-CoreML：Hugging Face 下载、缓存机制与 local_files_only 详解

基于1Panel 部署MySQL MCP 服务：TaoToken 统一 Key 接入与 config.toml 配置实战

崇州园区营销网站建设避坑指南：5大注意事项保备案安全

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？