Agno 流式事件系统与可观测性

Agno 流式事件系统与可观测性

学习目标

本章将分析 Agno 的事件系统和可观测性：

• RunOutputEvent 体系与多种具体事件
• events_to_skip 事件过滤和 store_events 持久化
• OpenTelemetry Tracing 集成
• Session Metrics 和 Run Metrics 双层指标

前置知识

建议先阅读：

• 流式传输架构 — Token/块/预览三层流式模型

下文假设你已理解上述概念，直接聚焦 Agno 的具体实现。

---

项目实践

RunOutputEvent 体系

Agno 的事件系统围绕 RunOutputEvent 基类构建：

# 伪代码：事件体系
class RunOutputEvent:
    """所有 Agent 运行事件的基类"""
    event: str  # 事件类型标识

class RunStartedEvent(RunOutputEvent):
    event = "run.started"

class RunCompletedEvent(RunOutputEvent):
    event = "run.completed"

class RunErrorEvent(RunOutputEvent):
    event = "run.error"

class RunPausedEvent(RunOutputEvent):
    event = "run.paused"

class RunCancelledEvent(RunOutputEvent):
    event = "run.cancelled"

class RunContentEvent(RunOutputEvent):
    event = "run.content"  # 流式内容增量

关键设计：每个事件类型有唯一的字符串标识（event），便于过滤和日志追踪。

事件过滤（events_to_skip）

调用方可以指定不需要的事件类型：

# 伪代码：事件过滤
agent = Agent(
    stream_events=True,
    events_to_skip=[RunStartedEvent, RunCompletedEvent],  # 跳过开始/结束事件
)

事件处理逻辑：

# 伪代码：handle_event
def handle_event(event: RunOutputEvent, run_response, events_to_skip=None):
    if events_to_skip and type(event) in events_to_skip:
        return None  # 跳过该事件
    if store_events:
        run_response.events.append(event)  # 持久化
    return event

事件持久化（store_events）

agent = Agent(
    store_events=True,  # 将事件存储到 run_response.events
)

启用后，所有未被跳过的事件都追加到 run_response.events 列表中。这允许：

• 重放整个执行过程
• 调试和分析
• 在 AgentOS 中通过 API 暴露给前端

OpenTelemetry Tracing

Agno 的 tracing/ 模块集成 OpenTelemetry：

# 伪代码：OpenTelemetry 集成
from opentelemetry import trace

tracer = trace.get_tracer("agno.agent")

def _run(agent, run_response, ...):
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.id", agent.id)
        span.set_attribute("agent.name", agent.name)
        span.set_attribute("run.id", run_response.run_id)
        span.set_attribute("model.id", agent.model.id)

        # 模型调用
        with tracer.start_as_current_span("model.call") as model_span:
            model_span.set_attribute("model.id", agent.model.id)
            model_response = call_model_with_fallback(...)
            model_span.set_attribute("tokens.input", model_response.usage.prompt_tokens)
            model_span.set_attribute("tokens.output", model_response.usage.completion_tokens)

        # 工具调用
        for tool_call in tool_calls:
            with tracer.start_as_current_span(f"tool.{tool_call.name}") as tool_span:
                tool_span.set_attribute("tool.name", tool_call.name)
                result = execute_tool(tool_call)
                tool_span.set_attribute("tool.status", "success")

双层 Metrics

Agno 提供两层指标：

# 伪代码：双层 Metrics
class RunMetrics:
    """单次运行的指标"""
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    tool_calls: int
    time_seconds: float

class SessionMetrics:
    """整个 Session 的累计指标"""
    total_runs: int
    total_prompt_tokens: int
    total_completion_tokens: int
    total_tokens: int
    avg_time_seconds: float

后台指标合并：记忆提取、学习提取等后台任务的指标在步骤 12 合并到主 Metrics：

# 伪代码：后台指标合并
merge_background_metrics(
    run_response.metrics,
    collect_background_metrics(memory_future, cultural_future, learning_future),
)

问题与规避

1. 事件数量过多

store_events=True 时，每次 run 可能产生数十个事件（started、content×N、completed 等）。

规避：

• 使用 events_to_skip 过滤不需要的事件
• 只存储关键事件（如 error、paused）
• 在 AgentOS 中限制存储的事件数量

2. OpenTelemetry 的性能开销

每个 span 创建和导出都有开销。

规避：

• 使用批量导出（而非每条 span 立即导出）
• 生产环境采样（如 10% 的请求）
• 禁用调试级别的 span

3. 后台任务指标的延迟

后台任务的 metrics 在步骤 12 才合并，如果后台任务尚未完成，metrics 不完整。

规避：wait_for_open_threads 等待所有后台任务完成后再合并，确保 metrics 完整性。

设计取舍

为什么用事件体系而非回调函数？

优势：事件可以持久化、重放、过滤；回调函数是瞬时的 代价：事件体系需要更多基础设施 替代方案：回调函数——更简单但不支持重放和持久化

为什么 Metrics 分两层？

优势：Run 级别追踪单次执行，Session 级别追踪长期趋势 代价：需要合并逻辑 替代方案：只有一层——要么只追踪单次（失去趋势），要么只追踪累计（失去细节）

参考来源

• Agno run/ 源码
• Agno tracing/ 源码