15 KiB
15 KiB
第08章：智能体评估
📌 本章目标
- 理解 Agent 评估的重要性和方法
- 掌握评估集（Eval Set）的创建方法
- 学习使用
adk eval命令进行自动化评估 - 了解评估指标和结果分析
- 掌握持续评估的最佳实践
8.1 评估概述
8.1.1 为什么需要评估？
Agent 的行为是非确定性的（由 LLM 驱动），同样的输入可能产生不同的输出。因此，需要系统化的评估来确保：
| 目标 | 说明 |
|---|---|
| 正确性 | Agent 是否给出了正确的回答 |
| 工具使用 | Agent 是否正确地使用了工具 |
| 指令遵循 | Agent 是否遵循了系统指令 |
| 安全性 | Agent 是否避免了不当输出 |
| 一致性 | Agent 在不同场景下表现是否一致 |
8.1.2 ADK 评估方法
ADK 提供了两种评估维度：
┌────────────────────────────────────────────┐
│ ADK 评估体系 │
│ │
│ ┌──────────────────────────────────────┐ │
│ │ 最终响应评估（Response Evaluation） │ │
│ │ - 评估 Agent 的最终输出质量 │ │
│ │ - 准确性、完整性、格式等 │ │
│ └──────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────┐ │
│ │ 执行轨迹评估（Trajectory Evaluation）│ │
│ │ - 评估 Agent 的执行过程 │ │
│ │ - 工具调用顺序、决策逻辑等 │ │
│ └──────────────────────────────────────┘ │
└────────────────────────────────────────────┘
8.2 创建评估集
8.2.1 评估集格式
评估集是一个 JSON 文件，包含多个测试用例：
{
"eval_cases": [
{
"case_id": "test_001",
"description": "测试天气查询功能",
"user_input": "北京今天天气怎么样？",
"expected_function_calls": [
{
"name": "get_weather",
"args": {
"city": "北京"
}
}
],
"expected_keywords": ["北京", "天气"]
},
{
"case_id": "test_002",
"description": "测试时间查询功能",
"user_input": "现在几点了？",
"expected_function_calls": [
{
"name": "get_current_time",
"args": {
"city": "北京"
}
}
]
}
]
}
8.2.2 创建评估集文件
"""
创建评估集文件
使用 Python 代码生成 .evalset.json 文件
"""
import json # 导入 JSON 模块
def create_eval_set():
"""创建评估集"""
# 定义评估用例列表
eval_cases = [ # 评估用例列表
{
# 用例标识
"case_id": "weather_beijing", # 用例唯一 ID
# 用例描述
"description": "测试北京天气查询", # 用例说明
# 用户输入
"user_input": "北京今天天气怎么样？", # 模拟用户提问
# 期望的工具调用（可选）
"expected_function_calls": [ # 期望 Agent 调用的工具
{
"name": "get_weather", # 工具名称
"args": { # 期望的参数
"city": "北京",
},
},
],
# 期望的关键词（可选）
"expected_keywords": ["北京", "天气"], # 回复中应包含的关键词
},
{
"case_id": "weather_shanghai", # 用例 ID
"description": "测试上海天气查询", # 描述
"user_input": "帮我查一下上海的天气", # 用户输入
"expected_function_calls": [ # 期望工具调用
{
"name": "get_weather",
"args": {"city": "上海"},
},
],
},
{
"case_id": "greeting", # 用例 ID
"description": "测试问候功能", # 描述
"user_input": "你好！", # 用户输入
"expected_keywords": ["你好"], # 期望关键词
"not_expected_keywords": ["工具", "错误"], # 不应出现的关键词
},
{
"case_id": "unknown_city", # 用例 ID
"description": "测试未知城市处理", # 描述
"user_input": "查询一下月球基地的天气", # 用户输入
"expected_keywords": ["找不到", "不支持", "无法"], # 应包含的错误提示
},
]
# 构建评估集对象
eval_set = { # 评估集
"eval_cases": eval_cases, # 用例列表
}
# 写入 JSON 文件
with open("my_agent_eval_set.evalset.json", "w", encoding="utf-8") as f: # 打开文件
json.dump( # 写入 JSON
eval_set, # 评估集数据
f, # 文件对象
ensure_ascii=False, # 允许中文
indent=2, # 缩进格式化
)
print("评估集已创建: my_agent_eval_set.evalset.json") # 打印成功信息
# 执行创建
create_eval_set() # 调用函数
8.3 运行评估
8.3.1 使用 CLI 运行评估
# 基本评估命令
adk eval my_agent my_agent_eval_set.evalset.json
# 指定输出格式
adk eval my_agent my_agent_eval_set.evalset.json --output-format json
# 指定输出文件
adk eval my_agent my_agent_eval_set.evalset.json --output-file results.json
8.3.2 评估输出示例
{
"summary": {
"total_cases": 4,
"passed": 3,
"failed": 1,
"pass_rate": 0.75
},
"results": [
{
"case_id": "weather_beijing",
"status": "passed",
"response": "北京今天天气晴朗，气温25°C。",
"function_calls_matched": true
},
{
"case_id": "weather_shanghai",
"status": "passed",
"response": "上海今天多云，气温28°C。",
"function_calls_matched": true
},
{
"case_id": "greeting",
"status": "passed",
"response": "你好！有什么可以帮助你的吗？"
},
{
"case_id": "unknown_city",
"status": "failed",
"response": "月球基地的天气信息暂时无法获取。",
"reason": "Missing expected keywords: ['找不到', '不支持', '无法']"
}
]
}
8.4 代码方式评估
"""
通过代码方式运行评估
适用于集成到 CI/CD 流水线
"""
from google.adk.agents import Agent # 导入 Agent 类
from google.adk.runners import Runner # 导入运行器
from google.adk.sessions import InMemorySessionService # 导入会话服务
from google.genai import types # 导入类型定义
import asyncio # 导入异步库
import json # 导入 JSON 模块
# ========================================
# 定义 Agent 和工具
# ========================================
def get_weather(city: str) -> dict:
"""获取天气信息"""
weather_data = { # 天气数据
"北京": {"temp": "25°C", "condition": "晴天"},
"上海": {"temp": "28°C", "condition": "多云"},
}
data = weather_data.get(city) # 查找数据
if not data: # 如果找不到
return {"status": "error", "error_message": f"未找到'{city}'的天气信息"}
return {"status": "success", "city": city, **data} # 返回天气
agent = Agent(
name="weather_agent", # Agent 名称
model="gemini-2.0-flash", # 模型
instruction="你是天气助手，使用 get_weather 工具查询天气。", # 指令
tools=[get_weather], # 工具
)
# ========================================
# 定义评估逻辑
# ========================================
async def evaluate_agent(test_cases: list) -> dict:
"""
评估 Agent
Args:
test_cases: 测试用例列表
Returns:
dict: 评估结果
"""
session_service = InMemorySessionService() # 会话服务
runner = Runner( # 运行器
agent=agent, # Agent
app_name="eval_app", # 应用名称
session_service=session_service, # 会话服务
)
results = [] # 结果列表
passed = 0 # 通过计数
for i, case in enumerate(test_cases): # 遍历测试用例
# 创建独立会话
session_id = f"eval_session_{i}" # 会话 ID
await session_service.create_session( # 创建会话
app_name="eval_app", # 应用名称
user_id="eval_user", # 用户 ID
session_id=session_id, # 会话 ID
)
# 构造用户消息
content = types.Content(
role='user', # 角色
parts=[types.Part(text=case["user_input"])], # 内容
)
# 运行 Agent
events = runner.run_async(
user_id="eval_user", # 用户 ID
session_id=session_id, # 会话 ID
new_message=content, # 消息
)
# 收集响应
response_text = "" # 初始化响应文本
async for event in events: # 遍历事件
if event.is_final_response(): # 最终响应
response_text = event.content.parts[0].text # 提取文本
# 检查是否通过
case_passed = True # 默认通过
failure_reason = "" # 失败原因
# 检查期望关键词
if "expected_keywords" in case: # 如果有关键词要求
for keyword in case["expected_keywords"]: # 遍历关键词
if keyword not in response_text: # 如果关键词不存在
case_passed = False # 标记为失败
failure_reason = f"缺少关键词: '{keyword}'" # 记录原因
break # 跳出循环
if case_passed: # 如果通过
passed += 1 # 递增通过计数
# 记录结果
results.append({ # 添加结果
"case_id": case.get("case_id", f"case_{i}"), # 用例 ID
"status": "passed" if case_passed else "failed", # 状态
"response": response_text[:200], # 响应（截断）
"reason": failure_reason, # 失败原因
})
# 返回汇总结果
return { # 返回评估结果
"total": len(test_cases), # 总用例数
"passed": passed, # 通过数
"failed": len(test_cases) - passed, # 失败数
"pass_rate": passed / len(test_cases), # 通过率
"results": results, # 详细结果
}
# ========================================
# 运行评估
# ========================================
async def main():
"""主函数"""
# 定义测试用例
test_cases = [ # 测试用例列表
{
"case_id": "test_001", # 用例 ID
"user_input": "北京天气怎么样？", # 用户输入
"expected_keywords": ["北京", "天气"], # 期望关键词
},
{
"case_id": "test_002",
"user_input": "上海天气如何？",
"expected_keywords": ["上海"],
},
{
"case_id": "test_003",
"user_input": "你好",
"expected_keywords": ["你好"],
},
]
# 执行评估
result = await evaluate_agent(test_cases) # 运行评估
# 打印结果
print("=" * 50) # 分隔线
print("评估结果") # 标题
print("=" * 50) # 分隔线
print(f"总用例: {result['total']}") # 总数
print(f"通过: {result['passed']}") # 通过数
print(f"失败: {result['failed']}") # 失败数
print(f"通过率: {result['pass_rate']:.1%}") # 通过率
print("-" * 50) # 分隔线
for r in result["results"]: # 遍历详细结果
status_icon = "✅" if r["status"] == "passed" else "❌" # 状态图标
print(f"{status_icon} {r['case_id']}: {r['status']}") # 打印结果
if r["reason"]: # 如果有失败原因
print(f" 原因: {r['reason']}") # 打印原因
asyncio.run(main()) # 执行主函数
8.5 评估最佳实践
8.5.1 评估用例设计原则
| 原则 | 说明 | 示例 |
|---|---|---|
| 覆盖核心功能 | 每个工具至少一个正向用例 | 天气查询 → 查询有效城市 |
| 边界情况 | 测试异常输入的处理 | 查询不存在的城市 |
| 多轮对话 | 测试上下文记忆能力 | 先说名字，再问名字 |
| 安全测试 | 测试不当请求的处理 | 注入攻击、敏感信息 |
| 回归测试 | 修复 bug 后添加用例 | 之前出错的场景 |
8.5.2 持续评估流程
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 开发 Agent │───▶│ 编写评估集 │───▶│ 运行评估 │───▶│ 分析结果 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘
▲ │
│ │
└───────────────────────────────────────────────┘
根据结果优化 Agent
📌 本章小结
- Agent 评估是确保质量的关键环节
- 评估集是 JSON 格式的测试用例集合
- 使用
adk eval命令或代码方式运行评估 - 评估维度包括最终响应和执行轨迹
- 好的评估集应覆盖核心功能、边界情况和安全场景
下一章：第09章 - 部署指南 → 学习如何将 Agent 部署到生产环境。