Opik Python SDK 中 OpikTracer 的完整用法与源码解析为 LangChain / LangGraph 应用构建结构化追踪【免费下载链接】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 文档的OpikTracerAPI 参考页apps/opik-documentation/python-sdk-docs/source/integrations/langchain/OpikTracer.rst展开完整讲解opik.integrations.langchain.OpikTracer的构造参数、公开方法与底层运行机制读完你可以直接把 LangChain / LangGraph 应用接入 Opik 平台拿到带层级 Span、Token 用量、成本与错误信息的 Trace并理解 Run 事件到 Trace/Span 的映射原理。一、OpikTracer 是什么定位与快速上手OpikTracer是 Opik Python SDK 提供的 LangChain 集成入口定义在 opik_tracer.py。它是一个 LangChain 的BaseTracer实现继承自langchain_core.tracers.BaseTracer作为回调callback传入 LangChain 的callbacks参数后LangChain 运行时的每一次 LLM 调用、Chain 执行、Tool 调用都会以回调事件的形式通知到 TracerTracer 再把这些事件映射为 Opik 的 Trace 与 Span 上报到 Opik 服务端。官方文档页 OpikTracer.rst 使用 Sphinx 的autoclass指令自动从源码 docstring 生成 API 参考因此下面所有参数语义均直接取自 opik_tracer.py 中的构造函数文档。同目录下的 index.rst 给出了最小可用示例from langchain.chains import LLMChain from langchain_openai import OpenAI from langchain.prompts import PromptTemplate from opik.integrations.langchain import OpikTracer # Initialize the tracer opik_tracer OpikTracer() # Create the LLM Chain using LangChain llm OpenAI(temperature0) prompt_template PromptTemplate( input_variables[input], templateTranslate the following text to French: {input} ) llm_chain LLMChain(llmllm, promptprompt_template) # Generate the translations translation llm_chain.run(Hello, how are you?, callbacks[opik_tracer]) print(translation)仓库中还提供了一个基于 LCEL 的可运行示例 langchain_integration_example.py展示了带tags与metadata的 tracer 创建、chain.invoke(input..., config{callbacks: [callback]})的调用方式以及显式调用callback.flush()保证数据落库from langchain_community.llms import fake from langchain.prompts import PromptTemplate from opik.integrations.langchain.opik_tracer import OpikTracer llm fake.FakeListLLM(responses[Im sorry, I dont think Im talented enough to write a synopsis]) prompt_template PromptTemplate( input_variables[title], templateGiven the title of play, write a synopsys for that. Title: {title}. ) synopsis_chain prompt_template | llm callback OpikTracer(tags[tag1, tag2], metadata{a: b}) result synopsis_chain.invoke(input{title: Documentary about Bigfoot in Paris}, config{callbacks: [callback]}) callback.flush()二、构造函数参数详解OpikTracer.__init__的完整签名与参数语义见 opik_tracer.py#L96-L108如下参数类型默认值作用tagsOptional[List[str]]None附加到所有记录的 Trace 上的标签列表metadataOptional[Dict[str, Any]]None附加到 Trace 上的元数据字典Tracer 会自动写入created_from: langchain标记graphOptional[Graph]NoneLangGraph 的 Graph 对象用于在 Opik UI 中可视化图结构通常为graph.get_graph(xrayTrue)project_nameOptional[str]None该 Tracer 产生的 Trace 所属的 Opik 项目名distributed_headersOptional[DistributedTraceHeadersDict]None分布式追踪上下文头opik_trace_id/opik_parent_span_id用于跨进程串联 Tracethread_idOptional[str]None会话线程唯一标识将 Trace 关联到同一对话线程若未显式传入会从 Run 的 metadata 中自动探测thread_idskip_error_callbackOptional[Callable[[str], bool]]None接收错误字符串的回调返回True表示该错误应被跳过视为预期错误而非故障opik_context_read_only_modeboolFalse是否以只读模式运行False时 Tracer 会向 Opik 上下文栈压入 Span使 LangChain 内部再调用opik.track装饰的函数时自动挂到父 SpanTrue时不修改上下文栈仅从 LangChain 的 Run 对象创建 Span/TraceproviderOptional[Union[str, LLMProvider, Callable]]None记录在 LLM Span 上的 provider供后端计算成本。既可以是固定字符串或opik.LLMProvider单一 provider 场景也可以是回调函数多 provider 混用场景详见下文**kwargsAny—透传给父类BaseTracer的其余参数几个值得注意的实现细节均来自 opik_tracer.py参数校验构造函数通过parameters_validator对thread_id、project_name字符串、metadata字典、tags列表做类型校验非法值会在构造期直接抛出验证错误而不是在运行中静默失败。自动标记来源构造时立即执行self._trace_default_metadata[created_from] langchain使 Opik 中一眼可识别该 Trace 来自 LangChain 集成。graph 延迟注入若传入了graph构造期会立即调用set_graph(graph)见下文方法一节。provider 参数成本计算的关键provider的设计目标是解决调用经由 OpenAI 兼容代理如 LiteLLM 网关时provider 会被自动识别为代理主机名、导致无法计算成本的问题。源码中定义了两个类型opik_tracer.py#L52-L72ProviderOverride Union[str, LLMProvider] # 固定 provider ProviderResolver Callable[[ProviderResolverContext], Optional[ProviderOverride]] # 按 run 动态解析ProviderResolverContext是一个NamedTuple包含model从 run 解析出的模型名通常作为路由键和run原始 LangChain run 字典作为兜底路由手段。解析逻辑见_resolve_provideropik_tracer.py#L664-L691# 回调形式按模型名路由 provider def resolve(ctx): if claude in (ctx.model or ): return anthropic return openai opik_tracer OpikTracer(providerresolve)实现上做了两处稳健性处理回调抛出异常时仅记录 warning 并回退到自动探测的 provider用户回调永远不会破坏追踪上报LLMProvider枚举会被归一化为.value字符串避免LLMProvider.OPENAI这类值泄漏到 Span 中。三、公开方法set_graph、flush、created_tracesautoclass :members:指令会渲染的公开成员方法有1.set_graph(graph: Graph) - Noneopik_tracer.py#L192-L205提取 LangGraph 图结构并存入 Trace 元数据使 Opik UI 能可视化该图。实现上调用graph.draw_mermaid()并以固定结构写入self._trace_default_metadata[_opik_graph_definition] { format: mermaid, data: graph.draw_mermaid(), }也就是说图定义是以 Mermaid 文本形式嵌在 Trace 的metadata._opik_graph_definition里上报的。2.flush() - Noneopik_tracer.py#L748-L752将数据强制发送flush到 Opik 服务端。由于 Opik 客户端采用批量/后台线程上报在脚本结束时调用flush()可确保数据完整落库官方示例正是这样使用的。3.created_traces() - List[trace.Trace]opik_tracer.py#L754-L761返回该 Tracer 已创建的 Trace 对象列表方便在测试或脚本断言中直接检查上报结果。4.get_current_span_data_for_run(run_id: UUID) - Optional[span.SpanData]opik_tracer.py#L763-L764按 LangChain 的 run id 查询对应的 Opik Span 数据是 LangGraph 异步场景下配合 extract_current_langgraph_span_data 做上下文桥接的内部支撑接口。四、从 Run 到 Trace/Span核心映射原理OpikTracer的所有回调入口都遵循同一套骨架先经_skip_tracking()判断全局追踪开关再进入_process_start_span/_process_end_span/_process_end_span_with_error。_skip_tracking基于tracing_runtime_config.is_tracing_active()opik_tracer.py#L766-L767意味着运行时可以通过 Opik 的运行时配置整体关闭追踪而不影响业务逻辑。4.1 回调事件与 Span 类型映射从源码结构看Tracer 覆盖了 LangChain 三大类事件opik_tracer.py#L769-L890LLM_on_llm_start/_on_llm_end/_on_llm_errorChat 模型on_chat_model_start_on_chat_model_start。这里有个专门的 workaround——LangChain 核心默认对 tracer 关闭on_chat_model_start事件因此 Tracer 自行构造了一个Runrun_typellminputs 为messages的model_dump()序列化结果再走_start_trace保证 Chat 模型消息被完整记录Chain / Tool_on_chain_start/_on_chain_end/_on_chain_error与_on_tool_start/_on_tool_end/_on_tool_error。Span 类型由 run_parse_helpers.py 中的get_span_type决定if run.get(run_type) in [llm, tool]: return cast(SpanType, run.get(run_type)) if run.get(run_type) in [prompt]: return cast(SpanType, tool) # prompt run 映射为 tool 类型 return cast(SpanType, general)即 LangChain 的llm/toolrun 原样映射promptrun 归入tool其余一律为general。4.2 根 Run 的特殊处理为什么 LangGraph 里不会多出一个根 SpanLangChain 的回调机制保证_persist_run只在每个 run 树的根上调用一次OpikTracer利用这一点做两件事1根 run 创建 Trace且刻意跳过根 Span。_create_root_trace_and_spanopik_tracer.py#L407-L443在创建新 Trace 时不创建对应的根 Span并在RunStateStore中将该 run 标记为 skipped LangGraph root其子 run 随后由_attach_span_to_local_or_distributed_traceopik_tracer.py#L505-L580直接挂到 Trace 下parent_span_idNone。这样可以避免Trace 与同名根 Span 内容完全重复的冗余。对纯 LLM/Tool 这种以根 run 为叶子的工作负载LLM/Tool 事件传入allow_duplicating_root_spanTrue会保留根 Span 本身。2Trace 终态只在 Tracer 拥有 该 Trace 时提交。_persist_runopik_tracer.py#L207-L257中只有span_data is Nonetrace-only 的根或owns_trace(trace_id)成立时才走_finalize_trace调用trace_data.init_end_time().update(output..., error_info...)后经__internal_api__trace__上报并从上下文栈弹出 Trace。若根 run 运行在外部 Trace 之下比如外层有opik.track函数或分布式头则该 Tracer 只向其贡献 Span终态交由真正拥有 Trace 的一方提交——这是避免重复 finalize 的关键边界。4.3 错误处理LangGraph 控制流不算错误LangGraph 的GraphInterrupt人工介入中断与ParentCommand子图路由到父图的 supervisor 模式在 LangChain 回调层表现为 error但语义上是正常控制流。_persist_run与_process_end_span_with_error都优先解析这两种情况opik_tracer.py#L213-L235GraphInterrupt由 parse_graph_interrupt_value 用正则 括号/引号配对扫描从 traceback 中提取Interrupt(value...)的值支持嵌套结构与转义字符串解码写入output的__interrupt__键并打上_langgraph_interrupt: True元数据不设置 error_infoParentCommand由 is_langgraph_parent_command 识别匹配langgraph.errors.ParentCommand或以ParentCommand(开头仅打上_langgraph_parent_command: True元数据真实错误包装为ErrorInfoDict(exception_typeException, tracebackerror_str)上报被跳过的错误skip_error_callback返回True时output 被替换为占位字典{warning: Error output skipped by skip_error_callback.}常量ERROR_SKIPPED_OUTPUTS用户也可事后用opik_context.get_current_span_data().update(output...)手动补写真实输出。4.4 Token 用量与成本提取Span 结束时的_process_end_spanopik_tracer.py#L582-L643依次完成用量提取provider_usage_extractors.try_extract_provider_usage_data(run_dict)按 provider 选择具体抽取器——目录 provider_usage_extractors/ 下分别实现了 OpenAI、Anthropic含 VertexAI 变体、Bedrock、Google Generative AI、Groq、VertexAI 等抽取器provider 覆盖调用_resolve_provider应用用户传入的provider参数见第二节成本提取response_cost_extractors.try_extract_response_cost(run_dict)目录 response_cost_extractors/ 中目前实现了 Litellm 响应的成本抽取更新与上报span_data.init_end_time().update(output..., usage..., provider..., model..., total_cost...)然后在追踪激活时经__internal_api__span__上报。此外还有两处针对.astream()的 workaround当 Span input 是{input: }/{input: {}}这类占位值时用 run 的真实inputs回填当 input 中携带 LangGraphCommand的 resume 值时extract_resume_value_from_command会改写为{__resume__: ...}以便 UI 展示。节点返回Command对象时extract_command_update 会把output从{output: Command(...)}解包为实际的 state update 字典。4.5 上下文栈管理与只读模式非只读模式下每个 Span 创建都会add_span_data压入 Opik 上下文栈结束时经_release_ended_span_stateopik_tracer.py#L645-L662弹出根 run 结束时还会release_run_tree释放整棵 run 树的状态。_ensure_no_hanging_opik_tracer_spansopik_tracer.py#L309-L321则负责清理悬挂 Span如果链调用前不存在外部 Span直接清空 Span 栈否则裁剪到记录的外部父 Span。这正是opik_context_read_only_mode的存在意义在并发环境下若上下文隔离不可靠例如某些事件循环/线程复用场景可开启只读模式让 Tracer 完全不碰上下文栈——代价是 LangChain 内部再嵌套调用opik.track装饰函数时不会自动挂到 LangChain 的父 Span。4.6 分布式追踪与运行时开关distributed_headers传入opik_trace_id/opik_parent_span_id后根 run 不再新建 Trace而是把 Span 直接挂到远程 Trace 之下见 opik_tracer.py#L534-L547 的SpanData(trace_id..., parent_span_id...)构造用于多服务间的 Trace 串联起始事件上报_emit_start_trace/_emit_start_span仅在客户端配置log_start_trace_span开启且追踪激活时才发出 start 参数用于实时观察进行中的 Spanopik_tracer.py#L445-L457。五、LangGraph 场景的配套工具LangGraph 的一个已知噪音问题是一个 agent run 会产生大量框架内部 runRunnableSequence 包装器、条件边路由、channel 写入、__start__/__end__标记等。OpikTracer通过 is_internal_langgraph_run 将其识别为内部 plumbingLLM/Tool run 永远有意义带langgraph_*元数据的 chain run 中只有 run 名等于langgraph_node元数据的节点边界算有意义其余包括__开头的图出入口标记被打上metadata._opik.is_internal TrueUI 可据此默认隐藏。针对 LangGraph 的另外两个同包工具见init.py 的导出清单track_langgraph(graph, opik_tracer)langgraph_tracer_injector.py把 tracer 注入编译后图的默认 config一次注入、后续所有graph.invoke(...)自动追踪无需每次传config{callbacks: [opik_tracer]}同时自动执行graph.get_graph(xrayTrue)set_graph完成图结构可视化。文档页见 track_langgraph.rstextract_current_langgraph_span_data文档页 extract_current_langgraph_span_data.rst解决ainvoke()异步调用中opik.track函数无法感知当前 Span 的上下文桥接问题。六、验证与测试参考该集成的行为在仓库中有对应的集成测试覆盖可作为阅读源码后的验证入口test_opik_tracer.pyLangChain 库集成测试含端到端上报校验以及 Google ADK 场景下的 test_opik_tracer.py。七、小结OpikTracer用约 900 行代码opik_tracer.py 辅助模块run_parse_helpers.py、provider_usage_extractors/、run_state.py完成了 LangChain 事件体系到 Opik Trace/Span 模型的完整映射参数层面通过project_name/tags/metadata/thread_id控制归属与检索skip_error_callback/provider/distributed_headers解决错误噪声、代理网关成本、跨服务串联三类生产痛点机制层面以 根 run 建 Trace 不建 Span、只由 Trace 拥有者 finalize、LangGraph 控制流不记为错误 三条核心规则保证上报结构干净。配合track_langgraph的一次性注入与flush()的落库保障它构成了在 Opik 平台上调试与监控 LangChain / LangGraph 应用的完整链路。【免费下载链接】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），仅供参考
Opik Python SDK 中 OpikTracer 的完整用法与源码解析：为 LangChain / LangGraph 应用构建结构化追踪
Opik Python SDK 中 OpikTracer 的完整用法与源码解析为 LangChain / LangGraph 应用构建结构化追踪【免费下载链接】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 文档的OpikTracerAPI 参考页apps/opik-documentation/python-sdk-docs/source/integrations/langchain/OpikTracer.rst展开完整讲解opik.integrations.langchain.OpikTracer的构造参数、公开方法与底层运行机制读完你可以直接把 LangChain / LangGraph 应用接入 Opik 平台拿到带层级 Span、Token 用量、成本与错误信息的 Trace并理解 Run 事件到 Trace/Span 的映射原理。一、OpikTracer 是什么定位与快速上手OpikTracer是 Opik Python SDK 提供的 LangChain 集成入口定义在 opik_tracer.py。它是一个 LangChain 的BaseTracer实现继承自langchain_core.tracers.BaseTracer作为回调callback传入 LangChain 的callbacks参数后LangChain 运行时的每一次 LLM 调用、Chain 执行、Tool 调用都会以回调事件的形式通知到 TracerTracer 再把这些事件映射为 Opik 的 Trace 与 Span 上报到 Opik 服务端。官方文档页 OpikTracer.rst 使用 Sphinx 的autoclass指令自动从源码 docstring 生成 API 参考因此下面所有参数语义均直接取自 opik_tracer.py 中的构造函数文档。同目录下的 index.rst 给出了最小可用示例from langchain.chains import LLMChain from langchain_openai import OpenAI from langchain.prompts import PromptTemplate from opik.integrations.langchain import OpikTracer # Initialize the tracer opik_tracer OpikTracer() # Create the LLM Chain using LangChain llm OpenAI(temperature0) prompt_template PromptTemplate( input_variables[input], templateTranslate the following text to French: {input} ) llm_chain LLMChain(llmllm, promptprompt_template) # Generate the translations translation llm_chain.run(Hello, how are you?, callbacks[opik_tracer]) print(translation)仓库中还提供了一个基于 LCEL 的可运行示例 langchain_integration_example.py展示了带tags与metadata的 tracer 创建、chain.invoke(input..., config{callbacks: [callback]})的调用方式以及显式调用callback.flush()保证数据落库from langchain_community.llms import fake from langchain.prompts import PromptTemplate from opik.integrations.langchain.opik_tracer import OpikTracer llm fake.FakeListLLM(responses[Im sorry, I dont think Im talented enough to write a synopsis]) prompt_template PromptTemplate( input_variables[title], templateGiven the title of play, write a synopsys for that. Title: {title}. ) synopsis_chain prompt_template | llm callback OpikTracer(tags[tag1, tag2], metadata{a: b}) result synopsis_chain.invoke(input{title: Documentary about Bigfoot in Paris}, config{callbacks: [callback]}) callback.flush()二、构造函数参数详解OpikTracer.__init__的完整签名与参数语义见 opik_tracer.py#L96-L108如下参数类型默认值作用tagsOptional[List[str]]None附加到所有记录的 Trace 上的标签列表metadataOptional[Dict[str, Any]]None附加到 Trace 上的元数据字典Tracer 会自动写入created_from: langchain标记graphOptional[Graph]NoneLangGraph 的 Graph 对象用于在 Opik UI 中可视化图结构通常为graph.get_graph(xrayTrue)project_nameOptional[str]None该 Tracer 产生的 Trace 所属的 Opik 项目名distributed_headersOptional[DistributedTraceHeadersDict]None分布式追踪上下文头opik_trace_id/opik_parent_span_id用于跨进程串联 Tracethread_idOptional[str]None会话线程唯一标识将 Trace 关联到同一对话线程若未显式传入会从 Run 的 metadata 中自动探测thread_idskip_error_callbackOptional[Callable[[str], bool]]None接收错误字符串的回调返回True表示该错误应被跳过视为预期错误而非故障opik_context_read_only_modeboolFalse是否以只读模式运行False时 Tracer 会向 Opik 上下文栈压入 Span使 LangChain 内部再调用opik.track装饰的函数时自动挂到父 SpanTrue时不修改上下文栈仅从 LangChain 的 Run 对象创建 Span/TraceproviderOptional[Union[str, LLMProvider, Callable]]None记录在 LLM Span 上的 provider供后端计算成本。既可以是固定字符串或opik.LLMProvider单一 provider 场景也可以是回调函数多 provider 混用场景详见下文**kwargsAny—透传给父类BaseTracer的其余参数几个值得注意的实现细节均来自 opik_tracer.py参数校验构造函数通过parameters_validator对thread_id、project_name字符串、metadata字典、tags列表做类型校验非法值会在构造期直接抛出验证错误而不是在运行中静默失败。自动标记来源构造时立即执行self._trace_default_metadata[created_from] langchain使 Opik 中一眼可识别该 Trace 来自 LangChain 集成。graph 延迟注入若传入了graph构造期会立即调用set_graph(graph)见下文方法一节。provider 参数成本计算的关键provider的设计目标是解决调用经由 OpenAI 兼容代理如 LiteLLM 网关时provider 会被自动识别为代理主机名、导致无法计算成本的问题。源码中定义了两个类型opik_tracer.py#L52-L72ProviderOverride Union[str, LLMProvider] # 固定 provider ProviderResolver Callable[[ProviderResolverContext], Optional[ProviderOverride]] # 按 run 动态解析ProviderResolverContext是一个NamedTuple包含model从 run 解析出的模型名通常作为路由键和run原始 LangChain run 字典作为兜底路由手段。解析逻辑见_resolve_provideropik_tracer.py#L664-L691# 回调形式按模型名路由 provider def resolve(ctx): if claude in (ctx.model or ): return anthropic return openai opik_tracer OpikTracer(providerresolve)实现上做了两处稳健性处理回调抛出异常时仅记录 warning 并回退到自动探测的 provider用户回调永远不会破坏追踪上报LLMProvider枚举会被归一化为.value字符串避免LLMProvider.OPENAI这类值泄漏到 Span 中。三、公开方法set_graph、flush、created_tracesautoclass :members:指令会渲染的公开成员方法有1.set_graph(graph: Graph) - Noneopik_tracer.py#L192-L205提取 LangGraph 图结构并存入 Trace 元数据使 Opik UI 能可视化该图。实现上调用graph.draw_mermaid()并以固定结构写入self._trace_default_metadata[_opik_graph_definition] { format: mermaid, data: graph.draw_mermaid(), }也就是说图定义是以 Mermaid 文本形式嵌在 Trace 的metadata._opik_graph_definition里上报的。2.flush() - Noneopik_tracer.py#L748-L752将数据强制发送flush到 Opik 服务端。由于 Opik 客户端采用批量/后台线程上报在脚本结束时调用flush()可确保数据完整落库官方示例正是这样使用的。3.created_traces() - List[trace.Trace]opik_tracer.py#L754-L761返回该 Tracer 已创建的 Trace 对象列表方便在测试或脚本断言中直接检查上报结果。4.get_current_span_data_for_run(run_id: UUID) - Optional[span.SpanData]opik_tracer.py#L763-L764按 LangChain 的 run id 查询对应的 Opik Span 数据是 LangGraph 异步场景下配合 extract_current_langgraph_span_data 做上下文桥接的内部支撑接口。四、从 Run 到 Trace/Span核心映射原理OpikTracer的所有回调入口都遵循同一套骨架先经_skip_tracking()判断全局追踪开关再进入_process_start_span/_process_end_span/_process_end_span_with_error。_skip_tracking基于tracing_runtime_config.is_tracing_active()opik_tracer.py#L766-L767意味着运行时可以通过 Opik 的运行时配置整体关闭追踪而不影响业务逻辑。4.1 回调事件与 Span 类型映射从源码结构看Tracer 覆盖了 LangChain 三大类事件opik_tracer.py#L769-L890LLM_on_llm_start/_on_llm_end/_on_llm_errorChat 模型on_chat_model_start_on_chat_model_start。这里有个专门的 workaround——LangChain 核心默认对 tracer 关闭on_chat_model_start事件因此 Tracer 自行构造了一个Runrun_typellminputs 为messages的model_dump()序列化结果再走_start_trace保证 Chat 模型消息被完整记录Chain / Tool_on_chain_start/_on_chain_end/_on_chain_error与_on_tool_start/_on_tool_end/_on_tool_error。Span 类型由 run_parse_helpers.py 中的get_span_type决定if run.get(run_type) in [llm, tool]: return cast(SpanType, run.get(run_type)) if run.get(run_type) in [prompt]: return cast(SpanType, tool) # prompt run 映射为 tool 类型 return cast(SpanType, general)即 LangChain 的llm/toolrun 原样映射promptrun 归入tool其余一律为general。4.2 根 Run 的特殊处理为什么 LangGraph 里不会多出一个根 SpanLangChain 的回调机制保证_persist_run只在每个 run 树的根上调用一次OpikTracer利用这一点做两件事1根 run 创建 Trace且刻意跳过根 Span。_create_root_trace_and_spanopik_tracer.py#L407-L443在创建新 Trace 时不创建对应的根 Span并在RunStateStore中将该 run 标记为 skipped LangGraph root其子 run 随后由_attach_span_to_local_or_distributed_traceopik_tracer.py#L505-L580直接挂到 Trace 下parent_span_idNone。这样可以避免Trace 与同名根 Span 内容完全重复的冗余。对纯 LLM/Tool 这种以根 run 为叶子的工作负载LLM/Tool 事件传入allow_duplicating_root_spanTrue会保留根 Span 本身。2Trace 终态只在 Tracer 拥有 该 Trace 时提交。_persist_runopik_tracer.py#L207-L257中只有span_data is Nonetrace-only 的根或owns_trace(trace_id)成立时才走_finalize_trace调用trace_data.init_end_time().update(output..., error_info...)后经__internal_api__trace__上报并从上下文栈弹出 Trace。若根 run 运行在外部 Trace 之下比如外层有opik.track函数或分布式头则该 Tracer 只向其贡献 Span终态交由真正拥有 Trace 的一方提交——这是避免重复 finalize 的关键边界。4.3 错误处理LangGraph 控制流不算错误LangGraph 的GraphInterrupt人工介入中断与ParentCommand子图路由到父图的 supervisor 模式在 LangChain 回调层表现为 error但语义上是正常控制流。_persist_run与_process_end_span_with_error都优先解析这两种情况opik_tracer.py#L213-L235GraphInterrupt由 parse_graph_interrupt_value 用正则 括号/引号配对扫描从 traceback 中提取Interrupt(value...)的值支持嵌套结构与转义字符串解码写入output的__interrupt__键并打上_langgraph_interrupt: True元数据不设置 error_infoParentCommand由 is_langgraph_parent_command 识别匹配langgraph.errors.ParentCommand或以ParentCommand(开头仅打上_langgraph_parent_command: True元数据真实错误包装为ErrorInfoDict(exception_typeException, tracebackerror_str)上报被跳过的错误skip_error_callback返回True时output 被替换为占位字典{warning: Error output skipped by skip_error_callback.}常量ERROR_SKIPPED_OUTPUTS用户也可事后用opik_context.get_current_span_data().update(output...)手动补写真实输出。4.4 Token 用量与成本提取Span 结束时的_process_end_spanopik_tracer.py#L582-L643依次完成用量提取provider_usage_extractors.try_extract_provider_usage_data(run_dict)按 provider 选择具体抽取器——目录 provider_usage_extractors/ 下分别实现了 OpenAI、Anthropic含 VertexAI 变体、Bedrock、Google Generative AI、Groq、VertexAI 等抽取器provider 覆盖调用_resolve_provider应用用户传入的provider参数见第二节成本提取response_cost_extractors.try_extract_response_cost(run_dict)目录 response_cost_extractors/ 中目前实现了 Litellm 响应的成本抽取更新与上报span_data.init_end_time().update(output..., usage..., provider..., model..., total_cost...)然后在追踪激活时经__internal_api__span__上报。此外还有两处针对.astream()的 workaround当 Span input 是{input: }/{input: {}}这类占位值时用 run 的真实inputs回填当 input 中携带 LangGraphCommand的 resume 值时extract_resume_value_from_command会改写为{__resume__: ...}以便 UI 展示。节点返回Command对象时extract_command_update 会把output从{output: Command(...)}解包为实际的 state update 字典。4.5 上下文栈管理与只读模式非只读模式下每个 Span 创建都会add_span_data压入 Opik 上下文栈结束时经_release_ended_span_stateopik_tracer.py#L645-L662弹出根 run 结束时还会release_run_tree释放整棵 run 树的状态。_ensure_no_hanging_opik_tracer_spansopik_tracer.py#L309-L321则负责清理悬挂 Span如果链调用前不存在外部 Span直接清空 Span 栈否则裁剪到记录的外部父 Span。这正是opik_context_read_only_mode的存在意义在并发环境下若上下文隔离不可靠例如某些事件循环/线程复用场景可开启只读模式让 Tracer 完全不碰上下文栈——代价是 LangChain 内部再嵌套调用opik.track装饰函数时不会自动挂到 LangChain 的父 Span。4.6 分布式追踪与运行时开关distributed_headers传入opik_trace_id/opik_parent_span_id后根 run 不再新建 Trace而是把 Span 直接挂到远程 Trace 之下见 opik_tracer.py#L534-L547 的SpanData(trace_id..., parent_span_id...)构造用于多服务间的 Trace 串联起始事件上报_emit_start_trace/_emit_start_span仅在客户端配置log_start_trace_span开启且追踪激活时才发出 start 参数用于实时观察进行中的 Spanopik_tracer.py#L445-L457。五、LangGraph 场景的配套工具LangGraph 的一个已知噪音问题是一个 agent run 会产生大量框架内部 runRunnableSequence 包装器、条件边路由、channel 写入、__start__/__end__标记等。OpikTracer通过 is_internal_langgraph_run 将其识别为内部 plumbingLLM/Tool run 永远有意义带langgraph_*元数据的 chain run 中只有 run 名等于langgraph_node元数据的节点边界算有意义其余包括__开头的图出入口标记被打上metadata._opik.is_internal TrueUI 可据此默认隐藏。针对 LangGraph 的另外两个同包工具见init.py 的导出清单track_langgraph(graph, opik_tracer)langgraph_tracer_injector.py把 tracer 注入编译后图的默认 config一次注入、后续所有graph.invoke(...)自动追踪无需每次传config{callbacks: [opik_tracer]}同时自动执行graph.get_graph(xrayTrue)set_graph完成图结构可视化。文档页见 track_langgraph.rstextract_current_langgraph_span_data文档页 extract_current_langgraph_span_data.rst解决ainvoke()异步调用中opik.track函数无法感知当前 Span 的上下文桥接问题。六、验证与测试参考该集成的行为在仓库中有对应的集成测试覆盖可作为阅读源码后的验证入口test_opik_tracer.pyLangChain 库集成测试含端到端上报校验以及 Google ADK 场景下的 test_opik_tracer.py。七、小结OpikTracer用约 900 行代码opik_tracer.py 辅助模块run_parse_helpers.py、provider_usage_extractors/、run_state.py完成了 LangChain 事件体系到 Opik Trace/Span 模型的完整映射参数层面通过project_name/tags/metadata/thread_id控制归属与检索skip_error_callback/provider/distributed_headers解决错误噪声、代理网关成本、跨服务串联三类生产痛点机制层面以 根 run 建 Trace 不建 Span、只由 Trace 拥有者 finalize、LangGraph 控制流不记为错误 三条核心规则保证上报结构干净。配合track_langgraph的一次性注入与flush()的落库保障它构成了在 Opik 平台上调试与监控 LangChain / LangGraph 应用的完整链路。【免费下载链接】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），仅供参考
