""" Google ADK 智能体评估完整示例 展示评估集创建和评估运行方法 对应教程：第08章 - 智能体评估 """ # 导入 ADK 核心模块 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: """ 获取天气信息 Args: city (str): 城市名称 Returns: dict: 天气信息 """ weather_data = { # 天气数据 "北京": {"temp": "25°C", "condition": "晴天"}, "上海": {"temp": "28°C", "condition": "多云"}, "广州": {"temp": "32°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], # 工具 ) # ======================================== # 示例一：创建评估集 # ======================================== def create_eval_set(): """ 创建评估集文件 生成 .evalset.json 文件供 adk eval 使用 """ # 定义评估用例 eval_cases = [ # 用例列表 { "case_id": "weather_beijing", # 用例 ID "description": "测试北京天气查询", # 描述 "user_input": "北京今天天气怎么样？", # 用户输入 "expected_keywords": ["北京", "天气"], # 期望关键词 }, { "case_id": "weather_shanghai", # 用例 ID "description": "测试上海天气查询", # 描述 "user_input": "帮我查一下上海的天气", # 用户输入 "expected_keywords": ["上海"], # 期望关键词 }, { "case_id": "greeting", # 用例 ID "description": "测试问候功能", # 描述 "user_input": "你好！", # 用户输入 "expected_keywords": ["你好"], # 期望关键词 }, { "case_id": "unknown_city", # 用例 ID "description": "测试未知城市处理", # 描述 "user_input": "查询月球基地的天气", # 用户输入 "expected_keywords": ["找不到", "无法", "不支持"], # 期望关键词 }, { "case_id": "multiple_cities", # 用例 ID "description": "测试多城市查询", # 描述 "user_input": "北京和上海的天气对比", # 用户输入 "expected_keywords": ["北京", "上海"], # 期望关键词 }, ] # 构建评估集 eval_set = { # 评估集对象 "eval_cases": eval_cases, # 用例列表 } # 写入文件 filepath = "weather_agent_eval_set.evalset.json" # 文件路径 with open(filepath, "w", encoding="utf-8") as f: # 打开文件 json.dump( # 写入 JSON eval_set, # 数据 f, # 文件对象 ensure_ascii=False, # 允许中文 indent=2, # 格式化 ) print(f"✅ 评估集已创建: {filepath}") # 打印成功信息 print(f" 用例数量: {len(eval_cases)}") # 打印用例数 return filepath # 返回文件路径 # ======================================== # 示例二：代码方式运行评估 # ======================================== 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): # 遍历用例 case_id = case.get("case_id", f"case_{i}") # 用例 ID print(f"\n📋 测试用例: {case_id}") # 打印用例 ID print(f" 输入: {case['user_input']}") # 打印输入 # 创建独立会话 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: # 如果有关键词要求 missing = [] # 缺失的关键词 for kw in case["expected_keywords"]: # 遍历关键词 if kw not in response_text: # 如果缺失 missing.append(kw) # 记录缺失 if missing: # 如果有缺失 case_passed = False # 标记失败 failure_reason = f"缺少关键词: {missing}" # 失败原因 # 检查不应出现的关键词 if "not_expected_keywords" in case: # 如果有排除关键词 for kw in case["not_expected_keywords"]: # 遍历 if kw in response_text: # 如果出现 case_passed = False # 标记失败 failure_reason = f"不应包含关键词: '{kw}'" # 失败原因 if case_passed: # 如果通过 passed += 1 # 递增通过数 print(f" ✅ 通过") # 打印通过 else: # 如果失败 print(f" ❌ 失败: {failure_reason}") # 打印失败原因 # 记录结果 results.append({ # 添加结果 "case_id": case_id, # 用例 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) if test_cases else 0, # 通过率 "results": results, # 详细结果 } # ======================================== # 示例三：打印评估报告 # ======================================== def print_report(result: dict): """ 打印评估报告 Args: result: 评估结果 """ print("\n" + "=" * 60) # 分隔线 print("📊 评估报告") # 标题 print("=" * 60) # 分隔线 print(f"总用例数: {result['total']}") # 总数 print(f"通过: {result['passed']} ✅") # 通过数 print(f"失败: {result['failed']} ❌") # 失败数 print(f"通过率: {result['pass_rate']:.1%}") # 通过率 print("-" * 60) # 分隔线 for r in result["results"]: # 遍历详细结果 icon = "✅" if r["status"] == "passed" else "❌" # 状态图标 print(f" {icon} {r['case_id']}: {r['status']}") # 打印结果 if r["reason"]: # 如果有失败原因 print(f" 原因: {r['reason']}") # 打印原因 if r["status"] == "passed": # 如果通过 print(f" 响应: {r['response'][:100]}...") # 打印响应片段 print("=" * 60) # 分隔线 # ======================================== # 主函数 # ======================================== async def main(): """主函数""" # 第一步：创建评估集 filepath = create_eval_set() # 创建评估集文件 # 第二步：加载评估用例 with open(filepath, "r", encoding="utf-8") as f: # 读取文件 eval_set = json.load(f) # 解析 JSON test_cases = eval_set["eval_cases"] # 获取用例列表 # 第三步：运行评估 result = await evaluate_agent(test_cases) # 执行评估 # 第四步：打印报告 print_report(result) # 打印评估报告 if __name__ == "__main__": # 直接运行 asyncio.run(main()) # 执行主函数