跳转到主要内容

​
中断与取消

AbortSignal 为运行中的智能体调用提供协作式取消机制。它能干净地终止 LLM 流式传输和工具执行。

​
基本用法

from SimpleLLMFunc.hooks import AbortSignal

signal = AbortSignal()


@llm_chat(llm_interface=llm, toolkit=[...], stream=True)
async def agent(message: str, history: list | None = None):
    """An agent that can be cancelled."""
    pass


async def run():
    async for output in agent("do something long", history, _abort_signal=signal):
        if is_response_yield(output):
            history = output.messages


# 从另一个协程或回调中取消：
signal.abort("user cancelled")

​
中断后发生什么

​
LLM 流式传输期间

  1. 流式连接被终止
  2. 已接收的内容会被保留
  3. 运行时产生内部截断补丁，包含：
    • partial_content — 中断前已接收的内容
    • abort_reason — 你提供的原因
  4. ReAct 循环通过终结路径退出

​
工具执行期间

  1. 正在运行的工具被取消（asyncio 任务取消）
  2. 对每个被取消的工具，运行时产生内部取消补丁，包含：
    • tool_call_id — 被取消的调用 ID
    • tool_name — 正在运行的工具名称
    • abort_reason — 取消原因
  3. 这确保对话记录在结构上保持有效（每个工具调用都有对应结果）

​
一致性保证

中断操作始终产生有效的最终状态。对话记录不会遗留悬空的工具调用或不完整的消息。内部对话记录补丁确保了结构一致性。

​
在 @llm_function 中使用

用法相同： 
@llm_function(llm_interface=llm, toolkit=[research_tool])
async def research(question: str) -> Report:
    """Research the question."""
    pass

signal = AbortSignal()

async for output in research.stream("...", _abort_signal=signal):
    ...

​
TUI 集成

内置的 @tui 装饰器自动处理中断——按 Ctrl+C 即可触发信号： 
from SimpleLLMFunc.utils.tui import tui

@tui
@llm_chat(llm_interface=llm, toolkit=[...], stream=True)
async def agent(message: str, history: list | None = None, _abort_signal=None):
    """My agent."""
    pass

​
检查中断状态

signal = AbortSignal()

# 检查是否已中断
if signal.is_aborted:
    print(f"Aborted: {signal.reason}")

​
实用模式：超时控制

import asyncio

signal = AbortSignal()

async def with_timeout():
    task = asyncio.create_task(consume(agent("...", history, _abort_signal=signal)))

    try:
        await asyncio.wait_for(task, timeout=60.0)
    except asyncio.TimeoutError:
        signal.abort("timeout after 60s")
        await task  # 等待清理完成
→ API 参考：事件