第 21 章 · 可观测性与监控集成

本章目标：读懂 usage_metrics 里的每一项，接入 Langtrace / Langfuse 两个第三方观测平台，并用事件监听把数据导出到自建监控系统。

21.1 先看懂 `usage_metrics`：不接平台也能算成本

每次 kickoff() 的返回值都带 token_usage（即 usage metrics），这是零依赖的成本核算起点：

python

import os
from crewai import Agent, Task, Crew, LLM

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

researcher = Agent(role="研究员", goal="调研主题", backstory="行业分析专家", llm=llm)
task = Task(
    description="调研 {topic} 的市场格局",
    expected_output="300 字中文简报",
    agent=researcher,
)
crew = Crew(agents=[researcher], tasks=[task])

result = crew.kickoff(inputs={"topic": "国产大模型"})

usage = result.token_usage
print(f"成功请求数:   {usage.successful_requests}")
print(f"提示词 tokens: {usage.prompt_tokens}")
print(f"补全 tokens:  {usage.completion_tokens}")
print(f"总 tokens:    {usage.total_tokens}")
# 按 DeepSeek 定价即可折算单次运行成本，
# 多次运行取均值后就能估算出每千次调用的预算

四个字段的含义与用法：

successful_requests：LLM 被调用了几次——它往往远大于任务数（一次 Agent 执行 = 至少一次调用 + 每次工具调用后再来一轮），是发现"Agent 循环打转"的第一指标；
prompt_tokens / completion_tokens：输入与输出分别计费，输出侧单价通常高数倍，优化优先砍输出长度；
total_tokens：总量，直接乘单价得成本。

Task 级指标

单个任务的 TaskOutput 同样携带用量信息；配合第 14 章的 task_callback 可以做到"每个环节花多少钱"的粒度归因。

21.2 接入 Langtrace：一行初始化的全链路追踪

Langtrace 是开源 LLM 观测平台，官方文档给出的接入方式非常轻量。注意一个关键顺序：init 必须在导入 CrewAI 之前：

bash

pip install langtrace-python-sdk

python

from langtrace_python_sdk import langtrace

# 必须放在所有 crewai 导入之前，才能拦截到全部调用链
langtrace.init(api_key=os.environ["LANGTRACE_API_KEY"])

# 之后照常编写 CrewAI 代码
from crewai import Agent, Task, Crew  # noqa: E402

在 Langtrace 控制台把项目类型设为 CrewAI 后，你可以获得三类视图：Token 与成本追踪（每次 Agent/工具交互的花费）、执行步骤 Trace 图（谁调了哪个工具、耗时几何）、性能回归对比（不同版本的运行横向比较）。

21.3 接入 Langfuse：OpenTelemetry 标准路线

Langfuse 是另一个主流开源平台，官方推荐通过 OpenTelemetry（经 OpenLit SDK） 接入——这条路的好处是你的埋点不绑定任何一家厂商：

bash

pip install langfuse openlit crewai crewai_tools

python

import os

# Langfuse 项目凭证（从控制台设置页获取）
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # 或 US 区

# 初始化 OpenLit 的 OpenTelemetry 埋点，trace 自动上报 Langfuse
import openlit
openlit.init()

from langfuse import get_client
langfuse = get_client()
assert langfuse.auth_check(), "Langfuse 凭证校验失败"

# 之后正常创建并运行 crew，trace 即出现在 Langfuse 控制台

选型参考：只想快速看到"花了多少钱、哪步慢"，Langtrace 一行 init 最省事；团队已有 OTel 基础设施、或需要自托管 Langfuse，走 OpenLit/OpenTelemetry 更可移植。两者可以并存灰度切换。

21.4 事件监听：导出到自建监控

不想依赖第三方平台时，可以用 CrewAI 的事件总线自建导出器。核心三件套：CrewAIEventsBus（单例事件总线）、事件类（如 CrewKickoffStartedEvent）、BaseEventListener（监听器基类）：

python

import time
from collections import defaultdict

from crewai.events import (
    BaseEventListener,
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)


