Agent 可观测性与评估

可观测性的重要性

Agent 系统比传统 API 复杂得多：一次任务可能经历数十次 LLM 调用、工具调用和状态转换。没有可观测性，你无法知道：Agent 在哪一步失败了？工具被错误调用了几次？哪个 Prompt 导致了幻觉？为什么同样的问题有时成功有时失败？

LangSmith 集成

LangSmith 是 LangChain 官方的可观测性平台，对 LangGraph Agent 有原生支持，配置极简：

import os
from langsmith import Client, traceable

# ── 环境变量配置（.env 文件）─────────────────────────────
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "ls__xxxxxxxxxxxxxxxx"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

# 配置后，所有 LangChain/LangGraph 调用自动被追踪
# 无需修改任何业务代码——零侵入

# ── 手动追踪（非 LangChain 代码）───────────────────────
@traceable(
    name="research_agent_run",
    tags=["production", "v2"],
    metadata={"agent_version": "2.1.0"}
)
def run_research_agent(user_query: str, user_id: str) -> str:
    """用 @traceable 装饰器将函数自动添加到追踪树中。"""
    result = graph.invoke(
        {"messages": [HumanMessage(content=user_query)]},
        config={
            "metadata": {       # 这些元数据会出现在 LangSmith UI 中
                "user_id": user_id,
                "query_type": "research",
                "session_id": generate_session_id()
            }
        }
    )
    return result["messages"][-1].content

# ── 程序化分析追踪数据 ──────────────────────────────────
client = Client()

# 获取最近 100 次运行
runs = list(client.list_runs(
    project_name="my-agent-project",
    run_type="chain",
    limit=100
))

# 分析失败的运行
failed_runs = [r for r in runs if r.error]
print(f"失败率：{len(failed_runs)/len(runs):.1%}")

# 计算成本（按 Token 估算）
total_tokens = sum(r.total_tokens for r in runs if r.total_tokens)
avg_tokens = total_tokens / len(runs)
print(f"平均 Token/次：{avg_tokens:.0f}")
print(f"估算成本/次：${avg_tokens / 1000 * 0.003:.4f}（Sonnet 定价）")

# 找出最慢的10次运行（延迟分析）
slow_runs = sorted(
    [r for r in runs if r.end_time and r.start_time],
    key=lambda r: (r.end_time - r.start_time).total_seconds(),
    reverse=True
)[:10]
for run in slow_runs:
    duration = (run.end_time - run.start_time).total_seconds()
    print(f"  [{duration:.1f}s] {run.name}: {str(run.inputs)[:60]}")

OpenTelemetry 追踪（自托管）

对于不能将数据发送到第三方平台的合规场景，可以使用 OpenTelemetry 自托管追踪：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 配置 OTLP 导出器（可对接 Jaeger、Tempo、Zipkin 等）
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",  # Jaeger/Tempo 的 gRPC 端口
    insecure=True
)
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent.research", "1.0.0")

# ── 手动 Span：追踪 LangGraph 节点 ──────────────────────
def traced_agent_node(state: dict) -> dict:
    """带 OTel Span 的 LangGraph 节点。"""
    with tracer.start_as_current_span("agent_reasoning") as span:
        # 记录节点的关键属性
        span.set_attribute("messages.count", len(state["messages"]))
        span.set_attribute("iteration", state.get("iteration", 0))
        span.set_attribute("model", "claude-sonnet-4-6")

        response = llm_with_tools.invoke(state["messages"])

        # 记录 LLM 响应属性
        span.set_attribute("tool_calls.count", len(response.tool_calls))
        span.set_attribute("output_tokens",
                           response.usage_metadata.get("output_tokens", 0))

        # 如果有错误，标记 Span 为错误状态
        if not response.content:
            span.set_status(trace.status.Status(trace.status.StatusCode.ERROR))

        return {"messages": [response]}

# ── 自定义结构化日志（与 Span 关联）───────────────────
import structlog

log = structlog.get_logger()

def log_tool_call(tool_name: str, tool_input: dict, result: str):
    """记录工具调用到结构化日志（可与 Span 关联）。"""
    span = trace.get_current_span()
    trace_id = format(span.get_span_context().trace_id, "032x")

    log.info(
        "tool_call",
        tool_name=tool_name,
        input_keys=list(tool_input.keys()),
        result_length=len(result),
        trace_id=trace_id  # 关联 trace，日志与追踪可以互相跳转
    )

Agent 评估指标体系

不同于传统系统的延迟/错误率指标，Agent 评估需要业务特定的语义指标：

LangSmith 评估框架

LangSmith 提供了完整的评估工作流：构建数据集 → 定义评估器 → 运行评估 → 对比实验：

from langsmith import Client, evaluate
from langsmith.schemas import Run, Example
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage
from typing import Dict
import re

client = Client()

# ── Step 1: 构建评估数据集 ────────────────────────────
# 数据集是评估的"标准答案库"，应覆盖典型案例+边界情况+回归用例
dataset = client.create_dataset(
    "agent_eval_v1",
    description="Agent 核心功能评估集，涵盖查询/计算/综合任务"
)

# 添加测试用例（问题 + 期望答案 + 元数据）
test_cases = [
    {"question": "Python 3.12 的主要新特性有哪些？",
     "expected": "f-string嵌套、类型参数语法、新的perf指标",
     "category": "knowledge"},
    {"question": "1024 的平方根是多少？",
     "expected": "32",
     "category": "calculation"},
    {"question": "今天的日期用代码获取，然后告诉我今天是星期几",
     "expected": "正确的星期几",
     "category": "code_execution"},
]

