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(nameMy 42 dataset) results evaluate( experiment_nameMy experiment, datasetdataset, taskllm_task, # 见下文“task 函数” nb_samples2, scoring_metrics[IsJson(), Hallucination()], )完整参数表以下为evaluate()签名中的全部参数源码签名见 evaluator.py参数类型 / 默认值说明datasetDataset或DatasetVersion必填。Opik 数据集实例或其某个版本快照。传入TestSuite时会向后兼容地取其底层 datasettaskLLMTask即Callable[[Dict], Dict]必填。接收数据集条目内容的 dict返回将被评分的 dictscoring_metricsList[BaseMetric]默认None评估指标列表每个指标有score(...)方法所需键名从任务输出中取scoring_functionsList[ScorerFunction]默认None打分函数列表无需scoring_key_mapping用保留参数接收入参experiment_namestr默认None实验名为None时自动生成experiment_name_prefixstr默认None自动命名实验的前缀如my-experiment-随机后缀project_namestr默认None已弃用若数据集本身设置了project_name则始终优先使用数据集的值并打印警告否则 trace/span 记到该 project缺省为Default Projectexperiment_configDict[str, Any]默认None描述实验参数的字典随实验一起存到后端verboseint默认10无输出1输出摘要与 tqdm 进度条默认2额外打印详细分数统计nb_samplesint默认None评估的样本数不提供则评估全部条目task_threadsint默认16并发执行任务的线程数设为1时在当前线程顺序执行。任务对象需支持跨线程共享prompt/promptsBasePrompt/List[BasePrompt]与实验关联的 Prompt 对象prompt已弃用应使用promptsscoring_key_mappingDict[str, Union[str, Callable]]默认None将数据集条目或任务输出中的键重命名为指标期望的键值也可以是 callabledataset_item_idsList[str]默认None只评估指定 id 的数据条目dataset_samplerBaseDatasetSampler默认None采样器实例用于抽样数据条目trial_countint默认1每个数据集条目执行任务并评分的次数experiment_scoring_functionsList[ExperimentScoreFunction]默认None实验级打分函数接收全部TestResult列表返回实验级ScoreResultexperiment_tagsList[str]默认None实验标签dataset_filter_stringstr默认NoneOQL 过滤字符串按 tags、data 字段、metadata 等过滤数据条目blueprint_idstr默认None蓝图 id其配置会合并进experiment_configerror_toleranceErrorTolerance或等值 int默认ErrorTolerance.METRIC_ERRORS失败容错级别见下文专节返回值为 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/spantrack() def llm_task(item: Dict[str, Any]) - Dict[str, Any]: response openai_client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: item[input][message]}], ) return { output: response.choices[0].message.content, reference: test, }scoring_metrics 与 scoring_functions 两种打分方式二选一即可。scoring_metrics每个指标有score(...)方法其形参名必须能从“数据集条目内容 任务输出”经scoring_key_mapping重命名后中解析到。比如Equals()需要output和reference两个键。scoring_functions自定义打分函数按保留参数从引擎接收数据无需scoring_key_mapping。其协议ScorerFunctionProtocol定义在 scorer_function.pydef my_scorer( dataset_item: Dict[str, Any], # 数据集条目内容 task_outputs: Dict[str, Any], # 任务输出 task_spanNone, # 可选任务执行期间收集的 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实验命名与项目解析_use_or_create_experiment_name在experiment_name为空时生成带前缀的随机名helpers.resolve_project_name按“数据集project_name优先、用户参数兜底”解析目标项目。创建实验client.create_experiment(...)带上experiment_config若给了blueprint_id会先通过merge_blueprint_into_config合并蓝图配置、关联的 prompts、tags 与数据集版本 id。解析数据项helpers.resolve_dataset_items根据nb_samples/dataset_item_ids/dataset_sampler/dataset_filter_string解析出条目迭代器随后_materialize_for_checkpoint按三种情况处理——使用 sampler 时先物化列表以便断点续跑记录精确条目 id仅显式指定 id 时保持惰性流式两者皆无时直接流式处理且不写断点。包装打分函数若提供了scoring_functions_wrap_scoring_functions将其包装成指标形态。引擎执行_evaluate_task构造ExecutionPolicyruns_per_itemtrial_count并实例化EvaluationEngine并发度即task_threads调用run_and_score完成“跑任务 打分”主循环。实验级打分compute_experiment_scores在全部TestResult收集完毕后逐个执行experiment_scoring_functions单个函数抛异常只告警不中断见 evaluation_result.py。展示与回写verbose 1时打印摘要并给出实验 URLclient.flush()后调用finish_experiments通知后端实验完成实验级分数经experiment.log_experiment_scores写回verbose 2时额外打印分数统计report.display_evaluation_scores_statistics。一个细节实验完成通知被包在_try_notifying_about_experiment_completion的 try/except 中通知失败只记录 debug 日志、不影响评估结果本身——说明实验 URL 与完成通知都属于“尽力而为”的收尾动作。结果对象 EvaluationResultEvaluationResultevaluation_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(nameCustom metric, valuemax(scores))这正是ExperimentScoreFunction契约Callable[[List[TestResult]], Union[ScoreResult, List[ScoreResult]]]的完整示范。选择评估哪些数据项evaluate()提供四组互有优先级的筛选参数nb_samples只取前 N 条适合冒烟验证dataset_item_ids精确指定条目 id 列表dataset_sampler采样器实例。SDK 内置RandomDatasetSamplerrandom_dataset_sampler.py基类BaseDatasetSampler是抽象类可自行扩展采样策略dataset_filter_stringOQL 过滤字符串支持按tagscontains运算符、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枚举成员或其等值 intIntEnum非法值抛ValueError定义见 types.pyErrorTolerance.METRIC_ERRORS10默认score方法内部抛出的异常被记录为“失败的打分结果”scoring_failedTrue评估继续但在进入score之前发生的失败——数据集缺少指标必需的键、条目级评估器无法构建——会使整个运行中止。ErrorTolerance.ALL_SCORING_ERRORS20额外容忍“导致某指标完全无法打分”的错误——必需的打分参数缺失、条目级评估器构建失败。此时拿到的是带失败记录的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( nameMy 42 dataset, descriptionFor storing stuff ) dataset.insert_from_json( json_arrayjson.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( modelgpt-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( nameCustom metric, valuemax(scores) if len(scores) 1 else 0.0, ) ] results evaluate( experiment_nameMy experiment, datasetdataset, taskllm_task, nb_samples2, scoring_metrics[IsJson()], experiment_scoring_functions[compute_hallucination_stats], verbose2, ) print(results.experiment_url) aggregated results.aggregate_evaluation_scores() print(aggregated.aggregated_scores)运行前提已配置 Opik 客户端连接Opik()会读取默认/环境变量配置且track_openai等集成已导入并初始化示例中省略。verbose2会在结尾打印每个分数的详细统计。中断续跑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.rstPrompt 对比评估。【免费下载链接】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），仅供参考