Opik Python SDK 的 OpenAI 集成：用 track_openai 一行代码追踪 Chat Completions、流式与结构化输出-编程实验室

Opik Python SDK 的 OpenAI 集成：用 track_openai 一行代码追踪 Chat Completions、流式与结构化输出

【免费下载链接】comet-llmDebug, evaluate, and monitor your LLM applications, RAG systems, and agentic workflows with comprehensive tracing, automated evaluations, and production-ready dashboards.项目地址: https://gitcode.com/GitHub_Trending/co/comet-llm

本文围绕 Opik Python SDK 的 OpenAI 集成文档展开，讲解如何用track_openai装饰函数包装openai.OpenAI/openai.AsyncOpenAI客户端，将每一次模型调用自动记录为 Opik 平台上的 trace/span。读完本文，你将掌握集成的完整用法（参数、支持的方法范围、provider 推断规则），并理解底层补丁机制——包括流式响应的聚合逻辑与版本兼容分支——以便在排查“为什么我的流式调用 token 用量没记上”这类问题时能快速定位原因。

一、基本用法：包装客户端即开始记录

官方文档 OpenAI 集成页给出的核心用法只有四行代码：用track_openai包住 OpenAI 客户端，之后所有经过该客户端的调用都会被记录：

from opik.integrations.openai import track_openai from openai import OpenAI openai_client = OpenAI() openai_client = track_openai(openai_client) response = openai_client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello, world!"}], )

两点需要注意：

track_openai不是装饰器，而是客户端包装函数：它接收一个openai.OpenAI或openai.AsyncOpenAI实例，返回同一个被原地打补丁（patched）的实例（类型定义见 opik_tracker.py）。
补丁在调用track_openai时就已生效，但每次被包装的调用执行时会先检查opik.is_tracing_active()——追踪被关闭时调用照常执行，只是不发 span/trace（该行为写在 track_openai 的 docstring 中）。这意味着你可以放心在启动阶段无条件包装客户端，通过开关控制是否上报。

二、track_openai 参数详解

track_openai的完整签名（见 opik_tracker.py#L24-L28）：

def track_openai( openai_client: OpenAIClient, project_name: Optional[str] = None, provider: Optional[Union[str, LLMProvider]] = None, ) -> OpenAIClient:

参数	默认值	说明
`openai_client`	必填	`OpenAI`或`AsyncOpenAI`实例
`project_name`	`None`	数据上报到的 Opik 项目名称；不传则沿用当前`opik.track`/上下文所在项目
`provider`	`None`	记录在每条 LLM span 上的模型供应商标识；接受任意字符串，或 opik.types.LLMProvider 枚举中的已识别供应商

provider 的推断与覆盖

OpenAI SDK 常被当作访问其他 OpenAI 兼容 API（Together、OpenRouter、vLLM、DeepSeek 等）的通用客户端，因此provider参数用来标注真实的模型供应商而不是 base URL 主机名：

不传provider时：由 _get_provider 从客户端的base_url推断——host 为api.openai.com记为"openai"，否则直接使用 host 字符串。
传字符串时：原样记录，例如track_openai(client, provider="vllm")。
传LLMProvider枚举时：取.value归一化为纯字符串，避免枚举成员本身泄漏到日志中。

LLMProvider枚举中与成本追踪相关的取值包括openai、anthropic、google_vertexai、google_ai、groq、bedrock、anthropic_vertexai（见 types.py#L20-L32）。使用已识别的 provider 名可以让 Opik 按对应供应商的价格体系计算 token 成本。

幂等性保护

if hasattr(openai_client, "opik_tracked"): return openai_client openai_client.opik_tracked = True

track_openai 的实现用opik_tracked属性标记已包装的客户端，重复调用会直接返回原对象，不会双重包装。但要注意一个实现细节：functools.wraps会把__wrapped__等属性复制到被装饰函数上，因此 audio 补丁处专门调整了打补丁顺序（with_streaming_response.create必须先于speech.create被包装），否则幂等检查会误判跳过——见源码注释。

三、哪些 OpenAI 调用会被追踪

track_openai按客户端实际具备的属性按需打补丁（hasattr判断，见 opik_tracker.py#L82-L91），覆盖范围如下（同样列在 docstring 中）：

方法	span 名称	备注
`chat.completions.create()`	`chat_completion_create`	含`stream=True`流式模式
`beta.chat.completions.parse()`	`chat_completion_parse`	结构化输出
`beta.chat.completions.stream()`	`chat_completion_stream`	仅 OpenAI SDK ≥ 1.92.0 时单独包装
`chat.completions.parse()`	`chat_completion_parse`	同上版本分支
`responses.create()`/`responses.parse()`	`responses_create`/`responses_parse`	Responses API
`videos.create()`/`remix()`/`poll()`/`list()`/`delete()`/`create_and_poll()`/`download_content()`	`videos.*`	`retrieve`有意不包装，避免轮询期间产生过多 span
`audio.speech.create()`	`audio.speech.create`	TTS
`audio.speech.with_streaming_response.create()`	同名 span	流式 TTS

所有补丁后的 span 统一带有created_from: "openai"、type: "openai_chat"（chat 类）或openai_videos（视频类）的 metadata 与openai标签，方便在 Opik 中按来源过滤。

为什么 beta 分支与 OpenAI SDK 版本相关

_patch_openai_chat_completions 中有明确的版本分支：

