LangChain 系列 · 第十篇:LangSmith——调试和监控你的 LLM 应用

🎯 适合人群:已构建 LangChain 应用,想用可观测性工具系统性排查问题、量化优化效果的工程师
⏱️ 阅读时间:约 25 分钟
💬 本文介绍 LangSmith 的链路追踪、评估数据集管理与实验对比功能,从"靠日志猜问题"升级为"用数据找问题"


一、为什么 LLM 应用难以调试

传统 Web 应用出了问题,看日志和异常堆栈基本能定位。但 LLM 应用有三类传统工具处理不了的问题:

问题一:输出不确定,难以复现

同一个请求发两次,可能得到两个不同的答案。某次输出质量差,但日志里只有一行 response.content = "...",无从知道模型收到了什么、用了哪些 context、Token 消耗了多少。

问题二:调用链路深,中间步骤不可见

一个 RAG + Agent 请求的背后,可能包含:检索 → 重排 → 多次 LLM 调用 → 工具执行 → 再次 LLM 调用。哪一步慢了?哪一步的输出有问题?传统日志难以追踪完整链路。

问题三:优化效果无法量化

改了一个 Prompt,感觉好像好一点了——但到底提升了多少?改动有没有引入新的问题?没有量化对比,优化就是在凭感觉赌博。

LangSmith 正是为了解决这三类问题而设计的 LLM 应用可观测性平台,由 LangChain 官方出品,与 LangChain/LangGraph 深度集成。


二、LangSmith 的核心能力

  LangSmith Core Capabilities

  +------------------+  +------------------+  +------------------+
  |    Tracing       |  |   Datasets &     |  |   Experiments    |
  |                  |  |   Evaluation     |  |                  |
  | - Full call tree |  | - Test cases     |  | - A/B comparison |
  | - Inputs/outputs |  | - Ground truth   |  | - Metric trends  |
  | - Latency/tokens |  | - RAGAS/custom   |  | - Prompt diffs   |
  | - Error details  |  |   evaluators     |  | - Version mgmt   |
  +------------------+  +------------------+  +------------------+
         |                      |                      |
    "What happened?"      "Is it correct?"    "Which version wins?"

三个核心能力解决三类问题:

  • Tracing(链路追踪):记录每次请求的完整调用树,包含每一步的输入输出、耗时和 Token 消耗
  • Datasets & Evaluation(数据集与评估):管理测试用例,运行自动化评估
  • Experiments(实验对比):对比不同版本的 Prompt 或配置,量化优化效果

三、环境配置

3.1 注册与获取 API Key

访问 smith.langchain.com 注册账号,在 Settings → API Keys 创建一个 API Key。

pip install langsmith langchain-openai

3.2 环境变量配置

只需设置以下环境变量,LangChain 会自动将所有调用数据发送到 LangSmith,无需修改任何业务代码

# .env 文件
LANGCHAIN_TRACING_V2=true          # 开启链路追踪
LANGCHAIN_API_KEY=ls__xxxxxxxx     # LangSmith API Key
LANGCHAIN_PROJECT=my-rag-project   # 项目名称(在 LangSmith 中用于分组)
OPENAI_API_KEY=sk-xxxxxxxx
from dotenv import load_dotenv
load_dotenv()

