LiteLLM Levo AI 集成实战：基于 OpenTelemetry 的一行配置实现 LLM 可观测性上报

LiteLLM 的 Levo AI 集成文档 介绍了如何在 LiteLLM 中以最小成本将 LLM 调用链路上报至 Levo AI：无需改动任何业务代码，只需在配置中声明回调 callbacks: ["levo"] 并设置四个环境变量，LiteLLM 便会借助 OpenTelemetry OTLP 协议把 Trace 自动发往 Levo 的 Collector。本文以该集成文档为骨架，结合 levo.py 与 test_levo.py 等仓库源码，逐层拆解其配置读取、鉴权 Header 构造、回调注册链路与异常处理，帮助你从“会配”进阶到“懂原理”，可直接在自有 LiteLLM 网关中复现这一观测方案。

集成概览：为什么选择 OTLP 通道对接 Levo

Levo AI 提供 LLM 可观测性分析能力，需要把网关侧产生的请求/响应追踪数据持续送入其 Collector。LiteLLM 复用其成熟的 OpenTelemetry 能力栈，通过 OTLP 协议把数据推送到 Levo 的采集端点，并在传输层携带 Levo 专属的鉴权与路由 Header。

从实现来看，整个集成只是对 LiteLLM 通用 OpenTelemetry 回调的一次“定制化配置”，具备以下特点：

• 自动 OTLP 导出：调用产生的 Trace 由 OpenTelemetry SDK 自动导出，无需手写任何埋点；
• Levo 专属 Header：自动注入 x-levo-organization-id 与 x-levo-workspace-id，实现多组织/多工作区路由；
• 极简配置：只需在 litellm_settings.callbacks 中声明 "levo" 一个字符串；
• 环境变量驱动：鉴权凭据与端点全部通过环境变量注入，避免把密钥写死在配置里。

这种“字符串回调 + 环境变量”的模式，与仓库内 custom_logger_registry.py 中 logfire、arize、langfuse_otel 等一批 OTEL 兼容回调的接入方式保持一致。

快速开始：四步完成 Levo 上报接入

第一步：安装 OpenTelemetry 依赖

Levo 集成依赖 Python OpenTelemetry 组件。文档给出的安装方式使用 uv：

uv add opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-grpc

其中：

• opentelemetry-api / opentelemetry-sdk：提供 Trace API、TracerProvider 与 Span 处理链路；
• opentelemetry-exporter-otlp-proto-http：提供 OTLPSpanExporter 的 HTTP/Protobuf 实现（Levo 集成默认即走 otlp_http 通道）；
• opentelemetry-exporter-otlp-proto-grpc：gRPC 通道选项，按需保留。

第二步：在 LiteLLM 配置中声明回调

编辑你的 LiteLLM 配置文件（例如 config.yaml）：

litellm_settings:
  callbacks: ["levo"]

litellm_settings.callbacks 是 LiteLLM 声明式回调入口。配置中以字符串 "levo" 引用，由核心模块负责解析成真正的回调实例（解析细节见下文“回调注册链路”）。

第三步：设置环境变量

export LEVOAI_API_KEY="<your-levo-api-key>"
export LEVOAI_ORG_ID="<your-levo-org-id>"
export LEVOAI_WORKSPACE_ID="<your-workspace-id>"
export LEVOAI_COLLECTOR_URL="<your-levo-collector-url>"

前三个变量用于鉴权与租户路由，LEVOAI_COLLECTOR_URL 是 Levo 支持侧下发的采集端点完整地址。

第四步：启动 LiteLLM

litellm --config config.yaml

启动后，所有经过该网关的 LLM 请求都会自动产生并发送 Trace 到 Levo——这就是该集成交付的核心价值：零业务改动获得可观测性。

配置说明：环境变量语义与默认值

必填环境变量

变量	说明
LEVOAI_API_KEY	用于 OTLP 请求鉴权的 Levo API Key，会拼装为 Authorization: Bearer {key}
LEVOAI_ORG_ID	Levo 组织 ID，用于数据路由，对应 x-levo-organization-id Header
LEVOAI_WORKSPACE_ID	Levo 工作区 ID，用于数据路由，对应 x-levo-workspace-id Header
LEVOAI_COLLECTOR_URL	Levo 下发的 Collector 端点完整 URL（含协议与端口）

可选环境变量

变量	说明	默认值
LEVOAI_ENV_NAME	用于给 Trace 打环境标签的环境名	None

需要说明的细节：原集成文档将 LEVOAI_ENV_NAME 列为可选变量，但当前仓库中 levo.py 的 get_levo_config() 仅读取并校验上述四个必填变量，尚未实际读取 LEVOAI_ENV_NAME；测试代码在清理环境变量时已为其预留位置（见 test_levo.py），可以推断该变量是为将来给 Trace 打环境标签预留的接口。

