ADK 中止执行要先封口工具调用

先看结论

Google ADK Python 2.11.0 把“停止生成”从调用方自己关闭异步生成器，变成了 Runner、Workflow 和节点都能理解的协作式取消：调用方传入 asyncio.Event 作为 abort_signal，服务端在 SSE 客户端断开时触发它，运行中的工作流停止派生任务；如果取消发生在工具调用期间，ADK 会为悬空的 function call 补写合成的 function response，再结束事件流。对开发者来说，最重要的变化不是多了一个停止按钮，而是停止之后的会话历史仍有机会保持协议完整。

这项变化适用于 ADK Python 2.11.0。官方发布说明把它列为执行取消能力，并同时说明 /run_sse 已经接入客户端断连处理。升级后可以把取消信号接在用户点击“停止”、浏览器关闭流、网关主动断开和服务端超时这些入口上，而不是直接对任务调用 cancel()，也不要把 async for 的 break 当成完整的业务语义。

如果你正在排查流式 Agent 的中止、工具重复执行或下一轮请求报错，可以先参考站内的 LLM 流式输出工程排查，再从 模型与平台目录 核对实际接入的模型和 API 入口。这里的关键在 Agent 运行时如何收尾，不在某个上游模型是否返回了最后一个 token。

2.11.0 具体补上了什么

从“关掉流”变成“通知运行时停止”

过去，调用方想停止 Runner.run_async()，常见做法是退出 async for，或者直接取消外层 asyncio.Task。这两种动作只描述了调用方不再消费结果，并没有明确告诉 Agent：用户主动中止了本次执行，后续节点和工具都不应继续推进。ADK 仓库中的 issue #4796 记录了这个缺口：当时 break 可能触发异步生成器关闭，但中止是否被记录、工具调用是否封口、下一轮会话是否一致，都缺少公开保证。

2.11.0 的接口是一个可复用的信号对象：

abort_signal = asyncio.Event()

async for event in runner.run_async(
    user_id="user_123",
    session_id=session.id,
    new_message=message,
    abort_signal=abort_signal,
):
    yield event

需要停止时，在同一个事件循环中调用：

abort_signal.set()

官方指南说明，Runner 会在后续事件周期检查信号，取消活动的子任务并干净地结束生成器。信号还会沿着 InvocationContext 传给子 Agent、Workflow 和节点，工具可以通过 ctx.is_aborted 得知外部取消已经到达。这样做的好处是“请求结束”和“业务中止”被区分开：前者是传输层事实，后者是 Agent 生命周期中的一个可处理状态。

/run_sse 已经把断连接进取消链路

只在业务代码里传 abort_signal 还不够，SSE 服务端必须知道浏览器或中转层何时断开。2.11.0 的 CLI 改动在 /run_sse 的事件生成器中创建取消事件，并在生成器收到 GeneratorExit 或 asyncio.CancelledError 时设置它。也就是说，客户端主动关闭连接、点击停止或上层响应流被关闭时，生产 Agent 事件的任务会收到取消信号，而不是继续在后台把模型和工具跑完。

这个设计与 ASGI 的事件模型相符：HTTP 连接可以收到 http.disconnect，应用需要接收并作出反应。ADK 的服务端提交记录明确展示了两个关键动作：把事件交给客户端后推进下一步，以及在生成器关闭时设置 abort_signal 并清理生产任务。对自建 SSE 入口的开发者而言，不能只在 StreamingResponse 外面捕获异常；要保证断连事件最终能触发同一个运行时取消信号。

正确接入的三层写法

第一层：在调用边界持有信号

把 abort_signal 放在一次 Agent 执行的生命周期里，不要做成跨请求共享的全局变量。一个请求一个 asyncio.Event，这样用户 A 关闭页面不会中止用户 B 的会话，也不会把已经结束的信号带到下一轮。

async def run_agent(runner, session, message):
    abort_signal = asyncio.Event()
    async for event in runner.run_async(
        user_id="user_123",
        session_id=session.id,
        new_message=message,
        abort_signal=abort_signal,
    ):
        yield event

真实项目里通常需要把信号和“停止请求”关联起来。停止请求可能来自同一个 SSE handler，也可能来自 WebSocket、后台超时任务或前端的独立 HTTP 请求。若停止请求来自另一个线程，不能直接假设跨线程调用 Event.set() 就能立即唤醒正确的事件循环。官方指南给出的做法是保存运行时的 loop，并使用 loop.call_soon_threadsafe(abort_signal.set) 把操作投递回运行 Agent 的 loop。

第二层：让长工具主动检查状态

abort_signal 能取消 ADK 管理的异步子任务，但同步阻塞代码不会凭空变成可抢占代码。Python 的 asyncio 任务采用协作式调度；调用 Task.cancel() 会在协程下一次合适的暂停点抛出 CancelledError，如果代码一直在事件循环线程里执行 CPU 密集循环，循环期间就没有机会处理取消。

因此，长时间运行的同步工具应在可接受的粒度上检查 ctx.is_aborted：

def process_records(records: list[dict], ctx) -> str:
    for index, record in enumerate(records):
        if ctx.is_aborted:
            return f"已在第 {index} 条记录前停止处理。"
        process_one(record)
    return "处理完成。"

