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 查询进度和取消；完成事件注入主轨迹时带真实结果（退出码、产出哈希），不要只报"已完成"。 6. **触达用户**：用户沟通工具支持异步消息、已读/未读、多渠道（IM/短信/邮件/电话/推送），按紧急程度、用户状态、内容性质、用户偏好选择渠道——通知机制同时是召回机制，用于长任务完成与周期性任务的习惯养成。 7. **身份与环境**：给 Agent 独立虚拟身份（专属账号、存储、计算环境）跑在隔离的 VM/容器或虚拟手机里；通过共享卷（如 `/workspace/shared`）以**文件路径**而非内容传数据，避免占满上下文；必须登录用户真实账号时走 Human-in-the-Loop（VNC/RDP 让用户亲自登录，会话令牌有效期内复用）。 8. **验收三层**：事件能否及时到达 → 系统是否按策略执行（取消真的终止了子进程、批量是否齐全）→ 模型是否正确使用了这批更新（结果归属、未完成时不编造、恢复原任务）。可加状态检查点做中断后恢复验证。 ## 常见陷阱 - **把"任务已启动"当成"任务已完成"**：占位符让模型误以为结果已到，在真实结果返回前做出依赖它的判断。必须用明确的任务状态和结果校验，并在评估中检查是否编造了未到达的数据。 - **注意力分散**：批量事件只回应最后一条，前面的要求被静默丢弃。 - **轮询冒充异步**：让 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 框架的运行时设计）