Opik（comet-llm）DSPy 集成指南：OpikCallback 的配置方法、埋点机制与源码级解析

Opik（comet-llm）DSPy 集成指南：OpikCallback 的配置方法、埋点机制与源码级解析

【免费下载链接】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（comet-llm 仓库）Python SDK 中 DSPy 集成文档及其源码实现，讲解如何将 DSPy 应用的运行过程（模块调用、LM 调用、工具调用、token 用量、成本与异常）完整记录到 Opik 平台。读完本文，你将掌握OpikCallback的完整配置方式，理解它在 DSPy 回调钩子中如何构建 Trace/Span 树、如何从 DSPy LM 历史中提取用量与真实 provider，并能参考仓库中的集成测试验证自己的埋点是否正确。

快速开始：三行代码接入 Opik

官方文档（index.rst）给出的接入方式非常简洁：只需创建OpikCallback实例并挂到dspy.settings的 callbacks 上：

import dspy from opik.integrations.dspy.callback import OpikCallback project_name = "DSPY" lm = dspy.LM( model="openai/gpt-4o-mini", ) dspy.configure(lm=lm) opik_callback = OpikCallback(project_name=project_name, log_graph=True) dspy.settings.configure( callbacks=[opik_callback], ) cot = dspy.ChainOfThought("question -> answer") cot(question="What is the meaning of life?")

执行一次cot(...)后，Opik 中会生成一条以顶层模块名（如ChainOfThought）命名的 Trace，其下嵌套Predict（type 为llm）等 Span，Predict之下再嵌套LMSpan，携带 provider、model、usage 等字段。

OpikCallback 参数说明

OpikCallback定义在 callback.py，并通过init.py 导出，文档站则通过 OpikCallback.rst 的 autoclass 指令生成其 API 页。两个构造参数如下：

两点补充（均来自源码）：

• 构造函数中会执行analytics.track_event("integration", "dspy")，即接入行为会被 SDK 的 analytics 记录（见 callback.py#L37）；
• 所有由该回调生成的数据都会附带metadata = {"created_from": "dspy"}，用于在 Opik 侧区分数据来源（见 callback.py#L43）。

埋点机制：DSPy 回调钩子如何构建 Trace/Span 树

OpikCallback继承自dspy.utils.callback.BaseCallback，DSPy 在模块、LM、工具的开始与结束时分别触发on_module_start/on_module_end、on_lm_start/on_lm_end、on_tool_start/on_tool_end五组钩子。

on_module_start 的上下文决策链

顶层模块调用是否要“新建 Trace”还是“挂到已有 Trace/Span 上”，由 on_module_start 中的决策链决定，优先级为：

1. 若全局追踪未激活（tracing_runtime_config.is_tracing_active()为假），直接跳过；
2. 检查回调自身的上下文栈：若已有 Span 数据（self._context_storage.top_span_data()），则将当前模块作为其子 Span；
3. 否则检查回调上下文中是否有 Trace 数据，有则挂为该 Trace 的直接 Span；
4. 再检查 Opik 自身的上下文（opik_context.get_current_span_data()/get_current_trace_data()）——这就是@opik.track装饰器或手动context_storage能“接住”DSPy 数据的原因；
5. 以上都没有时，调用_start_trace新建一条 Trace，Trace 名取模块类名（instance.__class__.__name__），输入即模块的inputs。

Span 的type字段由 get_span_type 判定：dspy.Predict与dspy.LM记为llm，dspy.Tool记为tool，其余（如ChainOfThought这类组合模块）记为general。因此一次典型调用的树形结构是：

Trace: ChainOfThought (general, 顶层模块) └── Span: Predict (llm) └── Span: LM: openai - gpt-4o-mini (llm, provider/model/usage/total_cost)

LM Span 的命名规则在 on_lm_start 中：以instance.model.split("/", 1)拆出 provider 与 model，并将 Span 名重写为f"{span_data.name}: {provider} - {model}"，即LM: openai - gpt-4o-mini。

token 用量、成本与缓存状态的提取

DSPy 把每次 LM 调用的usage、cost等写入 LM 实例的history列表。回调在on_lm_start时记录(lm_instance, expected_messages)，在on_lm_end时由 extract_lm_info_from_history 完成提取，关键设计有三点：

1. 并发竞态保护：取history[-1]前先校验其messages与本次调用预期消息一致，不一致则跳过提取（可能是并发的另一个 LM 调用污染了 history），只输出 debug 日志；
2. 路由器的真实 provider/model 识别：对 OpenRouter 这类 LLM 路由器，response对象（LiteLLMModelResponse）带有provider/model属性（如@preset/qwen实际解析为具体模型）。回调会把 Span 上的 provider 更新为真实服务方，并把原始值存入 metadata：llm_router记录原 provider、original_model记录原 model（见 callback.py#L231-L254），从而保证成本核算落在真实计费方上；
3. cache_hit 推断：优先取response.cache_hit；没有时以“usage 为空”推断为缓存命中。缓存命中时不产生新的 API 调用，对应 Span 的usage为None、metadata["cache_hit"]为True；未命中则为False且带完整 usage。

