LangSmith 实战指南：从链路追踪到自动化评测

LangChain  →  生产 Agent
LangSmith  →  观测、调试、评测、优化 Agent

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls__your_api_key"
os.environ["LANGSMITH_PROJECT"] = "travel-agent-v3"

# 之后所有 LangChain 调用自动记录
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("推荐成都3个景点")
# → 自动出现在 LangSmith Trace 面板

Run Tree:
  📊 AgentExecutor (3.2s, $0.04)
  ├── 🤖 LLM Call (0.8s, $0.02) → "需要调 search_attractions"
  ├── 🔧 Tool: search_attractions (0.5s)
  ├── 🤖 LLM Call (0.6s, $0.01) → "需要调 get_weather"
  ├── 🔧 Tool: get_weather (0.3s)
  └── 🤖 LLM Call (1.0s, $0.01) → 最终回答

from langsmith import traceable

@traceable(run_type="tool")
def search_attractions(city: str, keywords: list):
    """搜索景点，自动记录到 LangSmith"""
    results = amap_api.search(city, keywords)
    return results

@traceable(run_type="chain")
def plan_itinerary(user_profile):
    pois = search_attractions(user_profile["destination"], user_profile["interests"])
    weather = get_weather(user_profile["destination"])
    schedule = or_tools_solve(pois, user_profile)
    return schedule

问题：用户反馈 Agent 推荐了周二闭馆的博物馆

LangSmith Trace 分析：
├── RAG 检索 → 返回了博物馆（正确）✅
├── LLM 规划 → 安排在周二（正确，用户要求周二）✅
├── FactCheck → 应该拦截但没执行 ❌ ← 根因在这
└── 输出 → 安排了闭馆景点

诊断：FactCheck 节点的条件路由写错了，
"周二闭馆"条件判断有 off-by-one 错误

from langsmith import Client

client = Client()

# 手工上传
examples = [
    {
        "input": "推荐成都2天亲子游",
        "expected_output": {
            "days": 2,
            "has_children_friendly": True,
            "budget_ok": True
        }
    },
    {
        "input": "三亚5天，预算500",
        "expected_output": {
            "feasibility_conflict": True,
            "suggested_alternative": True
        }
    },
]

dataset = client.create_dataset(
    "itinerary-qa-v2",
    description="行程生成质量测试集"
)
client.create_examples(inputs=[e["input"] for e in examples],
                       outputs=[e["expected_output"] for e in examples],
                       dataset_id=dataset.id)

# 筛选线上跑过的 Trace，挑出满意和不满意的
runs = client.list_runs(
    project_name="travel-agent-production",
    execution_order=1,
    filter='eq(feedback_score, 1)',  # 用户点赞的
)

# 转为测试用例
for run in runs:
    client.create_example(
        inputs={"question": run.inputs["question"]},
        outputs={"answer": run.outputs["answer"]},
        dataset_id=dataset.id,
    )

itinerary-qa-v1 → 100 条基础用例（手动标注）
itinerary-qa-v2 → 500 条（导入线上数据）
itinerary-qa-v3 → 1000 条（加入边缘 case）

from langsmith.evaluation import evaluate, LangChainStringEvaluator

# 定义评测器
evaluators = [
    # 正确性：回答是否符合预期
    LangChainStringEvaluator("cot_qa", prepare_data=lambda r, e: {
        "input": r.inputs["question"],
        "prediction": r.outputs["answer"],
        "reference": e.outputs["answer"],
    }),

    # 简洁性：回答是否废话太多
    LangChainStringEvaluator(
        "criteria",
        config={"criteria": "回答简洁，没有冗余信息，直接回应用户问题"}
    ),
]

# 运行评测
results = evaluate(
    lambda inputs: my_agent.run(inputs["question"]),
    data="itinerary-qa-v3",
    evaluators=evaluators,
    experiment_prefix="v3-optimized-",
)

