企业官网建设流程全解析

在 ADK 中实现自定义 Evaluator：从自定义指标函数到 Evaluator 子类的完整指南

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

导读

google.adk.evaluation.evaluator.Evaluator是 ADK 评估体系中所有内置评估指标（共 13 个）背后的统一接口。本文讲解如何为你的 Agent 编写专属评估规则：一种轻量级路径是编写固定四参数签名的自定义指标函数并通过 eval config 中的点路径接入；另一种重量级路径是编写Evaluator子类以支持每次运行前的构造、昂贵的模型加载与自定义 criterion 类型。读完本文，你将掌握两种自定义评估方式的完整写法、底层注册与解析机制、EvaluationResult/PerInvocationResult结果契约，以及那些最容易踩坑的边界行为。

为什么内置指标不够，你需要自定义 Evaluator

ADK 的 13 个内置评估指标回答的是通用问题：Agent 是否调用了记录中预期的工具（tool_trajectory_avg_score）、最终回答是否与金标准答案相似（response_match_score）、裁判模型是否认为回答有依据（final_response_match_v2）等等。但真实 Agent 往往存在只属于它自己的业务规则：

• 温控 Agent 绝不能把温度设在安全区间之外；
• 客服 Agent 绝不能引用没有查询过的价格；
• 预订 Agent 在确认可用性之前绝不能先确认订单。

这些规则在 Python 里用几行代码就能检查，却无法表达成某个通用指标的阈值。ADK 的评估体系因此允许你提供自己的评分逻辑。有两条入口，区别在于你接管多少框架机制：

Evaluator子类之所以必须注册，是因为 config 文件只能命名一个函数，不能命名一个类。因此除非命中上述三种需求之一，优先使用函数：一个函数除了函数本身之外不增加任何成本，而子类多出来的注册步骤必须与评估运行在同一个进程内执行。

无论走哪条路径，运行期真正被调用的对象都是一个Evaluator，它必须产出一个EvaluationResult。源码中Evaluator是一个只声明了criterion_type类变量与evaluate_invocations方法的抽象基类，见 evaluator.py。

快速开始：编写一个自定义指标函数

自定义指标函数接收四个参数并返回EvaluationResult。下面这个函数会让任何把温度设置在安全区间之外的调用失败：

from typing import Optional from google.adk.evaluation.eval_case import ConversationScenario from google.adk.evaluation.eval_case import get_all_tool_calls from google.adk.evaluation.eval_case import Invocation from google.adk.evaluation.eval_metrics import EvalMetric from google.adk.evaluation.evaluator import EvalStatus from google.adk.evaluation.evaluator import EvaluationResult from google.adk.evaluation.evaluator import PerInvocationResult _SAFE_MIN = 18 _SAFE_MAX = 30 def _is_safe(invocation: Invocation) -> bool: for call in get_all_tool_calls(invocation.intermediate_data): if call.name != "set_temperature": continue temperature = (call.args or {}).get("temperature") if temperature is not None and not (_SAFE_MIN <= temperature <= _SAFE_MAX): return False return True def temperature_safety_score( eval_metric: EvalMetric, actual_invocations: list[Invocation], expected_invocations: Optional[list[Invocation]], conversation_scenario: Optional[ConversationScenario], ) -> EvaluationResult: """Scores 1.0 unless a set_temperature call left the safe range.""" per_invocation_results = [] for invocation in actual_invocations: safe = _is_safe(invocation) per_invocation_results.append( PerInvocationResult( actual_invocation=invocation, score=1.0 if safe else 0.0, eval_status=EvalStatus.PASSED if safe else EvalStatus.FAILED, ) ) if not per_invocation_results: return EvaluationResult() overall_score = sum(r.score for r in per_invocation_results) / len( per_invocation_results ) return EvaluationResult( overall_score=overall_score, overall_eval_status=( EvalStatus.PASSED if overall_score == 1.0 else EvalStatus.FAILED ), per_invocation_results=per_invocation_results, )

关键点解析：

• get_all_tool_calls是一个工具函数，从Invocation.intermediate_data中提取按时间顺序排列的工具调用列表。它同时支持两种中间数据结构：IntermediateData（直接读tool_uses字段）和InvocationEvents（逐个事件扫描function_callpart），见 eval_case.py。
• 空调用列表的兜底：当per_invocation_results为空时直接返回EvaluationResult()，即默认状态NOT_EVALUATED，而不是抛除零异常。
• 该示例改编自仓库样例 temperature_safety.py，同一份完整可运行的版本。

在 eval config 中接线

在 eval config 的criteria中命名该指标，并用custom_metrics指向它的点路径：