# 之后所有 LangChain 调用都会自动被追踪,无需其他改动
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = (
    ChatPromptTemplate.from_messages([("human", "{question}")])
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# 这次调用会自动出现在 LangSmith 的 Tracing 面板中
result = chain.invoke({"question": "什么是向量数据库?"})

💡 LANGCHAIN_TRACING_V2=true 是开关,LANGCHAIN_PROJECT 决定数据归属哪个项目。同一个代码库在开发、测试、生产用不同的 project 名称,可以完全隔离追踪数据。


四、Tracing:链路追踪

4.1 在 LangSmith 中看到什么

开启追踪后,每次 chain.invoke 都会在 LangSmith 生成一条完整的 Run 记录,展示:

  Run Tree (example for a RAG request):

  chain.invoke (total: 2.3s, $0.0032)
  ├── retriever.invoke (0.4s)
  │   └── vectorstore.similarity_search (0.4s)
  │       └── [returns 4 chunks]
  ├── reranker.invoke (0.6s)
  │   └── [returns top 2 chunks]
  └── ChatOpenAI.invoke (1.3s, 847 tokens, $0.0031)
      ├── input:  [SystemMessage, HumanMessage with context]
      └── output: AIMessage("向量数据库是...")

每个节点都可以展开查看:

  • 完整的输入输出:Prompt 发给模型的确切内容,模型返回的原始响应
  • 耗时分布:哪一步是瓶颈
  • Token 用量:input tokens、output tokens、总费用
  • 错误详情:如果某步失败,完整的异常信息

4.2 手动添加追踪标签

默认的追踪数据已经很丰富,但可以手动添加标签来方便后续过滤:

from langsmith import traceable

# @traceable 装饰器将普通函数纳入追踪体系
@traceable(name="rag_pipeline", tags=["rag", "production"], metadata={"version": "v2.1"})
def run_rag(question: str) -> str:
    docs = retriever.invoke(question)
    context = "\n".join([d.page_content for d in docs])
    return chain.invoke({"question": question, "context": context})

result = run_rag("LangGraph 的 Checkpointer 是什么?")

tagsmetadata 会出现在 LangSmith 的过滤面板,方便批量筛选特定版本或场景的 Run。

4.3 在代码中获取 Run ID

有时需要在代码中拿到当前 Run 的 ID,用于后续关联用户反馈:

from langchain_core.callbacks import collect_runs

# collect_runs 上下文管理器捕获 Run 信息
with collect_runs() as cb:
    result = chain.invoke({"question": "什么是 LCEL?"})
    run_id = cb.traced_runs[0].id

print(f"Run ID: {run_id}")
# 可以将 run_id 存储,后续关联用户点赞/踩等反馈

4.4 记录用户反馈

将用户的显式反馈(点赞/踩)关联到对应的 Run,用于构建评估数据集:

from langsmith import Client

client = Client()

# 用户点赞
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1,          # 1 = 正面反馈,0 = 负面反馈
    comment="回答准确,有代码示例",
)

# 用户踩
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=0,
    comment="答案和问题不相关",
)

五、Datasets:管理测试数据集

Dataset 是 LangSmith 中的测试用例集合,每条记录包含输入(input)和可选的期望输出(output)。

📝 LangSmith 中的 Dataset 与第六篇 RAGAS 评估中的评估数据集概念相同:都是"问题 + 标准答案"的集合。区别在于 LangSmith Dataset 存储在云端,可以版本管理并跨实验复用。

5.1 创建 Dataset

from langsmith import Client

client = Client()

# 创建数据集
dataset = client.create_dataset(
    dataset_name="langchain-qa-v1",
    description="LangChain 系列文章的问答评估集",
)

# 批量添加测试用例
examples = [
    {
        "inputs": {"question": "LCEL 的管道操作符是什么?"},
        "outputs": {"answer": "LCEL 使用 | 操作符将 Runnable 组件串联,前一个组件的输出作为后一个的输入。"},
    },
    {
        "inputs": {"question": "LangGraph 的 Checkpointer 有什么作用?"},
        "outputs": {"answer": "Checkpointer 负责持久化 Graph 的状态,支持工作流在中断后从断点恢复。"},
    },
    {
        "inputs": {"question": "RAG 中 Faithfulness 指标衡量什么?"},
        "outputs": {"answer": "Faithfulness 衡量 LLM 的回答是否忠实于检索到的 context,即答案中有多少比例的陈述可以从 context 中推导。"},
    },
]

client.create_examples(
    inputs=[e["inputs"] for e in examples],
    outputs=[e["outputs"] for e in examples],
    dataset_id=dataset.id,
)

print(f"Dataset 已创建:{dataset.id}")

5.2 从已有 Run 添加到 Dataset

在 LangSmith UI 中,可以直接将追踪到的 Run 一键添加到 Dataset——这是从生产流量中构建评估集的最便捷方式:

# 也可以通过代码将指定 Run 添加到 Dataset
client.create_example_from_run(
    run_id=run_id,               # 某次 Run 的 ID
    dataset_id=dataset.id,
    note="从生产流量中收集的高质量问答",
)

5.3 查看和更新 Dataset

# 列出所有 Dataset
for ds in client.list_datasets():
    print(f"{ds.name}: {ds.example_count} 条用例")

# 列出某个 Dataset 中的用例
for example in client.list_examples(dataset_id=dataset.id):
    print(f"输入:{example.inputs}")
    print(f"期望输出:{example.outputs}")
    print("---")