# 结果对比
print(f"正确率: {results['cot_qa']}")
print(f"简洁度: {results['criteria']}")

def evaluate_itinerary_completeness(run, example):
    """评估行程的完整性"""
    output = run.outputs["answer"]

    score = 0
    # 检查是否包含天数
    if any(f"{d}天" in output or f"{d}日" in output for d in range(1,8)):
        score += 0.25
    # 检查是否包含预算
    if "¥" in output or "元" in output:
        score += 0.25
    # 检查是否包含具体景点
    if output.count("**") >= 2:
        score += 0.25
    # 检查是否有时段安排
    if ":" in output and ("上午" in output or "下午" in output or ":" in output.split("\n")[3]):
        score += 0.25

    return {"key": "completeness", "score": score}

from langsmith import Client
client = Client()

# 用户点赞后上报
def on_user_thumbs_up(run_id: str):
    client.create_feedback(
        run_id,
        key="user_satisfaction",
        score=1,
        comment="用户满意"
    )

# 用户点踩后记录原因
def on_user_thumbs_down(run_id: str, reason: str):
    client.create_feedback(
        run_id,
        key="user_satisfaction",
        score=0,
        comment=reason  # "推荐的餐厅已倒闭" ← 宝贵的数据
    )

v1: "你是一个旅游助手..."                    → 效果一般
v2: "你是一个资深旅游规划师，有10年经验..."      → 效果好了一点
v3: 加了 Few-shot 示例                      → 效果显著提升
v4: 调整了输出格式要求                        → 格式对了但内容变差
v5: 回退到 v3 + 小改                        → 👍 最佳版本

from langchain import hub

# 从 Hub 拉取 Prompt
prompt = hub.pull("my-org/itinerary-planner:production")

# 本地调试
response = llm.invoke(prompt.format(destination="成都"))

# 推送到 Hub（新版本）
hub.push("my-org/itinerary-planner:dev", prompt)

# CI/CD 集成：跑过评测的标记为 production
# ❯ langchain hub promote my-org/itinerary-planner:dev → production

命名规范：
  my-org/intent-classifier:production   ← 线上版本
  my-org/intent-classifier:dev          ← 开发中
  my-org/intent-classifier:v3.1         ← 里程碑版本

版本说明：
  每次 push 写清楚：
  - 改了什么参数（temperature, max_tokens）
  - 加了什么 Few-shot 示例
  - 评测结果变化（正确率 85% → 92%）

from langsmith import Client
import datetime

client = Client()

# 统计本周成本
runs = client.list_runs(
    project_name="travel-agent-v3",
    start_time=datetime.datetime.now() - datetime.timedelta(days=7),
)

total_cost = sum(
    run.total_cost or 0
    for run in runs
    if run.total_cost
)

total_tokens = sum(
    run.total_tokens or 0
    for run in runs
    if run.total_tokens
)

print(f"本周: {total_cost:.2f} 美元, {total_tokens:,} tokens")
print(f"平均每任务: {total_cost/len(list(runs)):.4f} 美元")

@traceable(run_type="chain")
def my_agent(query: str):
    # LangSmith 自动记录 cost
    response = llm.invoke(query)
    return response

# 在 LangSmith UI 设置:
# 告警规则: 单次 LLM 调用 > $0.10 → 通知
# 告警规则: 每小时总成本 > $5 → 通知

import os
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"  # 只记录 10%

# 或者关键链路全记录，非关键采样
if user.is_vip:
    os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "1.0"

travel-agent-v3-production     ← 线上流量
travel-agent-v3-staging        ← 预发布评测
travel-agent-v4-experiment     ← 实验对比

@traceable(
    run_type="chain",
    tags=["v3", "production", "vip_user"],
    metadata={"user_tier": user.tier, "region": "cn"}
)
def run_agent(query):
    ...

# 每周一自动跑评测
0 9 * * 1 cd /app && python eval_all.py --dataset itinerary-qa-v3 --commit HEAD