{ "criteria": { "temperature_safety_score": 1.0 }, "custom_metrics": { "temperature_safety_score": { "code_config": {"name": "temperature_safety.temperature_safety_score"}, "description": "Fails if any set_temperature call is outside 18-30 Celsius." } } }

真实配置见 custom_metric/eval_config.json。criteria中的1.0是阈值简写（criteria决定哪些指标运行），custom_metrics中的code_config.name是点路径（决定指标名如何解析为函数）。该路径的模块部分由importlib导入，因此必须能从评估运行处被 import——本例中temperature_safety.py与 config、eval 数据放在同一目录。

工作原理：从点路径到被调用

当一个评估运行启动时，eval config 会被遍历，custom_metrics下的每个条目都会以点路径为键注册进MetricEvaluatorRegistry。具体注册逻辑在register_custom_metrics_from_config中实现：每个条目被包装为_CustomMetricEvaluator并记录其code_config.name（若未提供metric_info，则自动生成一个值区间为[0.0, 1.0]的默认MetricInfo），见 metric_evaluator_registry.py。

注册目标分两种场景，这正是fork()方法存在的意义（见 metric_evaluator_registry.py）：

• AgentEvaluator（pytest 场景）会把指标注册进进程级默认注册表的副本（fork），从而让一次测试声明的指标不会泄漏到下一次测试；
• adk eval（命令行场景）直接注册进默认注册表本身，这对一次性命令没有问题。

评分时解析点路径：在最后一个点处分割，导入模块并获取属性。模块无法导入、属性不存在、路径中没有点，三种情况都会以同样的方式暴露——ImportError: Could not import custom metric function from <path>；属性存在但不可调用则抛出TypeError。该逻辑见 custom_metric_evaluator.py。

随后你的函数以位置参数方式被调用，恰好四个参数，因此参数的数量重要而名字无关紧要。若返回值是 awaitable，则会被 await——这就是为什么async def指标无需额外声明即可工作（见 custom_metric_evaluator.py）。

关于这次调用，有三件事最容易让人踩坑

1. eval_metric.threshold被有意置为None。你拿到的 metric 的该字段被清空了，因为“把分数与阈值比较”是框架的职责，不是指标的职责。需要时请读取 metric 的名称（eval_metric.metric_name）和它的 criterion，而不要读threshold。置空操作发生在 custom_metric_evaluator.py 的eval_metric.threshold = None。

2. 返回结果的契约比看上去更严格。必须设置overall_eval_status，并且为actual_invocations中的每一个条目按相同顺序返回恰好一个PerInvocationResult。数量不符会抛出ValueError: Eval metric should return results for each invocation.并中止运行——该校验位于 local_eval_service.py。把overall_eval_status留在NOT_EVALUATED默认值上比报错更糟，因为它是静默的：你算出的 per-invocation 结果会被丢弃并替换成空结果，指标最终不报告任何分数。

3. 指标内部的异常不会大声失败。评估服务会捕获异常、记录 traceback，并替换为一个状态为NOT_EVALUATED的空结果，这样单个指标损坏不会拖垮其他指标。该行为在 local_eval_service.py 中实现：except Exception之后构造EvaluationResult(overall_eval_status=EvalStatus.NOT_EVALUATED)并写入日志。在AgentEvaluator下这仍会使测试失败，但你看到的报错是Expected 1.0, but got None.，看不出任何信息。看到这个报错时，去日志里找真正的错误。

结果对象

EvaluationResult是指标返回的对象（基于 pydanticBaseModel，定义见 evaluator.py）：

PerInvocationResult是列表中的一行：

EvalStatus有三个成员：PASSED、FAILED、NOT_EVALUATED（定义见 eval_metrics.py）。

两个层级的裁决如何合并

合并方式取决于谁在驱动评估：

• LocalEvalService（即adk eval的后端）读取状态：只要任一指标报告FAILED，该 case 即为FAILED；至少一个PASSED且没有FAILED则为PASSED；否则为NOT_EVALUATED。
• AgentEvaluator额外做一件事：把每个 invocation 的score跨所有 run 求平均，再与配置的阈值比较，其失败消息正是来自这里。

同时设置分数与状态（如上面的示例那样）能同时满足两者。

高级应用：编写Evaluator子类

当指标需要每次运行一次（而非每次调用一次）的初始化时，就继承Evaluator。构造函数以唯一关键字参数eval_metric=调用，evaluate_invocations可以是同步也可以是异步的：

