Agent 可观测性与评估

可观测性的重要性

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

LangSmith 集成

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

import os
from langsmith import Client

# ── 环境变量配置（.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 调用自动被追踪
# 无需修改任何业务代码

# ── 手动添加追踪元数据 ────────────────────────────────────
from langsmith import traceable

@traceable(name="research_agent_run", tags=["production", "v2"])
def run_research_agent(user_query: str, user_id: str) -> str:
    # 自动追踪这个函数内的所有 LLM 调用
    result = graph.invoke(
        {"messages": [HumanMessage(content=user_query)]},
        config={
            "metadata": {  # 追加到追踪记录的元数据
                "user_id": user_id,
                "query_type": "research"
            }
        }
    )
    return result["messages"][-1].content

# ── 使用 LangSmith Client 程序化访问追踪数据 ────────────
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 使用量
avg_tokens = sum(r.total_tokens for r in runs if r.total_tokens) / len(runs)
print(f"平均 Token/次：{avg_tokens:.0f}")

OpenTelemetry 追踪（自托管）

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

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 配置 Jaeger 导出器（可换成 Zipkin、Tempo 等）
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(provider)

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

# 手动 Span
def traced_agent_node(state):
    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))

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

        span.set_attribute("tool_calls.count", len(response.tool_calls))
        span.set_attribute("response.tokens",
                           response.usage_metadata.get("output_tokens", 0))
        return {"messages": [response]}

Agent 评估指标体系

from langsmith import Client, evaluate
from langsmith.schemas import Run, Example
from typing import Dict

client = Client()

# ── 构建评估数据集 ────────────────────────────────────────
dataset = client.create_dataset("agent_eval_v1")

# 添加测试用例（问题 + 期望答案）
test_cases = [
    {"question": "Python 3.11 比 3.10 快多少？",
     "expected": "约快 10-60%，平均25%"},
    {"question": "LangGraph 和 LangChain 的关系？",
     "expected": "LangGraph 是 LangChain 的 Agent 编排库"},
]

for case in test_cases:
    client.create_example(
        inputs={"question": case["question"]},
        outputs={"answer": case["expected"]},
        dataset_id=dataset.id
    )

# ── 定义评估器（LLM-as-Judge）────────────────────────────
def correctness_evaluator(run: Run, example: Example) -> Dict:
    """用 LLM 评估答案是否正确。"""
    actual = run.outputs.get("answer", "")
    expected = example.outputs.get("answer", "")

    judge_llm = ChatOpenAI(model="gpt-4o", temperature=0)
    score = judge_llm.invoke([
        SystemMessage(content="""你是公正的答案评估者。
        判断实际答案是否实质上符合期望答案（不需要完全一致）。
        只回复 'yes' 或 'no'，不要解释。"""),
        HumanMessage(content=
            f"期望：{expected}\n实际：{actual}"
        )
    ])
    is_correct = "yes" in score.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分
    score = max(0, 1.0 - max(0, steps - 5) * 0.1)
    return {"key": "tool_efficiency", "score": score}

# ── 运行评估 ──────────────────────────────────────────────
def agent_runner(inputs: dict) -> dict:
    result = graph.invoke({"messages": [HumanMessage(content=inputs["question"])]})
    return {"answer": result["messages"][-1].content}

results = evaluate(
    agent_runner,
    data="agent_eval_v1",
    evaluators=[correctness_evaluator, tool_efficiency_evaluator],
    experiment_prefix="v2-gpt4o-mini"
)
print(results.to_pandas()[["correctness", "tool_efficiency"]].describe())

Prompt 优化方法论

1. 建立基线：先用最简单的 Prompt 运行，记录关键指标（正确率、步骤数、成本）。
2. 错误分析：从 LangSmith 中筛选失败/低质量的运行，分析失败模式（工具选错？推理出错？格式问题？）。
3. 逐步改进：每次只改变一个变量（Prompt、模型、工具描述），对比前后指标。
4. A/B 测试：对两个版本同时运行相同数据集，统计显著性差异。
5. 持续回归：新版本上线前，必须通过历史评估数据集的回归测试。