六、Evaluation:自动化评估

有了 Dataset,就可以运行自动化评估。LangSmith 的 evaluate 函数将 Dataset 中的每条用例发送给目标函数,并用评估器打分。

6.1 内置评估器

LangSmith 提供了开箱即用的 LLM-as-Judge 评估器:

from langsmith.evaluation import evaluate, LangChainStringEvaluator
from langchain_openai import ChatOpenAI

# 定义被评估的目标函数(接收 Dataset 的 inputs,返回 outputs)
def my_rag_app(inputs: dict) -> dict:
    question = inputs["question"]
    answer = chain.invoke({"question": question})  # 你的 RAG chain
    return {"answer": answer}

# 内置评估器:
# - "cot_qa":Chain-of-Thought QA 评估,对比答案与 reference 的事实准确性
# - "labeled_criteria":根据指定标准评分(helpfulness、correctness、conciseness 等)
# - "embedding_distance":计算答案与 reference 的语义相似度
evaluators = [
    LangChainStringEvaluator(
        "cot_qa",
        config={"llm": ChatOpenAI(model="gpt-4o-mini", temperature=0)},
        prepare_data=lambda run, example: {
            "prediction": run.outputs["answer"],
            "reference": example.outputs["answer"],
            "input": example.inputs["question"],
        },
    ),
]

# 运行评估
results = evaluate(
    my_rag_app,
    data="langchain-qa-v1",       # Dataset 名称
    evaluators=evaluators,
    experiment_prefix="rag-v1",   # 实验名称前缀,方便在 UI 中对比
    metadata={"version": "1.0", "model": "gpt-4o-mini"},
)

print(results.to_pandas())

6.2 自定义评估器

当内置评估器不满足需求时,可以编写自定义评估函数:

from langsmith.schemas import Run, Example

def check_answer_length(run: Run, example: Example) -> dict:
    """评估答案长度是否合理(不超过 200 字)"""
    answer = run.outputs.get("answer", "")
    char_count = len(answer)
    score = 1 if char_count <= 200 else max(0, 1 - (char_count - 200) / 200)
    return {
        "key": "answer_length_score",
        "score": score,
        "comment": f"答案长度 {char_count}{'(合理)' if char_count <= 200 else '(偏长)'}",
    }

def check_contains_code(run: Run, example: Example) -> dict:
    """评估技术问题的答案是否包含代码示例"""
    question = example.inputs.get("question", "")
    answer = run.outputs.get("answer", "")
    is_technical = any(kw in question for kw in ["如何", "怎么", "代码", "实现"])
    has_code = "```" in answer
    score = 1 if (not is_technical or has_code) else 0
    return {
        "key": "code_example_score",
        "score": score,
        "comment": "包含代码示例" if has_code else "技术问题缺少代码示例",
    }

# 使用自定义评估器
results = evaluate(
    my_rag_app,
    data="langchain-qa-v1",
    evaluators=[check_answer_length, check_contains_code],
    experiment_prefix="rag-v1-custom-eval",
)

6.3 结合 RAGAS 进行评估

第六篇介绍的 RAGAS 框架可以直接集成到 LangSmith 评估流程中:

from ragas.metrics import faithfulness, answer_relevancy
from ragas.integrations.langchain import EvaluatorChain

# 将 RAGAS 指标包装为 LangSmith 评估器
faithfulness_evaluator = EvaluatorChain(metric=faithfulness)

def my_rag_app_with_context(inputs: dict) -> dict:
    question = inputs["question"]
    docs = retriever.invoke(question)
    context = [d.page_content for d in docs]
    answer = rag_chain.invoke({"question": question, "context": "\n".join(context)})
    return {
        "answer": answer,
        "contexts": context,   # RAGAS 需要 context 字段
    }

results = evaluate(
    my_rag_app_with_context,
    data="langchain-qa-v1",
    evaluators=[faithfulness_evaluator],
    experiment_prefix="rag-ragas-eval",
)

七、Experiments:A/B 对比实验

Experiments 是 LangSmith 最核心的工程价值所在——在相同的 Dataset 上运行多个版本,直接对比指标,让"哪个版本更好"有数据支撑。

7.1 对比两个 Prompt 版本

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

# Prompt 版本 A:简单直接
prompt_v1 = ChatPromptTemplate.from_messages([
    ("system", "你是一个技术助手,请回答用户的问题。"),
    ("human", "{question}"),
])