from google.adk.evaluation.eval_metrics import BaseCriterion from google.adk.evaluation.eval_metrics import EvalMetric from google.adk.evaluation.evaluator import EvaluationResult from google.adk.evaluation.evaluator import Evaluator from google.adk.evaluation.evaluator import EvalStatus from google.adk.evaluation.evaluator import PerInvocationResult class ResponseLengthEvaluator(Evaluator): """Scores 1.0 when the final response stays under a character budget.""" criterion_type = BaseCriterion def __init__(self, eval_metric: EvalMetric): self._threshold = eval_metric.criterion.threshold def evaluate_invocations( self, actual_invocations, expected_invocations=None, conversation_scenario=None, ) -> EvaluationResult: results = [] for invocation in actual_invocations: response = invocation.final_response parts = (response.parts or []) if response else [] length = len("".join(part.text or "" for part in parts)) score = 1.0 if length <= 200 else 0.0 results.append( PerInvocationResult( actual_invocation=invocation, score=score, eval_status=( EvalStatus.PASSED if score else EvalStatus.FAILED ), ) ) overall = sum(r.score for r in results) / len(results) return EvaluationResult( overall_score=overall, overall_eval_status=( EvalStatus.PASSED if overall >= self._threshold else EvalStatus.FAILED ), per_invocation_results=results, )

注册到进程级默认注册表

运行前把它注册到进程级默认注册表——必须是默认注册表：

from google.adk.evaluation.eval_metrics import Interval from google.adk.evaluation.eval_metrics import MetricInfo from google.adk.evaluation.eval_metrics import MetricValueInfo from google.adk.evaluation.metric_evaluator_registry import DEFAULT_METRIC_EVALUATOR_REGISTRY DEFAULT_METRIC_EVALUATOR_REGISTRY.register_evaluator( metric_info=MetricInfo( metric_name="response_length", description="Penalizes over-long final responses.", metric_value_info=MetricValueInfo( interval=Interval(min_value=0.0, max_value=1.0) ), ), evaluator=ResponseLengthEvaluator, )

为什么必须是默认注册表？因为AgentEvaluator运行时会fork该注册表而非新建一个——正是为了让这里注册的类保持可解析。fork 的隔离语义保证了 eval config 中基于函数的指标只对单次运行可见，而你基于类的指标对所有运行可用。注册完成后，该指标只需在criteria中有一条普通记录，不需要custom_metrics条目——注册表已经知道这个名字了。

注册必须在与评估相同的进程中发生，这在实际中意味着程序化运行：pytest 套件里的conftest.py，或调用 eval 服务的脚本。没有任何钩子能让adk eval从 config 文件中加载一个类。

声明你自己的 criterion 类型

criterion_type是一个ClassVar，声明你的 evaluator 期望的 criterion 类。内置 evaluator 都会设置它，然后把传入的 criterion 重新校验成该类型——这正是match_type能到达TrajectoryEvaluator、judge_model_options能到达 judge 类指标的原因。子类化BaseCriterion并加上自己的字段、把criterion_type指向它、在构造函数中做校验，即可获得同样的行为。额外键能存活过 config 解析，是因为BaseCriterion允许 extras（extra="allow"，见 eval_metrics.py），它们已经在等你了。两阶段校验的细节见 eval config 指南：解析阶段按宽松的BaseCriterion放行所有额外键，运行期解析出 evaluator 后才由其criterion_type重新校验。

局限性

• config 只能命名函数。基于类的指标完全无法声明在eval_config.json中；它们需要先于评估运行的 Python 代码。
• 注册是进程全局的。DEFAULT_METRIC_EVALUATOR_REGISTRY是模块级单例，重复注册同一指标名会替换首次注册并记录日志（register_evaluator会打印 "Updating Evaluator class for ..."）。任何注册表的构造都会发出实验性功能警告——MetricEvaluatorRegistry带有@experimental标记（见 metric_evaluator_registry.py）。
• 包级别不做任何再导出。请直接从google.adk.evaluation.evaluator和google.adk.evaluation.metric_evaluator_registry导入；包的__init__只导出AgentEvaluator。
• 错误被吞掉。指标抛出异常会降级为NOT_EVALUATED并仅记录一条日志，而不是把异常抛给调用方。
• 运行评估需要安装额外依赖。解析 config 在基础安装下即可工作，但指标注册表会经由google-cloud-aiplatform[evaluation]引入vertexai。运行评估前请安装google-adk[eval]。

相关示例与指南

• custom_metric 样例 是上文示例所改编的完整可运行函数，配套的 eval_config.json 展示了接线方式。整个 evaluation 样例集 复用同一个 home-automation Agent，逐一对展示确定性指标、自定义指标、LLM-as-a-judge、rubrics 与用户模拟五种评估技巧，便于横向对比。
• EvalConfig 与 eval config 文件 讲解指标名与其 criterion 如何抵达注册表，包括criteria/custom_metrics/user_simulator_config/live_model_config四个顶层键与两阶段校验机制。
• BaseEvalService 与 LocalEvalService 是构造并调用你的 evaluator 的服务，也是把抛异常的指标转成空NOT_EVALUATED结果的地方。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python