作者:PySuper | 来源:zhengxingtao.com | 更新日期:2026-07-15
系列第 62 篇 · Python/AI工程化 | 本篇为 LangSmith 指南
关联阅读:[第6篇 OpenTelemetry + LangFuse 实践] · [第24篇 Agent 可观测性] · [第42篇 LangGraph 快速上手] · [第43篇 Tool Calling 实战]
零、60 秒了解全貌
如果你赶时间,先看这张图,对 LangSmith 建立整体印象:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith 全局视角 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Tracing │ │ Evaluation │ │ Prompt Hub │ │ Monitoring │ │
│ │ 链路追踪 │ │ 质量评估 │ │ 提示词管理 │ │ 生产监控 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ LangSmith Platform │ │
│ │ │ │
│ │ ┌──────────┐ ┌───────────┐ ┌───────────┐ ┌──────────────────┐ │ │
│ │ │ SmithDB │ │ Engine │ │ Context │ │ LLM Gateway │ │ │
│ │ │ 专用数据库│ │ 自动修复 │ │ Hub │ │ 消费限制/脱敏 │ │ │
│ │ └──────────┘ └───────────┘ └───────────┘ └──────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ 设置 LANGSMITH_API_KEY 即自动追踪 │
│ │ │
│ ┌──────┴───────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ LangChain │ │ LangGraph │ │ 任意框架 │ │
│ │ 零配置接入 │ │ 节点级追踪 │ │ @traceable │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
本文目标:从零接入 LangSmith,掌握 Tracing、Evaluation、Prompt Hub、Monitoring 四大核心能力,并在生产环境稳定运行。
一、LangSmith 是什么:为什么 LangGraph 开发者离不开它
1.1 一句话定义
LangSmith = LangChain 官方出品的 LLM 可观测性 + 评估平台。
它解决的核心问题就一个:把 AI 应用内部每一步执行细节"透明化"。
你写了一个 Agent,用户说"回答有问题",你能看到什么?传统方式下,你只能看到输入和输出,中间发生了什么——模型调了几次?工具返回了什么?哪一步出了偏差?全是黑盒。
LangSmith 就是打开这个黑盒的工具。
1.2 核心价值
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ 没有 LangSmith vs 有 LangSmith │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 没有 LangSmith: │
│ ┌────────────────────────────────────────────────────────────────────┐ │
│ │ 用户输入 ──→ [???黑盒???] ──→ 输出不对 │ │
│ │ │ │
│ │ 调试方式:print() 大法、猜、反复跑 │ │
│ │ 成本追踪:月末看账单,心惊肉跳 │ │
│ │ 质量评估:人肉看,凭感觉 │ │
│ │ Prompt 管理:Git 里存个 txt,改了啥全靠注释 │ │
│ └────────────────────────────────────────────────────────────────────┘ │
│ │
│ 有 LangSmith: │
│ ┌────────────────────────────────────────────────────────────────────┐ │
│ │ 用户输入 ──→ [Step1: LLM] ──→ [Step2: Tool] ──→ [Step3: LLM] │ │
│ │ ↓ ↓ ↓ │ │
│ │ Token: 1,247 Args: {...} Token: 523 │ │
│ │ Time: 1.2s Result: ... Time: 0.8s │ │
│ │ Cost: $0.02 Error: None Cost: $0.01 │ │
│ │ │ │
│ │ 调试方式:可视化 Trace 树,点哪看哪 │ │
│ │ 成本追踪:实时 Dashboard,按项目/模型/用户 │ │
│ │ 质量评估:自动化 Evaluator,数据说话 │ │
│ │ Prompt 管理:版本化、A/B 测试、一键回滚 │ │
│ └────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
1.3 与 LangGraph 的关系
LangSmith 和 LangGraph 的关系简单到离谱:设置一个环境变量,LangGraph 的每个节点、每条边、每次工具调用就自动被追踪了。
不需要改代码,不需要加装饰器,不需要手动埋点。
python
# 就这三行,LangGraph 的整个执行过程就透明了
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
# 你的 LangGraph 代码完全不用改
这就是"框架原生"的优势——LangSmith 和 LangGraph 是同一团队出品,追踪逻辑直接内置在 LangGraph 的运行时里。
1.4 2026 年新动态
2026 年 5 月的 Interrupt 大会上,LangChain 放了一堆大招(据《Everything we shipped at Interrupt》https://www.langchain.com/blog/interrupt-2026-overview):
表格
后面第八章会深度解析这些新功能。先搞定基础。
二、核心概念:Run Tree 追踪体系
2.1 Trace 和 Span/Run
理解 LangSmith 的追踪体系,只需要记住两个概念:
Trace:一次完整的用户请求,从输入到输出,对应一棵树
Run(也叫 Span) :Trace 中的每一个步骤,是树中的一个节点
一个 Trace 就是一个树状结构,每个 Run 可以有子 Run,形成嵌套的父子关系。
2.2 Run 类型
LangSmith 定义了几种 Run 类型,每种对应不同的执行阶段:
表格
2.3 Run Tree 示例
一个典型的 Agent 请求展开后的 Run Tree 长这样:
plaintext
┌─ Trace: "用户查询北京天气" ──────────────────────────────────────────────────┐
│ │
│ Run [agent] ChatAgent ─────────────────── 总耗时: 3.2s ─── 总Token: 2,847 │
│ │ │
│ ├── Run [llm] GPT-4o (第1次推理) ──────── 耗时: 1.2s ─── Token: 1,247/42 │
│ │ └── 输出: 决定调用 get_weather 工具 │
│ │ │
│ ├── Run [tool] get_weather ────────────── 耗时: 0.3s ─── Token: 0 │
│ │ ├── 输入: {"city": "北京", "date": "2026-07-15"} │
│ │ └── 输出: {"temp": "32°C", "condition": "晴", "humidity": "45%"} │
│ │ │
│ └── Run [llm] GPT-4o (第2次推理) ──────── 耗时: 0.9s ─── Token: 3,102/156 │
│ └── 输出: "北京今天32°C,晴天,湿度45%..." │
│ │
└────────────────────────────────────────────────────────────────────────────┘
这棵树让你一眼就能看到:
Agent 做了几次 LLM 调用(2 次)
调了什么工具(get_weather),参数和返回值是什么
每一步的耗时和 Token 消耗
哪一步是瓶颈(第 1 次 LLM 调用最慢)
2.4 异步无阻塞遥测
LangSmith 的追踪是异步批处理的:你的业务代码正常执行,Trace 数据在后台线程里攒批发送。不影响业务性能,不阻塞主流程。
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith 遥测机制 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 业务线程 后台线程 │
│ ───────── ────────── │
│ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Agent │ Run 事件写入内存队列 │ Batch │ │
│ │ 执行 │ ────────────────────────→│ Processor │ │
│ │ │ (不阻塞,微秒级) │ │ │
│ └──────────┘ │ 定时/定量 │ │
│ │ 批量发送 │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ LangSmith │ │
│ │ Server │ │
│ └──────────────┘ │
│ │
│ 注意:Serverless 环境需设置 LANGCHAIN_CALLBACKS_BACKGROUND=false │
│ 确保函数退出前 Trace 已发送 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
重要:如果你在 Serverless 环境(AWS Lambda、Cloud Functions)运行,必须设置
LANGCHAIN_CALLBACKS_BACKGROUND=false,否则函数可能在 Trace 发送前就退出了。
三、5 分钟上手:从零接入 LangSmith
3.1 注册账号、创建 API Key
访问 smith.langchain.com,用 GitHub/Discord/邮箱注册
进入 Settings → API Keys → Create API Key
复制 API Key(格式:
ls_...),只显示一次,务必保存
3.2 环境变量配置
bash
# 必填
export LANGSMITH_TRACING=true # 开启追踪
export LANGSMITH_API_KEY="ls_..." # 你的 API Key
# 可选
export LANGSMITH_PROJECT="my-agent-project" # 项目名,默认 "default"
export LANGSMITH_ENDPOINT="https://api.smith.langchain.com" # 默认值,欧盟用户改 eu.api
或者写在 .env 文件里:
bash
# .env
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=ls_...
LANGSMITH_PROJECT=my-agent-project
3.3 实战1:LangChain 应用接入(最简示例)
python
"""
LangSmith 接入实战1:LangChain 应用
3 行配置,零代码改动
"""
import os
# 第 1 步:设置环境变量(也可以在 .env 中配置)
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..." # 替换为你的 Key
os.environ["LANGSMITH_PROJECT"] = "langchain-demo"
# 第 2 步:正常写你的 LangChain 代码,完全不用改
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的 Python 编程助手。"),
("human", "{question}"),
])
chain = prompt | llm
# 第 3 步:正常调用,Trace 自动上传到 LangSmith
response = chain.invoke({"question": "Python 的 GIL 是什么?"})
print(response.content)
运行后,打开 smith.langchain.com,在 langchain-demo 项目下就能看到这次 Trace 了。
3.4 实战2:LangGraph Agent 接入
基于第 42 篇的第一个 Agent,加上 LangSmith 后的效果:
python
"""
LangSmith 接入实战2:LangGraph Agent
和实战1一样,只需配置环境变量,LangGraph 自动追踪每个节点
"""
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "langgraph-agent-demo"
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
# ---- 定义 State ----
class State(TypedDict):
messages: Annotated[list, add_messages]
# ---- 定义工具 ----
@tool
def get_weather(city: str) -> str:
"""查询指定城市的天气信息"""
# 模拟天气 API
weather_data = {
"北京": "32°C,晴天,湿度45%",
"上海": "28°C,多云,湿度72%",
"深圳": "35°C,雷阵雨,湿度85%",
}
return weather_data.get(city, f"{city}暂无天气数据")
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression) # 生产环境请用更安全的方式
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {e}"
# ---- 构建 Agent ----
tools = [get_weather, calculate]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
def chatbot(state: State) -> dict:
"""Agent 节点:调用 LLM 决定下一步"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
# ---- 构建图 ----
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools))
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
# ---- 运行 ----
result = graph.invoke({
"messages": [{"role": "user", "content": "北京今天天气怎么样?"}]
})
print(result["messages"][-1].content)
打开 LangSmith UI,你会看到这个 Agent 的 Trace 树:
plaintext
┌─ Trace: "北京天气查询" ────────────────────────────────────────────────────┐
│ │
│ Run [chain] LangGraph ──────────────────── 总耗时: 2.8s │
│ │ │
│ ├── Run [chain] chatbot ────────────────── 耗时: 1.1s │
│ │ └── Run [llm] ChatOpenAI ──────────── Token: 847/28 │
│ │ └── 决定调用 get_weather │
│ │ │
│ ├── Run [chain] tools ──────────────────── 耗时: 0.02s │
│ │ └── Run [tool] get_weather ────────── 输入: {city: "北京"} │
│ │ 输出: "32°C,晴天..." │
│ │ │
│ └── Run [chain] chatbot ────────────────── 耗时: 0.8s │
│ └── Run [llm] ChatOpenAI ──────────── Token: 1,102/86 │
│ └── "北京今天32°C,晴天,湿度45%..." │
│ │
└────────────────────────────────────────────────────────────────────────────┘
零代码改动,每个节点的输入/输出/耗时/Token 全都看到了。 这就是 LangSmith + LangGraph 的威力。
3.5 实战3:非 LangChain 应用接入
如果你不用 LangChain/LangGraph,LangSmith 也能用。通过 @traceable 装饰器和 wrap_openai 包装器手动追踪:
python
"""
LangSmith 接入实战3:非 LangChain 应用
用 @traceable 装饰器手动追踪
"""
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "custom-framework-demo"
import openai
from langsmith import traceable
from langsmith.wrappers import wrap_openai
# wrap_openai 让 OpenAI 调用自动成为子 Run
client = wrap_openai(openai.Client())
@traceable(name="retrieve_docs", run_type="retriever")
def retrieve_docs(query: str) -> list[str]:
"""检索相关文档(模拟向量库检索)"""
docs = [
"Python GIL 是全局解释器锁...",
"GIL 确保同一时刻只有一个线程执行 Python 字节码...",
"多线程 I/O 密集型任务不受 GIL 影响...",
]
return docs
@traceable(name="generate_answer", run_type="chain")
def generate_answer(query: str, docs: list[str]) -> str:
"""基于检索文档生成回答"""
context = "\n".join(docs)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": f"根据以下上下文回答问题:\n{context}"},
{"role": "user", "content": query},
],
)
return response.choices[0].message.content
@traceable(name="rag_pipeline", run_type="chain")
def rag_pipeline(query: str) -> str:
"""RAG 主流程"""
docs = retrieve_docs(query) # 自动成为子 Run
answer = generate_answer(query, docs) # 自动成为子 Run
return answer
# 调用
result = rag_pipeline("Python GIL 是什么?")
print(result)
@traceable 的嵌套调用会自动形成父子 Run 关系,最终在 LangSmith UI 中看到和 LangGraph 类似的树状结构。
3.6 在 LangSmith UI 中查看 Trace
左侧选择你的 Project
在 Trace 列表里点击任意一条
中间面板是 Run Tree,点击每个 Run 可以展开详情
右侧面板是选中 Run 的输入/输出/Metadata/Tags
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith UI 布局示意 │
├──────────────────┬──────────────────────────┬───────────────────────────────┤
│ Trace 列表 │ Run Tree │ Run 详情 │
│ │ │ │
│ ▶ Trace 1 ✓ │ ● agent (3.2s) │ ┌─ Input ─────────────────┐ │
│ ▶ Trace 2 ✗ │ ├── ● llm (1.1s) │ │ messages: [...] │ │
│ ▶ Trace 3 ✓ │ ├── ● tool (0.3s) │ └─────────────────────────┘ │
│ ▶ Trace 4 ✓ │ └── ● llm (0.9s) │ ┌─ Output ────────────────┐ │
│ │ │ │ content: "北京今天..." │ │
│ │ │ │ tokens: 523 │ │
│ │ │ └─────────────────────────┘ │
│ │ │ ┌─ Metadata ───────────────┐ │
│ │ │ │ model: gpt-4o-mini │ │
│ │ │ │ user_id: u-123 │ │
│ │ │ └─────────────────────────┘ │
└──────────────────┴──────────────────────────┴───────────────────────────────┘
四、Tracing 深度实战
第三章是"能用",这章是"用好"。
4.1 LangGraph 节点级追踪
LangGraph 的每个节点(Node)在 LangSmith 中自动成为一个 Run,你可以看到:
节点的输入 State(进入节点时的状态)
节点的输出 State(节点返回的状态更新)
节点的耗时
节点内调用的子步骤(LLM、Tool 等)
python
"""
LangGraph 节点级追踪示例
每个节点的输入/输出/耗时一目了然
"""
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "langgraph-tracing-deep"
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
class State(TypedDict):
messages: Annotated[list, add_messages]
search_results: list[str] # 新增:搜索结果存储
@tool
def web_search(query: str) -> str:
"""搜索互联网信息"""
# 模拟搜索 API
return f"搜索结果:{query}的相关信息包括...(模拟数据)"
@tool
def code_execute(code: str) -> str:
"""执行 Python 代码"""
try:
result = eval(code)
return f"执行结果: {result}"
except Exception as e:
return f"执行错误: {e}"
tools = [web_search, code_execute]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
def analyze_intent(state: State) -> dict:
"""节点1:分析用户意图"""
messages = state["messages"]
response = llm.invoke(
messages + [{"role": "system", "content": "分析用户意图,决定需要搜索还是计算"}]
)
return {"messages": [response]}
def process_results(state: State) -> dict:
"""节点2:处理工具结果"""
messages = state["messages"]
# 对工具结果进行总结
response = llm.invoke(
messages + [{"role": "system", "content": "基于工具结果,给出最终回答"}]
)
return {"messages": [response]}
# 构建图
builder = StateGraph(State)
builder.add_node("analyze_intent", analyze_intent)
builder.add_node("tools", ToolNode(tools))
builder.add_node("process_results", process_results)
builder.add_edge(START, "analyze_intent")
builder.add_conditional_edges("analyze_intent", tools_condition)
builder.add_edge("tools", "process_results")
builder.add_edge("process_results", END)
builder.add_edge("analyze_intent", END)
graph = builder.compile()
# 运行
result = graph.invoke({
"messages": [{"role": "user", "content": "搜索 LangGraph 最新版本号"}]
})
在 LangSmith UI 中,你会看到三个节点各自成为独立的 Run:
plaintext
┌─ Trace ────────────────────────────────────────────────────────────────────┐
│ │
│ Run [chain] LangGraph ─────────────────── 总耗时: 4.1s │
│ │ │
│ ├── Run [chain] analyze_intent ──────────── 耗时: 1.3s │
│ │ │ 输入: {messages: [{"role":"user","content":"搜索..."}]} │
│ │ │ 输出: {messages: [..., tool_call: web_search]} │
│ │ └── Run [llm] ChatOpenAI ──────────── Token: 1,247/42 │
│ │ │
│ ├── Run [chain] tools ──────────────────── 耗时: 0.5s │
│ │ └── Run [tool] web_search ────────── 输入: {query: "LangGraph..."} │
│ │ 输出: "搜索结果:LangGraph..." │
│ │ │
│ └── Run [chain] process_results ─────────── 耗时: 1.8s │
│ │ 输入: {messages: [..., tool_result]} │
│ │ 输出: {messages: [..., "根据搜索结果..."]} │
│ └── Run [llm] ChatOpenAI ──────────── Token: 2,102/156 │
│ │
└────────────────────────────────────────────────────────────────────────────┘
4.2 工具调用追踪
ToolNode 的每次调用都被详细记录:函数名、参数、返回值、耗时。如果你的工具调用出问题了,在 LangSmith 里点开对应的 Tool Run 就能一目了然。
4.3 条件边追踪
条件边(Conditional Edge)的路由决策也会在 Trace 中体现。当你用 tools_condition 时,LangSmith 会显示 LLM 是否决定调用工具(路由到 tools 节点)还是直接回答(路由到 END)。
4.4 自定义 Metadata:给 Trace 打标签
默认的追踪已经够用了,但在生产环境,你需要更多的上下文信息来定位问题。比如:
这个请求是哪个用户的?
跑的是哪个环境(dev/staging/prod)?
用的是哪个版本(v1.2.3)?
python
"""
自定义 Metadata 和 Tags
"""
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "metadata-demo"
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的编程助手。"),
("human", "{question}"),
])
chain = prompt | llm
# 通过 config 传入 Metadata 和 Tags
response = chain.invoke(
{"question": "什么是 RAG?"},
config={
"metadata": {
"user_id": "u-12345",
"environment": "production",
"version": "v2.1.0",
"session_id": "sess-abc-def",
},
"tags": ["production", "rag-qa", "v2"],
},
)
print(response.content)
在 LangSmith UI 中,这些 Metadata 和 Tags 会显示在 Run 详情的 Metadata 区域,而且可以用它们来过滤和搜索 Trace。
4.5 Tags 和 Metadata 的过滤与搜索
LangSmith 支持按 Metadata 和 Tags 过滤 Trace:
在 UI 中:点击 Filter → 按 Tags/Metadata 筛选
通过 API:
python
"""
通过 API 搜索和过滤 Trace
"""
from langsmith import Client
client = Client()
# 按项目列出 Trace
runs = client.list_runs(
project_name="metadata-demo",
# 按 Tag 过滤
filter='tags = "production"',
# 按 Metadata 过滤
# filter='metadata.environment = "production"',
# 只看错误的
# is_root=True,
# error=True,
)
for run in runs:
print(f"Run: {run.name} | Status: {'ERROR' if run.error else 'OK'} | Time: {run.total_time}")
4.6 实战:给复杂 LangGraph Agent 加上完整追踪配置
把前面学到的全用上:
python
"""
实战:完整的 LangGraph Agent 追踪配置
包含 Metadata、Tags、Project 分离
"""
import os
import uuid
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "agent-production"
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
class State(TypedDict):
messages: Annotated[list, add_messages]
@tool
def search_knowledge_base(query: str) -> str:
"""搜索知识库"""
return f"知识库搜索结果:{query}相关内容..."
@tool
def query_database(sql: str) -> str:
"""查询数据库"""
return f"数据库查询结果:执行 {sql} 返回3条记录..."
@tool
def send_notification(to: str, message: str) -> str:
"""发送通知"""
return f"已发送通知给 {to}:{message[:50]}..."
tools = [search_knowledge_base, query_database, send_notification]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
def agent_node(state: State) -> dict:
"""Agent 决策节点"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
builder.add_edge("agent", END)
graph = builder.compile()
# ---- 运行时传入追踪配置 ----
session_id = str(uuid.uuid4())
result = graph.invoke(
{"messages": [{"role": "user", "content": "查一下本月销售额,然后通知运营团队"}]},
config={
# Metadata:结构化键值对,用于过滤
"metadata": {
"user_id": "u-67890",
"environment": "production",
"version": "v3.0.0",
"session_id": session_id, # 多轮对话追踪
"team": "sales-ops",
},
# Tags:标签,用于分类
"tags": [
"production",
"sales-agent",
"v3",
"multi-tool",
],
# Runnable 名称前缀(在 Trace 中更易识别)
"run_name": "SalesAgent-QueryAndNotify",
},
)
print(result["messages"][-1].content)
五、Evaluation:让 Agent 质量可量化
5.1 为什么需要评估
Agent 的输出不可预测——同一句话,今天回答正确,明天可能就错了。你不能只靠人看,你需要自动化评估。
"改了 Prompt 感觉效果变好了"是最危险的判断。LangSmith 的 Evaluation 模块让你用数据说话。
5.2 LangSmith 评估体系
三个核心概念:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith 评估体系 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Dataset │ │ Evaluator │ │ Experiment │ │
│ │ 评估数据集 │ │ 评估器 │ │ 评估运行 │ │
│ ├──────────────┤ ├──────────────┤ ├──────────────┤ │
│ │ 输入+期望输出 │ │ 评分逻辑 │ │ 运行结果+评分 │ │
│ │ │ │ │ │ │ │
│ │ Q: 首都是? │ │ 精确匹配? │ │ 得分: 0.85 │ │
│ │ A: 北京 │ │ 语义相似度? │ │ 通过: 8/10 │ │
│ │ │ │ LLM-as-Judge │ │ 平均: 0.82 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └───────────┬───────┘ │ │
│ │ │ │
│ ▼ │ │
│ ┌──────────────┐ │ │
│ │ evaluate() │ ──────────────────→│ │
│ │ 执行评估 │ │ │
│ └──────────────┘ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ LangSmith │ │
│ │ UI 对比查看 │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
5.3 创建评估数据集
三种方式:
方式1:手动创建(适合小规模基准测试)
python
"""
方式1:代码创建评估数据集
"""
from langsmith import Client
client = Client()
# 创建数据集
dataset = client.create_dataset(
dataset_name="qa-benchmark-v1",
description="QA 基准测试数据集",
)
# 添加样例
examples = [
{
"inputs": {"question": "中国的首都是哪里?"},
"outputs": {"answer": "北京"},
},
{
"inputs": {"question": "Python 的创建者是谁?"},
"outputs": {"answer": "Guido van Rossum"},
},
{
"inputs": {"question": "LangGraph 是什么?"},
"outputs": {"answer": "LangGraph 是一个用于构建有状态、多步骤 AI Agent 工作流的 Python 框架"},
},
{
"inputs": {"question": "RAG 的全称是什么?"},
"outputs": {"answer": "Retrieval-Augmented Generation(检索增强生成)"},
},
{
"inputs": {"question": "GIL 的全称是什么?"},
"outputs": {"answer": "Global Interpreter Lock(全局解释器锁)"},
},
]
for ex in examples:
client.create_example(
inputs=ex["inputs"],
outputs=ex["outputs"],
dataset_id=dataset.id,
)
print(f"数据集已创建:{dataset.name},共 {len(examples)} 个样例")
方式2:从生产 Trace 导入(最实用)
python
"""
方式2:从生产环境的 Trace 创建数据集
把真实用户的问题和正确答案保存下来
"""
from langsmith import Client
client = Client()
# 获取项目中的成功 Trace
runs = client.list_runs(
project_name="agent-production",
execution_order=1, # 只取顶层 Run
error=False, # 只取成功的
limit=50,
)
# 创建数据集
dataset = client.create_dataset(
dataset_name="production-qa-v1",
description="从生产环境 Trace 导入的评估数据",
)
for run in runs:
if run.inputs and run.outputs:
client.create_example(
inputs=run.inputs,
outputs=run.outputs,
dataset_id=dataset.id,
)
print(f"已从生产环境导入 {dataset.name}")
方式3:在 LangSmith UI 中手动创建
在 LangSmith UI → Datasets & Experiments → Create Dataset,直接在网页上添加样例。适合非技术人员参与评估数据的构建。
5.4 内置评估器
LangSmith 提供了几种常用的内置评估器:
表格
5.5 自定义评估器
内置评估器不够用时,写你自己的:
python
"""
自定义评估器实战
评估一个 RAG Agent 的回答质量
"""
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["OPENAI_API_KEY"] = "sk-..."
import json
from langsmith import Client
from langsmith.evaluation import evaluate, EvaluationResult
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
client = Client()
# ---- 定义被评估的 RAG Pipeline ----
def rag_pipeline(inputs: dict) -> dict:
"""被评估的 RAG Pipeline"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个准确的知识问答助手。请简洁准确地回答问题。"),
("human", "{question}"),
])
chain = prompt | llm
response = chain.invoke({"question": inputs["question"]})
return {"answer": response.content}
# ---- 定义评估器 ----
def exact_match_evaluator(run, example) -> EvaluationResult:
"""精确匹配评估器:输出是否与期望完全一致"""
predicted = run.outputs.get("answer", "").lower().strip()
expected = example.outputs.get("answer", "").lower().strip()
score = 1.0 if predicted == expected else 0.0
return EvaluationResult(
key="exact_match",
score=score,
comment=f"预测: {predicted[:50]}... | 期望: {expected[:50]}...",
)
def contains_key_info_evaluator(run, example) -> EvaluationResult:
"""关键信息覆盖评估器:输出是否包含期望答案中的关键信息"""
predicted = run.outputs.get("answer", "").lower()
expected = example.outputs.get("answer", "").lower()
# 简单实现:检查期望答案中的关键词是否出现在预测中
keywords = [w for w in expected.split() if len(w) > 1]
if not keywords:
return EvaluationResult(key="key_info_coverage", score=0.0)
hits = sum(1 for kw in keywords if kw in predicted)
score = hits / len(keywords)
return EvaluationResult(
key="key_info_coverage",
score=score,
comment=f"关键词命中 {hits}/{len(keywords)}",
)
def llm_judge_evaluator(run, example) -> EvaluationResult:
"""LLM-as-a-Judge 评估器:用 GPT-4o-mini 评估回答质量"""
judge_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
question = example.inputs["question"]
reference = example.outputs.get("answer", "")
prediction = run.outputs.get("answer", "")
judge_prompt = ChatPromptTemplate.from_messages([
("system", (
"你是一个评估专家。请评估 AI 回答的质量。\n"
"从以下维度打分(0-1):\n"
"1. 准确性:回答是否与参考答案一致\n"
"2. 完整性:回答是否涵盖了关键信息\n"
"返回 JSON: {\"score\": 0.0-1.0, \"reason\": \"一句话说明\"}"
)),
("human", (
"问题: {question}\n"
"参考答案: {reference}\n"
"AI回答: {prediction}\n"
"请评估。"
)),
])
chain = judge_prompt | judge_llm
response = chain.invoke({
"question": question,
"reference": reference,
"prediction": prediction,
})
try:
parsed = json.loads(response.content)
score = float(parsed.get("score", 0.0))
reason = parsed.get("reason", "")
except (json.JSONDecodeError, ValueError):
score = 0.0
reason = "解析评估结果失败"
return EvaluationResult(
key="llm_judge",
score=score,
comment=reason,
)
# ---- 创建数据集并运行评估 ----
dataset_name = "rag-eval-demo"
# 检查数据集是否已存在
existing = [d.name for d in client.list_datasets()]
if dataset_name not in existing:
dataset = client.create_dataset(
dataset_name=dataset_name,
description="RAG Agent 评估数据集",
)
examples = [
{"inputs": {"question": "RAG 的全称是什么?"},
"outputs": {"answer": "Retrieval-Augmented Generation(检索增强生成)"}},
{"inputs": {"question": "Python GIL 是什么?"},
"outputs": {"answer": "GIL 是全局解释器锁,确保同一时刻只有一个线程执行 Python 字节码"}},
{"inputs": {"question": "LangGraph 的核心概念是什么?"},
"outputs": {"answer": "LangGraph 的核心概念包括 State(状态)、Node(节点)、Edge(边),用有向图构建有状态的 Agent 工作流"}},
]
for ex in examples:
client.create_example(
inputs=ex["inputs"],
outputs=ex["outputs"],
dataset_id=dataset.id,
)
else:
dataset = client.read_dataset(dataset_name=dataset_name)
# 运行评估
results = evaluate(
rag_pipeline,
data=dataset, # 传入 dataset 对象,不是字符串
evaluators=[
exact_match_evaluator,
contains_key_info_evaluator,
llm_judge_evaluator,
],
experiment_prefix="rag-eval-v1",
description="RAG Agent 第一轮评估",
max_concurrency=2,
)
# 打印结果
print(f"\n评估完成: {results.experiment_name}")
for result in results:
print(f"问题: {result.reference_example.inputs['question']}")
print(f" 精确匹配: {result.evaluation_results[0].score}")
print(f" 关键信息: {result.evaluation_results[1].score}")
print(f" LLM Judge: {result.evaluation_results[2].score}")
print()
5.6 将评估集成到 CI/CD
把评估放进 GitHub Actions,每次 PR 都自动跑评估,回归问题一目了然:
yaml
# .github/workflows/llm-eval.yml
name: LLM Evaluation
on:
pull_request:
paths:
- "src/chains/ **"
- "prompts/** "
jobs:
eval:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dependencies
run: pip install langsmith langchain-openai
- name: Run evaluation
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python scripts/run_eval.py --fail-below 0.75
对应的评估脚本:
python
# scripts/run_eval.py
"""
CI/CD 集成评估脚本
低于阈值则失败
"""
import argparse
from langsmith import Client
from langsmith.evaluation import evaluate, EvaluationResult
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
parser = argparse.ArgumentParser()
parser.add_argument("--fail-below", type=float, default=0.75)
args = parser.parse_args()
client = Client()
# 被测函数
def qa_chain(inputs: dict) -> dict:
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "简洁准确地回答问题。"),
("human", "{question}"),
])
chain = prompt | llm
response = chain.invoke({"question": inputs["question"]})
return {"answer": response.content}
# 评估器
def relevance_evaluator(run, example) -> EvaluationResult:
predicted = run.outputs.get("answer", "").lower()
expected = example.outputs.get("answer", "").lower()
keywords = [w for w in expected.split() if len(w) > 1]
if not keywords:
return EvaluationResult(key="relevance", score=0.0)
hits = sum(1 for kw in keywords if kw in predicted)
return EvaluationResult(
key="relevance",
score=hits / len(keywords),
)
# 运行
dataset = client.read_dataset(dataset_name="qa-benchmark-v1")
results = evaluate(
qa_chain,
data=dataset,
evaluators=[relevance_evaluator],
experiment_prefix="ci-pr-eval",
)
# 检查阈值
df = results.to_pandas()
mean_score = df["feedback.relevance"].mean()
print(f"平均得分: {mean_score:.3f}")
if mean_score < args.fail_below:
print(f"FAIL: 得分 {mean_score:.3f} 低于阈值 {args.fail_below}")
exit(1)
print("PASS")
六、Prompt Hub:提示词版本管理
6.1 Prompt 版本管理的痛点
你有没有经历过:
改了 Prompt 之后线上效果变差了,想回滚但不知道改了啥
多人协作时,有人改了 Prompt 没通知,其他人都不知道
A/B 测试不同 Prompt,但版本混乱分不清
测试环境和生产环境用的 Prompt 不一致
Prompt Hub 就是解决这些问题的。
6.2 Prompt Hub 核心功能
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ Prompt Hub 核心功能 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 版本管理 │ │ A/B 测试 │ │ 一键回滚 │ │ 团队协作 │ │
│ │ │ │ │ │ │ │ │ │
│ │ 每次 push │ │ 不同版本 │ │ 指定 commit │ │ 共享、评论 │ │
│ │ 创建新 commit│ │ 同时运行对比 │ │ hash 即可 │ │ 变更历史 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ push_prompt() ←── 上传 pull_prompt() ←── 拉取 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
6.3 代码中拉取 Prompt:pull_prompt()
python
"""
从 Prompt Hub 拉取 Prompt
"""
import os
os.environ["LANGSMITH_API_KEY"] = "ls_..."
from langsmith import Client
from langchain_openai import ChatOpenAI
client = Client()
# 拉取最新版本
prompt = client.pull_prompt("my-org/tech-writer")
# 拉取特定版本(生产环境推荐,锁定版本)
# prompt = client.pull_prompt("my-org/tech-writer:abc123de")
# 也可以用 hub.pull(快捷方式)
# from langchain import hub
# prompt = hub.pull("my-org/tech-writer")
# 使用拉取的 Prompt
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm
response = chain.invoke({
"question": "什么是 RAG?",
"response_style": "one sentence",
})
print(response.content)
6.4 从代码推送 Prompt:push_prompt()
python
"""
向 Prompt Hub 推送 Prompt
"""
import os
os.environ["LANGSMITH_API_KEY"] = "ls_..."
from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate
client = Client()
# 创建 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{style}风格的写作助手。"),
("human", "{question}"),
])
# 推送到 Prompt Hub
url = client.push_prompt(
"my-org/tech-writer",
object=prompt,
readme="""
## tech-writer
按指定风格回答问题的写作助手。
**输入变量:**
- `question` — 用户问题
- `style` — 写作风格(如"简洁"、"详细"、"幽默")
**已测试模型:** gpt-4o-mini, claude-3-5-sonnet
""",
)
print(f"推送成功: {url}")
6.5 版本回滚
python
"""
Prompt 版本回滚
"""
from langsmith import Client
from langchain import hub
client = Client()
# 查看所有版本
commits = client.list_prompt_commits("my-org/tech-writer")
for commit in commits:
print(f"{commit.commit_hash} | {commit.created_at} | {commit.message or '(无备注)'}")
# 回滚到上一个已知良好版本
prompt = hub.pull("my-org/tech-writer:PREVIOUS_HASH")
核心原则:生产环境永远用带 hash 的版本号,不要用 latest。 这样改了 Prompt 不用重新部署,只换 hash 就行。
6.6 团队协作
Prompt Hub 支持团队协作功能:
共享 Prompt:设置
is_public=True让组织内所有人可见评论:在 Prompt 页面直接评论讨论
变更历史:每次 push 都有完整的 commit 记录
推送验证:确保 push/pull 一致性
python
"""
推送验证:确保 Push/Pull 一致
"""
from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate
client = Client()
test_prompt = ChatPromptTemplate.from_messages([
("human", "用{language}说你好。"),
])
client.push_prompt("my-org/hello-test", object=test_prompt)
pulled = client.pull_prompt("my-org/hello-test")
assert pulled.input_variables == ["language"], "输入变量不一致!"
print("✅ Push/Pull 验证通过")
七、Monitoring:生产监控
7.1 Dashboard:核心指标
LangSmith 的 Monitoring Dashboard 提供以下核心指标:
表格
7.2 告警规则
LangSmith 支持配置告警规则,在指标异常时通知你:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ 典型告警规则配置 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌────────────────────────────────────────────────────────────────────┐ │
│ │ 告警 1:错误率超阈值 │ │
│ │ 条件: 错误率 > 5% 持续 10 分钟 │ │
│ │ 通知: 邮件 → oncall@team.com │ │
│ └────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────────────────┐ │
│ │ 告警 2:延迟升高 │ │
│ │ 条件: P95 延迟 > 10s 持续 5 分钟 │ │
│ │ 通知: Slack → #agent-alerts │ │
│ └────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────────────────┐ │
│ │ 告警 3:Token 消耗暴增 │ │
│ │ 条件: 过去 1 小时 Token 消耗 > 过去 24 小时日均的 3 倍 │ │
│ │ 通知: 邮件 + Slack │ │
│ └────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
7.3 反馈收集:用户 👍👎 关联到 Trace
在生产环境中,你可以收集用户反馈(点赞/点踩),并关联到对应的 Trace:
python
"""
用户反馈收集
"""
from langsmith import Client
client = Client()
# 记录用户反馈(在收到用户 👍 或 👎 时调用)
def record_feedback(run_id: str, score: float, comment: str = ""):
"""
score: 1.0 = 👍, 0.0 = 👎
run_id: 对应 Trace 的 Run ID
"""
client.create_feedback(
run_id=run_id,
key="user_rating",
score=score,
comment=comment,
)
# 在你的 Web 应用中
# 当用户点击 👍 时:
# record_feedback(run_id="trace_run_id_here", score=1.0, comment="回答很准确")
# 当用户点击 👎 时:
# record_feedback(run_id="trace_run_id_here", score=0.0, comment="回答无关")
反馈数据会在 LangSmith Dashboard 中显示,也可以用来过滤低评分的 Trace,进行针对性优化。
7.4 Messages View(2026 新功能)
Messages View 是 2026 年新增的多轮对话可读视图。传统的 Trace 视图按 Run 树展示,对于多轮对话场景不太直观。Messages View 把同一个 Thread 的多轮对话按时间线展示,像聊天记录一样可读。
通过 session_id、thread_id 或 conversation_id 元数据关联同一会话的多个 Trace:
python
"""
使用 Thread ID 关联多轮对话
"""
import uuid
# 同一个会话的所有请求使用相同的 thread_id
thread_id = str(uuid.uuid4())
# 第 1 轮
result1 = graph.invoke(
{"messages": [{"role": "user", "content": "你好"}]},
config={"metadata": {"thread_id": thread_id}},
)
# 第 2 轮
result2 = graph.invoke(
{"messages": [{"role": "user", "content": "帮我查天气"}]},
config={"metadata": {"thread_id": thread_id}},
)
# 在 LangSmith 的 Messages View 中,这两轮会话会被关联在一起
7.5 Annotate Queue:人工标注队列
Annotate Queue 是 LangSmith 的人工标注功能,用于构建评估数据集:
从生产 Trace 中选取需要标注的样本
将它们放入 Annotate Queue
人工标注员在 UI 中逐个查看并标注
标注结果自动成为评估数据集的一部分
特别适合以下场景:
从生产数据中构建黄金数据集
评估 Agent 回答的质量
监督学习标注
八、2026 新功能深度解析
2026 年 5 月 Interrupt 大会发布的新功能,按实际影响排序。
8.1 SmithDB:专用数据库,核心负载快 15x
据《We built SmithDB, the data layer for agent observability》(https://www.langchain.com/blog/introducing-smithdb),SmithDB 是 LangSmith 专门为 Agent 可观测性构建的分布式数据库。
为什么需要专用数据库?
Agent Trace 数据有两个特点传统数据库搞不定:
嵌套深:一个 Agent Trace 可能有几百个嵌套 Span
长运行:一个 Span 可能持续几分钟甚至几小时
性能数据:
表格
技术架构:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ SmithDB 架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 无状态服务层 │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Ingestion│ │ Query │ │Compaction│ │ │
│ │ │ 写入服务 │ │ 查询服务 │ │ 压缩服务 │ │ │
│ │ └─────┬────┘ └─────┬────┘ └─────┬────┘ │ │
│ │ │ │ │ │ │
│ └─────────┼──────────────┼──────────────┼─────────────────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Object Store │ │ Postgres │ │ Rust + │ │
│ │ 持久化数据 │ │ 元数据存储 │ │ DataFusion │ │
│ │ (S3兼容) │ │ (小而精) │ │ + Vortex │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ 核心优势: │
│ • 无本地磁盘 → 加计算即可扩容 │
│ • 对象存储 → 多云/自托管友好 │
│ • Rust + DataFusion → 高性能查询 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
对用户的影响:SmithDB 已经在为所有 LangSmith 云用户服务,Trace 加载速度大幅提升。自托管版本即将推出。
8.2 LangSmith Engine:自动发现故障并提 PR
据《Everything we shipped at Interrupt》介绍,LangSmith Engine 是一个自主 Agent,自动完成"发现问题→诊断原因→修复代码"的循环。
工作流程:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith Engine 工作流程 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 监控 │ │ 聚类 │ │ 诊断 │ │ 修复 │ │
│ │ Traces │────→│ 故障 │────→│ 根因 │────→│ 提 PR │ │
│ │ │ │ │ │ │ │ │ │
│ │ 实时监控 │ │ 相似失败 │ │ 定位代码 │ │ 代码/提示 │ │
│ │ 生产流量 │ │ 聚类分组 │ │ 错误原因 │ │ 词修复 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │
│ ┌──────────┐ │ │
│ │ 验证 │←───────────────────────┘ │
│ │ │ │
│ │ 自动创建 │ │
│ │ Evaluator│ │
│ │ 防止回归 │ │
│ └──────────┘ │
│ │
│ 你只需要:Review + Merge │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
对于每个发现的 Issue,Engine 会:
提 PR:包含针对性的代码或 Prompt 修复
创建 Evaluator:针对该问题的自定义在线评估器,后续复发自动上报
添加测试用例:将失败的 Trace 加入离线评估数据集
状态:公开 Beta,Cogent 和 Campfire 已经用它修复了影响数千条 Trace 的问题。
8.3 Context Hub:Agent 指令的版本管理
据《Introducing LangSmith Context Hub》(https://www.langchain.com/blog/introducing-context-hub),Context Hub 管理 Agent 的"上下文文件"——AGENTS.md、Skills、Policies、Examples 等。
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ Context Hub vs Prompt Hub │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ Prompt Hub Context Hub │
│ ────────── ─────────── │
│ 管理内容:提示词模板 管理内容:Agent 指令文件 │
│ 格式:ChatPromptTemplate 格式:AGENTS.md、SKILL.md、tools.json │
│ 谁在用:开发者 谁在用:产品经理、设计师、运营等 │
│ 变更频率:中等 变更频率:高 │
│ 存储位置:LangSmith Hub 存储位置:Context Hub(独立 Repo) │
│ │
│ 核心功能: 核心功能: │
│ • 版本化(Commits) • 版本化(Commits) │
│ • Pull/Push • Tags(dev/staging/prod) │
│ • 团队共享 • Comments(协作评论) │
│ • CLI 推送/拉取 │
│ • Agent 自创建上下文 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
Context Hub 的使用方式:
bash
# CLI 操作
langsmith auth login
langsmith hub init --type skill --dir ./skills/support-style --name support-style
langsmith hub push support-style --type skill --dir ./skills/support-style
langsmith hub pull support-style --tag prod --output ./skills/support-style
也可以通过 Python SDK 操作:
python
"""
通过 SDK 操作 Context Hub
"""
from langsmith import Client
from langsmith.schemas import FileEntry
client = Client()
# 推送 Agent 上下文
url = client.push_agent(
"email-assistant",
files={
"AGENTS.md": FileEntry(content="你是一个邮件分类助手。"),
"tools.json": FileEntry(content='{"tools": []}'),
},
description="邮件分类和回复助手",
tags=["email", "productivity"],
is_public=False,
)
print(f"推送成功: {url}")
# 拉取 Agent 上下文
agent = client.pull_agent("email-assistant", tag="prod")
for filename, file in agent.files.items():
print(f"{filename}: {file.content[:100]}...")
8.4 LLM Gateway:LLM 网关
据《LangSmith LLM Gateway: runtime governance built into the agent lifecycle》(https://www.langchain.com/blog/introducing-llm-gateway),LLM Gateway 是一个运行时治理层,位于 Agent 和 LLM 提供商之间。
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LLM Gateway 架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Agent │ │ LLM 提供商 │ │
│ │ 应用 │ │ │ │
│ └────┬─────┘ │ ┌────────┐ │ │
│ │ │ │ OpenAI │ │ │
│ │ 请求 │ ├────────┤ │ │
│ ▼ │ │Claude │ │ │
│ ┌──────────────────────────────────┐ │ ├────────┤ │ │
│ │ LLM Gateway │ │ │ Gemini │ │ │
│ │ │ │ └────────┘ │ │
│ │ ┌────────────┐ ┌─────────────┐ │ └──────────────┘ │
│ │ │ 消费限制 │ │ PII 脱敏 │ │ ▲ │
│ │ │ 硬上限 402 │ │ 请求/响应 │ │ │ 转发 │
│ │ └────────────┘ └─────────────┘ │───────┘ │
│ │ ┌────────────┐ ┌─────────────┐ │ │
│ │ │ 审计日志 │ │ Trace 关联 │ │ │
│ │ └────────────┘ └─────────────┘ │ │
│ └──────────────────────────────────┘ │
│ │
│ 接入方式:只需改 base_url,其他代码不变 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
核心功能(私有 Beta):
消费限制:在组织/工作区/用户/API Key 级别设置硬上限,超限返回 402
实时消费可见性:按工作区、用户、API Key 的实时费用汇总
PII 和密钥检测:请求和响应中的敏感数据自动脱敏
Trace 连续性:网关代理的调用仍然出现在同一个工作区
Engine 集成:策略违规事件直接出现在 Engine 中
接入方式超级简单——只需改 base_url:
python
"""
LLM Gateway 接入示例
只需把 base_url 指向 Gateway 即可
"""
from openai import OpenAI
# 之前:直接调 OpenAI
# client = OpenAI()
# 现在:通过 LLM Gateway
client = OpenAI(
base_url="https://gateway.smith.langchain.com/v1", # Gateway 端点
api_key="ls_...", # 使用 LangSmith API Key
)
# 其他代码完全不变
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Hello"}],
)
8.5 Sandboxes GA:安全沙箱正式版
LangSmith Sandboxes 为 Agent 提供安全的代码执行环境:
每个沙箱运行在硬件虚拟化的 microVM 中
支持文件系统、Shell、包管理器、持久状态
空闲自动暂停,不浪费资源
支持快照和 Fork(copy-on-write)
8.6 Managed Deep Agents:托管式 Agent 运行时
为需要长时间运行的 Agent 提供托管运行时:
持久线程、流式运行、Checkpoint
人在回路(Human-in-the-loop)工作流
通过
/v1/deepagentsAPI 管理和运行集成 Context Hub 和 Sandboxes
九、LangSmith vs LangFuse:怎么选
第 6 篇介绍了 LangFuse,这篇介绍 LangSmith。很多读者问:到底选哪个?这里做个客观对比。
9.1 架构差异
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ 架构差异 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ LangSmith LangFuse │
│ ────────── ────────── │
│ 框架原生集成 框架无关 + OTel 原生 │
│ │
│ ┌───────────┐ ┌───────────┐ │
│ │ LangGraph │──→ 自动追踪 │ 任意框架 │──→ OTel Span │
│ │ LangChain │──→ 自动追踪 │ OpenAI │──→ SDK 手动埋点 │
│ │ 其他框架 │──→ @traceable │ LangChain │──→ 集成可用 │
│ └───────────┘ │ 其他 │──→ OTel Exporter │
│ │ └───────────┘ │
│ ▼ │ │
│ ┌───────────┐ ▼ │
│ │ LangSmith │ ┌───────────┐ │
│ │ Platform │ │ LangFuse │ │
│ │ (闭源) │ │ (开源MIT) │ │
│ └───────────┘ └───────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
9.2 详细对比
表格
9.3 选型决策
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ 选型决策树 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 你的项目用什么框架? │
│ │ │
│ ├── LangGraph / LangChain ────────────────→ 选 LangSmith │
│ │ (零配置追踪,深度集成,没有理由不用) │
│ │ │
│ ├── 多框架混合 ──────────────────────────→ 看情况 │
│ │ │ │
│ │ ├── 需要自托管 / 数据合规 ────────→ 选 LangFuse │
│ │ ├── 团队大(10+人)/ 预算敏感 ───→ 选 LangFuse │
│ │ └── LangGraph 为主 + 少量其他 ──→ LangSmith 为主 + LangFuse 辅助 │
│ │ │
│ └── 非 LangChain 框架为主 ────────────────→ 选 LangFuse │
│ (OTel 原生集成更灵活,不用绑死 LangChain 生态) │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
9.4 可以同时用
LangSmith 和 LangFuse 不互斥。你可以:
LangGraph Agent → LangSmith 追踪(零配置)
其他微服务(搜索/推荐/缓存)→ LangFuse 追踪(OTel 原生)
两套系统通过
trace_id关联
LangSmith SDK 本身也支持 OTel 模式:
python
"""
LangSmith 的 OTel 混合模式
同时发送到 LangSmith 和 OTel 后端
"""
from langsmith import Client
# Hybrid 模式:同时发送到 LangSmith 和 OTel
hybrid_client = Client(tracing_mode="hybrid")
# 纯 OTel 模式:只发送到 OTel 后端
# otel_client = Client(tracing_mode="otel")
十、生产最佳实践
10.1 Project 组织
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ Project 组织策略 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 按环境分离: │
│ ├── sales-agent-dev (开发环境) │
│ ├── sales-agent-staging (测试环境) │
│ └── sales-agent-prod (生产环境) │
│ │
│ 按服务分离: │
│ ├── rag-qa-prod (RAG 问答服务) │
│ ├── code-review-prod (代码审查服务) │
│ └── customer-support-prod (客服 Agent) │
│ │
│ 按团队分离(通过 Workspace): │
│ ├── Workspace: AI-Team │
│ │ ├── project-a-prod │
│ │ └── project-b-prod │
│ └── Workspace: Data-Team │
│ ├── etl-agent-prod │
│ └── analysis-agent-prod │
│ │
│ 推荐策略:环境 × 服务 组合命名 │
│ 格式:<service-name>-<environment> │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
10.2 采样策略:高流量下的 Trace 采样
Agent 工作流中,一个用户请求可能产生多个 Span。在高流量场景下,全量采集 Trace 的成本很高。虽然 LangSmith 目前没有内置采样功能,但你可以通过代码控制:
python
"""
自定义 Trace 采样策略
"""
import os
import random
# 采样率:10% 的请求记录 Trace
SAMPLE_RATE = 0.1
def should_trace() -> bool:
"""决定是否采集当前请求的 Trace"""
return random.random() < SAMPLE_RATE
# 动态控制是否开启追踪
def invoke_with_sampling(graph, inputs, config=None):
"""带采样控制的 Agent 调用"""
if should_trace():
os.environ["LANGSMITH_TRACING"] = "true"
else:
os.environ["LANGSMITH_TRACING"] = "false"
return graph.invoke(inputs, config=config)
更精细的采样策略(基于错误率、延迟等):
python
"""
智能采样:错误全量采集,正常按比例采样
"""
import os
import random
class SmartSampler:
def __init__(
self,
normal_rate: float = 0.05, # 正常请求采样 5%
error_rate: float = 1.0, # 错误请求 100% 采集
slow_threshold_ms: float = 5000, # 超过 5s 视为慢请求
):
self.normal_rate = normal_rate
self.error_rate = error_rate
self.slow_threshold_ms = slow_threshold_ms
def should_trace(self, is_error: bool = False, latency_ms: float = 0) -> bool:
if is_error:
return random.random() < self.error_rate
if latency_ms > self.slow_threshold_ms:
return True # 慢请求全量采集
return random.random() < self.normal_rate
10.3 数据保留与合规
表格
GDPR 合规建议:
使用客户端 anonymizer 脱敏 PII
或使用 LLM Gateway(Beta)自动脱敏
设置合理的数据保留策略
定期清理过期数据
python
"""
客户端 PII 脱敏
使用 LangSmith 的 anonymizer 在数据发送前脱敏
"""
from langsmith.anonymizer import create_anonymizer
from langsmith import Client
anonymizer = create_anonymizer([
{"pattern": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "replace": "<EMAIL>"},
{"pattern": r"\b\d{3}-\d{2}-\d{4}\b", "replace": "<SSN>"},
{"pattern": r"\b1[3-9]\d{9}\b", "replace": "<PHONE>"},
])
client = Client(anonymizer=anonymizer)
10.4 API Key 安全
python
"""
API Key 安全最佳实践
"""
# ❌ 错误:硬编码
# os.environ["LANGSMITH_API_KEY"] = "ls_abc123..."
# ✅ 正确:从环境变量读取
# 系统环境变量、.env 文件、K8s Secret、云平台 Secret Manager
# .env 文件(加入 .gitignore!)
# LANGSMITH_API_KEY=ls_...
# 或在运行时注入
import os
api_key = os.environ.get("LANGSMITH_API_KEY")
if not api_key:
raise ValueError("LANGSMITH_API_KEY 环境变量未设置")
10.5 成本控制
据《LangSmith Pricing 2026: Plans, Costs & Real TCO》(https://checkthat.ai/brands/langsmith/pricing)的分析,LangSmith 的实际成本需要考虑以下因素:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith 成本估算示例 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 场景:5 人团队,每月 50,000 条 Trace │
│ │
│ Plus Tier: │
│ ├── 基础订阅: 5 × $39 = $195/月 │
│ ├── Trace 超量: (50,000 - 10,000) × $0.50/1K = $20/月 │
│ └── 总计: $215/月 │
│ │
│ 注意隐藏成本: │
│ ├── 扩展保留(400天): 基础价格的 9-10 倍 │
│ ├── Agent Builder 运行: $0.05/次(超 500 次/月后) │
│ └── 按席位收费: 10 人团队 = $390/月 基础 │
│ │
│ 省钱建议: │
│ 1. 开发环境用 Developer Tier(免费 5K traces) │
│ 2. 生产环境用采样策略,降低 Trace 量 │
│ 3. 定期清理无用 Project │
│ 4. 评估是否真的需要扩展保留 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
10.6 团队协作
Workspace:按团队/项目隔离,最多 3 个(Plus Tier)
角色权限:Owner / Admin / Member / Viewer
共享 Dashboard:Team Tier 支持团队级共享
Prompt/Context 协作:通过 Prompt Hub 和 Context Hub 共享和评论
十一、踩坑记录
11.1 Trace 不出现的常见原因
按概率排序,最常见的 Trace 丢失原因:
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ Trace 不出现排查清单 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ □ 1. 环境变量没设置 │
│ │ LANGSMITH_TRACING=true 了吗?(大小写敏感) │
│ │ LANGSMITH_API_KEY 正确吗?(ls_ 开头) │
│ │ Python 代码中 os.environ 设置在 import langchain 之前了吗? │
│ │ │
│ □ 2. 看错 Project 了 │
│ │ 默认 Project 是 "default",不是你设的那个 │
│ │ 检查 LANGSMITH_PROJECT 环境变量 │
│ │ │
│ □ 3. Serverless 环境下函数提前退出 │
│ │ 设置 LANGCHAIN_CALLBACKS_BACKGROUND=false │
│ │ 或在函数结束前调用 client.await_pending_trace_batches() │
│ │ │
│ □ 4. 网络问题 │
│ │ 防火墙拦截了 api.smith.langchain.com │
│ │ 代理设置不正确 │
│ │ │
│ □ 5. 异步代码中 Trace 上下文丢失 │
│ │ asyncio 中需要确保 tracing_context 正确传递 │
│ │ │
│ □ 6. 免费 Tier 额度用完了 │
│ │ Developer Tier 只有 5,000 traces/月 │
│ │ 超了之后新 Trace 会被丢弃 │
│ │ │
└─────────────────────────────────────────────────────────────────────────────┘
11.2 Run Tree 嵌套过深导致 UI 卡顿
复杂 Agent(尤其是多 Agent 协作)的 Trace 可能有几百个嵌套 Span。SmithDB 上线后这个问题好了很多(P50 92ms 加载),但如果你的 Trace 特别大,还是可能卡。
解决方案:
避免在 Trace 中存储大块二进制数据(图片、PDF 内容等)
用
run_type="retriever"标记检索步骤,只记录元数据不记录全文分 Project 拆分不同服务的 Trace
11.3 评估数据集与生产数据分布不一致
这是最常见的评估陷阱:你的评估数据集太"干净"了,生产数据的分布完全不同。
解决方案:
定期从生产 Trace 中导入新样例到评估数据集
使用 Annotate Queue 让人工标注生产数据
监控评估得分与生产用户反馈的相关性
11.4 Prompt Hub 版本冲突
多人同时修改同一个 Prompt 时可能出现版本冲突。
解决方案:
生产环境永远使用带 hash 的版本号
修改前先 pull 最新版本
使用 commit message 记录变更原因
团队约定:一个人负责 Prompt 变更
11.5 免费 Tier 额度用完后的处理
Developer Tier 只有 5,000 traces/月,对于认真的开发来说很容易用完。
处理方式:
升级到 Plus Tier($39/seat/月,10K traces)
使用采样策略减少 Trace 量
在开发环境设置
LANGSMITH_TRACING=false,只在需要调试时开启考虑 LangFuse(50K events/月免费)作为开发环境替代
11.6 LangSmith 与 OpenTelemetry 同时接入的兼容性
LangSmith 和 OTel 可以同时工作,但需要注意:
LangSmith SDK 从 v0.7.35+ 支持
tracing_mode参数tracing_mode="hybrid"同时发送到 LangSmith 和 OTeltracing_mode="otel"只发送到 OTel 后端确保不要产生重复的 Span
python
"""
LangSmith + OTel 混合模式
"""
from langsmith import Client
# 方式 1:混合模式,同时发送
hybrid_client = Client(tracing_mode="hybrid")
# 方式 2:只发送到 OTel
otel_client = Client(tracing_mode="otel")
# 方式 3:默认只发送到 LangSmith
ls_client = Client() # tracing_mode="langsmith"
总结
plaintext
┌─────────────────────────────────────────────────────────────────────────────┐
│ LangSmith 使用路线图 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 第 1 天:5 分钟上手 │
│ ├── 注册账号,获取 API Key │
│ ├── 设置环境变量 │
│ └── 跑通第一个 Trace,在 UI 中查看 │
│ │
│ 第 1 周:Tracing 深度实践 │
│ ├── 给 LangGraph Agent 加 Metadata/Tags │
│ ├── 学会用过滤和搜索定位问题 │
│ └── 配置 Project 分离(dev/staging/prod) │
│ │
│ 第 2 周:Evaluation 上线 │
│ ├── 创建评估数据集 │
│ ├── 写自定义 Evaluator │
│ └── 集成到 CI/CD │
│ │
│ 第 3 周:Prompt Hub + Monitoring │
│ ├── 迁移 Prompt 到 Prompt Hub │
│ ├── 配置告警规则 │
│ └── 上线用户反馈收集 │
│ │
│ 持续优化: │
│ ├── 定期从生产 Trace 导入评估数据 │
│ ├── 使用 Context Hub 管理 Agent 指令 │
│ ├── 评估 LLM Gateway 的治理能力 │
│ └── 关注 LangSmith Engine 的自动修复能力 │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
一句话总结:如果你用 LangGraph,LangSmith 是标配——零配置追踪、可视化调试、自动化评估,没有理由不用。如果你不用 LangChain 生态,LangFuse 是更灵活的选择。两者不互斥,可以共存。
参考资料
LangSmith 官方文档:https://docs.smith.langchain.com
LangSmith SDK (Python):https://pypi.org/project/langsmith/
Everything we shipped at Interrupt:https://www.langchain.com/blog/interrupt-2026-overview
We built SmithDB:https://www.langchain.com/blog/introducing-smithdb
Introducing LangSmith Context Hub:https://www.langchain.com/blog/introducing-context-hub
LangSmith LLM Gateway:https://www.langchain.com/blog/introducing-llm-gateway
LangSmith Pricing 2026:https://checkthat.ai/brands/langsmith/pricing
Implementing Compliant AI Tracing with LangSmith:https://hshahin.com/blog/langsmith-tracing-redaction/
评论区