SDK < 1.92.0：beta.chat.completions.stream()底层调用chat.completions.create(stream=True)，装饰create就自动覆盖了 stream，因此只需单独包装beta.chat.completions.parse。
SDK ≥ 1.92.0：OpenAI 重构了 beta API——chat.completions.stream仍走create无需重复装饰，但beta.chat.completions.stream不再经过create，必须显式包装；同时parse同时出现在chat.completions与beta.chat.completions两个路径下，两处都要装饰。

版本判断使用SemanticVersion.parse(openai.__version__) < "1.92.0"完成，所以升级 OpenAI SDK 大版本时追踪行为会自动切换，无需改代码。

四、span 记录了什么：input、output 与 token 用量

chat 类调用的 span 内容由 OpenaiChatCompletionsTrackDecorator 负责组装：

开始时（_start_span_inputs_preprocessor）：

input：仅取 kwargs 中的messages和function_call键（KWARGS_KEYS_TO_LOG_AS_INPUTS）；
metadata：其余 kwargs 全部归入 metadata，并合并created_from/type标识；
model、provider、tags=["openai"]；
若stream=True，span 名自动改写为chat_completion_stream，并过滤掉 OpenAI SDK 内部的NOT_GIVEN/Omit哨兵值（_remove_not_given_sentinel_values），避免无意义的占位参数进入日志。

结束时（_end_span_inputs_preprocessor）：

output：响应体中的choices；其余字段进 metadata；
usage：当响应含usage时，用 OpenAI 格式的用量解析器构建 Opik 用量对象——注意这里的 "openai" 指的是用量 payload 格式，与 span 上的 provider（可能已被provider=参数覆盖为别的供应商）是两回事，源码注释特别说明了这一点；
model：取自响应体的model字段（实际模型名，而非请求时的别名）。

流式响应：分块聚合再记录

流式调用的难点在于 span 的 output 和 usage 只有在流读完之后才完整。实现分两层：

流补丁层（stream_patchers.py）：替换openai.Stream.__iter__/openai.AsyncStream.__aiter__（以及ChatCompletionStreamManager的__enter__/__aenter__），在迭代过程中累积所有 chunk，捕获中途异常作为error_info，并在流耗尽或退出时结束对应 span/trace。四种流形态（同步/异步 × 裸流/流管理器）各有对应补丁函数，分派逻辑见 _streams_handler。
聚合层（chat_completion_chunks_aggregator.py）：aggregate()把一组ChatCompletionChunk还原为一个类ChatCompletion的 Pydantic 对象——拼接delta.content得到完整文本、保留首个 chunk 的id/model/created、取最后一个finish_reason和最后一个非空usage（第 59-60 行）。聚合失败时只记录错误日志并返回None，不会打断业务调用。

一个直接推论：流式 span 的 output 和 token 用量要等生成器被完全消费（或提前退出）后才更新；若你拿到流但从未迭代完，span 可能停留在中间状态。这也是官方示例中建议开启stream_options={"include_usage": True}的原因——否则流式响应里根本没有usage字段可聚合。

五、完整可运行的参考示例

仓库自带示例 openai_integration_example.py 覆盖了三种典型调用形态与四条 trace 的划分方式，可直接复制改造：

from openai import OpenAI from opik import flush_tracker, track from opik.integrations.openai import opik_tracker from pydantic import BaseModel client = OpenAI() client = opik_tracker.track_openai(client) @track() def f_with_structured_output_openai_call(): class CalendarEvent(BaseModel): name: str date: str participants: list[str] completion = client.beta.chat.completions.parse( model="gpt-4o-2024-08-06", messages=[ {"role": "system", "content": "Extract the event information."}, {"role": "user", "content": "Alice and Bob are going to a science fair on Friday."}, ], response_format=CalendarEvent, ) print(completion) @track() def f_with_streamed_openai_call(): stream = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Tell a fact"}], max_tokens=10, stream=True, stream_options={"include_usage": True}, ) for item in stream: print(item) f_with_streamed_openai_call() # trace 1 f_with_structured_output_openai_call() # trace 2（嵌套 span 挂在 @track 之下） flush_tracker()

示例展示了两个关键实践：

与@track组合：OpenAI 调用发生在@track()装饰的函数内部时，会作为嵌套 span挂到当前 trace 下（示例注释明确写道 "will create one more nested span"）；脱离@track上下文直接调用则各自成 trace（示例第 76-83 行的裸调用即为独立 trace 4）。
进程退出前调用flush_tracker()：确保缓冲中的 span/trace 全部发往 Opik 后端。

六、适用前提与限制小结

需要openai包已安装且版本语义可解析；beta 相关追踪路径的行为随 OpenAI SDK 1.92.0 分界（见第三节的版本分支说明）。
track_openai对responses、videos、audio等命名空间采用hasattr探测，老版本 OpenAI SDK 上这些补丁会自动跳过，只追踪 chat completions 部分。
该集成会无条件上报一次analytics.track_event("integration", "openai")匿名使用事件（opik_tracker.py#L67 附近，实际位于补丁前），如介意可在配置层面关闭 Opik 的 analytics。
视频retrieve方法有意不追踪、流式 span 需完整消费生成器后才落盘 output/usage——这两点都是源码中明确的取舍，排查数据“缺失”时应优先核对此处。

相关源码与文档入口：集成文档、track_openai API 页（该页通过autofunction直接从 opik_tracker.py 的 docstring 生成，二者内容始终一致）、集成示例。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考