关于 Collector URL 的处理语义

集成文档明确强调：LEVOAI_COLLECTOR_URL 会原样使用，不做任何路径拼接/改写。这一点在源码层得到了印证——levo.py 直接把 collector_url 赋给 endpoint，配置阶段零加工。参数化测试也专门断言了自定义 URL 与本地 HTTP 端点均被“原样透传”（见 test_levo.py）。

需要补充一个底层细节：该“原样使用”约束作用于 LevoLogger 配置层。真正构造 OTLP 导出器时，基类 OpenTelemetry 会调用 _normalize_otel_endpoint() 将端点规整到信号路径（如把缺省路径的端点补上 /v1/traces），若你提供的 Collector URL 已包含标准信号路径（如 /v1/traces），则保持不变（见 opentelemetry.py）。因此在向 Levo 支持获取地址时，建议直接使用其下发的完整端点。

运行原理：从环境变量到 OTLP Trace 的完整链路

集成文档给出了四步工作原理，结合源码可以进一步还原完整链路：

1. LevoLogger 继承 LiteLLM 的 OpenTelemetry 基类。LevoLogger(OpenTelemetry) 定义在 levo.py，因此天然具备基类全套的 Span 生命周期管理能力。
2. 配置由 get_levo_config() 从环境变量读取。这是整个集成的“大脑”，见 levo.py。
3. OTLP Header 自动拼接。源码按如下格式构造：

Authorization=Bearer {LEVOAI_API_KEY},x-levo-organization-id={LEVOAI_ORG_ID},x-levo-workspace-id={LEVOAI_WORKSPACE_ID}

三个键值对以逗号分隔拼成一个字符串（见 levo.py），最终会被基类的 _get_headers_dictionary() 按“先按逗号切分、再按第一个 = 切分”的方式解析成字典交给 OTLP 导出器（见 opentelemetry.py）。 4. 协议固定为 otlp_http，Trace 以 OTLP/HTTP 格式发往 Collector 端点。协议在 levo.py 硬编码为 "otlp_http"，测试同样断言了该值（见 test_levo.py）。

完整数据流可以概括为：

环境变量 ──► get_levo_config() 校验并组装
                 │
                 ▼
      LevoConfig(otlp_auth_headers, protocol="otlp_http", endpoint)
                 │
                 ▼
   OpenTelemetryConfig(headers/endpoint/exporter)
                 │
                 ▼
    OTLP HTTP SpanExporter ──► Levo Collector

源码结构与关键类解析

集成模块位于 litellm/integrations/levo/，文件组织如下：

litellm/integrations/levo/
├── __init__.py          # 导出 LevoLogger
├── levo.py              # LevoLogger / LevoConfig 实现
└── README.md            # 本集成文档

LevoLogger

集成入口，继承 OpenTelemetry，职责是“把 Levo 环境变量翻译成标准的 OpenTelemetry 配置”：

• 静态方法 get_levo_config() 负责读取并校验环境变量、构造 OTLP Header、返回 LevoConfig（levo.py）；
• 异步方法 async_health_check() 用于健康检查：配置齐全返回 {"status": "healthy", ...}，缺配置返回 {"status": "unhealthy", ...}（levo.py），供网关侧就绪探测使用。

LevoConfig

一个轻量配置载体（Pydantic 元数据注释保留在 README，实际实现为普通类，见 levo.py），承载三个字段：

字段	含义
otlp_auth_headers	逗号分隔的 OTLP 鉴权/路由 Header 字符串
protocol	导出协议，恒为 otlp_http
endpoint	Collector 端点（原样透传的环境变量值）

注意：仓库同时提供了一个基于 pydantic 模型的 OTEL V2 适配层 levo_preset()，其内部同样调用 LevoLogger.get_levo_config() 并构造 ExporterSpec(kind="otlp_http", owner=LEVO)，说明新旧两套 OTEL 适配层共用同一套环境变量逻辑（见 otel/presets/levo.py 与 otel/presets/init.py）。

错误处理与健康检查

get_levo_config() 采用“缺什么报什么”的策略：四个必填环境变量缺失时分别抛出信息明确的 ValueError，便于快速定位：

缺失变量	报错内容（摘自源码）
LEVOAI_API_KEY	LEVOAI_API_KEY environment variable is required for Levo integration.
LEVOAI_ORG_ID	LEVOAI_ORG_ID environment variable is required for Levo integration.
LEVOAI_WORKSPACE_ID	LEVOAI_WORKSPACE_ID environment variable is required for Levo integration.
LEVOAI_COLLECTOR_URL	提示必须设置并联系 Levo support 获取 Collector URL