上述行为均有对应测试：test_dspy__cache_disabled__usage_present_and_cache_hit_false、test_dspy__cache_enabled_and_response_cached__no_usage_and_cache_hit_true等（见 test_dspy.py）。

log_graph：模块结构的 Mermaid 可视化

开启log_graph=True后，每个模块 Span 的 metadata 会额外包含一个_opik_graph_definition键：

{ "created_from": "dspy", "_opik_graph_definition": { "format": "mermaid", "data": "graph TD\n..." # Mermaid 图 } }

图由 build_mermaid_graph_from_module 生成：它递归遍历模块的__dict__收集子模块与lm属性（_get_dspy_module_heirarchy），把每个节点渲染为A(<b>类名</b><br><i>instructions 前100字符</i>)，边为A --> B，工具集合渲染为Toolssubgraph，最后附加 ReAct / Predict / ChainOfThought 等类的颜色样式。生成失败只记录 warning，不影响追踪本身。test_dspy_log_graph/test_dspy_no_log_graph分别验证了开启时 metadata 中存在以graph TD开头的 mermaid 数据、关闭时该键不存在（见 test_dspy.py#L385-L446）。

异常捕获

模块执行抛异常时，on_module_end会收到exception参数，回调通过error_info_collector.collect(exception)将其写入 Span 的error_info（含exception_type等字段）。集成测试test_dspy__openai_llm_is_used__error_occurred_during_openai_call__error_info_is_logged验证了：API key 错误导致调用失败时，PredictSpan 与所有下游LMSpan 都会带error_info（见 test_dspy.py#L95-L147）。

与 Opik 自身追踪体系的嵌套

由于上下文决策链会检查 Opik 上下文，DSPy 集成可以自然嵌套进@opik.track或其他 Opik 埋点中，形成统一的 Trace 树：

import opik from opik.integrations.dspy.callback import OpikCallback @opik.track(project_name="dspy-integration-test", capture_output=True) def f(x): ... opik_callback = OpikCallback(project_name="dspy-integration-test") dspy.settings.configure(callbacks=[opik_callback]) cot = dspy.ChainOfThought("question -> answer") cot(question="What is the meaning of life?") opik_callback.flush() return "the-output"

测试test_dspy_callback__used_inside_another_track_function__data_attached_to_existing_trace_tree断言了此时的树形：Tracef→ Spanf（general）→ SpanChainOfThought→ SpanPredict（llm）→LMSpan。同样，预先通过context_storage手动创建的 Trace 或 Span 也能作为 DSPy 数据的挂载点（对应..._existing_trace_without_span_...与..._existing_span_without_trace_...两个测试，见 test_dspy.py#L150-L375）。在 LM 调用内部，也可以直接通过opik_context.get_current_span_data()/get_current_trace_data()访问当前上下文（test_dspy_callback__opik_context_api_accessible_during_execution验证了这一点）。

数据发送与实战注意事项

1. flush 时机：OpikCallback.flush()会透传到全局 Opik 客户端发送待发送数据（callback.py#L372-L374）。在脚本式（非服务长驻）场景下，建议在流程结束时显式调用，例如opik_callback.flush()（若同时使用@opik.track还需opik.flush_tracker()）；
2. Trace 输入格式：顶层 Trace 的input是 DSPy 对模块调用的原始包装，形如{"args": [], "kwargs": {"question": "..."}}（测试中的断言即为证据），阅读 Opik 页面时注意这一结构；
3. LM Span 数量不固定：DSPy 的 ChatAdapter 在解析失败时会静默回退到 JSONAdapter 重试，因此Predict下的 LM Span 可能是 1 个也可能是 2 个——这是 SDK 测试注释中明确说明的行为，属于预期现象而非 bug；
4. 追踪开关：若全局关闭了 tracing（tracing_runtime_config），回调会整体跳过，不产生任何数据，可用于在测试环境中零成本移除埋点；
5. 验证埋点：建议参考 test_dspy.py 中的断言方式自查——Trace 名是否为顶层模块类名、Predict的 type 是否为llm、LM Span 是否带usage（含prompt_tokens/completion_tokens/total_tokens）、metadata是否含created_from: dspy。

小结

DSPy 集成通过一个轻量回调把 DSPy 的模块层次、LM 调用与工具调用映射为 Opik 的 Trace/Span 树，并额外解决了路由器场景下真实 provider/model 的成本归属、缓存命中标记与模块拓扑的 Mermaid 可视化。核心实现集中在 sdks/python/src/opik/integrations/dspy/ 下的callback.py、parsers.py、graph.py三个文件，行为验证见 sdks/python/tests/library_integration/dspy/ 下的集成测试；需要 API 细节时可查阅由 OpikCallback.rst 生成的文档页。

【免费下载链接】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