























本文以一个基于 LangGraph 构建的多步骤 Agent 为案例,完整演示如何在 LLM 应用中集成 LangSmith,覆盖追踪调试、测试数据集构建、自动化评估、OpenEvals 评估器对接等全流程。
LangSmith 是 LangChain 官方提供的 LLM 应用可观测性平台。当你的应用涉及多步 LLM 调用、Agent 编排、工具链组合时,LangSmith 能帮你:
我们将使用一个基于 LangGraph 的多步骤 Agent,具备条件分支路由能力:
用户输入 → 任务评估节点 → 条件路由
├── 简单任务 → 快速响应节点 → 结束
└── 复杂任务 → 数据收集 → 深度分析 → 结果生成 → 结束
这种架构在实际项目中非常常见,理解它的追踪和评估方式,可以帮助你快速将 LangSmith 集成到自己的项目中。
pip install langchain langchain-core langsmith langgraph langchain-community
LangSmith 的启用完全依赖环境变量,无需修改任何业务代码即可开启基础追踪。
| 环境变量 | 说明 | 是否必须 |
|---|---|---|
LANGSMITH_API_KEY |
从 LangSmith 平台获取的 API 密钥 | 是 |
LANGCHAIN_TRACING_V2 |
设置为 true 启用追踪 |
是 |
LANGCHAIN_PROJECT |
项目名称,用于在 LangSmith 中组织追踪记录 | 否(有默认值) |
LANGCHAIN_ENDPOINT |
LangSmith API 端点,一般使用默认即可 | 否 |
$env:LANGSMITH_API_KEY="lsv2_pt_你的密钥"
$env:LANGCHAIN_TRACING_V2="true"
$env:LANGCHAIN_PROJECT="my-agent-project"
export LANGSMITH_API_KEY="lsv2_pt_你的密钥"
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_PROJECT="my-agent-project"
如果不想每次打开终端都重新设置,可以写入用户级环境变量:
[System.Environment]::SetEnvironmentVariable("LANGCHAIN_TRACING_V2", "true", "User")
[System.Environment]::SetEnvironmentVariable("LANGSMITH_API_KEY", "lsv2_pt_你的密钥", "User")
[System.Environment]::SetEnvironmentVariable("LANGCHAIN_PROJECT", "my-agent-project", "User")
import os
LANGSMITH_ENABLED = os.getenv("LANGCHAIN_TRACING_V2", "").lower() == "true"
LANGSMITH_PROJECT = os.getenv("LANGCHAIN_PROJECT", "my-agent-project")
if LANGSMITH_ENABLED:
print(f"LangSmith 追踪已启用,项目: {LANGSMITH_PROJECT}")
print(f"查看追踪: https://smith.langchain.com")
else:
print("LangSmith 追踪未启用,请配置环境变量")
LangSmith 最强大的特性之一是:只要你使用了 LangChain 生态的组件,设好环境变量后,追踪自动生效。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.llms import Tongyi
llm = Tongyi(model_name="qwen-turbo-latest", dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"))
prompt = ChatPromptTemplate.from_template("请用一句话回答:{question}")
chain = prompt | llm | StrOutputParser()
# 直接调用 —— 环境变量已配置时,这次调用会自动被 LangSmith 记录
result = chain.invoke({"question": "什么是机器学习?"})
print(result)
运行后登录 LangSmith,你会看到一条追踪记录,包含:
对于更复杂的多步骤 Agent,LangSmith 会自动记录每个节点的执行过程:
from langgraph.graph import StateGraph, END
from langchain_core.output_parsers import JsonOutputParser
from typing import TypedDict, Optional, Dict, Any
import json
class AgentState(TypedDict):
user_input: str
task_type: Optional[str]
result: Optional[str]
error: Optional[str]
def assess_task(state: AgentState) -> AgentState:
"""评估节点:判断任务类型"""
prompt = ChatPromptTemplate.from_template(
"判断以下任务的类型,返回JSON(含task_type字段,值为simple或complex):{input}"
)
chain = prompt | llm | JsonOutputParser()
result = chain.invoke({"input": state["user_input"]})
return {**state, "task_type": result.get("task_type", "simple")}
def quick_response(state: AgentState) -> AgentState:
"""快速响应节点"""
prompt = ChatPromptTemplate.from_template("请简洁回答:{input}")
chain = prompt | llm | StrOutputParser()
response = chain.invoke({"input": state["user_input"]})
return {**state, "result": response}
def deep_analysis(state: AgentState) -> AgentState:
"""深度分析节点(多步 Chain)"""
# 第一步:收集信息
step1 = ChatPromptTemplate.from_template("分析以下问题需要哪些信息:{input}")
info_chain = step1 | llm | StrOutputParser()
info = info_chain.invoke({"input": state["user_input"]})
# 第二步:生成深度回答
step2 = ChatPromptTemplate.from_template("基于以下背景信息回答问题。\n问题:{input}\n背景:{info}")
answer_chain = step2 | llm | StrOutputParser()
response = answer_chain.invoke({"input": state["user_input"], "info": info})
return {**state, "result": response}
# 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("assess", assess_task)
workflow.add_node("quick", quick_response)
workflow.add_node("deep", deep_analysis)
workflow.set_entry_point("assess")
workflow.add_conditional_edges(
"assess",
lambda x: "quick" if x.get("task_type") == "simple" else "deep",
{"quick": "quick", "deep": "deep"}
)
workflow.add_edge("quick", END)
workflow.add_edge("deep", END)
agent = workflow.compile()
# 运行 —— 每个节点的 LLM 调用都会自动被追踪
result = agent.invoke({
"user_input": "什么是机器学习?",
"task_type": None,
"result": None,
"error": None
})
在 LangSmith 中你会看到一棵完整的 Trace 树:
assess (评估节点)
└── LLM Call: 判断任务类型
quick (快速响应)
└── LLM Call: 生成回答
或者对于复杂任务:
assess (评估节点)
└── LLM Call: 判断任务类型
deep (深度分析)
├── LLM Call: 收集信息
└── LLM Call: 生成深度回答
每个 LLM Call 都可以展开查看完整的 Prompt、输出、耗时和 Token 消耗。
当你需要在 LangSmith 中对大量追踪记录进行分类、筛选和分析时,就需要用到 RunnableConfig。
from langchain_core.runnables import RunnableConfig
from datetime import datetime
config = RunnableConfig(
# 标签:用于分类筛选
tags=["my-agent", "production", "v2"],
# 元数据:结构化上下文信息,支持搜索
metadata={
"user_id": "U12345",
"session_id": "S67890",
"user_input": "用户输入的摘要"[:100],
"timestamp": datetime.now().isoformat()
},
# 运行名称:在追踪列表中快速识别
run_name=f"agent-U12345-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
)
# 将 config 传给 invoke
result = agent.invoke(initial_state, config=config)
| 配置项 | 作用 | 典型用法 |
|---|---|---|
| tags | 分类标签,支持在 LangSmith 中按标签筛选 | 按环境(dev/prod)、版本、用户类型分类 |
| metadata | 结构化键值对,支持搜索和过滤 | 记录用户ID、会话ID、输入摘要等 |
| run_name | 运行名称,显示在追踪列表 | 包含用户ID和时间戳,便于定位 |
在实际项目中,可以按需决定是否传入追踪配置:
LANGSMITH_ENABLED = os.getenv("LANGCHAIN_TRACING_V2", "").lower() == "true"
config = {}
if LANGSMITH_ENABLED:
config = RunnableConfig(
tags=["my-agent", "production"],
metadata={
"user_id": user_id,
"session_id": session_id,
"timestamp": datetime.now().isoformat()
},
run_name=f"agent-{user_id}-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
)
if config:
result = agent.invoke(initial_state, config=config)
else:
result = agent.invoke(initial_state)
这是 LangSmith 最核心的高级功能——创建测试数据集、运行批量评估、对比不同版本的效果。
from langsmith import Client
client = Client(api_key=os.getenv("LANGSMITH_API_KEY"))
根据 Agent 的不同场景设计测试用例,每个用例包含输入和期望输出:
TEST_CASES = [
# 简单任务用例
{
"inputs": {
"user_input": "什么是机器学习?",
"user_id": "test-user-1"
},
"expected_outputs": {
"task_type": "simple",
"should_contain": ["数据", "学习", "模型"]
}
},
{
"inputs": {
"user_input": "Python 的列表和元组有什么区别?",
"user_id": "test-user-1"
},
"expected_outputs": {
"task_type": "simple",
"should_contain": ["列表", "元组", "可变"]
}
},
# 复杂任务用例
{
"inputs": {
"user_input": "请帮我设计一个推荐系统的技术方案",
"user_id": "test-user-2"
},
"expected_outputs": {
"task_type": "complex",
"should_contain": ["协同过滤", "特征", "模型", "评估"]
}
},
{
"inputs": {
"user_input": "如何优化一个高并发的 Web 服务?",
"user_id": "test-user-2"
},
"expected_outputs": {
"task_type": "complex",
"should_contain": ["缓存", "负载均衡", "数据库", "优化"]
}
},
]
将测试用例上传到 LangSmith 平台:
def create_test_dataset(dataset_name: str = "agent-test-dataset"):
"""创建测试数据集"""
try:
client.read_dataset(dataset_name=dataset_name)
print(f"数据集已存在: {dataset_name}")
except:
client.create_dataset(
dataset_name=dataset_name,
description="Agent 测试数据集"
)
print(f"数据集已创建: {dataset_name}")
for test_case in TEST_CASES:
client.create_example(
inputs=test_case["inputs"],
outputs=test_case["expected_outputs"],
dataset_name=dataset_name
)
print(f"已添加 {len(TEST_CASES)} 个测试用例")
return dataset_name
dataset_name = create_test_dataset()
LangSmith 支持自定义评估器来检验 Agent 的输出质量:
(1)路由正确性评估器 —— 判断 Agent 是否选择了正确的处理路径
from langsmith import RunEvaluator
from langsmith.schemas import Run, Example
class TaskTypeEvaluator(RunEvaluator):
"""评估任务类型判断是否正确"""
def evaluate_run(self, run: Run, example: Example, **kwargs):
expected_type = example.outputs.get("task_type")
if not expected_type:
return {"key": "task_type", "score": None}
actual_type = run.outputs.get("task_type") if run.outputs else None
if actual_type == expected_type:
return {
"key": "task_type",
"score": 1.0,
"comment": f"任务类型正确: {actual_type}"
}
else:
return {
"key": "task_type",
"score": 0.0,
"comment": f"期望 {expected_type},实际 {actual_type}"
}
(2)响应完整性评估器 —— 检查回答是否包含关键信息
class ResponseCompletenessEvaluator(RunEvaluator):
"""评估响应是否包含期望的关键词"""
def evaluate_run(self, run: Run, example: Example, **kwargs):
response = run.outputs.get("result", "") if run.outputs else ""
expected_keywords = example.outputs.get("should_contain", [])
if not expected_keywords:
return {"key": "completeness", "score": None}
found = [kw for kw in expected_keywords if kw.lower() in response.lower()]
score = len(found) / len(expected_keywords)
return {
"key": "completeness",
"score": score,
"comment": f"找到 {len(found)}/{len(expected_keywords)} 个关键词"
}
使用 LangSmith 的 evaluate() 函数一键运行评估:
from langsmith.evaluation import evaluate
def predict(example: Example) -> dict:
"""预测函数:运行 Agent 并返回结果"""
user_input = example.inputs.get("user_input", "")
result = agent.invoke({
"user_input": user_input,
"task_type": None,
"result": None,
"error": None
})
return {
"result": result.get("result", ""),
"task_type": result.get("task_type", "unknown"),
}
results = evaluate(
predict, # 预测函数
data="agent-test-dataset", # 测试数据集
evaluators=[ # 评估器列表
TaskTypeEvaluator(),
ResponseCompletenessEvaluator()
],
experiment_prefix="agent-eval-v1", # 实验名称前缀
max_concurrency=1 # 串行执行,避免 API 限流
)
print("评估完成!登录 https://smith.langchain.com 查看详细结果")
运行后,在 LangSmith 的 Experiments 页面中可以看到:
除了自定义评估器,我们还可以使用 OpenEvals —— 一个开源评估器框架,提供大量预置的评估 Prompt,覆盖正确性、相关性、简洁性、幻觉检测、毒性检测等维度。
pip install openevals
from openevals.prompts import (
CORRECTNESS_PROMPT, # 正确性评估
ANSWER_RELEVANCE_PROMPT, # 回答相关性
CONCISENESS_PROMPT, # 简洁性
RAG_HELPFULNESS_PROMPT, # RAG 帮助性
HALLUCINATION_PROMPT, # 幻觉检测
TOXICITY_PROMPT, # 毒性检测
)
from openevals.llm import create_llm_as_judge
from langchain_community.chat_models import ChatTongyi
# 创建用于评估的 LLM(裁判模型)
eval_llm = ChatTongyi(
model_name="qwen-turbo",
dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
temperature=0
)
# 创建多种评估器
evaluators = [
create_llm_as_judge(
prompt=ANSWER_RELEVANCE_PROMPT,
feedback_key="relevance",
judge=eval_llm,
continuous=True,
use_reasoning=False,
),
create_llm_as_judge(
prompt=CONCISENESS_PROMPT,
feedback_key="conciseness",
judge=eval_llm,
continuous=True,
use_reasoning=False,
),
create_llm_as_judge(
prompt=RAG_HELPFULNESS_PROMPT,
feedback_key="helpfulness",
judge=eval_llm,
continuous=True,
use_reasoning=False,
),
create_llm_as_judge(
prompt=HALLUCINATION_PROMPT,
feedback_key="hallucination",
judge=eval_llm,
continuous=True,
use_reasoning=False,
),
create_llm_as_judge(
prompt=TOXICITY_PROMPT,
feedback_key="toxicity",
judge=eval_llm,
continuous=True,
use_reasoning=False,
),
]
| 评估器 | 评估维度 | 分数含义 |
|---|---|---|
CORRECTNESS_PROMPT |
正确性 | 输出与参考答案的一致程度(0-1) |
ANSWER_RELEVANCE_PROMPT |
相关性 | 回答是否切中问题要害 |
CONCISENESS_PROMPT |
简洁性 | 回答是否精炼,不冗余 |
RAG_HELPFULNESS_PROMPT |
帮助性 | 回答对用户是否有帮助 |
HALLUCINATION_PROMPT |
幻觉检测 | 是否包含编造的信息(分数越高幻觉越少) |
TOXICITY_PROMPT |
毒性检测 | 是否包含有害内容 |
OpenEvals 的评估器也可以独立使用,无需 LangSmith 平台:
correctness_evaluator = create_llm_as_judge(
prompt=CORRECTNESS_PROMPT,
feedback_key="correctness",
judge=eval_llm,
continuous=True,
use_reasoning=False,
)
result = correctness_evaluator(
inputs="什么是机器学习?",
outputs="机器学习是人工智能的一个分支,通过算法让计算机从数据中学习规律。",
reference_outputs="机器学习是让计算机从数据中自动学习模式的技术。"
)
print(f"正确性评分: {result.get('score', result):.3f}")
# 输出示例: 正确性评分: 0.800
将 OpenEvals 评估器与自定义评估器组合,接入 LangSmith 的 evaluate() 进行全面评估:
def predict_for_eval(inputs: dict) -> dict:
"""预测函数,包装 Agent"""
user_input = inputs.get("user_input", "")
result = agent.invoke({
"user_input": user_input,
"task_type": None,
"result": None,
"error": None
})
return {
"output": result.get("result", ""),
"task_type": result.get("task_type", "unknown"),
}
results = evaluate(
predict_for_eval,
data="agent-test-dataset",
evaluators=evaluators, # 包含 5 个 OpenEvals 评估器
experiment_prefix="agent-openevals-v1",
max_concurrency=1
)
print("评估完成!可在 LangSmith 查看多个维度的评估结果")
# 开发环境
$env:LANGCHAIN_PROJECT="my-agent-dev"
# 测试环境
$env:LANGCHAIN_PROJECT="my-agent-staging"
# 生产环境
$env:LANGCHAIN_PROJECT="my-agent-prod"
在 LangSmith 中通过切换项目名,可以轻松对比不同环境下的表现差异。
在元数据中避免记录完整的用户输入:
metadata={
"user_input": user_input[:100], # 只记录前100个字符
"user_id": user_id, # 用户ID可以记录
}
即使发生异常,也应让程序正常捕获并返回错误信息,这样 LangSmith 会记录到失败的 Trace:
try:
result = agent.invoke(initial_state, config=config)
return result
except Exception as e:
return {
"error": f"执行过程中发生错误: {str(e)}",
"result": "处理您的请求时出现了问题。"
}
通过不同的 experiment_prefix 来对比不同版本的效果:
# v1 版本
results_v1 = evaluate(predict_v1, data=dataset_name,
evaluators=evaluators, experiment_prefix="agent-v1")
# v2 版本
results_v2 = evaluate(predict_v2, data=dataset_name,
evaluators=evaluators, experiment_prefix="agent-v2")
# 在 LangSmith 的 Experiments 页面中可以直接对比两个版本的分数
LANGCHAIN_TRACING_V2 的值是小写的 "true"(不是 "True" 或 "TRUE")LANGSMITH_API_KEY 正确无误,且未过期langchain、langgraph 等原生支持追踪)api.smith.langchain.com通常几秒内就能在 LangSmith 平台看到,但高并发场景下可能有 10-30 秒的延迟。
追踪数据的上报是异步的,对主流程的性能影响极小,通常可以忽略不计。
每个评估用例都需要调用 LLM(例如 4 个用例 × 5 个评估器 = 20 次 LLM 调用)。建议:
max_concurrency=1 串行执行,避免 API 限流LangSmith 提供免费层级,适合开发调试使用。每月有一定的追踪次数上限。生产环境的大规模追踪建议查看官方的定价方案。
| 场景 | 核心操作 | 适用阶段 |
|---|---|---|
| 自动追踪 | 设置环境变量,零代码改动 | 开发调试 |
| 标签/元数据 | 使用 RunnableConfig 传入 tags/metadata |
测试验证 |
| 测试数据集+评估 | Client 创建数据集,evaluate() 批量评估 |
质量保障 |
| OpenEvals 评估器 | create_llm_as_judge() 创建多维度评估器 |
全面评估 |
完整的集成路径:
1. 设环境变量 → 自动获得追踪能力
2. 加 RunnableConfig → 追踪记录可分类、可搜索
3. 建测试数据集 → 标准化评估流程
4. 接入 OpenEvals → 多维度量化评估输出质量
5. 版本对比实验 → 持续迭代优化 Agent
LangSmith 的集成非常轻量——从设置两个环境变量开始,到构建完整的评估体系,每一步都是增量式的。这是它作为 LLM 应用可观测性工具最大的优势。
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。