# Prompt 版本 B:加入角色设定和输出格式约束
prompt_v2 = ChatPromptTemplate.from_messages([
    (
        "system",
        "你是一个专注于 LangChain 和 LLM 工程的资深工程师。\n"
        "回答时:先给出核心结论,再用代码示例佐证,最后说明适用场景。\n"
        "回答控制在 150 字以内。"
    ),
    ("human", "{question}"),
])

chain_v1 = prompt_v1 | llm | parser
chain_v2 = prompt_v2 | llm | parser

def run_v1(inputs: dict) -> dict:
    return {"answer": chain_v1.invoke({"question": inputs["question"]})}

def run_v2(inputs: dict) -> dict:
    return {"answer": chain_v2.invoke({"question": inputs["question"]})}

# 在相同 Dataset 上运行两个版本
evaluators = [check_answer_length, check_contains_code]

results_v1 = evaluate(
    run_v1,
    data="langchain-qa-v1",
    evaluators=evaluators,
    experiment_prefix="prompt-v1",
)

results_v2 = evaluate(
    run_v2,
    data="langchain-qa-v1",
    evaluators=evaluators,
    experiment_prefix="prompt-v2",
)

# 在 LangSmith UI 的 Experiments 面板中可以直接对比两次实验的得分

在 LangSmith UI 中,两次实验会并列展示:

  Experiment Comparison:

  Metric                  prompt-v1    prompt-v2    Delta
  ──────────────────────────────────────────────────────
  answer_length_score       0.72         0.91       +0.19 ✅
  code_example_score        0.50         0.83       +0.33 ✅
  cot_qa_score              0.68         0.74       +0.06 ✅
  avg_latency               1.2s         1.4s       +0.2s ⚠️
  avg_tokens                312          428        +116  ⚠️

7.2 对比不同模型

from langchain_anthropic import ChatAnthropic

# 用不同模型运行相同任务
def run_gpt4o_mini(inputs: dict) -> dict:
    chain = prompt_v2 | ChatOpenAI(model="gpt-4o-mini") | parser
    return {"answer": chain.invoke({"question": inputs["question"]})}

def run_claude_haiku(inputs: dict) -> dict:
    chain = prompt_v2 | ChatAnthropic(model="claude-haiku-4-5-20251001") | parser
    return {"answer": chain.invoke({"question": inputs["question"]})}

evaluate(run_gpt4o_mini, data="langchain-qa-v1",
         evaluators=evaluators, experiment_prefix="model-gpt4o-mini")
evaluate(run_claude_haiku, data="langchain-qa-v1",
         evaluators=evaluators, experiment_prefix="model-claude-haiku")

八、Prompt 版本管理

LangSmith 支持在云端托管和版本化 Prompt,团队成员可以共享和复用:

from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate

client = Client()

# 将 Prompt 推送到 LangSmith Hub(需要登录)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专注于 LangChain 的技术助手。回答简洁,附代码示例。"),
    ("human", "{question}"),
])

# 推送到 Hub,格式为 "用户名/prompt名称"
client.push_prompt("my-org/langchain-qa-prompt", object=prompt)

# 从 Hub 拉取(其他团队成员或生产环境)
from langchain import hub
pulled_prompt = hub.pull("my-org/langchain-qa-prompt")

# 拉取特定版本(commit hash)
specific_version = hub.pull("my-org/langchain-qa-prompt:abc1234")

💡 Prompt 版本管理的价值在于:每次实验都记录了使用的 Prompt 版本,出现问题时可以精确回溯到是哪一版的 Prompt 引发的。


九、生产环境最佳实践

9.1 按环境隔离项目

import os

# 根据环境变量切换 LangSmith 项目
env = os.getenv("APP_ENV", "development")
os.environ["LANGCHAIN_PROJECT"] = f"my-app-{env}"
# 开发:my-app-development
# 测试:my-app-staging
# 生产:my-app-production

9.2 采样追踪,控制成本

import random

# 对于高流量生产环境,只追踪一部分请求
TRACE_SAMPLE_RATE = 0.1  # 追踪 10% 的请求

def should_trace() -> bool:
    return random.random() < TRACE_SAMPLE_RATE

# 动态开关追踪
import langsmith
if should_trace():
    os.environ["LANGCHAIN_TRACING_V2"] = "true"
