Opik Python SDK AttachmentClient 实战指南为 Trace 与 Span 管理附件的上传、下载与查询【免费下载链接】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 Python SDK 中的AttachmentClient是面向 Trace追踪与 Span跨度附件操作的一站式客户端提供附件列表查询、字节流下载与文件上传三类能力。本文以该类的源码 docstring 为骨架结合仓库内的实现代码、端到端测试与完整示例系统讲解其构造方式、三个核心方法的参数语义、底层上传/下载管线以及典型应用场景如为 LLM-as-Judge 在线评估携带图片附件帮助读者在 RAG、Agentic 工作流的追踪与评估中正确落地附件能力。AttachmentClient 是什么AttachmentClient位于 client.py是 Opik Python SDK 中负责附件相关操作的客户端。其职责集中在三类操作列出与某条 Trace 或某个 Span 关联的附件以字节流Iterator[bytes]形式下载附件内容将本地文件上传为 Trace 或 Span 的附件。所有操作都在特定项目project上下文中执行因此调用时都必须显式提供project_name。从源码的类注释可以看到该客户端同时支持 Trace 与 Span 两种实体类型这也决定了附件功能在追踪系统中的基本使用粒度——既可以为整条链路Trace挂载文件也可以细化到链路中的单个调用Span。需要强调的是AttachmentClient通常不直接实例化而是通过Opik.get_attachment_client()工厂方法获取这样它会自动绑定当前Opik客户端已配置的 REST 连接、URL 与工作区信息避免手动装配底层依赖。获取客户端get_attachment_client 与构造参数在 opik_client.py 中get_attachment_client()的实现如下def get_attachment_client(self) - attachment_client.AttachmentClient: return attachment_client.AttachmentClient( rest_clientself._rest_client, url_overrideself._config.url_override, workspace_nameself._workspace, rest_httpx_clientself._httpx_client, )即它把当前Opik实例的四个关键依赖透传给AttachmentClient。AttachmentClient.__init__client.py的构造参数含义如下参数类型说明rest_clientrest_api_client.OpikApi用于发起后端 REST 请求的客户端实例项目解析、附件列表等均由它执行url_overridestrOpik 服务器的 Base URL下载与上传时会被 Base64 编码后拼入请求路径workspace_namestr工作区名称用于下载操作rest_httpx_clienthttpx.Client用于文件上传的httpx客户端实例实际使用只需import opik client opik.Opik(project_namemy-project) attachments_client client.get_attachment_client()get_attachment_list查询 Trace / Span 的附件列表get_attachment_list用于获取指定实体Trace 或 Span的全部附件元数据签名如下client.pydef get_attachment_list( self, project_name: str, entity_id: str, entity_type: Literal[span, trace], ) - List[RESTAttachmentDetails]:参数说明project_name实体所在项目的名称客户端会先将其解析为项目 IDentity_id目标 Trace 或 Span 的 IDentity_type实体类型仅允许trace或span。返回值RESTAttachmentDetails是 rest_api/types/attachment.py 中自动生成的 Pydantic 模型字段包括字段类型说明file_namestr附件文件名file_sizeint附件大小字节mime_typestr附件的 MIME 类型linkOptional[str]下载链接可为空从实现上看该方法有两个值得注意的细节项目名解析通过rest_helpers.resolve_project_id_by_name(self._rest_client, project_name)把项目名换算成项目 ID 后再请求后端URL 编码传递url_override被 Base64 编码后作为path参数传给self._rest_client.attachments.attachment_list(...)用于服务端定位正确的 API 前缀。典型调用attachments attachments_client.get_attachment_list( project_namemy-project, entity_idtrace_id, entity_typetrace, ) for attachment in attachments: print(attachment.file_name, attachment.file_size, attachment.mime_type)download_attachment以字节流下载附件内容download_attachment将附件内容作为字节迭代器返回适合流式消费大文件client.pydef download_attachment( self, project_name: str, entity_type: Literal[trace, span], entity_id: str, file_name: str, mime_type: str, ) - Iterator[bytes]:参数说明project_name实体所在项目名称entity_type实体类型trace/spanentity_id包含该附件的 Trace 或 Span 的 IDfile_name要下载的文件名mime_type文件的 MIME 类型。该方法内部的工作流程清晰体现了先查后下的设计先调用get_attachment_list拿到该实体的全部附件遍历匹配file_name与mime_type两者都一致的附件作为下载目标若未找到匹配项抛出ValueError(fAttachment not found: {file_name})若找到但link为空抛出ValueError(fNo download URL available for attachment: {file_name})根据链接类型选择下载通道若url_helpers.is_aws_presigned_url(link)判断为 AWS 预签名 URL则使用缓存复用、针对 S3 场景优化的s3_httpx_client.get_cached()否则使用实例自带的_rest_httpx_client通过httpx_client_upload.stream(GET, link)流式读取2xx 状态码下逐块yield字节非 2xx 时解析响应体并以rest_api_core.ApiError携带状态码、响应头与响应体抛出便于上层统一捕获处理。使用示例与 e2e 测试中的写法一致先拼装再对比内容attachment_data attachments_client.download_attachment( project_namemy-project, entity_typetrace, entity_idtrace_id, file_namereport.pdf, mime_typeapplication/pdf, ) content b.join(attachment_data)upload_attachment上传本地文件为附件upload_attachment将本地文件上传并挂载到指定 Trace 或 Span 上client.pydef upload_attachment( self, project_name: str, entity_type: Literal[trace, span], entity_id: str, file_path: str, file_name: Optional[str] None, mime_type: Optional[str] None, ) - None:参数说明project_name目标实体所在项目名称entity_type实体类型trace/spanentity_id目标 Trace 或 Span 的 IDfile_path本地文件的路径file_name上传后显示的文件名不传则默认取file_path的 basenamemime_type附件的 MIME 类型不传则通过mimetypes.guess_type(file_path)依据文件扩展名自动推断。方法开头有两处显式校验文件不存在时抛出FileNotFoundError随后计算文件大小并把url_overrideBase64 编码后一并封装进upload_options.FileUploadOptionsupload_options最后调用file_uploader.upload_attachment(...)执行真正的上传。底层上传管线upload_attachment并非简单 POST 一个文件而是走完整的分片上传管线file_uploader.py由FilePartsStrategy按最小分片大小计算文件需要拆分成几个 part通过RestFileUploadClient.start_upload(...)向后端申请上传元数据含预签名 URL 列表若后端返回should_use_s3_uploader()为真则走S3 直传用S3FileDataUploader把各 part 直接上传到对象存储再回传各 part 的e_tag与part_number完成s3_upload_completed确认否则走后端本地上传以UPLOAD_CHUNK_SIZE 8 * 1024 * 10248MB为块大小流式上传到后端。上传过程中还支持可选的FileUploadMonitor监控对象以及delete_after_upload、on_upload_success/on_upload_failed回调钩子上传失败时会在日志中记录文件名、路径与大小并抛出异常。这套机制意味着大附件也能以分片并发的方式高效落库。附件的对象模型Attachment 与 AttachmentWithContext除客户端本身外附件能力还涉及两个配套对象。Attachment用户侧输入模型attachment.py 中定义了用于声明附件的 Pydantic 模型class Attachment(pydantic.BaseModel): data: Union[str, bytes] file_name: Optional[str] None content_type: Optional[str] None create_temp_copy: bool Truedata附件内容支持三种形式——本地文件路径字符串、Base64 编码字符串或原始字节file_name自定义文件名缺省时使用原始文件名content_typeMIME 类型缺省时按文件推断create_temp_copy默认True会在上传前创建临时副本确保即使原始文件在上传前被删除也能成功上传临时文件会在上传后被清理。在创建 Trace / Span 时即可通过attachments[Attachment(...)]参数直接挂载见下文示例这是与upload_attachment并列的另一条上传路径。AttachmentWithContext内部上下文模型attachment_context.py 中的AttachmentWithContext是一个 dataclass将附件与其所属上下文绑定attachment_data附件本体、entity_typespan或trace、entity_id、project_name与context用途描述。该模型主要用于消息处理管线中为附件补充分类信息属于 SDK 内部流转结构。实战示例为在线评估携带图片附件仓库中的 thread_with_image_attachment.py 是一个完整的可运行示例展示了图片附件 在线 LLM-as-Judge 评估的组合用法评估器通过get_attachment拉取附件内容后再打分。其核心思路值得借鉴附件数据既可以是磁盘路径字符串也可以是内存中的字节示例内置了纯 Python 构造 1×1 红色 PNG的兜底逻辑保证脚本开箱即用单个独立 Trace 与多轮 Thread 两条路径都演示了client.trace(..., attachments[attachment])的挂载方式通过环境变量OPIK_API_KEY/OPIK_WORKSPACE/OPIK_URL_OVERRIDE配置连接本地部署可指向http://localhost:5173/api注意给 Trace 显式设置end_time否则在线评估采样器会将其视为进行中的不完整 Trace 而跳过评分。核心挂载代码节选import opik from opik import Attachment attachment Attachment( datapath/to/image.png, # 文件路径也可传 bytes / base64 字符串 file_nameimage.png, content_typeimage/png, ) client.trace( idtrace_id, namesingle-trace-vision-question, project_namePROJECT_NAME, end_timedatetime.datetime.now(tzdatetime.timezone.utc), input{role: user, content: Ive attached an image. Can you describe it?}, output{role: assistant, content: ...}, attachments[attachment], ) client.flush()端到端验证测试如何覆盖三大方法仓库的 e2e 测试 test_attachments_client.py 为AttachmentClient提供了完整的端到端覆盖可作为实现正确性的参考test_attachments_client__get_attachment_list_for_trace__happyflow创建带附件的 Trace 后查询列表断言数量为 1 且file_name、mime_type与上传时一致test_attachments_client__download_attachment_for_trace__happyflow下载附件后与原始文件逐字节比对验证内容一致性test_attachments_client__get_attachment_list_for_span__happyflow验证 Span 维度的列表查询test_attachments_client__upload_attachment_for_trace__happyflow与...__for_span__happyflow分别对 Trace、Span 执行upload_attachment再用verifiers.verify_attachments校验附件元数据与数据大小test_attachments_client__invalid_project_name__or_non_existing_entity_id__raises_error验证不存在的项目名或实体 ID 会抛出rest_api_core.ApiError。这些测试同时印证了三条 API 约定列表查询与下载是配套使用的下载内部依赖列表结果、Trace 与 Span 双实体均受支持、错误场景统一以ApiError暴露。另一条上传路径queue_attachment_upload 后台流式上传如果你的场景不需要同步阻塞等待上传完成可以改用Opik.queue_attachment_upload(entity_type, entity_id, project_name, file_path, file_nameNone, mime_typeNone)opik_client.py。该方法是非阻塞的文件进入后台流式上传管线streamer处理获得并行化、自动重试与监控能力之后调用client.flush()等待全部排队上传完成。选择建议需要在 Trace / Span 已创建后补充附件 →upload_attachment同步、确定性强希望在记录 Trace 的同时顺手异步携带附件 →queue_attachment_upload吞吐优先。小结AttachmentClient是 Opik Python SDK 附件能力的程序化入口get_attachment_list负责元数据查询、download_attachment负责流式取回内容、upload_attachment负责分片上传支持 S3 直传与后端上传两种通道。配合Attachment对象在创建 Trace / Span 时直接挂载、queue_attachment_upload异步补充以及示例与 e2e 测试所展示的图片附件评估场景开发者可以完整覆盖从记录附件到评估器读取附件的闭环。深入阅读 client.py、file_uploader.py 与 thread_with_image_attachment.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），仅供参考