如果工具调用的是外部服务，检查点还应放在每次分页、批量提交和重试之间。取消只能阻止尚未发生的后续动作，不能撤销已经发出的邮件、已经提交的订单或已经写入外部数据库的变更。因此，带副作用的工具仍需要幂等键、事务边界和可重试设计，不能把“Agent 被取消”理解为“外部操作自动回滚”。

第三层：把中止当成终态处理

消费事件流的代码不要只处理“正常完成”和“异常抛出”两种结果。ADK 指南说明，没有悬空工具调用时，运行时会追加带有 INVOCATION_ABORTED 的中止事件；如果有未完成的 function call，则会追加合成的 function response。调用方应把这类事件记录为一次明确的终态，而不是把它拼接成普通模型文本。

建议在应用层保存至少三种状态：completed、aborted 和 failed。其中 aborted 可以携带触发原因，例如 client_disconnect、user_stop、request_timeout 或 worker_shutdown。这样下一轮请求可以正常开始，审计和调试也能区分用户主动停止、网络断连与真实错误。

工具调用为什么必须“封口”

一次带工具的 Agent 交互，通常先由模型写入 function call，再由工具执行并写回 function response，之后模型才能决定下一步。如果只在中间把消费端关闭，历史里可能留下“已经请求工具但没有结果”的半对话。下一轮请求如果把这段历史原样发给模型，就可能出现协议校验失败、重复执行工具，或者模型一直等待一个永远不会到达的结果。

ADK 2.11.0 的处理是把取消分成两条路径。工具调用尚未开始或没有待配对的调用时，运行时追加中止事件；工具调用已经写入但响应尚未写回时，运行时为悬空调用追加合成响应，并把它写入会话历史。这样下一轮模型看到的不是一个未闭合的调用，而是一次明确失败或中止的工具结果。

这并不表示合成响应等同于真实工具结果。业务代码必须让模型知道工具没有完成，并在后续提示中避免把“取消”解释成“执行成功”。对于不可重复的副作用，更应该把工具设计成两阶段：第一步生成待确认操作，第二步在用户确认后提交；或者先写入任务表，再由可恢复的后台 worker 执行。这样即使流断开，也不会依赖模型自己猜测外部系统的真实状态。

三个容易误判的边界

断开连接不等于所有工作立即停止

官方文档的表述是协作式停止，并不是强制终止进程。已经进入不可中断的同步函数、正在等待没有取消语义的第三方 SDK，或者已经发到外部系统的请求，都可能继续到达一个安全边界后才结束。需要严格上限时，应同时配置请求超时、工具级超时和外部任务的幂等机制。

多进程部署不能只靠内存事件

asyncio.Event 只存在于创建它的 Python 进程和事件循环中。官方指南明确把跨进程取消列为边界：如果停止请求打到了另一台副本，应用层必须把取消消息路由到真正执行该 Agent 的 worker。可选做法包括按 invocation_id 做粘性路由、在 Redis 或数据库中记录取消标记，或者把长任务交给具备取消协议的工作队列。共享一个全局布尔变量不能解决这个问题。

主动结束和外部取消不是一回事

当工具或回调根据业务规则决定结束当前调用时，ADK 指南建议使用 ctx.end_invocation = True；当用户、SSE 断连或超时从外部要求停止时，使用 abort_signal。两者的审计原因、事件语义和是否需要通知上层都可能不同，把它们混成一个“返回空字符串”会让调试变得困难。

升级与验收步骤

1. 将 google-adk 升级到 2.11.0 或更高版本，并确认运行环境实际加载的版本，不要只改锁文件。
2. 给每次 runner.run_async() 创建独立的 asyncio.Event，在用户停止和服务端断连两个入口都触发它。
3. 用一个会持续数秒的异步工具和一个分段处理的同步工具分别测试，确认取消后不会继续启动新的工具调用。
4. 在工具调用已经产生 function call、但响应尚未返回的时间点断开 SSE，检查会话历史是否出现合成 function response。
5. 断开后使用同一个 session_id 发起下一轮文本请求，确认不会因为未配对的 function call 而失败，也不会自动重复刚才的副作用操作。
6. 在多 worker 环境中让停止请求故意打到不同副本，验证路由或共享取消机制；如果没有这项机制，不要宣称分布式停止已经生效。
7. 为 completed、aborted 和 failed 分别写日志和监控，至少记录 session_id、invocation_id、取消原因及是否存在待封口工具调用。

这套验收的重点是会话一致性，而不是“按钮按下后几毫秒内没有新 token”。速度当然重要，但对于带工具的 Agent，能否在中止后安全开始下一轮、能否避免重复副作用，才是取消实现是否可用的判断标准。

参考来源

• Google ADK Python 2.11.0 发布说明
• ADK Runner 执行取消指南
• ADK 将 SSE 断连接入 abort_signal 的提交
• ADK “Stop generating” 问题与原始需求
• Python asyncio 任务取消文档

资料核验时间：2026 年 10 月 5 日（UTC）。版本相关结论以 ADK Python 2.11.0 代码、发布说明和文档为准。