Opik Python SDK 的 Anthropic 集成：用 track_anthropic 为 Claude 调用自动注入全链路追踪

Opik Python SDK 的 Anthropic 集成用 track_anthropic 为 Claude 调用自动注入全链路追踪【免费下载链接】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-llmOpik 是 Comet 开源的 LLM 应用可观测性平台其 Python SDK 通过opik.integrations.anthropic.track_anthropic函数让用户只需一行包装即可将 Anthropic 官方客户端的所有messages调用同步/异步、流式/非流式、beta API转化为 Opik 中的 Trace 与 LLM Span自动记录输入输出、模型名、Token 用量与错误信息。本文基于仓库中的集成实现与测试用例详解该函数的用法、参数语义、被拦截的客户端方法范围以及底层 monkey-patch 与流式聚合的工作机制。快速上手一行代码接入追踪track_anthropic的完整 API 由 track_anthropic.rst 文档页自动从函数 docstring 生成其入口实现位于 opik_tracker.py。函数签名如下def track_anthropic( anthropic_client: AnthropicClient, project_name: Optional[str] None, ) - AnthropicClient:使用方式与仓库集成测试中的真实调用见 test_anthropic.py一致import anthropic from opik.integrations.anthropic import track_anthropic import opik client anthropic.Anthropic() client track_anthropic(anthropic_clientclient, project_namemy-anthropic-app) response client.messages.create( modelclaude-sonnet-4-20250514, messages[{role: user, content: Tell a short fact}], max_tokens1000, systemYou are a helpful assistant, ) opik.flush_tracker() # 将缓冲的 trace 数据推送到 Opik 后端包装完成后客户端的每个受支持调用都会自动上报外层生成一条 Trace内层生成一条typellm的 Span二者同名如anthropic_messages_create。支持哪些客户端类与哪些方法从 docstring 与TypeVar定义opik_tracker.py可以确认该函数接受以下 6 种 Anthropic SDK 客户端实例anthropic.Anthropic/anthropic.AsyncAnthropic标准 API同步/异步anthropic.AnthropicBedrock/anthropic.AsyncAnthropicBedrockAWS Bedrock 托管 Claudeanthropic.AnthropicVertex/anthropic.AsyncAnthropicVertexGCP Vertex AI 托管 Claude对上述所有类被追踪的方法范围为docstring 原文清单方法说明client.messages.create()标准消息创建client.messages.parse()消息解析结构化输出场景取决于 SDK 版本是否提供client.messages.stream()流式消息返回上下文管理器client.beta.messages.create()beta 版消息创建client.beta.messages.parse()beta 版消息解析client.beta.messages.stream()beta 版流式消息需要注意的是源码中 provider 统一硬编码为LLMProvider.ANTHROPIC并留有注释TODO: implement a proper support for vertex and bedrockopik_tracker.py即 Bedrock / Vertex 客户端目前也能接入追踪并写入区域元数据但 provider 标识仍记为anthropic。方法不支持时的降级行为并非所有 SDK 方法都会被完整追踪。源码中对两类方法只做了“告警 透传”处理实现见 messages_batch_decorator.pyclient.beta.messages.batches.create打印警告 “At the moment Opik Anthropic integration does not support tracking forclient.beta.messages.batches.createcalls”调用照常执行但不产生 Traceclient.completions.create打印警告 “Opik Anthropic integration does not support tracking forclient.completions.createcalls”。同时对可能不存在的 API 采用防御式 patch若 SDK 版本较老没有messages.parse或客户端没有beta.messages/completions属性patch 会以 debug 日志静默跳过opik_tracker.py保证track_anthropic对不同版本的 anthropic SDK 均能安全生效。幂等性与客户端元数据提取track_anthropic是可重复调用的函数在客户端实例上打opik_tracked True标记若检测到标记直接原样返回避免二次包装造成重复上报opik_tracker.py。此外包装时会从客户端实例提取连接元数据写入每条 Span 的 metadata见_extract_metadata_from_clientopik_tracker.py客户端类型额外写入的 metadata 字段全部客户端base_urlAnthropicBedrock/AsyncAnthropicBedrockaws_regionAnthropicVertex/AsyncAnthropicVertexregion、project_id这意味着在 Opik 界面中你可以通过 metadata 过滤出经由不同区域或 base_url 路由的 Claude 调用。Span 记录了什么输入、输出、usage 与 model真正决定上报内容的核心是 messages_create_decorator.py 中的AnthropicMessagesCreateDecorator它继承自 SDK 的BaseTrackDecorator框架。输入侧_start_span_inputs_preprocessor调用dict_utils.split_dict_by_keys(kwargs, KWARGS_KEYS_TO_LOG_AS_INPUTS)其中KWARGS_KEYS_TO_LOG_AS_INPUTS [messages, system, tools, output_format]messages_create_decorator.py。也就是说messages、system、tools、output_format四个参数进入 Span 的input其余所有 kwargs如model、max_tokens、temperature等进入metadataSpan 固定携带tags [anthropic]与metadata[created_from] anthropic以及provider字段。输出侧_end_span_inputs_preprocessor仅将响应对象中的content键提取为 Span 的outputRESPONSE_KEYS_TO_LOG_AS_OUTPUT [content]响应的model字段单独映射为 Span 的model其余字段id、role、stop_reason等进入 metadataToken 用量通过llm_usage.try_build_opik_usage_or_log_error从 Anthropic 的usage对象构建映射为 Opik 的标准 usage 结构若调用抛出异常输出为字符串时会被包装为{error: ...}记录。仓库集成测试对上报结构给出了精确断言test_anthropic.py例如 Span 的 usage 必须同时包含EXPECTED_ANTHROPIC_USAGE_DICT { completion_tokens: ANY_BUT_NONE, prompt_tokens: ANY_BUT_NONE, total_tokens: ANY_BUT_NONE, original_usage.input_tokens: ANY_BUT_NONE, original_usage.output_tokens: ANY_BUT_NONE, original_usage.cache_creation_input_tokens: ANY_BUT_NONE, original_usage.cache_read_input_tokens: ANY_BUT_NONE, original_usage.cache_creation.ephemeral_5m_input_tokens: ANY_BUT_NONE, original_usage.cache_creation.ephemeral_1h_input_tokens: ANY_BUT_NONE, }即 Opik 在计算成本/Token 指标时会同时保留 Anthropic 原始的缓存读写 Token 统计original_usage.*这对开启 prompt caching 的应用核算成本非常关键。流式调用如何被追踪stream patchers 机制流式路径比同步复杂得多messages.create(streamTrue)或messages.stream(...)返回的不是最终消息而是一个迭代器/上下文管理器。Opik 的做法是在 stream_patchers.py 中对 anthropic SDK 内部的流类做类级别的 dunder 方法 patch按返回对象类型分发见_streams_handlermessages_create_decorator.pyStream/AsyncStreamclient.messages.create(streamTrue)包装__iter__/__aiter__在迭代每个事件时用 anthropic 自带的_messages.accumulate_event累积事件快照finally中将完整累积消息作为 output 上报MessageStreamManager/AsyncMessageStreamManagerclient.messages.stream(...)上下文管理器因为MessageStream有get_final_message()等公开 API不能整体替换所以 patch__enter__/__aenter__把“待结束的 span/trace 引用”透传到内部 stream 对象再 patch 其__iter__/__aiter__迭代结束时调用get_final_message()取得完整消息后结束 SpanBetaMessageStreamManager/BetaAsyncMessageStreamManagerbeta 流同样逻辑若 SDK 版本没有 beta streaming 模块则自动禁用该分支。源码中的注释解释了一个 Python 细节必须装饰类方法而非实例方法因为解释器查找 dunder 方法时走类查找路径装饰实例属性不生效stream_patchers.py。同时每个 wrapper 用opik_tracked_instance标记 迭代结束后删除标记确保同一条流即使被多次迭代日志逻辑也只执行一次。所有流式 wrapper 都遵循相同的错误处理骨架捕获迭代过程中的异常、经error_info_collector.collect收集错误信息后重新抛出对业务无感并在finally中携带error_info结束 Span发生异常时 output 置为None仅保留错误信息。在更大的追踪上下文中使用docstring 明确说明 “Can be used within other Opik-tracked functions”即包装后的客户端可以放在opik.track装饰的函数或上下文管理器内部此时 LLM Span 会挂到当前 trace 之下而非新建 trace。这一行为由BaseTrackDecorator框架统一处理。仓库中还有两个典型使用场景可供参考评估模块内部封装的 Anthropic chat modelanthropic_chat_model.py直接复用track_anthropic上报评估过程中对 Claude 的调用演示脚本 demo_data.py 展示了用包装后的客户端生成 demo trace 数据。验证与测试集成行为的回归保障位于 test_anthropic.py共 1300 余行覆盖 create/stream/beta 等路径其断言模式值得注意测试通过TraceModel/SpanModel匹配完整上报结构——Trace 名为anthropic_messages_create、input 包含messages与system、tags 为[anthropic]、Span 的provider为anthropic、model以传入的模型短名开头。project_name参数为None时回落到 SDK 默认项目名OPIK_PROJECT_DEFAULT_NAME显式传入时则使用该值见测试参数化test_anthropic.py。测试使用tenacity对 5xx 内部服务器错误做最多 3 次重试说明这些用例依赖真实 Anthropic API 配置ensure_anthropic_configuredfixture。小结与适用前提适用前提已安装opik与anthropic两个 Python 包并完成 Opik 连接配置opik.init或环境变量track_anthropic返回同一个客户端实例属于非侵入式装饰可随时在应用生命周期任意时点调用能力边界beta.messages.batches.create与completions.create暂不追踪仅告警messages.parse与beta.*的 patch 取决于 anthropic SDK 版本是否实现对应方法Bedrock/Vertex 客户端可追踪但 provider 统一记为anthropic核心源码路径入口 opik_tracker.py、上报字段构造 messages_create_decorator.py、流式 patch stream_patchers.py、集成测试 test_anthropic.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），仅供参考