bojieli/ai-agent-book/async-event-agent

异步与事件驱动 Agent

何时使用

• 要让 Agent 在没有用户输入时也行动：定时生成报告、周期性检查服务器、跟进未回复的邮件、到工作时间再拨打电话。
• 要接入外部事件源：新邮件到达、API 回调、IM/短信消息、GitHub PR 变更、支付失败告警。
• 要支持执行期间被打断：长任务运行中用户说"停""改一下预算""先帮我查个天气"。
• 要同时管理多个并发任务：并行跑几个脚本、先完成的先看结果、未达进度就取消。
• 要从回合制 ReAct 循环升级为长期运行的事件循环 Agent。
• 要评估某个模型/接口是否支持原生异步，或为不支持的接口设计兼容层。
• 任务需要在"发起"与"收尾"之间跨越多个回合（如代打电话、长时审批流）。
• Agent 需要以"活人感"与用户异步沟通（发消息、图片、文件、按紧急度推送提醒），而非必须打开指定会话。

核心原则

• 回合制是交互约定，不是环境性质。 世界不会等模型反应完才变化；任务常在"发起"与"收尾"之间跨越多个回合。
• 架构是事件循环，不是轮询。 Agent 是一个长期运行的循环：每轮从输入队列取若干事件 → 追加到轨迹 → 调一次 LLM → 执行它决定的工具 → 回到开头等下一批。不要用"反复问有没有新消息"的轮询。
• "主动服务"的前提是事件能推给 Agent。 Hooks（框架内生命周期事件）、Cron（定时调度）、Heartbeat（每隔 N 分钟唤醒）三种机制里只有后两种让 Agent 自己动起来，且都是时间驱动的；对内置渠道之外的第三方事件源（新邮件、外部回调）必须建实时推送通道，否则只能等下一个周期才察觉——这个延迟在验证码等待、三方通话等场景里不可接受。
• 安全点决定一切。 同步接口中事件只能在循环边界（一段推理结束、一次工具返回）被消费；紧急事件的取消式处理本质是"主动提前制造一个安全点"。取消点必须是工具或推理能安全收尾的位置。
• 取消 ≠ 撤销。 发送取消信号只是让执行器停止；已经发生的动作不会回滚。未完成的工具结果用显式占位符表示，不能伪造成功。
• 三种处理策略按紧急度选择：
• 取消式（紧急事件：user.interrupt、supervisor.instruction、agent.interrupt、紧急告警）：停当前操作 → 清空队列 → 所有事件连同紧急事件一次性追加到轨迹 → 立即重新调 LLM。
• 队列式（常规事件：user.input、tool.result、timer.trigger）：放入队尾不打断，等本轮到达安全点后批量追加，减少往返。
• 并行式（独立、轻量、需快响的查询，如"今天天气怎么样"）：在独立推理会话中执行，结果追加进主轨迹并明确标记为并行，避免 LLM 混淆。
• 硬编码规则有局限；建议用轻量分类 LLM 做事件路由器。
• 所有输入统一建模为结构化事件：来源（谁）、渠道（方式）、内容（什么/紧急度）、上下文（与哪个任务相关）。这是防止把工具结果误当用户指令、以及提示注入的前提。异构触发器（webhook、定时器、邮件、数据库变更、文件监听）统一建模后，Agent 才能用一致方式处理不同来源的刺激。
• 异步语义要写进工具接口：把"启动"和"完成"解耦——initiate_phone_call 立即返回任务 ID 和初始状态，进展通过 phone_call_connected / phone_call_ended 等事件通知。工具名和描述本身就要传达异步语义（"任务发起后立即返回 ID，你可以继续处理其他事项，结束后会收到单独通知"）。
• 三条路线按模型能力选择：原生异步（工具挂起、模型继续、回合中途 steering）> 同步接口兼容（占位符 + 队列 + 重新请求）> 退化回纯串行。用了 asyncio 不等于模型具备异步能力，仍需应用层管理事件来源、工具生命周期与结果归属。
• 可靠处理比能接收更难：至少要检查——延迟结果能否归入正确任务且缺席时不编造；处理新要求后能否恢复原任务并区分"修改计划"与"停止执行"；多条更新能否同时遵守而不是只记住最后一条。
• 事件优先级要动态判断，不是静态排序：Agent 应像秘书按紧急程度决定先处理哪个、做到一半能否暂停切换，而不是固定先后。长任务异步执行不应阻塞用户交互，被打断的任务要能自然恢复。
• 占位符只在打断时引入：常态下 LLM 看到的就是标准同步轨迹；只有出现打断才插入"工具正在后台执行"的占位 tool result 修复配对格式，后台真实结果到达后再以带来源和任务 ID 的事件送回。

实践模式

