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

在构建复杂的 Agent 系统时，我们经常会遇到这样的场景：Agent 正在执行一个多步骤的任务，比如“下单购买商品”，但执行到一半时，我们需要它停下来，等待用户确认或输入额外信息。如果没有一种机制让 Agent 在中间暂停，整个过程就会失控——要么一次性执行完所有步骤（可能出错），要么不得不强行打断整个流程。今天，我们就来深入探讨如何用interrupt（中断）机制给 Agent 设置“关卡”，让它在执行到关键步骤时暂停、等待、再恢复。### 为什么需要 interrupt？想象一个电商场景：Agent 被要求“帮用户买一部手机”。正常流程是：搜索手机 → 选择型号 → 加入购物车 → 确认订单 → 支付。但如果没有 interrupt，Agent 可能会直接执行支付步骤，而用户可能还没确认价格或选择颜色。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, timeout=10): """创建一个新的中断点""" 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(), timeout=interrupt["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", timeout=5) self.create_interrupt("confirm_purchase", timeout=5) 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实现超时，超时后返回None，Agent 可以据此做降级处理。- 外部通过resolve_interrupt()传入数据，实现了灵活的双向通信。—## 总结通过以上两个示例，我们看到了 interrupt 机制在 Agent 系统中的强大作用。它不仅仅是暂停执行，更是一种流程控制和人机协作的桥梁。**核心要点：**1.中断是状态机的一部分：每个中断点代表 Agent 的一个状态，等待外部事件触发状态转换。2.异步事件是基础：asyncio.Event提供了简单可靠的阻塞/唤醒机制，适合大多数场景。3.超时处理必不可少：在真实系统中，用户可能无响应，超时机制能防止 Agent 永久挂起。4.多中断管理：复杂 Agent 需要多个中断点，建议用字典或类属性统一管理。5.生产环境扩展：在实际项目中，可以将中断点对外暴露为 REST API 或消息队列，实现分布式协作。最佳实践建议：- 每个中断点赋予唯一 ID，方便追踪。- 中断数据尽量序列化（如 JSON），以便在分布式系统中传递。- 考虑中断的幂等性，防止重复恢复导致错误。有了 interrupt 这个“关卡”，你的 Agent 就不再是“闭眼狂奔”的机器，而是一个可以随时沟通、灵活调整的智能助手。下次构建复杂流程时，记得给它加几个关卡！