Opik Python SDK 测试实战指南从 fake_backend 集成测试到 E2E 验证器【免费下载链接】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本文是 Opikcomet-llm 仓库Python SDK 的测试模式完整指南。你将从零掌握仓库内 Python SDK 的三层测试体系基于fake_backend的集成测试、基于verifiers的端到端E2E测试以及tests.testlib提供的断言与匹配工具同时学会在本地以 CI 等价方式或 dev-runner 方式运行tests/e2e全量套件并快速定位失败用例。读完本文你可以为 Opik Python SDK 的新功能写出与仓库现有测试风格一致、可稳定运行的高质量用例。测试命名规范让测试意图自解释仓库要求每个测试函数遵循三段式命名模式见 .agents/skills/python-sdk/testing.md# 模式test_WHAT__CASE_DESCRIPTION__EXPECTED_RESULT def test_tracked_function__error_inside_inner_function__caught_in_top_level_span(): pass # 正常路径test_WHAT__happyflow def test_optimization_lifecycle__happyflow(): passWHAT被测对象或行为例如tracked_function、optimization_lifecycleCASE_DESCRIPTION具体场景例如error_inside_inner_functionEXPECTED_RESULT期望结果例如caught_in_top_level_span正常路径统一使用happyflow后缀与仓库中大量test_*__happyflow用例保持一致性。这种命名规范让测试失败时的输出直接可读也方便通过pytest -k按关键词筛选相关用例。用 fake_backend 编写集成测试fake_backend是仓库集成测试的核心设施适用于一切会产生 trace/span 的测试尤其是第三方框架集成测试例如sdks/python/tests/library_integration/下的所有用例。它不发起真实网络请求而是把 SDK 内部消息管线替换为内存中的后端模拟器。fake_backend 的工作原理从 sdks/python/tests/conftest.py 的 fixture 源码可以看到pytest.fixture def fake_backend(patch_streamer): Patches the function that creates an instance of Streamer under the hood of Opik. As a result, instead of sending data to the backend, its being passed to the backend emulator, which uses this data to build span and trace trees. streamer, fake_message_processor_ patch_streamer ... mock_construct_online_streamer mock.Mock() mock_construct_online_streamer.return_value streamer with mock.patch.object( streamer_constructors, construct_online_streamer, mock_construct_online_streamer, ): yield fake_message_processor_即通过mock.patch替换streamer_constructors.construct_online_streamer让 SDK 以为连接了真实后端实际把数据交给BackendEmulatorMessageProcessor见 sdks/python/tests/testlib/backend_emulator_message_processor.py。该处理器继承自opik.message_processing.emulation.EmulatorMessageProcessor把收到的每条消息重建成TraceModel/SpanModel树测试可通过fake_backend.trace_trees或fake_backend.span_trees访问。配套 fixture 还有两个变体见 sdks/python/tests/conftest.pyfake_backend_without_batching当测试涉及 Span/Trace 的 update 请求不支持批处理时使用fake_backend_with_patched_environment以request.param指定的环境变量覆盖配合 fake_backend 使用适合验证环境变量对行为的影响。期望模型TraceModel / SpanModel测试中用于比对期望值的TraceModel/SpanModel定义在 sdks/python/tests/testlib/models.py。它们为默认值做了精心设计测试无需重复声明不关心的字段字段默认值含义project_nameANY默认不校验项目名除非测试显式指定last_updated_atANY_BUT_NONE只断言“不为 None”不比较具体时间attachmentsANY默认不校验附件sourcesdk默认来源标记完整示例嵌套函数追踪from tests.testlib import TraceModel, SpanModel, ANY_BUT_NONE, assert_equal from opik.decorator import tracker def test_track__one_nested_function__happyflow(fake_backend): tracker.track def f_inner(x): return inner-output tracker.track def f_outer(x): f_inner(inner-input) return outer-output f_outer(outer-input) tracker.flush_tracker() EXPECTED_TRACE_TREE TraceModel( idANY_BUT_NONE, namef_outer, input{x: outer-input}, output{output: outer-output}, start_timeANY_BUT_NONE, end_timeANY_BUT_NONE, spans[ SpanModel( idANY_BUT_NONE, namef_outer, input{x: outer-input}, output{output: outer-output}, spans[ SpanModel( idANY_BUT_NONE, namef_inner, input{x: inner-input}, output{output: inner-output}, spans[], ) ], ) ], ) assert len(fake_backend.trace_trees) 1 assert_equal(EXPECTED_TRACE_TREE, fake_backend.trace_trees[0])要点解读用tracker.track装饰器同时装饰内、外两层函数验证嵌套 span 树的构建测试结束时调用tracker.flush_tracker()确保内存中的消息被模拟器处理完成fake_backend.trace_trees返回List[TraceModel]期望树以递归spans列表表达层级关系assert_equal支持ANY_*通配符因此id、时间戳等动态字段用ANY_BUT_NONE占位。testlib 工具集ANY 系列与断言助手ANY 系列匹配器tests.testlib提供了四种通配匹配器实现在 sdks/python/tests/testlib/any_compare_helpers.pyfrom tests.testlib import ANY_BUT_NONE, ANY_STRING, assert_equal # ANY_BUT_NONE - 匹配任何非 None 的值 # ANY_STRING - 匹配任意字符串可附带前缀/包含条件 # assert_equal - 支持 ANY_* 的深度比较各匹配器的语义如下均来自源码实现ANY_BUT_NONEAnyButNone__eq__对一切非None值返回True用于“字段存在即可”的断言ANY_DICTAnyDict匹配任意 dict可通过containing({...})要求字典包含指定键值对ANY_LISTAnyList匹配任意 listANY_STRINGAnyString匹配任意字符串还支持链式starting_with(...)与containing(...)实现部分匹配ANY即unittest.mock.ANY见 any_compare_helpers.py用于精确类型之外的任意值。断言助手sdks/python/tests/testlib/assert_helpers.py 提供assert_equal(expected, actual)基于pytest_deepassert.equal的深度比较支持ANY_*通配assert_dicts_equal(dict1, dict2, ignore_keysNone)比较两个 dict可忽略指定键E2E 验证器中大量使用例如忽略动态生成的idassert_dict_has_keys(dic, keys)断言字典包含全部必需键assert_dict_keys_in_list(dic, keys)断言字典的所有键都在允许列表中assert_score_result(result, include_reasonTrue)针对ScoreResult的专用断言校验scoring_failed is False、value为[0.0, 1.0]区间的 float并可要求reason非空。E2E 测试与 verifiers 验证器当需要真正验证 SDK 与后端 API 的端到端行为时即发起真实 API 调用、数据落到 ClickHouse使用tests/e2e/verifiers.py提供的验证器。典型用法from tests.e2e import verifiers def test_trace_creation__e2e__happyflow(opik_client: opik.Opik): trace opik_client.trace( nametest-trace, input{query: test}, output{result: success} ) verifiers.verify_trace( opik_clientopik_client, trace_idtrace.id, nametest-trace, input{query: test}, output{result: success}, )验证器全家桶verifiers.py 覆盖了 SDK 的几乎所有核心对象每个验证器都接受opik_client、实体 ID 以及一组期望字段未指定的字段默认mock.ANY即不校验验证器验证对象关键能力verify_traceTracename/input/output/metadata/source/tags/error_info/project_name/feedback_scores/guardrails_validations/commentsverify_spanSpan除 Trace 字段外还校验trace_id、parent_span_id、model/provider/total_costverify_datasetDataset描述与条目数量忽略id键做逐条比较verify_dashboardDashboardwidget 配置、布局与version、section_countverify_experimentExperiment名称、元数据、反馈分数量、trace 数、prompt 版本与实验级分数verify_attachments附件大小、MIME 类型、下载链接前缀verify_threadThread按id搜索线程并校验反馈分verify_prompt_version/verify_chat_prompt_versionPrompt模板、类型、版本 ID、commit、环境归属底层机制轮询直到断言通过E2E 数据在 ClickHouse 中是最终一致性eventually-consistent创建操作落地快但后续 update例如离线队列重放的消息可能需要更长时间才被摄取。因此verify_trace/verify_span内部通过_retry_until_assertions_pass见 verifiers.py反复执行同一断言体断言体即_check()内部直接使用assert与testlib.assert_equal比较逻辑只维护一份轮询期间吞掉Exception404 未摄取、瞬时网络抖动等但pytest.fail.Exception继承自BaseException需显式转为返回值否则轮询会退化成单次尝试轮询受synchronization.until(..., max_try_seconds...)限制超时后重跑一次check()把真实异常与 traceback 抛给 pytest。E2E 环境的自动化配置sdks/python/tests/e2e/conftest.py 中的configure_e2e_tests_envfixture 会在每个测试模块期间通过testlib.patch_environ注入OPIK_PROJECT_NAME测试文件声明PROJECT_NAME常量则使用之否则生成e2e-module前缀的唯一项目名。同时opik_clientfixtureconftest.py每测试构建新的Opik(batchingTrue)客户端避免全局缓存客户端造成状态泄漏。参数化测试多场景驱动对于同一断言逻辑、多组输入输出的场景使用pytest.mark.parametrizepytest.mark.parametrize( text,expected_sentiment, [ (I love this product!, positive), (This is terrible., negative), (The sky is blue., neutral), ], ) def test_sentiment_classification(text, expected_sentiment): metric Sentiment() result metric.score(text) assert expected_sentiment in result.reason参数化与命名规范搭配后每个参数组合都会生成独立的测试节点失败时可直接定位到具体输入。四条核心规则仓库对新增测试有明确要求见 .agents/skills/python-sdk/testing.md只测试公共 APIpublic API only——不要触碰私有实现细节保证测试对重构有免疫力集成测试一律使用fake_backend——快速、无网络、无外部依赖E2E 测试使用verifiers验证器——统一轮询与断言逻辑避免每个用例自己重复实现“等数据落库”的样板代码动手前先研究已有相似测试——仓库tests/library_integration/129 个用例、tests/e2e/65 个用例是现成的模式库。本地运行 E2E 测试E2E 测试需要真实后端CI 工作流会隐式设置若干环境变量本地运行时需手动补齐。根据你的工作内容选择后端启动方式方案 ACI 等价方式推荐用于跑全量套件后端跑在 Docker 中行为与 GitHub Actions 一致# 后端以 Docker 方式启动与 GitHub Actions 匹配 TOGGLE_RUNNERS_ENABLEDtrue ./opik.sh --backend # 然后运行测试套件 cd sdks/python OPIK_URL_OVERRIDEhttp://localhost:5173/api/ \ venv/bin/pytest tests/e2e/ \ --ignoretests/e2e/test_guardrails.py \ -vv --durations20方案 Bdev-runner迭代后端代码时使用原生 Java 后端会继承你的 shell 环境变量因此启动前必须导出 MinIO 凭据与 runners 开关否则附件与 runner 相关测试会因环境问题而非真实回归失败export AWS_ACCESS_KEY_IDTHAAIOSFODNN7EXAMPLE export AWS_SECRET_ACCESS_KEYLESlrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY export TOGGLE_RUNNERS_ENABLEDtrue ./scripts/dev-runner.sh --restart cd sdks/python OPIK_URL_OVERRIDEhttp://localhost:8080/ \ venv/bin/pytest tests/e2e/ \ --ignoretests/e2e/test_guardrails.py \ -vv --durations20关于 AWS_* 凭据的说明AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY是 MinIO 的 root 用户/密码来源于 deployment/docker-compose/docker-compose.yaml其中MINIO_ROOT_USER与MINIO_ROOT_PASSWORD的默认值正是这两串EXAMPLE字符串参见该文件第 114-115、226-227 行。不涉及任何真实 AWS 账户——Java 后端的 S3 客户端在通过S3_URL指向 MinIO 时复用了标准 AWS 环境变量名。常见的坑gotchasTOGGLE_RUNNERS_ENABLED在 docker-compose 中默认为false。不开启的话tests/e2e/runner/目录下 8 个测试会因后端未启用 runners 功能而在 setup 阶段直接报错--ignoretests/e2e/test_guardrails.pyguardrails Python 服务不属于默认 compose 栈的一部分。CI 也显式忽略该文件见 .github/workflows/python_sdk_e2e_tests.ymlguardrails 的 E2E 由独立工作流 guardrails_e2e_tests.yml 负责其中同样设置了TOGGLE_RUNNERS_ENABLED: true并单独运行pytest tests/e2e/test_guardrails.pyMinIO 凭据只在方案 Bdev-runner中需要手动导出。方案 A 的 Docker 后端容器已内置这些凭据。调试失败的 E2E 测试当某个 E2E 用例失败时按以下步骤分层排查# 把 SDK 的 DEBUG 日志捕获到文件 OPIK_FILE_LOGGING_LEVELDEBUG OPIK_LOGGING_FILE/tmp/opik-sdk.log \ venv/bin/pytest tests/e2e/test_tracing.py::test_name -vvOPIK_FILE_LOGGING_LEVELDEBUG打开 SDK 的文件日志输出追踪 SDK 侧的上报行为与重试逻辑OPIK_LOGGING_FILE/tmp/opik-sdk.log指定日志落盘位置避免干扰终端输出。后端侧的错误日志取决于启动方式docker logs opik-backend-1 # 方案 A查看 Docker 后端容器日志 tail -f /tmp/opik-opik-backend.log # 方案 Bdev-runner 后端日志一个实用的判别技巧先判断失败是“环境问题”还是“真实回归”。若失败集中在 attachments、runner 相关用例优先检查TOGGLE_RUNNERS_ENABLED是否开启、MinIO 凭据是否导出若 SDK DEBUG 日志显示数据已成功上报而后端报错再结合后端日志定位服务端问题。此外E2E 验证器的轮询机制意味着偶发性失败往往与数据摄取延迟有关可先用--count或重跑确认是否稳定复现再深入分析。【免费下载链接】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），仅供参考