[!info] related notes

• 所属 MOC: Agent Runtime MOC
• 相关: Cancellation Propagation, Abort Generation, Timeout Control

Cancellation

一句话定义

Cancellation 是 Agent Runtime 中支持用户取消正在执行的 Run 的机制。取消不是简单地断开连接，而是要停止 LLM 推理、停止工具执行、保存已生成内容、释放资源。

核心原理

取消信号传播

用户点击"停止"
    │
    ▼
前端: AbortController.abort()
    │
    ▼
Go 后端: context cancelled
    │
    ▼
Python: cancel_event.set()
    │
    ├─ LLM 流式生成 → 检测到 cancel → 停止
    ├─ 工具执行 → 检测到 cancel → 尝试中断
    └─ Agent Loop → 检测到 cancel → 退出循环

Python 实现

class CancellableAgentLoop:
    def __init__(self, cancel_event: asyncio.Event):
        self.cancel_event = cancel_event

    async def run(self, context):
        messages = build_messages(context)

        for step in range(MAX_STEPS):
            # 检查取消信号
            if self.cancel_event.is_set():
                yield DoneEvent(status="cancelled")
                return

            # LLM 调用（可取消）
            response = await self.call_llm_cancellable(messages)

            if response.tool_calls:
                for tool_call in response.tool_calls:
                    # 再次检查取消
                    if self.cancel_event.is_set():
                        yield DoneEvent(status="cancelled")
                        return

                    result = await self.execute_tool_cancellable(tool_call)
                    messages.append(tool_result_message(tool_call, result))
            else:
                yield TextDeltaEvent(response.content)
                break

    async def call_llm_cancellable(self, messages):
        """可取消的 LLM 调用"""
        task = asyncio.create_task(self.llm.chat(messages))
        cancel_task = asyncio.create_task(self.cancel_event.wait())

        done, pending = await asyncio.wait(
            [task, cancel_task],
            return_when=asyncio.FIRST_COMPLETED,
        )

        for t in pending:
            t.cancel()

        if cancel_task in done:
            raise CancelledError()

        return task.result()

保存已生成内容

async def cancel_run(run_id: str):
    # 1. 设置取消信号
    cancel_events[run_id].set()

    # 2. 等待当前操作完成（带超时）
    await asyncio.wait_for(running_tasks[run_id], timeout=5.0)

    # 3. 保存已生成内容
    partial_content = get_partial_content(run_id)
    if partial_content:
        save_message(run_id, partial_content, status="cancelled")

常见坑

1. 取消后不保存内容: 用户已经等了 30 秒生成的内容全部丢失
2. 取消信号传播不到: Python 没有检测 cancel_event
3. 工具执行不可取消: 正在执行的数据库查询无法中断
4. 取消后资源泄漏: 连接、文件句柄没有释放

参考资料

• Cancellation Propagation
• Abort Generation