else:
    os.environ.pop("LANGCHAIN_TRACING_V2", None)

9.3 敏感数据脱敏

from langsmith import traceable

@traceable(
    name="user_query_handler",
    # 通过 process_inputs 对输入进行脱敏处理
    process_inputs=lambda inputs: {
        **inputs,
        "user_id": "***",         # 隐藏用户 ID
        "query": inputs["query"][:100] + "...",  # 截断长查询
    },
)
def handle_user_query(user_id: str, query: str) -> str:
    return chain.invoke({"question": query})

9.4 异步追踪,不阻塞主流程

# LangSmith 默认异步发送追踪数据,不阻塞请求响应
# 但在进程退出时需要确保数据已发送完毕
from langsmith import Client

client = Client()

# 在应用关闭时刷新缓冲区
import atexit
atexit.register(client.flush)

十、常见坑与最佳实践

坑一:忘记设置 LANGCHAIN_PROJECT,所有数据混在默认项目里

# ❌ 没有设置 LANGCHAIN_PROJECT
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxx
# 所有 Run 都进入默认的 "default" 项目,开发/测试/生产数据混在一起

# ✅ 明确指定项目名
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxx
LANGCHAIN_PROJECT=my-rag-app-production

坑二:评估器的 prepare_data 函数字段名与 Dataset 不匹配

# ❌ Dataset 中的字段名是 "answer",prepare_data 里写成了 "output"
evaluator = LangChainStringEvaluator(
    "cot_qa",
    prepare_data=lambda run, example: {
        "prediction": run.outputs["answer"],
        "reference": example.outputs["output"],  # ❌ 字段不存在,返回 None
    },
)

# ✅ 检查 Dataset example 的实际字段名
for example in client.list_examples(dataset_name="langchain-qa-v1"):
    print(example.outputs.keys())  # 先确认字段名
    break

坑三:在生产环境开启 verbose=True

# ❌ verbose=True 会将完整的输入输出打印到标准输出
# 生产环境日志量暴涨,还可能泄露敏感数据
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# ✅ 生产环境关闭 verbose,用 LangSmith Tracing 替代
executor = AgentExecutor(agent=agent, tools=tools, verbose=False)
# LangSmith 会自动记录所有中间步骤,不需要 verbose

坑四:evaluate 的目标函数签名不正确

# ❌ 目标函数接收的参数类型错误
def my_app(question: str) -> str:  # 错误:接收字符串
    return chain.invoke({"question": question})

# ✅ 必须接收 dict,返回 dict
def my_app(inputs: dict) -> dict:
    return {"answer": chain.invoke({"question": inputs["question"]})}

坑五:实验命名不规范,事后难以区分

# ❌ 命名随意,后期无法区分实验目的
evaluate(my_app, data="test", experiment_prefix="test1")
evaluate(my_app, data="test", experiment_prefix="test2")
evaluate(my_app, data="test", experiment_prefix="final")

# ✅ 命名包含:日期、模型、改动点
evaluate(my_app, data="langchain-qa-v1",
         experiment_prefix="2026-05-11-gpt4o-mini-prompt-v2-role-setting")

十一、总结

功能 使用场景 核心价值
Tracing 日常开发和生产监控 完整调用链路可视化,定位慢请求和错误根因
@traceable 非 LangChain 的自定义函数 将任意 Python 函数纳入追踪体系
Dataset 构建和管理测试用例 云端版本化管理,跨实验复用
evaluate 量化评估应用质量 自动化打分,替代人工抽查
Experiments 对比优化效果 A/B 测试不同 Prompt/模型,决策有数据支撑
Prompt Hub 团队协作和版本管理 Prompt 集中管理,可追溯每次实验使用的版本

🎯 LangSmith 的价值不在于"发现"问题,而在于让问题变得可重现、可量化、可对比。没有可观测性的 LLM 应用,每次优化都是在黑暗中摸索;有了 LangSmith,每次改动的效果都白纸黑字写在 Experiments 面板上。


参考资料


下期预告

工具、Agent、监控都就位了,最后一步是把应用真正部署到生产环境。

第十一篇《从 Demo 到上线:生产部署实战》 将介绍如何用 FastAPI 封装 LangChain 应用、实现 SSE 流式输出、处理并发与异步、设置重试与降级策略,以及通过 Token 用量监控控制成本——让 Demo 级别的代码变成生产可用的服务。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