class MetricsListener(BaseEventListener):
    """把 crew 运行指标推送到任意监控系统（示例打印 + 内存聚合）"""

    def __init__(self):
        super().__init__()
        self._start_ts = {}
        self.stats = defaultdict(float)

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_started(source, event):
            self._start_ts[event.crew_name] = time.time()
            print(f"[metrics] crew '{event.crew_name}' 启动")

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_done(source, event):
            print(f"[metrics] agent '{event.agent.role}' 完成")
            self.stats["agent_completions"] += 1
            # 生产中此处可推送到 Prometheus pushgateway / StatsD / Kafka ...

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_completed(source, event):
            cost = time.time() - self._start_ts.get(event.crew_name, time.time())
            self.stats["crew_wall_seconds"] += cost
            print(f"[metrics] crew 完成，耗时 {cost:.1f}s")
            # 可在此写入 result.token_usage 到时序数据库


# 关键：必须创建实例并保证它被 import（否则被 GC 后监听失效）
metrics_listener = MetricsListener()

注册方式：在你的 crew/flow 定义文件里 import 该模块并实例化监听器（或打包成 listeners/ 包统一引入）。事件是广播式的——同一个总线可以同时挂 Langtrace 导出器和你的 Prometheus 导出器，互不影响。

21.5 生产 dashboard 应该盯什么

指标	来源	异常信号
单次 kickoff 成本	`token_usage.total_tokens` × 单价	周环比突增 → 提示词膨胀或循环
`successful_requests`	usage metrics	远超任务数 → Agent 打转/委派风暴
P95 端到端延迟	事件时间差 / 平台 trace	尾部恶化 → 工具超时重试堆积
任务成功率	回调/事件统计	失败率抬头 → 上游 API 变更
工具错误分布	trace 中 tool span	集中在某工具 → 该依赖不稳定

实践建议：先用 21.1 的裸指标 + 日志跑通"能看见"；量起来后再上 Langtrace/Langfuse 看 trace 明细；最后用 21.4 的事件导出把核心四五个指标沉淀进公司统一的监控告警体系。

本章小结

result.token_usage 提供 successful_requests/prompt_tokens/completion_tokens/total_tokens 四个字段，是零依赖的成本起点；
Langtrace：pip install langtrace-python-sdk + 在导入 CrewAI 前 langtrace.init()；
Langfuse：走 OpenTelemetry，用 OpenLit 一行 openlit.init() 上报，可自托管、厂商无关；
BaseEventListener + crewai_event_bus.on(事件类) 可把任意事件导出自建监控，注意实例必须保持存活；
Dashboard 四板斧：成本、调用次数、P95 延迟、成功率，外加工具错误分布定位依赖问题。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 某 crew 有 3 个 task，但 token_usage 显示 successful_requests=17，最可能的原因是？

2. 按官方文档接入 Langtrace 时，最容易踩的坑是？

3. 官方推荐的"CrewAI + Langfuse"接入路径是？

4. 自定义 BaseEventListener 后，监听器没有收到任何事件，最可能的原因是？

🛠️ 动手实践

给第 23 章将完成的营销 crew 写一个 CostReporter 回调：每次 kickoff 结束打印本次 total_tokens 与按 DeepSeek 单价折算的人民币成本，累计 10 次运行输出平均值。
注册一个本地 Langtrace 账号（或自部署 OpenLit + Langfuse），给任意一个已有 crew 加上追踪，截图找出"哪个 Agent 花的 token 最多"。
基于 21.4 的 MetricsListener，扩展捕获 AgentExecutionCompletedEvent 中的输出长度，把「每个 Agent 的平均产出字数」聚合并随 kickoff 完成一起打印。

观测就绪后，最后一公里是把服务跑起来：第 22 章 · 部署与服务化。

第 21 章 · 可观测性与监控集成 ​

21.1 先看懂 usage_metrics：不接平台也能算成本 ​

21.2 接入 Langtrace：一行初始化的全链路追踪 ​

21.3 接入 Langfuse：OpenTelemetry 标准路线 ​

21.4 事件监听：导出到自建监控 ​

21.5 生产 dashboard 应该盯什么 ​

本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 21 章 · 可观测性与监控集成

21.1 先看懂 `usage_metrics`：不接平台也能算成本

21.2 接入 Langtrace：一行初始化的全链路追踪

21.3 接入 Langfuse：OpenTelemetry 标准路线

21.4 事件监听：导出到自建监控

21.5 生产 dashboard 应该盯什么

本章小结

🛠️ 动手实践