当你把一个 Agent 从 demo 推到生产环境,最大的问题不是"它能不能跑",而是"它跑得对不对、稳不稳"。深度 Agent——那种会多步推理、调用工具、自己纠错的 Agent——比单轮 LLM 调用难评估得多:一次对话可能触发 5 次工具调用,中间任何一步偏了,最终结果就废了。这篇文章把 LangChain 在深度 Agent 评估上的经验和 Anthropic 的 Agent 评估指南揉成一条可落地的路径,用 text-to-SQL Agent + Amazon Bedrock 走完从开发到生产的全流程。
深度 Agent 的评估为什么不一样
单轮问答的评估本质上是在比对"生成文本 vs 期望文本",BLEU、ROUGE、LLM-as-judge 都能凑合用。深度 Agent 的输出是一条执行轨迹——先想、再查表、再写 SQL、再纠错、再返回结果。你不仅要看最终答案对不对,还要看中间步骤有没有绕路、有没有幻觉式工具调用、有没有在死循环里烧 token。
所以评估维度至少要拆成三层:
- 终态正确性:最终返回的 SQL / 答案是否正确?
- 轨迹质量:中间步骤是否合理,有没有冗余或错误转折?
- 资源效率:用了多少步、多少 token,有没有更短的路径?
五种评估模式,从不同角度卡住 Agent 质量
结合 LangChain 和 Anthropic 的实践,以下五种模式覆盖了离线和线上最关键的评估场景:
1. 终态比对(Final Answer Comparison)
最直觉的模式:给定一个问题,Agent 返回的最终答案是否和期望答案一致。对于 text-to-SQL,期望答案可以是一条参考 SQL 或查询结果集。
关键细节:SQL 的"等价"比文本等价更难判断——SELECT * FROM orders WHERE status='active' 和 SELECT id, name FROM orders WHERE status='active' 语义上可能都对,但字段不同。建议用结果集比对而非字符串比对,或者用 LLM-as-judge 判断语义等价。
2. 轨迹步骤比对(Trajectory Evaluation)
不只看终点,看每一步。把 Agent 的实际轨迹和参考轨迹逐步比对:该查 schema 的时候查了吗?该纠错的时候纠了吗?有没有跳过必要步骤直接猜答案?
这种模式对调试最有价值——你能精准定位"Agent 在第 3 步开始跑偏"。
3. 步骤级单元测试(Step-level Unit Testing)
把 Agent 的每一步当成一个独立单元来测。比如:给定一个错误的 SQL,Agent 的纠错步骤能不能修好?给定一个 schema 描述,Agent 能不能写出正确的 WHERE 子句?
这种模式的好处是可复现、可隔离——不需要跑完整个 Agent 流程就能定位问题。
4. 回归测试集(Regression Suite)
维护一组固定的 question→expected_answer 对,每次 Agent 逻辑变更后全量跑一遍。这和传统软件的 CI 回归测试思路一致,但测试对象是 Agent 的推理链。
回归测试集要持续扩充:线上出问题的 case、边界场景、多表 join、嵌套子查询,都要逐步加进去。
5. 线上监控(Online Monitoring)
离线测试只能覆盖已知场景,线上才有未知的长尾问题。通过 LangSmith 的 tracing 实时捕获每次 Agent 执行的轨迹、token 消耗、工具调用次数,设置阈值告警。
线上监控的核心指标: - 单次执行的平均步骤数和 token 数 - 工具调用失败率 - 用户反馈的正面/负面比例 - 超时或超 token 限制的频率
用 pytest + LangSmith 搭离线评估
下面是一个可运行的离线评估框架,用 pytest 组织测试、LangSmith 记录轨迹和评分。假设你已经有 Amazon Bedrock 上的 Claude 模型和一个 text-to-SQL Agent。
先安装依赖:
pip install pytest langsmith langchain langchain-aws
项目结构
agent-eval/
├── conftest.py # pytest 配置,初始化 LangSmith
├── test_final_answer.py # 终态比对测试
├── test_trajectory.py # 轨迹比对测试
├── test_step_unit.py # 步骤级单元测试
├── eval_dataset.py # 测试数据集定义
└── agent.py # 你的 text-to-SQL Agent 实现
eval_dataset.py — 测试数据集
"""离线评估用的测试数据集,每个 case 包含问题、参考 SQL 和参考轨迹关键步骤。"""
EVAL_CASES = [
{
"id": "simple_select",
"question": "查询所有活跃订单的客户名称",
"reference_sql": "SELECT customer_name FROM orders WHERE status = 'active'",
"reference_result": [["张三"], ["李四"], ["王五"]],
"key_trajectory_steps": [
"list_tables", # 应先列出可用表
"get_schema", # 应查 orders 表的 schema
"write_sql", # 应写出 SQL
],
},
{
"id": "multi_table_join",
"question": "查询每个客户最近一笔订单的金额",
"reference_sql": """
SELECT c.name, o.amount
FROM customers c
JOIN orders o ON c.id = o.customer_id
WHERE o.id = (
SELECT MAX(id) FROM orders o2 WHERE o2.customer_id = c.id
)
""",
"reference_result": [["张三", 1200], ["李四", 850]],
"key_trajectory_steps": [
"list_tables",
"get_schema", # 应查 customers 和 orders 两张表
"write_sql",
"validate_sql", # 应验证 SQL 是否可执行
],
},
{
"id": "error_correction",
"question": "查询销售额超过一万的地区",
"reference_sql": "SELECT region FROM sales WHERE revenue > 10000",
"reference_result": [["华东"], ["华南"]],
"key_trajectory_steps": [
"list_tables",
"get_schema",
"write_sql",
"execute_sql", # 可能首次执行失败
"correct_sql", # 应触发纠错步骤
],
},
]
conftest.py — pytest + LangSmith 初始化
import os
import pytest
from langsmith import Client
# 确保 LangSmith 环境变量已设置
# 实际使用时把这些放到 .env 或 CI 的环境变量中
os.environ.setdefault("LANGSMITH_API_KEY", "your-langsmith-api-key")
os.environ.setdefault("LANGSMITH_PROJECT", "text-to-sql-agent-eval")
@pytest.fixture(scope="session")
def langsmith_client():
return Client()
@pytest.fixture(scope="session")
def agent():
"""初始化你的 text-to-SQL Agent,使用 Amazon Bedrock 上的模型。"""
from agent import TextToSQLAgent
# agent.py 中你用 langchain-aws 的 ChatBedrockConverse 配置 Claude
return TextToSQLAgent()
test_final_answer.py — 终态比对
"""终态比对:Agent 返回的 SQL 执行结果是否和参考结果一致。"""
import pytest
from eval_dataset import EVAL_CASES
from langsmith import evaluate
def final_answer_evaluator(run, example):
"""LangSmith 评分函数:比对 Agent 输出和期望输出。"""
output = run.outputs
expected = example.outputs or example.inputs
# 比对结果集(而非 SQL 字符串)
agent_results = output.get("query_results", [])
ref_results = expected.get("reference_result", [])
# 简化比对:结果集行数和首行内容
score = 1.0 if agent_results == ref_results else 0.0
return {"key": "final_answer_accuracy", "score": score}
@pytest.mark.parametrize("case", EVAL_CASES, ids=lambda c: c["id"])
def test_final_answer(agent, langsmith_client, case):
"""对每个测试 case,运行 Agent 并用 LangSmith 记录评估结果。"""
result = agent.run(case["question"])
# 通过 LangSmith evaluate API 记录评分
# 这里用直接比对做演示,生产环境可接入 LLM-as-judge
actual_results = result.get("query_results", [])
expected_results = case["reference_result"]
assert actual_results == expected_results, (
f"Case '{case['id']}' failed: "
f"expected {expected_results}, got {actual_results}"
)
test_trajectory.py — 轨迹关键步骤比对
"""轨迹评估:Agent 执行过程中是否经过了期望的关键步骤。"""
import pytest
from eval_dataset import EVAL_CASES
@pytest.mark.parametrize("case", EVAL_CASES, ids=lambda c: c["id"])
def test_trajectory_contains_key_steps(agent, case):
"""检查 Agent 的执行轨迹是否包含所有关键步骤。"""
result = agent.run(case["question"])
trajectory = result.get("trajectory", []) # 你的 Agent 应返回步骤列表
# 提取实际步骤名称
actual_steps = [step["action"] for step in trajectory]
expected_steps = case["key_trajectory_steps"]
missing = [s for s in expected_steps if s not in actual_steps]
assert not missing, (
f"Case '{case['id']}' missing trajectory steps: {missing}. "
f"Actual steps: {actual_steps}"
)
test_step_unit.py — 步骤级单元测试
"""步骤级单元测试:隔离测试 Agent 的单个能力。"""
import pytest
def test_sql_correction_step(agent):
"""给定一个有语法错误的 SQL,测试 Agent 的纠错能力。"""
broken_sql = "SELECT name FROM orders WHERE status = 'active'"
corrected = agent.correct_sql(broken_sql, error_msg="syntax error near 'WHERE'")
assert "WHERE" in corrected, f"Correction failed: got '{corrected}'"
assert "WHERE" not in corrected, f"Still contains typo: '{corrected}'"
def test_schema_lookup_step(agent):
"""给定表名,测试 Agent 能否正确获取 schema。"""
schema_info = agent.get_table_schema("orders")
assert "customer_name" in str(schema_info), (
f"Schema lookup missed expected column. Got: {schema_info}"
)
运行测试
# 跑全部离线评估
cd agent-eval
pytest -v --tb=short
# 只跑终态比对
pytest test_final_answer.py -v
# 只跑轨迹评估
pytest test_trajectory.py -v
每次运行后,LangSmith 的 dashboard 上会自动记录每次 Agent 执行的完整 trace,包括每步的输入输出、token 消耗和耗时。你可以在项目 text-to-sql-agent-eval 下查看。
线上监控:LangSmith + Bedrock 的生产配置
离线测试跑通了,不代表线上没问题。以下是生产环境的关键配置。
环境变量配置(部署到 AWS 时)
# LangSmith tracing — 所有 Agent 调用自动上报
export LANGSMITH_API_KEY="your-langsmith-api-key"
export LANGSMITH_PROJECT="text-to-sql-agent-prod"
export LANGSMITH_TRACING_V2="true"
# Amazon Bedrock 配置
export AWS_REGION="us-east-1"
export AWS_DEFAULT_REGION="us-east-1"
LangSmith 告警规则示例
在 LangSmith 项目设置中配置以下告警(也可通过 API 创建):
from langsmith import Client
client = Client()
# 创建一个监控仪表盘关注的指标
# LangSmith 目前通过项目页面的 Filters 和 Tags 实现筛选
# 生产环境建议按以下维度设置关注:
MONITORING_DIMENSIONS = {
"avg_steps_per_run": {
"description": "单次 Agent 执行的平均步骤数",
"threshold": 8, # 超过 8 步可能意味着 Agent 在绕路
"action": "检查是否有冗余工具调用或死循环",
},
"tool_call_failure_rate": {
"description": "工具调用失败率",
"threshold": 0.05, # 超过 5% 需要排查
"action": "检查 Bedrock 模型权限或数据库连接",
},
"avg_token_per_run": {
"description": "单次执行的平均 token 消耗",
"threshold": 2000,
"action": "优化 prompt 或减少不必要的 schema 查询",
},
}
# 实际告警通过 LangSmith 的 Web UI 配置更方便
# 也可以用 Python SDK 查询近期 runs 做自定义监控
recent_runs = client.list_runs(
project_name="text-to-sql-agent-prod",
limit=50,
)
# 统计近期执行的平均步骤数
step_counts = []
for run in recent_runs:
if run.child_runs:
step_counts.append(len(run.child_runs))
avg_steps = sum(step_counts) / len(step_counts) if step_counts else 0
print(f"近 50 次执行平均步骤数: {avg_steps:.1f}")
if avg_steps > 8:
print("⚠️ 平均步骤数偏高,检查 Agent 是否在绕路")
Agent 实现中嵌入 trace 标签
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langsmith import traceable
model = ChatBedrockConverse(
model="anthropic.claude-3-5-sonnet-20241022-v2:0",
region_name="us-east-1",
max_tokens=4096,
)
@traceable(name="list_tables", run_type="tool")
def list_tables():
"""列出数据库中所有可用表。"""
# 实际实现:查询数据库 metadata
return ["orders", "customers", "sales"]
@traceable(name="get_schema", run_type="tool")
def get_table_schema(table_name: str):
"""获取指定表的 schema 信息。"""
# 实际实现:DESCRIBE table_name
schemas = {
"orders": {"columns": ["id", "customer_name", "status", "amount", "created_at"]},
"customers": {"columns": ["id", "name", "region", "email"]},
"sales": {"columns": ["id", "region", "revenue", "date"]},
}
return schemas.get(table_name, {})
@traceable(name="text_to_sql_agent", run_type="chain")
def run_agent(question: str):
"""完整的 text-to-SQL Agent 流程,每步都会被 LangSmith trace。"""
tables = list_tables()
# 根据问题选择相关表(简化示例)
relevant_table = "orders" # 实际应由 LLM 判断
schema = get_table_schema(relevant_table)
prompt = f"""基于以下 schema,写一条 SQL 回答问题。
表: {relevant_table}
Schema: {schema}
问题: {question}
只返回 SQL,不要解释。"""
response = model.invoke([HumanMessage(content=prompt)])
sql = response.content.strip()
# 实际生产中会加入 execute_sql 和 correct_sql 步骤
return {"sql": sql, "trajectory_steps": ["list_tables", "get_schema", "write_sql"]}
@traceable 装饰器让每个步骤自动上报到 LangSmith,线上每次用户调用都会生成一条完整的 trace 链。你可以在 LangSmith dashboard 里按步骤名、耗时、token 数筛选和排查。
落地时的取舍和建议
| 决策点 | 建议 | 原因 |
|---|---|---|
| 终态比对用字符串还是结果集 | 结果集比对 | SQL 写法多样,字符串比对误判率高 |
| 轨迹比对用严格匹配还是宽松匹配 | 宽松匹配:只检查关键步骤是否存在 | Agent 合理的探索步骤不应被判为错误 |
| 回归测试集多大 | 从 20 个 case 开始,逐步扩充到 100+ | 太小覆盖不了边界,太大维护成本高 |
| 线上监控用 LangSmith 还是自建 | 先用 LangSmith,规模大了再考虑自建 | LangSmith 开箱即用,省掉 trace 管道开发 |
| Bedrock 模型选哪个 | Claude 3.5 Sonnet 做 Agent 主模型,Haiku 做轻量步骤 | Sonnet 推理能力强,Haiku 省 token 适合 schema 查询等简单步骤 |
几个容易踩的坑:
- 不要只测终态。一个 Agent 可能碰巧给出了正确 SQL,但中间查了 5 次无关的表。轨迹评估能抓住这种"运气好但不可靠"的情况。
- 回归测试集要包含"应该失败"的 case。比如问一个不存在的表,Agent 应该明确拒绝而非编造 SQL。
- 线上告警阈值要根据实际数据调。初始阈值可以宽松,跑一周真实流量后再收紧。
- Bedrock 的模型调用有速率限制。评估时如果并发跑大量 case,注意控制并发数,避免被限流导致测试结果失真。
评估不是一次性动作,而是持续工程。离线回归测试挡住已知问题,线上监控抓住未知问题,两者配合才能让深度 Agent 在生产环境里真正可靠。