明白，我会为你生成一篇完整的 Markdown 文章，标题为 《Eino 框架入门：Callback 可观测性——给 Agent 装个“监控摄像头”》，内容涵盖概念解释、应用场景、实例代码、最佳实践等，字数超过 5000 字。我会按结构化方式分章节撰写，方便阅读和查找。下面是完整内容：

Eino 框架入门：Callback 可观测性——给 Agent 装个“监控摄像头”

引言

在现代智能系统中，Agent 的执行流程可能非常复杂。尤其在使用 Eino 框架 构建多 Agent 协作系统时，理解每个 Agent 在运行时的行为尤为重要。为了更高效地调试、优化和监控 Agent，我们需要一种机制，让我们像给 Agent 装上“监控摄像头”一样，实时观察其行为。这就是 Callback 可观测性（Callback Observability） 的核心价值。

本文将从理论讲解到实际案例，系统介绍 Eino 框架的 Callback 可观测性，并给出可直接使用的代码示例和应用场景。

第一章：什么是 Callback 可观测性

1.1 Callback 的概念

在编程中，Callback（回调函数） 是一种函数，作为参数传递给另一个函数，由后者在合适的时机调用。它通常用于处理异步事件、流程拦截或状态更新。在 Eino 框架中，Callback 主要用来：

拦截 Agent 行为：在 Agent 执行任务前、执行中或完成后触发回调。
收集运行数据：记录输入输出、运行时状态、异常信息。
增强可观测性：让开发者和系统管理员实时看到 Agent 的内部行为。

1.2 可观测性的意义

“可观测性”源自监控学，它的核心目标是：

可视化系统行为：知道系统在做什么。
便于调试：发现错误、性能瓶颈和异常路径。
支持优化：通过收集数据进行分析和策略改进。

在 Eino 框架中，Callback 可观测性就像给 Agent 装了一个透明的“摄像头”，可以捕捉以下内容：

Agent 收到的指令或任务
执行的步骤和状态变化
输出结果与异常信息
与其他 Agent 的交互情况

这种机制不仅对开发者友好，也为运维团队提供了监控数据来源。

第二章：Eino Callback 可观测性核心原理

Eino 框架通过以下三类 Callback 提供可观测性：

Pre-Action Callback：在 Agent 执行任务之前触发，用于记录输入和初始化状态。
Post-Action Callback：在 Agent 完成任务之后触发，用于记录结果和耗时。
Error Callback：在 Agent 执行过程中发生异常时触发，用于捕获错误信息和堆栈。

2.1 Callback 的触发流程

textCopy Code
任务触发 → Pre-Action Callback → Agent 执行任务 → Post-Action Callback
                                  ↘ Error Callback (若有异常)

通过这种结构，我们可以在任务的整个生命周期中收集和分析数据，从而实现完整的可观测性。

2.2 Callback 数据结构

Eino 的 Callback 提供的数据结构一般包括：

jsonCopy Code
{
  "agent_id": "agent_001",
  "task_id": "task_20260401_001",
  "timestamp": "2026-04-01T10:00:00Z",
  "input": { "query": "分析用户行为" },
  "output": { "result": "用户活跃度分析完成" },
  "status": "success",
  "error": null
}

其中关键字段说明：

agent_id：Agent 唯一标识
task_id：任务唯一标识
timestamp：事件时间戳
input、output：任务输入和输出
status：任务状态（success、failed、running）
error：异常信息，如果有

第三章：如何在 Eino 中使用 Callback

3.1 基本使用方式

假设我们有一个简单 Agent，用于处理自然语言查询，我们可以给它注册 Callback：

pythonCopy Code
from eino import Agent, Callback

class LoggingCallback(Callback):
    def on_pre_action(self, agent, task):
        print(f"[PRE] Agent {agent.name} 正在执行任务 {task.id}，输入: {task.input}")

    def on_post_action(self, agent, task, output):
        print(f"[POST] Agent {agent.name} 完成任务 {task.id}，输出: {output}")

    def on_error(self, agent, task, error):
        print(f"[ERROR] Agent {agent.name} 执行任务 {task.id} 出错: {error}")

# 创建 Agent
agent = Agent(name="TestAgent")

# 注册 Callback
agent.add_callback(LoggingCallback())

# 执行任务
agent.run_task({"query": "分析用户行为"})

输出示例：

textCopy Code
[PRE] Agent TestAgent 正在执行任务 task_001，输入: {'query': '分析用户行为'}
[POST] Agent TestAgent 完成任务 task_001，输出: {'result': '用户活跃度分析完成'}

3.2 高级使用：数据采集与分析

我们可以把 Callback 扩展为数据收集器，将每次任务执行的详细信息保存到数据库或日志系统中：

pythonCopy Code
import datetime
import sqlite3

class DBLoggingCallback(Callback):
    def __init__(self, db_path="agent_logs.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute(
            """CREATE TABLE IF NOT EXISTS logs(
                agent_id TEXT, task_id TEXT, timestamp TEXT,
                input TEXT, output TEXT, status TEXT, error TEXT
            )"""
        )

    def _log(self, agent, task, output=None, status="success", error=None):
        self.conn.execute(
            "INSERT INTO logs VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                agent.name,
                task.id,
                datetime.datetime.utcnow().isoformat(),
                str(task.input),
                str(output),
                status,
                str(error),
            ),
        )
        self.conn.commit()

    def on_pre_action(self, agent, task):
        self._log(agent, task, status="running")

    def on_post_action(self, agent, task, output):
        self._log(agent, task, output=output, status="success")

    def on_error(self, agent, task, error):
        self._log(agent, task, status="failed", error=error)

通过这种方式，我们就可以在数据库中建立 Agent 行为日志，实现长期监控与分析。

第四章：Callback 可观测性的应用场景

4.1 场景一：任务调试

当 Agent 执行复杂任务时，如果没有 Callback，开发者只能看到最终结果。一旦出错，难以定位问题。Callback 可观测性可以让开发者看到任务每一步的输入、输出和状态，极大提升调试效率。

示例：

Agent 执行文本摘要任务
Callback 输出每一阶段的处理结果
错误时记录详细堆栈和上下文信息

4.2 场景二：性能分析

我们可以在 Pre/Post Callback 中记录时间戳，分析任务执行耗时，发现性能瓶颈。

pythonCopy Code
import time

class TimingCallback(Callback):
    def on_pre_action(self, agent, task):
        task.meta['start_time'] = time.time()

    def on_post_action(self, agent, task, output):
        duration = time.time() - task.meta['start_time']
        print(f"任务 {task.id} 执行耗时: {duration:.3f}s")

4.3 场景三：异常监控

在生产环境中，Callback 可以将异常信息发送到告警系统，如 Slack、邮件或监控平台：

pythonCopy Code
class AlertCallback(Callback):
    def on_error(self, agent, task, error):
        # 假设 send_alert 是发送告警的函数
        send_alert(f"Agent {agent.name} 出错: {error}")

4.4 场景四：用户行为分析

在多 Agent 协作系统中，我们可以通过 Callback 记录每个 Agent 的操作历史，从而进行用户行为分析、推荐策略优化和任务改进。

第五章：案例分析

5.1 案例：多 Agent 协作的客服系统

假设我们有一个智能客服系统，由多个 Agent 协作完成任务：

AgentA：接收用户消息
AgentB：分析意图
AgentC：生成回复

我们可以为每个 Agent 添加 LoggingCallback 和 TimingCallback，实现可观测性：

pythonCopy Code
agents = [Agent