telemetry 遥测:框架怎么"匿名地了解自己被怎么用"
阶段8 收官。前几天讲的都是"你调 LLM 花了多少",今天讲另一面——CrewAI 框架自己收集匿名使用数据:你建了几个 Agent、用了什么 process、跑了多久。这就是 telemetry/ 模块。我们会看:Telemetry 为什么是单例、三个环境变量怎么一键关掉它、它怎么基于 OpenTelemetry 初始化、_safe_telemetry_operation 这个"永不抛错"的包装怎么保证遥测挂了也绝不拖累你跑 crew、以及最重要的——它到底收集什么、不收集什么(隐私边界)。这是学习一个成熟开源框架"可观测性 + 隐私自律"的好样本。
痛点:开源框架怎么知道该往哪改?
share_crew=True);②三个环境变量任一即可一键关闭;③所有遥测操作都裹在 _safe_telemetry_operation 里——出任何错都吞掉,绝不影响你的 crew 运行。可观测、可关闭、永不添乱。模块文件头的自我声明就是它的"隐私承诺"(telemetry/telemetry.py:1):
# telemetry/telemetry.py:1
"""Telemetry module for CrewAI.
No prompts, task descriptions, agent backstories/goals, responses, or sensitive
data is collected. Users can opt-in to share more complete data using the
`share_crew` attribute."""
share_crew 开关。默认是最保守的。Telemetry:一个进程只有一个
它是单例——整个进程共享一个实例(telemetry/telemetry.py:90 与 :103):
# telemetry/telemetry.py:103
def __new__(cls) -> Self:
if cls._instance is None:
with cls._lock: # ★加锁防并发重复创建
if cls._instance is None: # 双重检查
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self) -> None: # :111
if hasattr(self, "_initialized") and self._initialized:
return # ★已初始化过就直接返回,不重跑
self.ready: bool = False
self.trace_set: bool = False
self._initialized: bool = True
if self._is_telemetry_disabled():
return # 被关掉 → 到此为止,什么都不建
try:
... # 初始化 OTel(L04)
self.ready = True
except Exception as e:
...
self.ready = False # ★初始化失败 → 标记未就绪,不抛错
__new__ + _lock 双重检查★经典线程安全单例:加锁 + 两次判 None,保证多线程下也只创建一个实例。_initialized 卫兵单例的 __init__ 每次 Telemetry() 都会被调,用这个标记确保真正的初始化只跑一次。_is_telemetry_disabled → return★第一道闸:被环境变量关掉了,直接返回,连 OTel 都不初始化,零开销。try...except → ready=False★初始化 OTel 万一失败(网络、依赖问题),不抛错,只把 ready 置 False。后续操作看到 not ready 就跳过。一键关闭:三个环境变量任选其一
关闭遥测的总开关(telemetry/telemetry.py:148):
# telemetry/telemetry.py:148
@classmethod
def _is_telemetry_disabled(cls) -> bool:
"""Check if telemetry should be disabled based on environment variables."""
return (
os.getenv("OTEL_SDK_DISABLED", "false").lower() == "true"
or os.getenv("CREWAI_DISABLE_TELEMETRY", "false").lower() == "true"
or os.getenv("CREWAI_DISABLE_TRACKING", "false").lower() == "true"
)
def _should_execute_telemetry(self) -> bool: # :156
"""Check if telemetry operations should be executed."""
return self.ready and not self._is_telemetry_disabled()
OTEL_SDK_DISABLEDOpenTelemetry 官方标准的关闭变量——尊重生态约定,用惯 OTel 的人一眼就懂。CREWAI_DISABLE_TELEMETRYCrewAI 自己的语义化开关,名字直白。CREWAI_DISABLE_TRACKING又一个别名——"tracking"是很多人第一反应会搜的词,多设一个提高"想关就能关到"的概率。任一为 "true" 即关三个 or 连接:只要有一个被设成 true 就关闭。宁可多几个入口,也要让用户轻松关掉。_should_execute_telemetry★每次遥测操作前都问它:ready(初始化成功) 且 没被禁用,才真执行。运行时也能实时响应关闭。export CREWAI_DISABLE_TELEMETRY=true(或 OTEL_SDK_DISABLED=true / CREWAI_DISABLE_TRACKING=true),再跑你的脚本。此后 _is_telemetry_disabled() 返回 True,__init__ 里直接 return 不初始化 OTel,所有 span 也不会上报。一行环境变量,完全静默。OTel 初始化:搭一条安全的上报管道
__init__ 里真正搭建 OpenTelemetry 的部分(telemetry/telemetry.py:126):
# telemetry/telemetry.py:126(__init__ 内的 try 块)
self.resource = Resource(attributes={SERVICE_NAME: CREWAI_TELEMETRY_SERVICE_NAME})
with suppress_warnings():
self.provider = TracerProvider(resource=self.resource)
processor = BatchSpanProcessor( # ★批量上报,不是每条都发
SafeOTLPSpanExporter( # ★用"安全版"导出器
endpoint=f"{CREWAI_TELEMETRY_BASE_URL}/v1/traces",
timeout=30,
)
)
self.provider.add_span_processor(processor)
self._register_shutdown_handlers() # 注册退出时 flush 的钩子
self.ready = True
那个"安全版导出器"是关键(telemetry/telemetry.py:67):
# telemetry/telemetry.py:67
class SafeOTLPSpanExporter(OTLPSpanExporter):
def export(self, spans: Any) -> SpanExportResult:
try:
return super().export(spans)
except Exception as e:
logger.error(e)
return SpanExportResult.FAILURE # ★上报失败也只返回 FAILURE,绝不抛出
BatchSpanProcessor★批量处理:把 span 攒一批再发,而不是每产生一条就发一次网络请求。省资源、不阻塞主流程。SafeOTLPSpanExporter★继承官方导出器,唯一改动是把 export 包进 try/except——网络挂了、服务器不通,只记日志返回 FAILURE,不让异常冒出来。timeout=30上报请求 30 秒超时——防止遥测服务器卡住时把你的进程也拖住。_register_shutdown_handlers注册进程退出/信号钩子,退出前把攒着的 span flush 出去(用短超时,不拖慢关闭)。_safe_telemetry_operation:永不抛错的安全气囊
所有遥测动作都必须穿过的"安全气囊"(telemetry/telemetry.py:250):
# telemetry/telemetry.py:250
def _safe_telemetry_operation(self, operation: Callable[[], Span | None]) -> Span | None:
"""Execute telemetry operation safely, checking both readiness and env vars."""
if not self._should_execute_telemetry():
return None # ① 未就绪/被禁用 → 直接不干
try:
return operation() # ② 真执行遥测动作
except Exception as e:
logger.debug(f"Telemetry operation failed: {e}")
return None # ③ ★出任何错 → 吞掉,返回 None
接收一个 operation 函数★高阶函数设计:把"具体要做的遥测动作"作为参数传进来,安全气囊只负责"安全地执行它"。职责分离。先 _should_execute_telemetry没就绪或被关 → 连试都不试,直接返回 None。遥测关闭时几乎零成本。try: operation()真跑遥测动作(建 span、加属性、上报)。except → 吞掉返回 None★核心承诺:无论遥测出什么错,都不往外抛。你的 crew 该跑跑该停停,绝不因为"上报统计"这种旁枝末节而崩。crew_creation / task_started / tool_usage / flow_execution_span...)。朴素做法:每个方法里各写一遍 try...except...return None——几十份重复,且难保证每处都写对(漏一个就可能让遥测崩到主流程)。源码做法:把"安全执行"抽成 _safe_telemetry_operation 这一个入口,每个记录方法把自己的核心逻辑定义成内部函数 _operation(),再交给它执行(你在 L06 会看到这个模式)。"永不抛错"这条铁律只在一处实现、一处保证——想改安全策略(比如加重试、改日志级别)也只改这里。这是"用高阶函数收敛横切关注点",比到处 try/except 优雅且可靠得多。一个 span 怎么记:以 crew_creation 为例
看一个真实的记录方法怎么用上面那套(telemetry/telemetry.py:269):
# telemetry/telemetry.py:269
def crew_creation(self, crew: Crew, inputs: dict | None) -> None:
"""Records the creation of a crew."""
def _operation() -> None: # ★把核心逻辑定义成内部函数
tracer = trace.get_tracer("crewai.telemetry")
span = tracer.start_span("Crew Created")
self._add_attribute(span, "crewai_version", version("crewai"))
self._add_attribute(span, "python_version", platform.python_version())
add_crew_attributes(span, crew, self._add_attribute)
self._add_attribute(span, "crew_process", crew.process) # 用了什么流程
self._add_attribute(span, "crew_memory", crew.memory) # 开没开记忆
self._add_attribute(span, "crew_number_of_tasks", len(crew.tasks)) # 几个任务
self._add_attribute(span, "crew_number_of_agents", len(crew.agents)) # 几个 Agent
if crew.share_crew: # ★只有用户主动开启才记详细内容
self._add_attribute(span, "crew_agents", json.dumps([
{"role": agent.role, "goal": agent.goal, "backstory": agent.backstory, ...}
for agent in crew.agents]))
...
self._safe_telemetry_operation(_operation) # ★交给安全气囊执行
def _operation(): ...★L05 说的模式:把"这次要记什么"写成一个内部函数,最后一行整个交给 _safe_telemetry_operation。这样它天然享受"永不抛错"。start_span("Crew Created")开一个 span(一段可观测的"事件"),后面往上面挂属性。process / memory / 数量★默认只记这些结构统计:什么流程、开没开记忆、几个任务几个 Agent——都是无隐私的计数/枚举。if crew.share_crew:★隐私分水岭:只有你主动把 share_crew=True,才会记 role/goal/backstory 这些可能含内容的字段。默认绝不记。_add_attribute连"往 span 加一个属性"这种小动作也走安全包装(:925),value 为 None 直接跳过。层层防御。隐私边界:默认收什么、开了 share_crew 才收什么
把隐私边界讲透——默认与 opt-in 两档收集内容对比:
| 字段 | 默认(匿名) | share_crew=True(主动分享) |
|---|---|---|
| crewai 版本 / python 版本 / 平台 | ✅ 收集 | ✅ 收集 |
| process 类型 / 是否开 memory | ✅ 收集 | ✅ 收集 |
| Agent 数量 / Task 数量 | ✅ 收集 | ✅ 收集 |
| Agent 的 role / goal / backstory | ❌ 不收 | ✅ 收集 |
| Task 描述 / 期望输出 | ❌ 不收 | ✅ 收集 |
| 你的 prompt / 模型回答 / 工具输入输出 | ❌ 永不收 | ❌ 永不收 |
结构 vs 内容默认只收"结构统计"(数量、类型、开关状态)——这些帮维护者了解用法,又不暴露你在做什么业务。share_crew 才收元信息role/goal/backstory/task 描述这类"配置内容",只有你主动 opt-in 才上报——用于更深入的产品分析。prompt/回答永不收★最敏感的运行时数据(你实际问了什么、模型答了什么)无论如何都不收集——这是硬底线。👶 小白:那 D53 讲的 token 数会被上报吗?
👨🏫 老师:token 数量属于"结构统计"(多少而非内容),可能进遥测的统计范畴,但它不暴露你的对话内容——知道"这次用了 3000 token"跟知道"你问了什么"是两码事。如果你连这类匿名统计都不想上报,直接 CREWAI_DISABLE_TELEMETRY=true 一键全关即可。选择权始终在你手里。
CREWAI_DISABLE_TELEMETRY=true,并在文档里记一笔,别依赖"应该没人注意"。另外注意:遥测上报走 telemetry.crewai.com:4319(telemetry/constants.py),内网防火墙若不放行,SafeOTLPSpanExporter 会安静地 FAILURE——不影响你跑,但也说明"没关但也没发出去",别误以为关了。要真关,就用环境变量明确关。阶段8 收官 + 今日小结
BaseLLM 接口 + LLM 工厂路由(一个类装下所有模型);D50 各家 provider 适配器(把统一接口翻译成方言);D51 函数调用与结构化输出(让模型吐可编程的结构);D52 上下文窗口管理(对话太长自动压缩);D53 token/成本追踪(算清每次调用花多少);D54 遥测(框架匿名了解自己、可关、永不添乱)。一条主线:把"调用大模型"这件充满差异和不确定性的事,收敛成上层可以放心依赖的统一、稳定、可观测的一层。🧠 今天你应该能回答
- 遥测收集什么、绝不收集什么?
share_crew改变了什么? - Telemetry 为什么做成线程安全单例?
- 哪三个环境变量能关掉遥测?为什么设这么多别名?
_safe_telemetry_operation怎么保证遥测永不拖垮主流程?为什么集中在一处?SafeOTLPSpanExporter相比官方导出器改了什么?- 企业环境该怎么正确关闭遥测?为什么不能只靠防火墙?
✋ 10 分钟动手
P=lib/crewai/src/crewai
sed -n '90,165p' $P/telemetry/telemetry.py # 单例 + 关闭开关
sed -n '250,268p' $P/telemetry/telemetry.py # _safe_telemetry_operation
sed -n '269,360p' $P/telemetry/telemetry.py # crew_creation 记录方法
cat $P/telemetry/constants.py # 上报地址/服务名
# 验证关闭生效:设变量后跑,不会有对外上报
export CREWAI_DISABLE_TELEMETRY=true
python -c "
from crewai.telemetry.telemetry import Telemetry
t=Telemetry(); print('ready =', t.ready) # 关闭时应为 False
"
project/ 目录,讲 @CrewBase 装饰器与 YAML 配置:怎么用 agents.yaml / tasks.yaml 把配置和代码分离、@agent / @task / @crew 装饰器怎么把方法自动注册成 crew 的组件。