for case in test_cases:
    client.create_example(
        inputs={"question": case["question"]},
        outputs={"answer": case["expected"]},
        metadata={"category": case["category"]},  # 用于分组分析
        dataset_id=dataset.id
    )

# ── Step 2: 定义评估器 ────────────────────────────────
judge_llm = ChatAnthropic(model="claude-opus-4-6", temperature=0)

def correctness_evaluator(run: Run, example: Example) -> Dict:
    """LLM-as-Judge：评估答案的实质正确性。"""
    actual = run.outputs.get("answer", "")
    expected = example.outputs.get("answer", "")

    # 让更强的模型（opus）来判断较弱模型（sonnet）的答案
    judgment = judge_llm.invoke([
        SystemMessage(content="""你是公正的答案评估者。
判断实际答案是否实质上符合期望答案（不需要措辞完全一致）。
对于计算类问题：数字必须精确匹配。
对于知识类问题：关键信息必须包含。
只回复 'yes' 或 'no'，不要任何解释。"""),
        HumanMessage(content=f"期望答案：{expected}\n\n实际答案：{actual}")
    ])
    is_correct = "yes" in judgment.content.lower()
    return {"key": "correctness", "score": 1.0 if is_correct else 0.0}

def tool_efficiency_evaluator(run: Run, example: Example) -> Dict:
    """评估工具调用效率：步骤数越少越好。"""
    steps = run.extra.get("tool_call_count", 0)
    # 5步以内满分，每多一步扣0.1分，最低0分
    score = max(0.0, 1.0 - max(0, steps - 5) * 0.1)
    return {"key": "tool_efficiency", "score": score}

def hallucination_evaluator(run: Run, example: Example) -> Dict:
    """检测答案中的幻觉：答案是否包含工具结果以外的信息。"""
    actual = run.outputs.get("answer", "")
    # 获取 Agent 使用的工具结果（从 run 的子 span 中提取）
    tool_results = run.extra.get("tool_results", [])
    tool_context = "\n".join(tool_results)

    if not tool_context:
        return {"key": "no_hallucination", "score": 1.0}  # 无工具调用跳过

    judgment = judge_llm.invoke([
        SystemMessage(content="""判断答案中是否包含工具结果以外的事实声明。
如果答案完全基于工具结果，回复 'grounded'。
如果答案包含工具结果中没有的事实，回复 'hallucinated'。"""),
        HumanMessage(content=f"工具结果：\n{tool_context}\n\n答案：{actual}")
    ])
    is_grounded = "grounded" in judgment.content.lower()
    return {"key": "no_hallucination", "score": 1.0 if is_grounded else 0.0}

# ── Step 3: 运行评估实验 ──────────────────────────────
def agent_runner(inputs: dict) -> dict:
    """将 Agent 包装为评估器可调用的格式。"""
    result = graph.invoke({"messages": [HumanMessage(content=inputs["question"])]})
    return {"answer": result["messages"][-1].content}

# experiment_prefix 用于区分不同版本的实验
results = evaluate(
    agent_runner,
    data="agent_eval_v1",
    evaluators=[correctness_evaluator, tool_efficiency_evaluator, hallucination_evaluator],
    experiment_prefix="v2-claude-sonnet",
    num_repetitions=3  # 每个用例重复3次，消除随机性
)

# 查看汇总结果
df = results.to_pandas()
print(df[["correctness", "tool_efficiency", "no_hallucination"]].describe())

Prompt 优化方法论

Agent 的系统 Prompt 是最强大但也最难优化的旋钮。以下是系统化的优化流程：

# ── 工具描述优化示例 ─────────────────────────────────────
# 差的工具描述（导致工具调用精确率低）：
BAD_DESCRIPTION = "搜索工具"  # 太模糊，Agent 不知道何时使用

# 好的工具描述（提供明确的使用场景和限制）：
GOOD_DESCRIPTION = """在互联网上搜索实时信息。

适用场景：
- 查找最新新闻、事件、产品发布
- 获取实时数据（股价、天气、体育比赛结果）
- 查找技术文档和教程

不适用：
- 数学计算（请用 calculate 工具）
- 代码执行（请用 run_python 工具）
- 基于已有信息做分析（LLM 直接推理即可）

参数：
- query: 搜索查询词，越具体越好（如 "Python 3.12 release notes" 优于 "Python"）"""

# ── 自动比较两个 Prompt 版本 ─────────────────────────────
def compare_prompts(
    system_v1: str,
    system_v2: str,
    test_queries: list[str],
    judge_fn
) -> dict:
    """A/B 比较两个系统 Prompt 的效果。"""
    v1_scores, v2_scores = [], []

    for query in test_queries:
        # 两个版本各运行一次
        answer_v1 = run_agent_with_system(query, system_v1)
        answer_v2 = run_agent_with_system(query, system_v2)

        # 评估器打分（LLM-as-Judge 或规则）
        score_v1 = judge_fn(query, answer_v1)
        score_v2 = judge_fn(query, answer_v2)
        v1_scores.append(score_v1)
        v2_scores.append(score_v2)

    import numpy as np
    from scipy import stats

    # t 检验判断差异是否显著（p < 0.05 认为有效）
    t_stat, p_value = stats.ttest_rel(v1_scores, v2_scores)

    return {
        "v1_mean": np.mean(v1_scores),
        "v2_mean": np.mean(v2_scores),
        "improvement": np.mean(v2_scores) - np.mean(v1_scores),
        "p_value": p_value,
        "significant": p_value < 0.05,  # 差异是否统计显著
        "recommendation": "使用 v2" if np.mean(v2_scores) > np.mean(v1_scores) else "保持 v1"
    }