1. 搭骨架：inbox 队列 → 分发器（按紧急度路由）→ 单线程事件循环 → 轨迹存储。参考 chapter6/async-agent/ 的 runtime.py::AgentRuntime（_dispatcher 路由紧急度、_handle_interrupt 在安全点取消、run_llm_turn 推进轨迹）。
2. 接事件源：三类事件触发工具——定时器（一次性 + 循环，循环定时器兼容只能主动查询的外部服务）、后台任务监控（监控新增输出或关键词，别让 Agent 反复盯命令行浪费 token）、外部事件通道（实时推送）。每个触发源都要定义过滤规则和足够上下文的 payload，避免无关事件烧算力。
3. 定义紧急度分类器：先列事件类型清单和默认策略，再用轻量 LLM 兜住语义边界（"马上停下来"→取消式，"报告用中文发我"→队列式，"天气如何"→并行式）。
4. 同步兼容层五规则：assistant 消息与工具调用项立即落轨迹；只有工具真正返回才记 tool result；执行中被打断则生成占位 tool result 后再追加新事件并重新请求；无 steering 时取消未完成生成、只保留已确认消息；非打断事件一律入队批处理。
5. 多条更新的综合：给每个未处理事件加显式标记（[未处理事件 2/4] ...），末尾附汇总，提示模型必须回应全部，防止只记住最后一条。
6. 并发任务的状态可观测：每个异步工具调用都要有任务 ID，支持按 ID 查询进度和取消；完成事件注入主轨迹时带真实结果（退出码、产出哈希），不要只报"已完成"。
7. 触达用户：用户沟通工具支持异步消息、已读/未读、多渠道（IM/短信/邮件/电话/推送），按紧急程度、用户状态、内容性质、用户偏好选择渠道——通知机制同时是召回机制，用于长任务完成与周期性任务的习惯养成。
8. 身份与环境：给 Agent 独立虚拟身份（专属账号、存储、计算环境）跑在隔离的 VM/容器或虚拟手机里；通过共享卷（如 /workspace/shared）以文件路径而非内容传数据，避免占满上下文；必须登录用户真实账号时走 Human-in-the-Loop（VNC/RDP 让用户亲自登录，会话令牌有效期内复用）。
9. 验收三层：事件能否及时到达 → 系统是否按策略执行（取消真的终止了子进程、批量是否齐全）→ 模型是否正确使用了这批更新（结果归属、未完成时不编造、恢复原任务）。可加状态检查点做中断后恢复验证。

常见陷阱

• 把"任务已启动"当成"任务已完成"：占位符让模型误以为结果已到，在真实结果返回前做出依赖它的判断。必须用明确的任务状态和结果校验，并在评估中检查是否编造了未到达的数据。
• 注意力分散：批量事件只回应最后一条，前面的要求被静默丢弃。
• 轮询冒充异步：让 Agent 每隔几秒查一次状态，既慢又烧 token；要么用推送通道，要么用循环定时器 + 增量监控。
• 明文拼接不可见推理：服务端管理的 reasoning 状态要按提供商续接协议保留，不要自行拼出"半截思考"回灌。
• 取消只发信号不检查：执行器若不响应取消（类似不检查 ctx.Done()），"停止"消息发出去任务还在跑。
• 给所有事件同一优先级：紧急告警和常规输入走同一条队列，要么该停的停不下来，要么鸡毛蒜皮打断主任务。
• 直接托管用户个人账号：Agent 出错或被攻破即暴露全部数字身份；独立身份 + 隔离环境更安全也更可审计。
• 忽略反机器人机制：数据中心 IP 的虚拟环境容易被识别，需住宅代理；平台侧还可能有生态级封禁。
• 用一次失败反推训练原因：未公开的训练过程不可证，先查系统层的任务状态、事件来源和执行反馈。
• 忘记维护多个对话线程的关联：第三方消息如何影响用户情绪、用户在不同线程扮演什么角色、何时综合多线程信息给建议——纯事件路由处理不了这类上下文。

配套代码

• chapter6/async-agent/ — 实验 6-2：事件驱动异步 Agent 框架（Flux），并行工具、打断取消、检查点持久化，含零依赖离线 demo。
• chapter6/agent-with-event-trigger/ — 实验 6-1：FastAPI 事件驱动 Agent + 42 个 MCP 工具，演示"事件到达 → Agent 处理 → 结果输出"最小闭环。
• chapter6/async-agent/ 同时给出离线验收 demo（并行 vs 串行墙钟时间、打断取消后恢复、检查点持久化与还原）与 LLM 场景复现两条路径，适合先跑离线 demo 验证运行时再上真实模型。
• chapter6/astra-async-steering/ — 实验 6-3：GPT-6 Astra 原生异步工具调用与回合中途引导，含 sync / async / steer_reasoning / async_steer / unsupported_steer 五组对照与真实 API 运行记录。
• 语音相关（时间尺度更细的"可打断"场景）：chapter6/end-to-end-speech/、chapter6/streaming-speech/、chapter6/controllable-tts/、chapter6/live-audio/、chapter6/phone-agent/。
• 机器人相关（毫秒级连续观察与动作分块）：chapter6/gemini-xlerobot-navigation/、chapter6/rgb-sim2real-grasping/、chapter6/xlerobot-teleoperation/。

深度阅读

• book/chapter6.md「模态与触发时机的扩展」
• book/chapter6.md「异步与事件驱动：当世界主动找上门」
• 相关设计文档：chapter6/async-agent/agent_framework_design.md（Flux 框架的运行时设计）