跳转到主要内容

​
事件流

@llm_chat 调用返回 AsyncGenerator[ReactOutput, None]；@llm_function 的事件流通过 fn.stream(...) 返回。每个产出的项要么是响应（最终输出），要么是事件（生命周期信号）。

​
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):
        # 最终响应——文本或类型化结果
        print(output.response)
        history = output.messages
    elif is_event_yield(output):
        # 生命周期事件
        handle_event(output.event)

​
ResponseYield

包含智能体的最终输出： 
@dataclass
class ResponseYield:
    response: Any              # The result (str, Pydantic model, raw dict)
    messages: NormalizedMessageList  # Updated conversation history

​
EventYield

包含一个生命周期事件： 
@dataclass
class EventYield:
    event: ReActEvent          # One of 14 event types
    origin: EventOrigin        # Source identification (main agent or fork)

​
14 种事件类型

​
循环生命周期

事件触发时机关键字段
ReactStartEventReAct 循环开始—
ReactIterationStartEvent新迭代开始iteration
ReactIterationEndEvent迭代完成iteration
ReactEndEvent循环终止final_messages, response

​
LLM 调用

事件触发时机关键字段
LLMCallStartEvent调用提供商之前messages（LLM 看到的内容）
LLMChunkArriveEvent每个流式数据块到达chunk（文本增量）
LLMCallEndEventLLM 响应完成usage, content, response, messages
LLMCallErrorEventLLM 调用失败error

​
工具执行

事件触发时机关键字段
ToolCallsBatchStartEvent工具批次开始tool_calls
ToolCallStartEvent单个工具开始执行tool_name, tool_call_id, arguments
ToolCallArgumentsDeltaEvent流式工具参数delta
ToolCallEndEvent工具执行完成tool_name, result, execution_time, success
ToolCallErrorEvent工具执行失败tool_name, error
ToolCallsBatchEndEvent批次中所有工具完成—

​
自定义事件

事件触发时机关键字段
CustomEvent工具发出自定义数据event_name, data, tool_call_id
自定义事件由工具通过 ToolEventEmitter 发出——用于流式输出工具结果（例如 PyRepl 的标准输出、shell 输出等）。

​
便捷过滤器

from SimpleLLMFunc.hooks import responses_only, events_only, filter_events

# 仅获取最终响应
async for resp in responses_only(agent("hello", history)):
    print(resp.response)

# 仅获取事件
async for evt in events_only(agent("hello", history)):
    handle(evt.event)

# 仅获取特定事件类型
from SimpleLLMFunc.hooks import LLMChunkArriveEvent

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

​
EventOrigin（分叉路由）

使用 SelfRef 分叉时，事件来自多个智能体。EventOrigin 标识事件来源： 
@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
可以利用它将事件路由到不同的 UI 面板（例如主智能体与子智能体）。

​
常见模式

​
流式文本输出到终端

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

​
进度追踪

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}]")

​
事件观察者装饰器

用于跨切面事件处理，无需修改消费逻辑： 
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 参考：事件