用于智能体评估的自定义指标¶

ADK 已支持Python v1.18.0

如果你需要针对特定用例或领域定制的专门指标，且内置选项无法覆盖这些需求，你可以定义自己的自定义指标。

定义自定义指标¶

自定义指标是一个 Python 函数，用于评估智能体在给定评估用例 (Eval Case) 中的表现，并返回一个 EvaluationResult。该函数接收 EvalMetric、智能体在评估运行期间生成的 Invocation 对象列表，以及可选的预期调用列表或在评估用例中定义的 ConversationScenario。

每个 Invocation 对象代表用户与智能体之间的一轮交互，包含该轮交互的工具轨迹、中间响应和最终响应等信息。

你的自定义指标函数必须符合以下签名：

from typing import Optional
from google.adk.evaluation.eval_case import Invocation
from google.adk.evaluation.eval_metrics import EvalMetric
from google.adk.evaluation.conversation_scenarios import ConversationScenario
from google.adk.evaluation.evaluator import EvaluationResult

def my_custom_metric_function(
    eval_metric: EvalMetric,
    actual_invocations: list[Invocation],
    expected_invocations: Optional[list[Invocation]],
    conversation_scenario: Optional[ConversationScenario],
) -> EvaluationResult:
  ...

该函数应返回一个 EvaluationResult 对象，并填充 overall_score（总分）、overall_eval_status（总评估状态）和 per_invocation_results（每次调用结果）字段。

示例¶

下面是一个简单的自定义指标示例，它检查智能体在每一轮中的最终响应是否与预期的最终响应完全匹配。

import statistics
from typing import Optional

from google.adk.evaluation.conversation_scenarios import ConversationScenario
from google.adk.evaluation.eval_case import Invocation
from google.adk.evaluation.eval_metrics import EvalMetric
from google.adk.evaluation.eval_metrics import EvalStatus
from google.adk.evaluation.evaluator import EvaluationResult, PerInvocationResult

def check_final_response_exact_match(
    eval_metric: EvalMetric,
    actual_invocations: list[Invocation],
    expected_invocations: Optional[list[Invocation]],
    conversation_scenario: Optional[ConversationScenario],
) -> EvaluationResult:
  """检查第一轮的最终响应是否与预期响应匹配。"""
  if not expected_invocations:
    return EvaluationResult(overall_score=0.0, overall_eval_status=EvalStatus.NOT_EVALUATED)

  per_invocation_results = []

  for actual, expected in zip(actual_invocations, expected_invocations):
    actual_final_response = "".join([part.text for part in actual.final_response.parts])
    expected_final_response = "".join([part.text for part in expected.final_response.parts])
    score = 1.0 if actual_final_response == expected_final_response else 0.0
    eval_status = EvalStatus.PASSED if score else EvalStatus.FAILED
    invocation_result = PerInvocationResult(
        actual_invocation=actual,
        expected_invocation=expected,
        score=score,
        eval_status=eval_status
    )
    per_invocation_results.append(invocation_result)

  average_score = statistics.mean(result.score for result in per_invocation_results)

  threshold = eval_metric.criterion.threshold
  overall_eval_status = (
    EvalStatus.PASSED if average_score >= threshold else EvalStatus.FAILED
  )
  return EvaluationResult(
      overall_score=average_score,
      overall_eval_status=overall_eval_status,
      per_invocation_results=per_invocation_results,
  )

异步指标¶

如果你的自定义指标需要进行异步调用（例如调用 API），你可以将其定义为 async 函数。

以下是一个自定义指标函数的示例，它使用模拟的异步违规词检查 API 来检查智能体响应是否包含违规词。

import asyncio
import statistics
from typing import Optional

from google.adk.evaluation.conversation_scenarios import ConversationScenario
from google.adk.evaluation.eval_case import Invocation
from google.adk.evaluation.eval_metrics import EvalMetric
from google.adk.evaluation.eval_metrics import EvalStatus
from google.adk.evaluation.evaluator import EvaluationResult, PerInvocationResult

