深入解读 Opik Python SDK 的 SpanDataSpan 数据模型、字段体系与源码级实现【免费下载链接】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导读SpanData是 Opikcomet-llm 仓库中的 LLM 可观测性平台Python SDK 中表示一次 Span跨度的核心数据类无论是用track装饰器追踪的函数、手动创建的 span还是从服务端读取的公开 Span 数据最终都会落到SpanData这个数据结构上。本文以官方 API 文档 SpanData.rst 为主线结合 SDK 源码逐一拆解其字段体系、类型约束、内置方法与底层实现帮助你彻底掌握在追踪上下文中读写 Span 数据这一核心能力。一、SpanData 是什么一张正在发生的观测快照SpanData是 Opik Python SDK 中定义在 span_data.py 的dataclasses.dataclass数据类其官方 docstring 明确说明The SpanData object is returned when callingopik.opik_context.get_current_span_datafrom a tracked function.也就是说它是对当前正在执行的 Span的内存态描述包含这条 Span 归属的 trace、父 Span、起止时间、输入输出、token 用量、成本、反馈分数等全部观测字段。SDK 在函数入口创建它、在函数执行过程中更新它、在函数结束时把它序列化并上报给 Opik 服务端。从类继承关系看SpanData继承自 ObservationDataTraceData也继承自它因此两者共享大量公共字段与update()、init_end_time()等通用方法。这也是Trace 与 Span 在数据模型上天然对齐的根基。二、字段全解定义、类型与默认值2.1 SpanData 自身声明的字段在 span_data.py 中SpanData直接声明了 7 个字段字段类型默认值含义trace_idstr必填无默认值该 Span 所属 Trace 的 ID是构建父子关系链的锚点idstr自动生成helpers.generate_idSpan 自身的唯一 IDparent_span_idOptional[str]None父 Span 的 ID为None表示根 SpantypeSpanTypegeneralSpan 类型见下文 2.3usageOptional[Union[Dict[str, Any], OpikUsage]]Nonetoken 用量信息见第 5 节modelOptional[str]NoneLLM 模型名称type llm时使用providerOptional[Union[str, LLMProvider]]NoneLLM 提供商影响成本计算total_costOptional[float]NoneSpan 总成本USD优先级高于由 usage 自动计算的成本注意trace_id是唯一没有默认值、必须显式传入的字段——这是数据类设计上刻意为之的约束因为任何 Span 都必须挂靠在一个 Trace 之下。2.2 继承自 ObservationData 的公共字段以下字段在 observation_data.py 中定义SpanData自动继承字段类型默认值含义nameOptional[str]NoneSpan 名称start_timeOptional[datetime]当前本地时间戳开始时间end_timeOptional[datetime]None结束时间函数结束/init_end_time()时填充metadataOptional[Dict[str, Any]]None任意元数据字典update()时会合并而非覆盖inputOptional[Dict[str, Any]]None输入数据outputOptional[Dict[str, Any]]None输出数据tagsOptional[List[str]]None标签列表feedback_scoresOptional[List[FeedbackScoreDict]]None反馈评分列表project_nameOptional[str]None所属项目名error_infoOptional[ErrorInfoDict]None异常信息见 2.5attachmentsOptional[List[Attachment]]None附件列表sourceTraceSourcesdk数据来源sdk/experiment/optimizationenvironmentOptional[str]None运行环境标识ObservationData使用 Python 3.10 的kw_onlyTrue这正是父类定义可选参数、子类可以定义必填参数如trace_id的设计技巧在类 docstring 中有明确说明。2.3 SpanType四种 Span 类型SpanType定义在 types.py是字面量联合类型SpanType Literal[general, tool, llm, guardrail]general通用 Span默认适合包裹任意函数或代码段tool工具调用类 SpanllmLLM 调用类 Span此时应同时提供model与provider以便展示模型名与计算成本guardrail护栏/校验逻辑类 Span。2.4 反馈评分 FeedbackScoreDictfeedback_scores中的每个元素是 types.py 中的TypedDict键必填说明id否评分归属对象的唯一标识trace_id / span_id / thread_idname是评分指标criterion名称value是数值评分floatcategory_name否类别名称用于分类评分reason否评分理由/解释2.5 异常信息 ErrorInfoDicterror_info的结构同样定义在 types.py键说明exception_type异常类名称如ValueErrormessage异常消息可选traceback异常堆栈字符串track装饰器在执行过程中捕获异常后会通过error_info_collector.collect(exception)填充该字段并继续抛出实现失败调用也可观测。三、内置方法从创建子 Span 到分布式追踪3.1create_child_span_data()便捷创建子 Span在 span_data.py 中该方法基于当前 Span 派生一个新的SpanData自动把trace_id透传给子 Span保证整条链路归属同一个 Trace自动把parent_span_id设置为当前 Span 的id从而串起父链继承project_name、source、environmentstart_time未传时自动取当前本地时间戳datetime_helpers.local_timestamp()支持传入name、type、end_time、metadata、input、output、tags、usage、feedback_scores、model、provider、error_info、total_cost、attachments等全部可选参数。这与 TraceData.create_child_span_data 形成对称设计Trace 创建根 SpanSpan 创建子 Span构建出完整的树形追踪结构。3.2as_start_parameters/as_parameters服务端序列化视图as_start_parametersL78-L102返回启动 Span时需要发给服务端的参数子集即id、start_time、project_name、trace_id、source并仅在值非 None 时附带parent_span_id、name、input、metadata、tags、environment。这样在 Span 启动阶段可以只上报已知信息。as_parametersL104-L129返回全量参数字典包含trace_id、id、parent_span_id、name、type、start_time、end_time、metadata、input、output、tags、usage、feedback_scores、project_name、model、provider、error_info、total_cost、attachments、source、environment。Span 结束上报时使用该视图。结合 opik_context.py 中的trace_context上下文管理器可以看到完整生命周期进入上下文时若开启log_start_trace_span则先发送as_start_parameters退出时捕获异常填充error_info、调用init_end_time()后发送as_parameters。3.3get_distributed_trace_headers()分布式追踪头get_distributed_trace_headers 返回DistributedTraceHeadersDict见 types.py仅包含两个键opik_trace_id: str # 当前 Trace ID opik_parent_span_id: str # 当前 Span 的 ID它用于把追踪上下文快递到远端节点远端调用方只要携带这两个头SDK 就能把远端新产生的 Span 挂接到当前链路上。SDK 层面的便捷封装是 get_distributed_trace_headers()后者从context_storage.top_span_data()读取当前 Span 并构造同样的字典若当前没有活跃 Span 则抛出OpikException(There is no span in the context.)。四、如何获取与更新当前 SpanData4.1 从追踪上下文中读取在任意被track装饰的函数内部调用import opik from opik import track track def my_function(x: int) - int: span_data opik.get_current_span_data() # 返回 SpanData 或 None print(span_data.id, span_data.trace_id, span_data.name) return x * 2其实现位于 opik_context.py先从context_storage.top_span_data()取出内部 Span 对象再通过SpanData(**span_data.__dict__)构造一个副本返回——这意味着你在函数内对返回对象的修改不会直接同步到内部追踪对象真正落地更新需要走update_current_span()。4.2 更新当前 Spanupdate_current_span 支持更新name、input、output、metadata、tags、usage、feedback_scores、model、provider、total_cost、attachments、error_info、prompts。其内部流程检查tracing_runtime_config.is_tracing_active()追踪被禁用时直接返回若传入prompts先转换为内部信息字典从context_storage取出当前 SpanData调用update(**new_params)完成字段合并。若当前上下文没有 Span会抛出异常。这也是函数中途记录中间结果的标准姿势import opik opik.track def chat(user_input: str) - str: opik.update_current_span(input{user_input: user_input}) # ... 业务逻辑 opik.update_current_span(output{reply: hello}) return hello4.3update()的合并语义ObservationData.update()observation_data.py是字段更新的统一入口有几个值得注意的行为value is None的键会被跳过不会把已有值置空metadata、input、output走data_helpers的merge 逻辑深度合并而非整体覆盖tags走merge_tags追加合并attachments走_update_attachments追加合并未知字段名只打印 debug 日志后跳过。五、usage、model 与 providerLLM Span 的成本观测链路当 Span 的type llm时usage、model、provider三个字段共同支撑 Opik 的 token 用量展示与成本核算。5.1 OpikUsage多提供商 usage 归一化usage字段的类型是Optional[Union[Dict[str, Any], OpikUsage]]。其中 OpikUsage 是一个 pydantic 模型核心结构class OpikUsage(pydantic.BaseModel): completion_tokens: Optional[int] None prompt_tokens: Optional[int] None total_tokens: Optional[int] None provider_usage: ProviderUsage它通过from_openai_completions_dict、from_google_dict、from_anthropic_dict、from_mistral_dict、from_bedrock_dict、from_openai_responses_dict、from_unknown_usage_dict等类方法opik_usage.py把不同厂商的原始 usage 字典转换为统一格式并尽量补出 OpenAI 风格的prompt_tokens/completion_tokens/total_tokens键供前后端直接消费。5.2 provider 与 LLMProvider 枚举provider字段建议使用 types.py 中的LLMProvider枚举Opik 内置了成本跟踪支持的提供商GOOGLE_VERTEXAI/GOOGLE_AIGemini 系列OPENAIANTHROPIC/ANTHROPIC_VERTEXAIGROQBEDROCKAWS BedrockMISTRALAI从源码注释看total_cost字段的优先级高于 SDK 依据 usage 自动计算出的成本若你的提供商不在枚举中仍可传任意字符串但将无法获得自动成本计算LLMProvider.has_value()可用于校验。六、服务端数据的回读span_public_to_span_data除了写入侧SpanData也承担读取侧的职责。转换函数 span_public_to_span_data 把 REST API 返回的SpanPublic对象转换为SpanData逐字段映射id、trace_id、parent_span_id、name、type、start_time、end_time、metadata、input、output、tags、usage、model、provider、error_info通过feedback_scores_public_to_feedback_scores_dict把公开评分对象转换为FeedbackScoreDict由于SpanPublic只携带project_idproject_name需由调用方显式传入源码 TODO 注释明确指出了这一点。这意味着内存中创建的 SpanData与从服务端读回的 SpanData在字段模型上是同构的读写两侧可以共用同一套字段理解。七、实战示例手动构建 Span 层级与分布式链路7.1 手动创建根 Span 与子 Span结合TraceData.create_child_span_data与SpanData.create_child_span_data可以完全手动搭建追踪结构import opik trace_data opik.TraceData(nameroot_trace, project_namemy-project) root_span trace_data.create_child_span_data(nameroot_span, typegeneral) child_span root_span.create_child_span_data( namellm_call, typellm, modelgpt-4o-mini, provideropenai, input{prompt: Hello}, ) print(child_span.trace_id root_span.trace_id) # True同属一个 Trace print(child_span.parent_span_id root_span.id) # True父子关系已挂接7.2 跨节点分布式追踪在服务 A 的追踪函数中取头传给远端服务 BB 侧收到后即可把新 Span 挂入同一链路# 服务 A追踪函数内部 headers opik.get_distributed_trace_headers() # headers {opik_trace_id: ..., opik_parent_span_id: ...} # 通过 HTTP 头 / 消息队列把 headers 传给服务 B # 服务 B入口函数携带 opik_distributed_trace_headers 参数 opik.track def remote_handler(opik_distributed_trace_headersNone): ... # 产生的 Span 自动挂接到服务 A 的链路上SDK 侧解析该参数的位置在 arguments_helpers.pyextract_distributed_trace_headers它会从 kwargs 中弹出opik_distributed_trace_headers用于上下文还原。八、源码中的实际创建场景SpanData在 SDK 内部被大量实例化了解这些调用点有助于理解其生命周期track装饰器create_span_data 在函数启动时构造SpanData用StartSpanParameters填充初始字段并默认sourcesdk上下文管理器分布式头上下文管理器也会基于传入参数构造SpanData见 distributed_headers_context_manager.py框架集成LangChain traceropik_tracer.py、DSPy callbackcallback.py都在回调中创建SpanData以桥接框架的 Run/模块生命周期上下文读取get_current_span_data 通过SpanData(**span_data.__dict__)返回副本。九、总结与参考路径SpanData是 Opik Python SDK 追踪体系的中枢数据结构它以 dataclass 形式统一了启动参数、全量参数、分布式头、子 Span 派生四种视图配合ObservationData的合并式update()让开发者既能细粒度观测 LLM 调用usage/model/provider/成本又能轻松构建多级 Span 树与跨节点分布式链路。继续深入阅读可参考以下文件官方 API 文档SpanData.rst核心实现span_data.py、observation_data.py上下文 APIopik_context.py类型定义types.pyusage 归一化opik_usage.py服务端回读转换converters.py【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考