企业官网建设流程全解析

Opik Python SDK 的 evaluate() 函数全解：数据集任务评估的参数、执行链路与错误容错机制

【免费下载链接】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.evaluation.evaluate为核心——它是 Opik Python SDK 中执行数据集评估的官方入口（API 参考页 通过 Sphinx 的autofunction指令直接引用该函数）。读完后，你将掌握evaluate()全部参数的取值与默认值、实验创建到结果回写的完整调用链、三种数据项选择方式（nb_samples/dataset_item_ids/dataset_sampler）、ErrorTolerance两级容错的语义差异，以及如何用evaluate_resume从中断处续跑评估。

evaluate() 是干什么的

evaluate()对给定数据集执行任务评估：它先在后端创建一个experiment（实验），然后并发执行传入的task函数（对每个数据集条目调用一次，拿到任务输出），再对所有任务输出执行评分（scoring_metrics或scoring_functions），最后把分数作为 feedback scores 写回实验，并返回一个EvaluationResult对象。

入口函数定义在 evaluator.py 中，官方示例 evaluation_example.py 展示了最小可用形态：

from opik.evaluation.metrics import IsJson, Hallucination from opik.evaluation import evaluate from opik import Opik client = Opik() dataset = client.get_or_create_dataset(name="My 42 dataset") results = evaluate( experiment_name="My experiment", dataset=dataset, task=llm_task, # 见下文“task 函数” nb_samples=2, scoring_metrics=[IsJson(), Hallucination()], )

完整参数表

以下为evaluate()签名中的全部参数（源码签名见 evaluator.py）：

返回值为 EvaluationResult，包含experiment_id、dataset_id、experiment_name、test_results、experiment_url、trial_count、experiment_scores七个字段。

task 函数的契约

task的类型别名LLMTask = Callable[[Dict[str, Any]], Dict[str, Any]]定义在 types.py。约定是：入参为该数据集条目的 data 字典，返回值是将被评分的字典。通常做法是用@track()装饰 task，这样任务内部的 LLM 调用会自动落成 trace/span：

@track() def llm_task(item: Dict[str, Any]) -> Dict[str, Any]: response = openai_client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": item["input"]["message"]}], ) return { "output": response.choices[0].message.content, "reference": "test", }

scoring_metrics 与 scoring_functions 两种打分方式

二选一即可。

1. scoring_metrics：每个指标有score(...)方法，其形参名必须能从“数据集条目内容 + 任务输出”（经scoring_key_mapping重命名后）中解析到。比如Equals()需要output和reference两个键。
2. scoring_functions：自定义打分函数，按保留参数从引擎接收数据，无需scoring_key_mapping。其协议ScorerFunctionProtocol定义在 scorer_function.py：

def my_scorer( dataset_item: Dict[str, Any], # 数据集条目内容 task_outputs: Dict[str, Any], # 任务输出 task_span=None, # 可选：任务执行期间收集的 span 数据 ) -> ScoreResult: ...

源码中validate_scorer_function会检查函数签名：必须同时具有dataset_item与task_outputs两个参数，或至少含task_span，否则抛ValueError。

scoring_key_mapping 的用法

当数据集条目或任务输出的键名与指标期望不一致时用它做映射。例如条目内容是{"user_question": "What is Opik ?"}而某指标要求input键，则写scoring_key_mapping={"input": "user_question"}。该参数还可以是 callable（在evaluate_experiment等场景做动态映射），源码中的类型别名为ScoringKeyMappingType = Dict[str, Union[str, Callable[[Dict[str, Any]], Any]]]。

执行链路：从创建实验到结果回写

evaluate()的内部执行顺序（均见 evaluator.py）：

1. 实验命名与项目解析：_use_or_create_experiment_name在experiment_name为空时生成带前缀的随机名；helpers.resolve_project_name按“数据集project_name优先、用户参数兜底”解析目标项目。
2. 创建实验：client.create_experiment(...)带上experiment_config（若给了blueprint_id，会先通过merge_blueprint_into_config合并蓝图配置）、关联的 prompts、tags 与数据集版本 id。
3. 解析数据项：helpers.resolve_dataset_items根据nb_samples/dataset_item_ids/dataset_sampler/dataset_filter_string解析出条目迭代器；随后_materialize_for_checkpoint按三种情况处理——使用 sampler 时先物化列表以便断点续跑记录精确条目 id，仅显式指定 id 时保持惰性流式，两者皆无时直接流式处理且不写断点。
4. 包装打分函数：若提供了scoring_functions，_wrap_scoring_functions将其包装成指标形态。
5. 引擎执行：_evaluate_task构造ExecutionPolicy（runs_per_item=trial_count）并实例化EvaluationEngine（并发度即task_threads），调用run_and_score完成“跑任务 + 打分”主循环。
6. 实验级打分：compute_experiment_scores在全部TestResult收集完毕后逐个执行experiment_scoring_functions（单个函数抛异常只告警不中断，见 evaluation_result.py）。
7. 展示与回写：verbose >= 1时打印摘要并给出实验 URL；client.flush()后调用finish_experiments通知后端实验完成，实验级分数经experiment.log_experiment_scores写回；verbose >= 2时额外打印分数统计（report.display_evaluation_scores_statistics）。

一个细节：实验完成通知被包在_try_notifying_about_experiment_completion的 try/except 中，通知失败只记录 debug 日志、不影响评估结果本身——说明实验 URL 与完成通知都属于“尽力而为”的收尾动作。

结果对象 EvaluationResult

EvaluationResult（evaluation_result.py）除直接携带test_results外，还提供两个聚合视图：

• aggregate_evaluation_scores()：对整个实验计算每个分数的聚合统计，返回EvaluationResultAggregatedScoresView（含aggregated_scores: Dict[str, ScoreStatistics]）；
• group_by_dataset_item_view()：按数据集条目分组（每组按trial_id排序），用于trial_count > 1时查看每个条目的多次试跑结果。

实验级打分函数就是基于这份test_results列表做全局聚合。官方示例中compute_hallucination_stats接收全部TestResult，提取第一个分数后返回一个ScoreResult(name="Custom metric", value=max(scores))，这正是ExperimentScoreFunction契约（Callable[[List[TestResult]], Union[ScoreResult, List[ScoreResult]]]）的完整示范。

选择评估哪些数据项

evaluate()提供四组互有优先级的筛选参数：

• nb_samples：只取前 N 条，适合冒烟验证；
• dataset_item_ids：精确指定条目 id 列表；
• dataset_sampler：采样器实例。SDK 内置RandomDatasetSampler（random_dataset_sampler.py），基类BaseDatasetSampler是抽象类，可自行扩展采样策略；
• dataset_filter_string：OQL 过滤字符串，支持按tags（contains运算符）、data字段（点号路径，如data.category）、created_at等 ISO 8601 时间字段过滤。docstring 给出的示例：

tags contains "failed" # 带 failed 标签的条目 data.category = "test" # 指定 data 字段值的条目 created_at >= "2024-01-01T00:00:00Z" # 某时间之后创建的条目

可过滤列包括id、source、trace_id、span_id、data（字典字段）、tags（列表字段）、created_at/last_updated_at（时间字段）、created_by/last_updated_by。

trial_count控制每个条目重复执行任务并评分的次数，默认1；配合group_by_dataset_item_view()可以拿到每条数据的多次试跑分数分布。

错误容错：ErrorTolerance 两级语义

error_tolerance参数接受ErrorTolerance枚举成员或其等值 int（IntEnum，非法值抛ValueError），定义见 types.py：

• ErrorTolerance.METRIC_ERRORS（10，默认）：score方法内部抛出的异常被记录为“失败的打分结果”（scoring_failed=True），评估继续；但在进入score之前发生的失败——数据集缺少指标必需的键、条目级评估器无法构建——会使整个运行中止。
• ErrorTolerance.ALL_SCORING_ERRORS（20）：额外容忍“导致某指标完全无法打分”的错误——必需的打分参数缺失、条目级评估器构建失败。此时拿到的是带失败记录的EvaluationResult而不是异常。

两点共同边界：无论哪一级，评估任务（task）本身的失败都会中止运行，scoring_key_mapping中 callable 抛异常也总是中止——因为它们不属于单个指标，无法归因。另注意容错并不提前止损：所有数据集条目都会先跑完，第一个失败才被重新抛出，所以影响全部条目的配置错误在任何级别下都会耗掉一整轮。被容忍的指标失败会记录在同名 span 的error_info上，在 trace 中可见；它们不会作为 feedback score 持久化，因此界面上分数单元格留空而不是显示 0，也从聚合统计中排除。

完整可运行示例

下面的例子整合了上面所有要点（数据集导入 → task → 指标 → 实验级打分 → 结果解析），风格与官方示例 evaluation_example.py 一致：

import json from typing import Any, Dict, List from opik import Opik, track from opik.evaluation import evaluate from opik.evaluation.metrics import IsJson, score_result from opik.evaluation import test_result client = Opik() dataset = client.get_or_create_dataset( name="My 42 dataset", description="For storing stuff" ) dataset.insert_from_json( json_array=json.dumps( [ {"Model inputs": {"message": "Greet me!"}}, {"Model inputs": {"message": "Give a json example!"}}, ] ), keys_mapping={"Model inputs": "input"}, ) @track() def llm_task(item: Dict[str, Any]) -> Dict[str, Any]: response = openai_client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": item["input"]["message"]}], ) return {"output": response.choices[0].message.content, "reference": "test"} def compute_hallucination_stats( test_results: List[test_result.TestResult], ) -> List[score_result.ScoreResult]: scores = [ x.score_results[0].value for x in test_results if x.score_results and len(x.score_results) > 0 ] if not scores: return [] return [ score_result.ScoreResult( name="Custom metric", value=max(scores) if len(scores) > 1 else 0.0, ) ] results = evaluate( experiment_name="My experiment", dataset=dataset, task=llm_task, nb_samples=2, scoring_metrics=[IsJson()], experiment_scoring_functions=[compute_hallucination_stats], verbose=2, ) print(results.experiment_url) aggregated = results.aggregate_evaluation_scores() print(aggregated.aggregated_scores)

运行前提：已配置 Opik 客户端连接（Opik()会读取默认/环境变量配置），且track_openai等集成已导入并初始化（示例中省略）。verbose=2会在结尾打印每个分数的详细统计。

中断续跑：evaluate_resume

大规模评估被网络抖动、限流或实例重启打断时，SDK 提供了opik.evaluate_resume(experiment_id=...)：它读取实验已完成的条目，只对剩余条目重新执行 task 与打分。由于evaluate()内部会通过resume_integration.write_checkpoint_if_needed在 sampler 或显式 id 场景下写入断点（记录引擎实际遍历过的条目 id），续跑时不会重放与原始运行不同的条目集合。示例 resume_evaluation.py 演示了完整流程：故意在第 12/20 条处崩溃的flaky_task先跑一轮，随后用修复后的healthy_task调用evaluate_resume；返回结果的test_results是续跑后整个实验的完整列表（历史条目由已存分数重建 + 本次新执行的条目），脚本最后校验全部 20 条已完成。注意 task 失败属于“始终中止”的一类，所以续跑用的 task 必须是修复过的版本。

小结

evaluate()把“创建实验 → 解析数据项 → 并发执行任务 → 逐项打分 → 实验级聚合 → 回写并展示”压缩成一次调用，其参数设计覆盖了实验命名（experiment_name/experiment_name_prefix）、数据选择（nb_samples/dataset_item_ids/dataset_sampler/dataset_filter_string）、并发（task_threads）、重复试跑（trial_count）与失败策略（error_tolerance）等评估运行时的关键决策点。想继续深入时可阅读 evaluator.py 中_evaluate_task的实现、evaluation_result.py 的聚合视图，以及同目录文档 evaluate_experiment.rst（对已有实验补打分）与 evaluate_prompt.rst（Prompt 对比评估）。

【免费下载链接】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