校验顺序与错误信息见 levo.py，每条信息都已在单测中通过 pytest.raises(ValueError, match=...) 精确断言（test_levo.py）。

async_health_check() 则把配置异常“吞”进结果字典而非直接抛错：正常时返回 healthy 与提示信息，异常时捕获 ValueError 并返回 unhealthy 与错误详情（levo.py）。集成测试验证了在 OpenTelemetry 依赖可用且注入 in-memory exporter 的前提下，健康检查能正确返回 healthy（test_levo.py）。

回调注册链路：一个字符串如何变成真正的 Logger

配置中的 "levo" 字符串，经由 LiteLLM 回调体系的三处关键机制完成“落地”：

1. 回调字面量白名单：litellm/__init__.py 中 _custom_logger_compatible_callbacks_literal 把 "levo" 登记为合法回调名（见 litellm/init.py），否则配置阶段即会被判定为非法回调。

2. 回调类注册表：custom_logger_registry.py 的 CALLBACK_CLASS_STR_TO_CLASS_TYPE 将 "levo" 与 OTEL 兼容回调类（OpenTelemetry，即 LevoLogger 的基类）建立映射——这是“回调名 → 类类型”的解析索引。

3. 回调实例化分支：核心日志模块 litellm_logging.py 为 "levo" 提供专门构造逻辑：

   • 先尝试复用已构造的 V2 适配实例（若 LITELLM_OTEL_V2 开启，见 litellm_logging.py）；
   • 否则调用 LevoLogger.get_levo_config() 生成配置，包成 OpenTelemetryConfig(exporter, endpoint, headers)；
   • 通过“实例类型 + callback_name == "levo"”双重判断去重，避免重复构造；
   • 最终 new LevoLogger(config=otel_config, callback_name="levo") 并加入内存 logger 列表。

值得注意的是去重逻辑：遍历内存中已有 logger，仅当“是 LevoLogger 实例”且“callback_name == 'levo'”时才复用，保证单例语义。集成测试对 callback_name 的赋值也做了断言（test_levo.py）。

测试：如何验证集成行为

仓库为 Levo 集成提供了完整的单测与集成测试，位于 tests/test_litellm/integrations/levo/：

• 配置单测（TestLevoConfig）：通过 unittest.mock.patch.dict 注入/清空环境变量，覆盖四种必填变量齐全、自定义 Collector URL、本地 HTTP 端点、逐变量缺失抛错、Header 逗号分隔格式等场景（test_levo.py）；
• 回调注册/集成测试（TestLevoIntegration）：在 OpenTelemetry 依赖可用时构造真实 LevoLogger，验证健康检查结果与 callback_name 语义（test_levo.py）；
• 参数化测试：用 @pytest.mark.parametrize 覆盖多组端点/Header 组合与逐一缺失必填变量的情况，是理解配置语义最快的“可执行文档”（test_levo.py）。

运行测试前需确保本机已安装 opentelemetry 相关包，否则依赖型用例会被 @pytest.mark.skipif(not OPENTELEMETRY_AVAILABLE, ...) 自动跳过。

使用注意事项与边界

基于源码核对，接入时请留意以下几点：

1. 四个必填环境变量缺一不可，否则网关启动/回调初始化阶段即抛 ValueError，错误信息会明确点名缺失变量；
2. LEVOAI_COLLECTOR_URL 请使用 Levo 下发的完整地址：配置层虽原样透传，但底层 OTLP 导出器仍会对缺失的信号路径做 /v1/traces 规整，地址最规范的做法是直接采用支持团队提供的形式；
3. Levo 集成采用静态凭据模型：与 arize、newrelic 等支持每请求动态 Header 的集成不同，Levo 不在 DYNAMIC_HEADERS_BY_CALLBACK 之列，走 logger 的默认 tracer（见 otel/presets/init.py），即鉴权信息全局共享；
4. LEVOAI_ENV_NAME 目前仅是文档化变量：当前 levo.py 尚未消费它，若需要环境标签请关注后续版本或在网关侧通过 Span 属性自行注入。

小结

LiteLLM 的 Levo 集成展示了一种“配置即集成”的可观测性接入范式：通过在 litellm_settings.callbacks 声明一个字符串、设置四个环境变量，就能把整个网关的 LLM 调用追踪以 OTLP/HTTP 协议持续送往 Levo Collector。其背后是 LevoLogger 对 OpenTelemetry 基类的轻量定制——环境变量读取、校验、鉴权 Header 拼装、协议与端点固定，全部收敛在 get_levo_config() 一个静态方法内，而回调的实例化、去重与健康检查则由 LiteLLM 核心回调体系与配套测试共同保障。理解这条链路后，你不仅能顺利接入 Levo，也能举一反三地看懂 arize、langfuse_otel 等同族 OTEL 集成的工作原理。