Skip to main content

​
Abort

AbortSignal provides cooperative cancellation for running agent invocations. It cleanly terminates LLM streaming and tool execution.

​
Basic Usage

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


# From another coroutine or callback:
signal.abort("user cancelled")

​
What Happens on Abort

​
During LLM Streaming

  1. The streaming connection is terminated
  2. Whatever content was received so far is preserved
  3. The runtime produces an internal truncation patch with:
    • partial_content — what was received before abort
    • abort_reason — the reason you provided
  4. The ReAct loop terminates via the finalize path

​
During Tool Execution

  1. Running tools are cancelled (asyncio task cancellation)
  2. For each cancelled tool, the runtime produces an internal cancellation patch with:
    • tool_call_id — which call was cancelled
    • tool_name — the tool that was running
    • abort_reason — your reason
  3. This ensures the transcript stays structurally valid (every tool call has a result)

​
The Guarantee

Abort always produces a valid final state. The transcript is never left with dangling tool calls or incomplete messages. Internal transcript patches ensure structural consistency.

​
Abort in @llm_function

Works the same way: 
@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 Integration

The built-in @tui decorator handles abort automatically — Ctrl+C triggers the signal: 
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

​
Checking Abort State

signal = AbortSignal()

# Check if aborted
if signal.is_aborted:
    print(f"Aborted: {signal.reason}")

​
Practical Pattern: Timeout

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  # Let it clean up
→ API Reference: Events