Opik Python SDK 评估核心数据结构TestResult 全解析【免费下载链接】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 的自动化评估Evaluation将「数据集样本 LLM 任务 评分指标」的运行结果沉淀为结构化的TestResult对象它是实验Experiment中一次完整评分单元的载体。本文基于 TestResult.rst 文档结合 SDK 源码深入拆解TestResult的字段语义、构建流程、聚合消费方式以及它在断点续跑与自定义实验评分函数中的用法帮助开发者理解 Opik 评估引擎的数据流转全貌。一、TestResult 的定位评估实验的最小评分单元Opik Python SDK 的评估管线大致是从数据集中取出样本 → 交由 LLM 任务生成输出 → 对输出执行一组评分指标 → 将评分结果写回 Trace 作为反馈分数。TestResult就是这组「样本 评分结果 试验编号」打包后的不可变数据容器定义在 test_result.pyfrom typing import List, Optional import dataclasses from . import test_case from .metrics import score_result dataclasses.dataclass class TestResult: test_case: test_case.TestCase score_results: List[score_result.ScoreResult] trial_id: int task_execution_time: Optional[float] None scoring_time: Optional[float] None它是一个标准dataclasses.dataclass字段说明如下字段类型含义test_casetest_case.TestCase本次评分对应的测试用例数据集样本、任务输出等上下文score_resultsList[score_result.ScoreResult]该用例上所有评分指标的结果列表一条用例可绑定多个指标trial_idint试验编号。同一数据集样本可多次运行trial用于评估输出的稳定性task_execution_timeOptional[float]任务执行耗时秒未记录时为Nonescoring_timeOptional[float]评分执行耗时秒未记录时为None从字段设计可以看出TestResult不直接保存原始文本而是引用test_case样本与输出和score_results各指标打分再通过trial_id区分同一用例的多轮试验天然支持「多次采样取均值/标准差」的评估统计。二、关联数据结构TestCase 与 ScoreResult要真正理解TestResult需要先看懂它引用的两个搭档。2.1 TestCase一次评分所需的全部输入定义在 test_case.pydataclasses.dataclass class TestCase: trace_id: str dataset_item_id: str task_output: Dict[str, Any] dataset_item_content: Dict[str, Any] dataclasses.field(default_factorydict) mapped_scoring_inputs: Optional[Dict[str, Any]] None dataset_item: Optional[dataset_item.DatasetItem] Nonetrace_id本次任务生成的 Trace 唯一标识评估结果最终通过它回写反馈分数dataset_item_id被评估的数据集样本 IDtask_outputLLM 任务的实际输出如回答文本、JSON 结构dataset_item_content数据集中该样本的原始内容mapped_scoring_inputs按评分函数所需字段映射后的输入由评估引擎在运行期填充见下文引擎源码dataset_item完整的数据集条目对象供需要读取样本元数据的评分函数使用。2.2 ScoreResult单个指标的打分结果评分指标Metric的score()方法返回ScoreResult轻量实现位于 _score_result.pydataclasses.dataclass class ScoreResult: name: str # 指标名如 equals、contains value: float # 数值分数 reason: Optional[str] None # 人类可读的判分理由 category_name: Optional[str] None # 可选分类标签 metadata: Optional[Dict[str, Any]] None # 附加元数据 scoring_failed: bool False # 评分是否失败scoring_failed字段尤其关键当指标因超时、模型不可用等原因无法评分时评估引擎不会丢弃该结果而是标记scoring_failedTrue并让下游统计逻辑将其排除避免污染均值详见第四节聚合逻辑。对外导出入口是 score_result.py 中的opik.evaluation.metrics.score_result.ScoreResult。三、TestResult 的诞生评估引擎中的构建与回写TestResult由评估引擎 engine.py 在每次评分后构造核心代码如下约第 120–161 行trace_tool_context self._build_trace_tool_context( trace_idtest_case_.trace_id, trace_datatrace_data ) score_results, mapped_scoring_inputs item_evaluator.compute_regular_scores( dataset_item_contenttest_case_.dataset_item_content, task_outputtest_case_.task_output, trace_tool_contexttrace_tool_context, ) test_case_.mapped_scoring_inputs mapped_scoring_inputs test_result_ test_result.TestResult( test_casetest_case_, score_resultsscore_results, trial_idtrial_id, ) rest_operations.log_test_result_feedback_scores( clientself._client, score_resultsscore_results, trace_idtest_case_.trace_id, project_nameself._project_name, ) return test_result_从源码可以梳理出TestResult的完整生命周期引擎按数据集样本构建TestCase并为其绑定指标求值器MetricsEvaluator调用compute_regular_scores()计算该用例上的全部常规指标同时把映射后的评分输入写回test_case.mapped_scoring_inputs组装TestResult(test_case, score_results, trial_id)——此时task_execution_time与scoring_time保持None属于可选计时字段调用log_test_result_feedback_scores()将score_results作为反馈分数写回对应 Tracetrace_id这就是评估结果能直接在 Opik 前端的 Trace 详情中看到评分的原因返回TestResult由引擎汇聚成EvaluationResult.test_results列表。注意引擎在构造TestResult时只显式传入三个必填字段task_execution_time/scoring_time两个可选计时字段在部分代码路径中保持默认值None从当前源码看并未在构造点赋值属预留的耗时统计位。四、TestResult 的消费聚合、分组与报表单个TestResult价值有限评估引擎会把一次实验的所有TestResult收集进EvaluationResult定义于 evaluation_result.py再提供两类视图。4.1 聚合统计aggregate_evaluation_scoresEvaluationResult.aggregate_evaluation_scores()调用 score_statistics.py 的calculate_aggregated_statistics()遍历每个TestResult.score_results跳过scoring_failedTrue的结果后按指标名分组计算mean、max、min、values以及标准差std样本数 2 时为Nonefor test_result_ in evaluation_results: for score_result in test_result_.score_results: # Only include successful scores with valid values if not score_result.scoring_failed and _is_valid_score_value( score_result.value ): scores_by_name[score_result.name].append(score_result.value)聚合结果封装为EvaluationResultAggregatedScoresView其aggregated_scores字典以指标名为键、ScoreStatistics为值直接对应 Opik 实验详情页中的汇总指标。4.2 按数据集样本分组group_by_dataset_item_view若需要按样本维度查看多轮试验的表现可调用EvaluationResult.group_by_dataset_item_view()。其内部实现会按test_result.test_case.dataset_item_id分组组内按trial_id排序并为每个样本单独计算聚合统计产出EvaluationResultGroupByDatasetItemsView——这在对比「同一 prompt 在不同样本上的稳定性」时非常实用。4.3 控制台表格report.py轻量评估场景下report.py 会把test_results转成终端表格先计算平均分与失败分数再逐条提取score_results生成行最后用对齐工具打印。这也是opik.evaluate在 Notebook 中输出评估摘要的实现基础。五、进阶用法基于 TestResult 的自定义实验评分TestResult不只是内部数据结构它还作为实验级评分函数Experiment Scoring Function的入参暴露给用户。在evaluate中传入experiment_scoring_functions时引擎会调用 evaluation_result.py 的compute_experiment_scores()def compute_experiment_scores( experiment_scoring_functions: List[ExperimentScoreFunction], test_results: List[test_result.TestResult], ) - List[score_result.ScoreResult]: for score_function in experiment_scoring_functions: try: scores score_function(test_results) # Handle Union[ScoreResult, List[ScoreResult]] if isinstance(scores, list): all_scores.extend(scores) else: all_scores.append(scores) except Exception as e: LOGGER.warning(Failed to compute experiment score: %s, e, ...)也就是说你的实验级评分函数收到的是整个实验的List[TestResult]可以跨样本计算整体指标如成功率、平均延迟、字段完整率返回单个或多个ScoreResult。典型签名如下def experiment_level_metric(test_results: List[TestResult]) - ScoreResult: valid [r for r in test_results if not any(s.scoring_failed for s in r.score_results)] return ScoreResult( namesuccess_rate, valuelen(valid) / len(test_results), reason..., )注意两点评分函数抛出的异常会被引擎捕获并记录 warningFailed to compute experiment score不会中断整个评估返回类型支持ScoreResult或List[ScoreResult]的联合形态。六、断点续跑中的 TestResult 合并EvaluationResult还提供一个纯函数merge_resume_results()用于把历史已完成的test_results与本次续跑新产生的结果合并返回一个覆盖整个实验的EvaluationResult它只拼接test_results列表并透传experiment_id、experiment_url、trial_count等身份字段它不会触碰后端也不会重新计算实验级评分——合并后的experiment_scores留空交由调用方用用户传入的experiment_scoring_functions重新计算调用方必须在续跑执行新试验之前对历史previous_test_results做快照否则续跑写入的新实验条目会被重新组装进合并结果造成重复计数。这保证了断点续跑后TestResult列表依然代表「完整实验」的评分事实便于统计口径一致。七、小结TestResult是 Opik 评估体系中连接「用例输入」「指标输出」「试验编号」「计时信息」四个维度的核心记录它由评估引擎在每次评分后构建engine.py并同步把评分回写到 Trace 反馈它经EvaluationResult聚合出实验级统计均值/最值/标准差也可按样本分组观察多轮试验它直接作为实验级评分函数的入参支持跨样本的整体指标计算它在断点续跑合并逻辑中充当「事实单位」保证续跑前后统计口径一致。对 SDK 用户而言理解TestResult的字段与流转就等于理解了 Opik 评估结果从「评分完成」到「前端展示」的完整链路。若要进一步深入可直接阅读 test_result.py、evaluation_result.py 与 score_statistics.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），仅供参考 Opik Python SDK 评估核心数据结构：TestResult 全解析
Opik Python SDK 评估核心数据结构TestResult 全解析【免费下载链接】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 的自动化评估Evaluation将「数据集样本 LLM 任务 评分指标」的运行结果沉淀为结构化的TestResult对象它是实验Experiment中一次完整评分单元的载体。本文基于 TestResult.rst 文档结合 SDK 源码深入拆解TestResult的字段语义、构建流程、聚合消费方式以及它在断点续跑与自定义实验评分函数中的用法帮助开发者理解 Opik 评估引擎的数据流转全貌。一、TestResult 的定位评估实验的最小评分单元Opik Python SDK 的评估管线大致是从数据集中取出样本 → 交由 LLM 任务生成输出 → 对输出执行一组评分指标 → 将评分结果写回 Trace 作为反馈分数。TestResult就是这组「样本 评分结果 试验编号」打包后的不可变数据容器定义在 test_result.pyfrom typing import List, Optional import dataclasses from . import test_case from .metrics import score_result dataclasses.dataclass class TestResult: test_case: test_case.TestCase score_results: List[score_result.ScoreResult] trial_id: int task_execution_time: Optional[float] None scoring_time: Optional[float] None它是一个标准dataclasses.dataclass字段说明如下字段类型含义test_casetest_case.TestCase本次评分对应的测试用例数据集样本、任务输出等上下文score_resultsList[score_result.ScoreResult]该用例上所有评分指标的结果列表一条用例可绑定多个指标trial_idint试验编号。同一数据集样本可多次运行trial用于评估输出的稳定性task_execution_timeOptional[float]任务执行耗时秒未记录时为Nonescoring_timeOptional[float]评分执行耗时秒未记录时为None从字段设计可以看出TestResult不直接保存原始文本而是引用test_case样本与输出和score_results各指标打分再通过trial_id区分同一用例的多轮试验天然支持「多次采样取均值/标准差」的评估统计。二、关联数据结构TestCase 与 ScoreResult要真正理解TestResult需要先看懂它引用的两个搭档。2.1 TestCase一次评分所需的全部输入定义在 test_case.pydataclasses.dataclass class TestCase: trace_id: str dataset_item_id: str task_output: Dict[str, Any] dataset_item_content: Dict[str, Any] dataclasses.field(default_factorydict) mapped_scoring_inputs: Optional[Dict[str, Any]] None dataset_item: Optional[dataset_item.DatasetItem] Nonetrace_id本次任务生成的 Trace 唯一标识评估结果最终通过它回写反馈分数dataset_item_id被评估的数据集样本 IDtask_outputLLM 任务的实际输出如回答文本、JSON 结构dataset_item_content数据集中该样本的原始内容mapped_scoring_inputs按评分函数所需字段映射后的输入由评估引擎在运行期填充见下文引擎源码dataset_item完整的数据集条目对象供需要读取样本元数据的评分函数使用。2.2 ScoreResult单个指标的打分结果评分指标Metric的score()方法返回ScoreResult轻量实现位于 _score_result.pydataclasses.dataclass class ScoreResult: name: str # 指标名如 equals、contains value: float # 数值分数 reason: Optional[str] None # 人类可读的判分理由 category_name: Optional[str] None # 可选分类标签 metadata: Optional[Dict[str, Any]] None # 附加元数据 scoring_failed: bool False # 评分是否失败scoring_failed字段尤其关键当指标因超时、模型不可用等原因无法评分时评估引擎不会丢弃该结果而是标记scoring_failedTrue并让下游统计逻辑将其排除避免污染均值详见第四节聚合逻辑。对外导出入口是 score_result.py 中的opik.evaluation.metrics.score_result.ScoreResult。三、TestResult 的诞生评估引擎中的构建与回写TestResult由评估引擎 engine.py 在每次评分后构造核心代码如下约第 120–161 行trace_tool_context self._build_trace_tool_context( trace_idtest_case_.trace_id, trace_datatrace_data ) score_results, mapped_scoring_inputs item_evaluator.compute_regular_scores( dataset_item_contenttest_case_.dataset_item_content, task_outputtest_case_.task_output, trace_tool_contexttrace_tool_context, ) test_case_.mapped_scoring_inputs mapped_scoring_inputs test_result_ test_result.TestResult( test_casetest_case_, score_resultsscore_results, trial_idtrial_id, ) rest_operations.log_test_result_feedback_scores( clientself._client, score_resultsscore_results, trace_idtest_case_.trace_id, project_nameself._project_name, ) return test_result_从源码可以梳理出TestResult的完整生命周期引擎按数据集样本构建TestCase并为其绑定指标求值器MetricsEvaluator调用compute_regular_scores()计算该用例上的全部常规指标同时把映射后的评分输入写回test_case.mapped_scoring_inputs组装TestResult(test_case, score_results, trial_id)——此时task_execution_time与scoring_time保持None属于可选计时字段调用log_test_result_feedback_scores()将score_results作为反馈分数写回对应 Tracetrace_id这就是评估结果能直接在 Opik 前端的 Trace 详情中看到评分的原因返回TestResult由引擎汇聚成EvaluationResult.test_results列表。注意引擎在构造TestResult时只显式传入三个必填字段task_execution_time/scoring_time两个可选计时字段在部分代码路径中保持默认值None从当前源码看并未在构造点赋值属预留的耗时统计位。四、TestResult 的消费聚合、分组与报表单个TestResult价值有限评估引擎会把一次实验的所有TestResult收集进EvaluationResult定义于 evaluation_result.py再提供两类视图。4.1 聚合统计aggregate_evaluation_scoresEvaluationResult.aggregate_evaluation_scores()调用 score_statistics.py 的calculate_aggregated_statistics()遍历每个TestResult.score_results跳过scoring_failedTrue的结果后按指标名分组计算mean、max、min、values以及标准差std样本数 2 时为Nonefor test_result_ in evaluation_results: for score_result in test_result_.score_results: # Only include successful scores with valid values if not score_result.scoring_failed and _is_valid_score_value( score_result.value ): scores_by_name[score_result.name].append(score_result.value)聚合结果封装为EvaluationResultAggregatedScoresView其aggregated_scores字典以指标名为键、ScoreStatistics为值直接对应 Opik 实验详情页中的汇总指标。4.2 按数据集样本分组group_by_dataset_item_view若需要按样本维度查看多轮试验的表现可调用EvaluationResult.group_by_dataset_item_view()。其内部实现会按test_result.test_case.dataset_item_id分组组内按trial_id排序并为每个样本单独计算聚合统计产出EvaluationResultGroupByDatasetItemsView——这在对比「同一 prompt 在不同样本上的稳定性」时非常实用。4.3 控制台表格report.py轻量评估场景下report.py 会把test_results转成终端表格先计算平均分与失败分数再逐条提取score_results生成行最后用对齐工具打印。这也是opik.evaluate在 Notebook 中输出评估摘要的实现基础。五、进阶用法基于 TestResult 的自定义实验评分TestResult不只是内部数据结构它还作为实验级评分函数Experiment Scoring Function的入参暴露给用户。在evaluate中传入experiment_scoring_functions时引擎会调用 evaluation_result.py 的compute_experiment_scores()def compute_experiment_scores( experiment_scoring_functions: List[ExperimentScoreFunction], test_results: List[test_result.TestResult], ) - List[score_result.ScoreResult]: for score_function in experiment_scoring_functions: try: scores score_function(test_results) # Handle Union[ScoreResult, List[ScoreResult]] if isinstance(scores, list): all_scores.extend(scores) else: all_scores.append(scores) except Exception as e: LOGGER.warning(Failed to compute experiment score: %s, e, ...)也就是说你的实验级评分函数收到的是整个实验的List[TestResult]可以跨样本计算整体指标如成功率、平均延迟、字段完整率返回单个或多个ScoreResult。典型签名如下def experiment_level_metric(test_results: List[TestResult]) - ScoreResult: valid [r for r in test_results if not any(s.scoring_failed for s in r.score_results)] return ScoreResult( namesuccess_rate, valuelen(valid) / len(test_results), reason..., )注意两点评分函数抛出的异常会被引擎捕获并记录 warningFailed to compute experiment score不会中断整个评估返回类型支持ScoreResult或List[ScoreResult]的联合形态。六、断点续跑中的 TestResult 合并EvaluationResult还提供一个纯函数merge_resume_results()用于把历史已完成的test_results与本次续跑新产生的结果合并返回一个覆盖整个实验的EvaluationResult它只拼接test_results列表并透传experiment_id、experiment_url、trial_count等身份字段它不会触碰后端也不会重新计算实验级评分——合并后的experiment_scores留空交由调用方用用户传入的experiment_scoring_functions重新计算调用方必须在续跑执行新试验之前对历史previous_test_results做快照否则续跑写入的新实验条目会被重新组装进合并结果造成重复计数。这保证了断点续跑后TestResult列表依然代表「完整实验」的评分事实便于统计口径一致。七、小结TestResult是 Opik 评估体系中连接「用例输入」「指标输出」「试验编号」「计时信息」四个维度的核心记录它由评估引擎在每次评分后构建engine.py并同步把评分回写到 Trace 反馈它经EvaluationResult聚合出实验级统计均值/最值/标准差也可按样本分组观察多轮试验它直接作为实验级评分函数的入参支持跨样本的整体指标计算它在断点续跑合并逻辑中充当「事实单位」保证续跑前后统计口径一致。对 SDK 用户而言理解TestResult的字段与流转就等于理解了 Opik 评估结果从「评分完成」到「前端展示」的完整链路。若要进一步深入可直接阅读 test_result.py、evaluation_result.py 与 score_statistics.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），仅供参考