Skip to main content

​
Event Stream

@llm_chat calls return an AsyncGenerator[ReactOutput, None]; @llm_function exposes the same event stream through fn.stream(...). Each yielded item is either a response (final output) or an event (lifecycle signal).

​
ReactOutput

ReactOutput = ResponseYield | EventYield
Use type guards to distinguish: 
from SimpleLLMFunc.hooks import is_response_yield, is_event_yield

async for output in agent("hello", history):
    if is_response_yield(output):
        # Final response — text or typed result
        print(output.response)
        history = output.messages
    elif is_event_yield(output):
        # Lifecycle event
        handle_event(output.event)

​
ResponseYield

Contains the final agent output: 
@dataclass
class ResponseYield:
    response: Any              # The result (str, Pydantic model, raw dict)
    messages: NormalizedMessageList  # Updated conversation history

​
EventYield

Contains a lifecycle event: 
@dataclass
class EventYield:
    event: ReActEvent          # One of 14 event types
    origin: EventOrigin        # Source identification (main agent or fork)

​
The 14 Event Types

​
Loop Lifecycle

EventWhenKey Fields
ReactStartEventReAct loop begins—
ReactIterationStartEventNew iteration startsiteration
ReactIterationEndEventIteration completesiteration
ReactEndEventLoop terminatesfinal_messages, response

​
LLM Call

EventWhenKey Fields
LLMCallStartEventBefore calling the providermessages (what the LLM sees)
LLMChunkArriveEventEach streaming chunkchunk (text delta)
LLMCallEndEventLLM response completeusage, content, response, messages
LLMCallErrorEventLLM call failederror

​
Tool Execution

EventWhenKey Fields
ToolCallsBatchStartEventTool batch beginstool_calls
ToolCallStartEventSingle tool startstool_name, tool_call_id, arguments
ToolCallArgumentsDeltaEventStreaming tool argsdelta
ToolCallEndEventTool completestool_name, result, execution_time, success
ToolCallErrorEventTool failedtool_name, error
ToolCallsBatchEndEventAll tools in batch done—

​
Custom Events

EventWhenKey Fields
CustomEventTool emits custom dataevent_name, data, tool_call_id
Custom events are emitted by tools via ToolEventEmitter — used for streaming tool output (e.g., PyRepl stdout, shell output).

​
Convenience Filters

from SimpleLLMFunc.hooks import responses_only, events_only, filter_events

# Only final responses
async for resp in responses_only(agent("hello", history)):
    print(resp.response)

# Only events
async for evt in events_only(agent("hello", history)):
    handle(evt.event)

# Only specific event types
from SimpleLLMFunc.hooks import LLMChunkArriveEvent

async for evt in filter_events(agent("hello", history), LLMChunkArriveEvent):
    print(evt.event.accumulated_content, end="")

​
EventOrigin (Fork Routing)

When using SelfRef forks, events come from multiple agents. EventOrigin identifies the source: 
@dataclass
class EventOrigin:
    session_id: str
    agent_call_id: str
    event_seq: int
    parent_agent_call_id: str | None = None
    fork_id: str | None = None
    fork_depth: int = 0
    fork_seq: int | None = None
    selfref_instance_id: str | None = None
    source_memory_key: str | None = None
    memory_key: str | None = None
    tool_name: str | None = None
    tool_call_id: str | None = None
Use this to route events to different UI panels (e.g., main agent vs. child agents).

​
Common Patterns

​
Streaming Text to Terminal

async for output in agent("hello", history):
    if is_event_yield(output):
        if isinstance(output.event, LLMChunkArriveEvent):
            print(output.event.accumulated_content, end="", flush=True)
    elif is_response_yield(output):
        print()  # newline
        history = output.messages

​
Progress Tracking

async for output in agent("do complex task", history):
    if is_event_yield(output):
        event = output.event
        if isinstance(event, ToolCallStartEvent):
            print(f"  → calling {event.tool_name}...")
        elif isinstance(event, ToolCallEndEvent):
            print(f"  ✓ {event.tool_name} ({event.execution_time}ms)")
        elif isinstance(event, ReactIterationStartEvent):
            print(f"[iteration {event.iteration}]")

​
Event Observer Decorator

For cross-cutting event handling without modifying consumption logic: 
from SimpleLLMFunc.hooks import with_event_observer

def log_events(event: ReActEvent, origin: EventOrigin):
    if isinstance(event, LLMCallEndEvent):
        print(f"Tokens: {event.usage}")

@with_event_observer(log_events)
@llm_chat(llm_interface=llm, toolkit=[...])
async def agent(message: str, history: list | None = None):
    """My agent."""
    pass
→ API Reference: Events