Agent 评估与测试
评估是 AI Agent 开发中至关重要但常被忽视的环节。与传统的软件测试不同,Agent 的输出具有不确定性,需要专门的评估框架和方法。本章将介绍 Agent 评估的核心概念、工具和最佳实践。
为什么需要 Agent 评估
AI Agent 的输出具有以下特点,使得评估变得复杂:
- 不确定性:同样的输入可能产生不同的输出
- 多步骤执行:Agent 可能经过多轮工具调用才完成任务
- 上下文依赖:输出质量受对话历史影响
- 主观性:某些任务的评判标准难以量化
没有系统的评估,我们无法:
- 判断 Agent 是否真正解决了问题
- 比较不同模型或提示词的效果
- 发现回归问题
- 持续改进 Agent 性能
评估类型
离线评估 vs 在线评估
离线评估在开发阶段进行,使用预先准备的测试数据集:
- 作为 Agent 的"单元测试"
- 在部署前捕获回归问题
- 比较不同版本的性能
在线评估在生产环境实时进行:
- 监控真实用户交互的质量
- 检测异常和性能下降
- 收集反馈用于模型改进
评估维度
Agent 评估通常关注以下维度:
| 维度 | 说明 | 示例指标 |
|---|---|---|
| 正确性 | 输出是否准确回答了问题 | 准确率、F1 分数 |
| 相关性 | 输出是否与问题相关 | 相关性评分 |
| 一致性 | 相似输入是否产生一致输出 | 一致性比率 |
| 效率 | 完成任务所需的资源 | Token 消耗、响应时间 |
| 安全性 | 输出是否安全合规 | 安全违规率 |
| 用户体验 | 用户满意度 | 反馈评分、完成率 |
LangSmith 评估框架
LangSmith 是 LangChain 官方的评估和可观测性平台,提供了完整的 Agent 评估解决方案。
基本设置
import os
# 配置 LangSmith
os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent-project"
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""搜索信息"""
return f"搜索结果:{query}"
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_agent(
model=llm,
tools=[search],
system_prompt="你是一个智能助手。"
)
创建评估数据集
from langsmith import Client
client = Client()
# 创建数据集
dataset = client.create_dataset(
dataset_name="agent-eval-dataset",
description="Agent 评估测试数据集"
)
# 添加测试用例
test_cases = [
{
"input": "北京今天天气怎么样?",
"expected_output": "包含天气信息"
},
{
"input": "帮我计算 123 * 456",
"expected_output": "56088"
},
{
"input": "搜索 Python 教程",
"expected_output": "包含 Python 教程相关内容"
}
]
for case in test_cases:
client.create_example(
inputs={"question": case["input"]},
outputs={"answer": case["expected_output"]},
dataset_id=dataset.id
)
定义评估器
LangSmith 支持多种评估器类型:
1. LLM-as-Judge 评估器
使用 LLM 来评判输出质量:
from langchain.evaluation import load_evaluator
from langchain_openai import ChatOpenAI
# 创建评估 LLM
eval_llm = ChatOpenAI(model="gpt-4", temperature=0)
# 加载预定义的评估器
correctness_evaluator = load_evaluator(
"labeled_score_string",
criteria={
"accuracy": "输出是否准确回答了问题?",
"completeness": "输出是否完整地回答了问题?",
"clarity": "输出是否清晰易懂?"
},
llm=eval_llm
)
# 自定义评估提示词
custom_criteria = """
请评估 Agent 的回答质量:
1. 准确性(1-5分):回答是否准确无误?
2. 完整性(1-5分):回答是否完整?
3. 有用性(1-5分):回答对用户是否有帮助?
请给出总分和具体评分理由。
"""
quality_evaluator = load_evaluator(
"score_string",
criteria=custom_criteria,
llm=eval_llm
)
2. 启发式评估器
基于规则的评估,适用于可程序化判断的场景:
from langchain.evaluation import EvaluatorType
from langchain.evaluation import load_evaluator
# 精确匹配评估器
exact_match = load_evaluator(EvaluatorType.EXACT_MATCH)
# 包含评估器
contains_evaluator = load_evaluator(
EvaluatorType.CONTAINS,
ignore_case=True,
ignore_punctuation=True
)
# 正则表达式评估器
regex_evaluator = load_evaluator(
EvaluatorType.REGEX_MATCH,
pattern=r"\d+" # 匹配数字
)
# JSON 有效性评估器
json_evaluator = load_evaluator(EvaluatorType.JSON_VALIDITY)
3. 自定义评估器
针对特定业务需求的自定义评估:
from langchain.evaluation import StringEvaluator
from typing import Any, Optional
class AgentQualityEvaluator(StringEvaluator):
"""Agent 质量评估器"""
@property
def requires_input(self) -> bool:
return True
@property
def requires_reference(self) -> bool:
return False
def _evaluate_strings(
self,
prediction: str,
input: Optional[str] = None,
reference: Optional[str] = None,
**kwargs: Any
) -> dict:
"""执行评估"""
scores = {}
# 检查响应长度
scores["length_score"] = min(len(prediction) / 500, 1.0)
# 检查是否包含关键信息
if input and "天气" in input:
scores["contains_weather"] = "°C" in prediction or "温度" in prediction
if input and "计算" in input:
import re
numbers = re.findall(r'\d+', prediction)
scores["contains_number"] = len(numbers) > 0
# 检查是否有错误信息
scores["has_error"] = "错误" in prediction or "失败" in prediction
# 计算总分
total_score = sum(v for v in scores.values() if isinstance(v, (int, float, bool)))
scores["total_score"] = total_score / len(scores)
return {
"score": scores["total_score"],
"reasoning": str(scores)
}
运行评估
from langchain.smith import RunEvalConfig, run_on_dataset
# 配置评估
eval_config = RunEvalConfig(
evaluators=[
# 预定义评估器
"exact_match",
"contains",
# 自定义评估器
AgentQualityEvaluator(),
],
custom_evaluators=[
AgentQualityEvaluator()
],
# 使用 LLM 作为裁判
eval_llm=ChatOpenAI(model="gpt-4", temperature=0)
)
# 在数据集上运行评估
results = run_on_dataset(
client=client,
dataset_name="agent-eval-dataset",
llm_or_chain_factory=lambda: agent,
evaluation=eval_config,
verbose=True
)
# 查看结果
print(results)
Agent 轨迹评估
Agent 的评估不仅要看最终输出,还需要评估整个执行轨迹:
from langchain.evaluation import load_evaluator
# 评估 Agent 轨迹
trajectory_evaluator = load_evaluator(
"criteria",
criteria={
"tool_selection": "Agent 是否选择了正确的工具?",
"tool_usage": "Agent 是否正确使用了工具参数?",
"reasoning": "Agent 的推理过程是否合理?",
"efficiency": "Agent 是否以最少的步骤完成任务?"
}
)
def evaluate_trajectory(run):
"""评估 Agent 执行轨迹"""
# 获取执行步骤
steps = run.child_runs
evaluation_results = []
for i, step in enumerate(steps):
if step.run_type == "tool":
result = {
"step": i,
"tool": step.name,
"input": step.inputs,
"output": step.outputs,
"success": step.error is None
}
evaluation_results.append(result)
return evaluation_results
单元测试
除了使用 LangSmith,也可以使用传统的测试框架:
import pytest
from unittest.mock import patch, MagicMock
class TestAgent:
"""Agent 单元测试"""
@pytest.fixture
def agent(self):
"""创建测试用的 Agent"""
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
return create_agent(model=llm, tools=[...])
def test_weather_query(self, agent):
"""测试天气查询"""
result = agent.invoke({
"messages": [{"role": "user", "content": "北京天气"}]
})
response = result["messages"][-1].content
# 检查响应包含天气相关信息
assert "温度" in response or "°C" in response
def test_calculation(self, agent):
"""测试计算功能"""
result = agent.invoke({
"messages": [{"role": "user", "content": "计算 123 * 456"}]
})
response = result["messages"][-1].content
# 检查响应包含正确结果
assert "56088" in response
def test_error_handling(self, agent):
"""测试错误处理"""
result = agent.invoke({
"messages": [{"role": "user", "content": "做一个不可能的任务"}]
})
response = result["messages"][-1].content
# Agent 应该优雅地处理无法完成的任务
assert response is not None
assert len(response) > 0
@patch('langchain_openai.ChatOpenAI.invoke')
def test_with_mock(self, mock_invoke, agent):
"""使用 Mock 测试"""
mock_invoke.return_value = MagicMock(
content="这是一个模拟的响应"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "测试"}]
})
assert result is not None
基准测试
AgentBench
AgentBench 是一个多维度 Agent 评估基准,涵盖:
- 推理能力:逻辑推理、数学问题
- 工具使用:API 调用、代码执行
- 多轮对话:上下文理解、任务延续
# AgentBench 评估示例
def run_agentbench(agent):
"""运行 AgentBench 基准测试"""
tasks = [
{"type": "reasoning", "question": "..."},
{"type": "tool_use", "task": "..."},
{"type": "dialogue", "history": [...]},
]
results = []
for task in tasks:
result = agent.invoke({"messages": [task]})
results.append({
"task": task,
"result": result,
"score": evaluate_result(result)
})
return results
其他基准
- WebShop:电商购物场景评估
- ALFWorld:家庭环境任务评估
- InterCode:代码执行能力评估
- OSWorld:操作系统交互评估
在线评估与监控
实时监控
在生产环境中持续监控 Agent 性能:
from langsmith import Client
client = Client()
def monitor_agent_performance():
"""监控 Agent 性能"""
# 获取最近的运行记录
runs = client.list_runs(
project_name="my-agent-project",
run_type="chain",
limit=100
)
metrics = {
"total_runs": 0,
"successful_runs": 0,
"failed_runs": 0,
"avg_latency": 0,
"avg_tokens": 0
}
latencies = []
tokens = []
for run in runs:
metrics["total_runs"] += 1
if run.error:
metrics["failed_runs"] += 1
else:
metrics["successful_runs"] += 1
if run.latency:
latencies.append(run.latency)
if run.total_tokens:
tokens.append(run.total_tokens)
if latencies:
metrics["avg_latency"] = sum(latencies) / len(latencies)
if tokens:
metrics["avg_tokens"] = sum(tokens) / len(tokens)
metrics["success_rate"] = metrics["successful_runs"] / metrics["total_runs"]
return metrics
用户反馈收集
def collect_feedback(run_id: str, user_rating: int, comment: str = None):
"""收集用户反馈"""
client = Client()
# 创建反馈记录
client.create_feedback(
run_id=run_id,
key="user_rating",
score=user_rating, # 1-5 分
comment=comment
)
# 在应用中集成
def handle_user_response(run_id: str, user_feedback: dict):
"""处理用户反馈"""
if user_feedback.get("thumbs_up"):
collect_feedback(run_id, 5, user_feedback.get("comment"))
elif user_feedback.get("thumbs_down"):
collect_feedback(run_id, 1, user_feedback.get("comment"))
评估最佳实践
1. 建立基线
在开始优化前,建立性能基线:
def establish_baseline(agent, test_cases):
"""建立性能基线"""
results = []
for case in test_cases:
result = agent.invoke({"messages": [{"role": "user", "content": case["input"]}]})
results.append({
"input": case["input"],
"output": result["messages"][-1].content,
"expected": case["expected"]
})
# 计算基线指标
baseline = {
"total": len(results),
"accuracy": calculate_accuracy(results),
"avg_latency": calculate_avg_latency(results)
}
return baseline
2. 增量评估
每次修改后重新评估,确保没有回归:
def compare_versions(agent_v1, agent_v2, test_cases):
"""比较两个版本的性能"""
results_v1 = evaluate_agent(agent_v1, test_cases)
results_v2 = evaluate_agent(agent_v2, test_cases)
comparison = {
"accuracy_change": results_v2["accuracy"] - results_v1["accuracy"],
"latency_change": results_v2["avg_latency"] - results_v1["avg_latency"],
"token_change": results_v2["avg_tokens"] - results_v1["avg_tokens"]
}
return comparison
3. 多维度评估
不要只关注单一指标:
def comprehensive_evaluation(agent, test_cases):
"""综合评估"""
return {
"accuracy": evaluate_accuracy(agent, test_cases),
"latency": evaluate_latency(agent, test_cases),
"cost": evaluate_cost(agent, test_cases),
"safety": evaluate_safety(agent, test_cases),
"user_satisfaction": evaluate_satisfaction(agent, test_cases)
}
4. 持续集成
将评估集成到 CI/CD 流程:
# .github/workflows/eval.yml
name: Agent Evaluation
on:
pull_request:
branches: [main]
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: pip install -r requirements.txt
- name: Run evaluation
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python -m pytest tests/evaluation/ -v
- name: Check thresholds
run: |
python scripts/check_eval_thresholds.py
小结
Agent 评估是确保质量和持续改进的关键环节:
- 区分离线评估和在线评估,两者都很重要
- 使用 LangSmith 进行系统化的评估管理
- 定义多维度的评估指标
- 建立基线并进行增量评估
- 将评估集成到开发流程中
下一章我们将学习如何将 Agent 部署到生产环境。