Documentation Index
Fetch the complete documentation index at: https://simplellmfunc.cn/llms.txt
Use this file to discover all available pages before exploring further.
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
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
| Event | When | Key Fields |
|---|
ReactStartEvent | ReAct loop begins | — |
ReactIterationStartEvent | New iteration starts | iteration |
ReactIterationEndEvent | Iteration completes | iteration |
ReactEndEvent | Loop terminates | final_messages, response |
LLM Call
| Event | When | Key Fields |
|---|
LLMCallStartEvent | Before calling the provider | messages (what the LLM sees) |
LLMChunkArriveEvent | Each streaming chunk | chunk (text delta) |
LLMCallEndEvent | LLM response complete | usage, content, response, messages |
LLMCallErrorEvent | LLM call failed | error |
| Event | When | Key Fields |
|---|
ToolCallsBatchStartEvent | Tool batch begins | tool_calls |
ToolCallStartEvent | Single tool starts | tool_name, tool_call_id, arguments |
ToolCallArgumentsDeltaEvent | Streaming tool args | delta |
ToolCallEndEvent | Tool completes | tool_name, result, execution_time, success |
ToolCallErrorEvent | Tool failed | tool_name, error |
ToolCallsBatchEndEvent | All tools in batch done | — |
Custom Events
| Event | When | Key Fields |
|---|
CustomEvent | Tool emits custom data | event_name, data, tool_call_id |
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
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