> ## Documentation Index
> Fetch the complete documentation index at: https://simplellmfunc.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# 中断与取消

> 使用 AbortSignal 取消正在运行的 LLM 调用和工具执行

# 中断与取消

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

## 基本用法

```python theme={null}
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 中使用

用法相同：

```python theme={null}
@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` 即可触发信号：

```python theme={null}
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
```

## 检查中断状态

```python theme={null}
signal = AbortSignal()

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

## 实用模式：超时控制

```python theme={null}
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 参考：事件](/zh/api/events)