class ProfanityChecker:
  """模拟异步 API 的虚拟违规词分析器。"""

  async def check(self, text: str) -> bool:
    """如果检测到违规词则返回 True，否则返回 False。"""
    await asyncio.sleep(0.01)
    return "profanity" in text.lower()

profanity_checker = ProfanityChecker()

async def check_for_profanity(
    eval_metric: EvalMetric,
    actual_invocations: list[Invocation],
    expected_invocations: Optional[list[Invocation]],
    conversation_scenario: Optional[ConversationScenario],
) -> EvaluationResult:
  """使用模拟异步 API 检查智能体响应是否包含违规词。"""
  per_invocation_results = []

  for invocation in actual_invocations:
    agent_response = "".join(part.text for part in invocation.final_response.parts)
    has_profanity = await profanity_checker.check(agent_response)
    score = 0.0 if has_profanity else 1.0
    eval_status = EvalStatus.FAILED if has_profanity else EvalStatus.PASSED

    invocation_result = PerInvocationResult(
        actual_invocation=invocation,
        score=score,
        eval_status=eval_status
    )
    per_invocation_results.append(invocation_result)

  scores = [
      result.score
      for result in per_invocation_results
      if result.eval_status != EvalStatus.NOT_EVALUATED
  ]

  average_score = statistics.mean(scores)

  threshold = eval_metric.criterion.threshold
  overall_eval_status = (
      EvalStatus.PASSED if average_score >= threshold else EvalStatus.FAILED
  )
  return EvaluationResult(
      overall_score=average_score,
      overall_eval_status=overall_eval_status,
      per_invocation_results=per_invocation_results,
  )

使用自定义指标¶

要在使用 adk eval 的评估运行中使用自定义指标，你需要在 EvalConfig JSON 文件中指定它。

将你的自定义指标添加为评估 Criteria（标准）之一。键是你的指标名称，值是 threshold（阈值）。
在 EvalConfig 中添加一个 custom_metrics 对象。在该对象内部，为每个自定义指标添加一个条目，其中键是指标名称（与 Criteria 中的名称匹配），值是包含 code_config 的对象。
code_config 对象应包含一个 name 字段，其字符串代表指向自定义指标函数的 Python 导入路径，格式为 my.module.my_function。

EvalConfig 示例¶

假设你的 check_final_response_match 函数定义在 my_agent.metrics.py 中，你的 EvalConfig 可能如下所示：

{
  "criteria": {
    "my_check_final_response_exact_match": {
      "threshold": 0.8
    },
    "tool_trajectory_avg_score": {
      "threshold": 1.0
    }
  },
  "custom_metrics": {
    "my_check_final_response_exact_match": {
      "code_config": {
        "name": "my_agent.metrics.check_final_response_exact_match"
      }
    }
  }
}

通过此配置，当你运行 adk eval --config_file_path=<path_to_this_config> 时，ADK 将为每个评估用例执行 check_final_response_exact_match，并检查返回的分数是否 >= 0.8，从而将 response_match 标准标记为通过或失败。

提供指标信息¶

你可以通过在 EvalConfig 的自定义指标定义中添加 MetricInfo 对象，选择性地提供有关自定义指标的元数据，例如其描述和值范围。如果未提供 metric_info，ADK 将使用默认值（min_value=0.0，max_value=1.0）。

ADK 工具可以使用此信息进行显示和结果汇总。

以下是为返回 -1.0 到 1.0 之间分数的自定义指标提供 metric_info 的示例：

{
  "criteria": {
    "my_metric": {
      "threshold": 0.5
    }
  },
  "custom_metrics": {
    "my_metric": {
      "code_config": {
        "name": "my_agent.metrics.my_metric_function"
      },
      "metric_info": {
        "metric_name": "my_metric",
        "description": "此指标评估 XYZ 并返回 -1.0 到 1.0 之间的分数。",
        "metric_value_info": {
          "interval": {
            "min_value": -1.0,
            "max_value": 1.0
          }
        }
      }
    }
  }
}