用 LangSmith 在 AWS 上评估深度 Agent:从离线测试到线上监控的完整路径

2026-05-29 20 预计阅读时间:1 分钟
来源:aws.amazon.com AI 摘要 原文链接

免责声明:本文为 AI 摘要整理,建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件,不作为官方说明。

预计阅读时间:17 分钟

当你把一个 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 查询等简单步骤

几个容易踩的坑:

  1. 不要只测终态。一个 Agent 可能碰巧给出了正确 SQL,但中间查了 5 次无关的表。轨迹评估能抓住这种"运气好但不可靠"的情况。
  2. 回归测试集要包含"应该失败"的 case。比如问一个不存在的表,Agent 应该明确拒绝而非编造 SQL。
  3. 线上告警阈值要根据实际数据调。初始阈值可以宽松,跑一周真实流量后再收紧。
  4. Bedrock 的模型调用有速率限制。评估时如果并发跑大量 case,注意控制并发数,避免被限流导致测试结果失真。

评估不是一次性动作,而是持续工程。离线回归测试挡住已知问题,线上监控抓住未知问题,两者配合才能让深度 Agent 在生产环境里真正可靠。


相关推荐