21c - LangSmith 设置指南

1. 点击右上角头像 → Settings
2. 左侧菜单选择 "API Keys"
3. 点击 "Create API Key"
4. 为 Key 命名（如 "dev-local"、"production"）
5. 复制生成的 Key（格式：lsv2_pt_xxxx...）
6. 妥善保存——Key 只显示一次

# 在 ~/.bashrc 或 ~/.zshrc 中添加
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2_pt_your_api_key_here"
export LANGSMITH_PROJECT="my-first-project"
 
# 可选：指定工作区（多工作区场景）
export LANGSMITH_WORKSPACE_ID="your-workspace-id"

$env:LANGSMITH_TRACING = "true"
$env:LANGSMITH_API_KEY = "lsv2_pt_your_api_key_here"
$env:LANGSMITH_PROJECT = "my-first-project"

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_pt_your_api_key_here
LANGSMITH_PROJECT=my-first-project
LANGSMITH_ENDPOINT=https://api.smith.langchain.com

你是一个 DevOps 工程师，请帮我生成一个安全的 LangSmith 环境变量配置方案：

项目类型：[Web 应用 / CLI 工具 / 微服务]
部署环境：[本地开发 / Docker / Kubernetes / Serverless]
密钥管理方式：[.env 文件 / AWS Secrets Manager / Vault / 环境变量]

要求：
1. 区分开发和生产环境的配置
2. 确保 API Key 不会泄露到版本控制
3. 包含 .gitignore 配置

# 使用 pip
pip install langsmith
 
# 使用 poetry
poetry add langsmith
 
# 如果同时使用 LangChain
pip install langsmith langchain langchain-openai

import langsmith
print(langsmith.__version__)  # 应输出 0.3.x 或更高版本
 
# 验证连接
client = langsmith.Client()
print(client.info)  # 应返回服务器信息

import langsmith as ls
from langsmith.wrappers import wrap_openai
import openai
 
# 包装 OpenAI 客户端，自动追踪所有 LLM 调用
client = wrap_openai(openai.Client())
 
@ls.traceable(run_type="tool", name="检索上下文")
def retrieve_context(question: str) -> str:
    """模拟文档检索"""
    # 实际项目中这里会调用向量数据库
    return "LangSmith 是 LangChain 团队开发的可观测性平台..."
 
@ls.traceable(run_type="chain", name="RAG 问答管线")
def rag_pipeline(question: str) -> str:
    context = retrieve_context(question)
    messages = [
        {"role": "system", "content": f"基于以下上下文回答问题：\n{context}"},
        {"role": "user", "content": question}
    ]
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=messages
    )
    return response.choices[0].message.content
 
# 执行——trace 自动上报到 LangSmith
result = rag_pipeline("什么是 LangSmith？")
print(result)

import langsmith as ls
from langsmith.wrappers import wrap_openai
import openai
 
client = wrap_openai(openai.Client())
 
def chat_pipeline(question: str) -> str:
    context = "LangSmith 提供端到端的 LLM 可观测性..."
    messages = [
        {"role": "system", "content": f"上下文：{context}"},
        {"role": "user", "content": question}
    ]
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=messages
    )
    return response.choices[0].message.content
 
# 使用上下文管理器手动控制 trace
app_inputs = {"input": "LangSmith 有什么功能？"}
with ls.trace("Chat Pipeline", "chain", project_name="my-project", inputs=app_inputs) as rt:
    output = chat_pipeline("LangSmith 有什么功能？")
    rt.end(outputs={"output": output})

@ls.traceable(
    run_type="chain",
    name="客服问答",
    tags=["production", "customer-service"],
    metadata={"version": "2.1", "team": "support"}
)
def customer_service_agent(query: str, user_id: str) -> str:
    # 元数据和标签会显示在 LangSmith UI 中，方便过滤和分析
    ...

# npm
npm install langsmith
 
# pnpm
pnpm add langsmith
 
# yarn
yarn add langsmith
 
# 如果同时使用 LangChain.js
npm install langsmith @langchain/core @langchain/openai

import { traceable } from "langsmith/traceable";
import { wrapOpenAI } from "langsmith/wrappers";
import OpenAI from "openai";
 
// 包装 OpenAI 客户端
const client = wrapOpenAI(new OpenAI());
 
const retrieveContext = traceable(
  async (question: string): Promise<string> => {
    // 模拟检索
    return "LangSmith 是一个 LLM 可观测性平台...";
  },
  { name: "检索上下文", run_type: "tool" }
);
 
const ragPipeline = traceable(
  async (question: string): Promise<string> => {
    const context = await retrieveContext(question);
    const response = await client.chat.completions.create({
      model: "gpt-4.1-mini",
      messages: [
        { role: "system", content: `基于上下文回答：\n${context}` },
        { role: "user", content: question },
      ],
    });
    return response.choices[0].message.content ?? "";
  },
  { name: "RAG 问答管线", run_type: "chain" }
);
 
// 执行
const result = await ragPipeline("什么是 LangSmith？");
console.log(result);

import { AISDKExporter } from "langsmith/vercel";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";
 
const result = await streamText({
  model: openai("gpt-4.1-mini"),
  prompt: "用一句话解释 AgentOps",
  experimental_telemetry: AISDKExporter.getSettings({
    runName: "Vercel AI 调用",
    metadata: { framework: "vercel-ai-sdk" },
  }),
});

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_xxx"
os.environ["LANGSMITH_PROJECT"] = "my-langchain-app"
 
# 所有 LangChain 调用自动追踪——无需额外代码
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
llm = ChatOpenAI(model="gpt-4.1-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    ("user", "{input}")
])
chain = prompt | llm
 
# 这个调用会自动生成完整的 trace
response = chain.invoke({"input": "什么是 AgentOps？"})

# 安装 OTel 依赖
# pip install opentelemetry-sdk opentelemetry-exporter-otlp
 
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 
# 配置 LangSmith 作为 OTel 后端
exporter = OTLPSpanExporter(
    endpoint="https://api.smith.langchain.com/otel/v1/traces",
    headers={"x-api-key": "lsv2_pt_xxx"}
)
 
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

import langsmith as ls
import random
 
# 方法 1：通过环境变量控制采样率
# 在代码中动态决定是否追踪
@ls.traceable(enabled=random.random() < 0.1)  # 10% 采样率
def high_volume_endpoint(query: str) -> str:
    ...
 
# 方法 2：按条件追踪
@ls.traceable
def conditional_trace(query: str, debug: bool = False) -> str:
    if not debug:
        # 生产环境可通过 LANGSMITH_TRACING 环境变量全局控制
        pass
    ...

1. 登录 LangSmith → 左侧菜单选择 "Projects"
2. 点击 "+ New Project"
3. 输入项目名称（如 "customer-service-bot"）
4. 选择描述和标签
5. 点击 "Create"

from langsmith import Client
 
client = Client()
# 创建新项目
client.create_project("customer-service-bot", description="客服机器人生产环境")

1. 进入项目 → 点击右上角 "Dashboard"
2. 点击 "Create Dashboard" 或 "Add Chart"
3. 选择图表类型（折线图、柱状图、数值卡片等）
4. 配置指标和过滤条件
5. 设置时间范围和刷新频率

常用过滤条件：
- 按标签过滤：tag = "production"
- 按元数据过滤：metadata.user_id = "user_123"
- 按延迟过滤：latency > 5s
- 按状态过滤：status = "error"
- 按时间范围：last 24 hours / last 7 days
- 按反馈分数：feedback.score < 0.5

1. 左侧菜单选择 "Prompts"
2. 点击 "+ New Prompt"
3. 输入 Prompt 名称（如 "customer-service-v1"）
4. 编写 Prompt 模板内容
5. 设置模型参数（temperature、max_tokens 等）
6. 点击 "Save" 保存

from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate
 
client = Client()
 
# 创建 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}。请用{language}回答用户问题。"),
    ("user", "{question}")
])
 
# 推送到 LangSmith Prompt Hub
client.push_prompt("customer-service-prompt", object=prompt)
 
# 从 Hub 拉取特定版本
prompt_v1 = client.pull_prompt("customer-service-prompt")
# 拉取特定 commit
prompt_specific = client.pull_prompt("customer-service-prompt:abc123")

1. 进入 Prompts → 选择一个 Prompt
2. 点击 "Open in Playground"
3. 填入变量值（如 role="客服专家"、language="中文"）
4. 选择模型（GPT-4、Claude 3.5 等）
5. 点击 "Run" 测试效果
6. 对比不同版本的输出质量
7. 满意后点击 "Commit" 保存新版本

请帮我设计一个 LangSmith Prompt 版本管理策略：

应用场景：[客服机器人 / 代码助手 / 数据分析]
团队规模：[1-3 人 / 5-10 人 / 10+ 人]
发布频率：[每天 / 每周 / 每月]

要求：
1. 定义 Prompt 命名规范（如 {team}-{function}-{version}）
2. 设计审批流程（谁可以推送到生产？）
3. 制定回滚策略（发现问题如何快速回退？）
4. 设置 A/B 测试方案（如何对比新旧版本效果？）

from langsmith import Client
 
client = Client()
 
# 创建数据集
dataset = client.create_dataset(
    "customer-service-eval",
    description="客服机器人评估数据集"
)
 
# 添加示例
examples = [
    {
        "inputs": {"question": "如何退款？"},
        "outputs": {"answer": "您可以在订单页面点击'申请退款'按钮..."}
    },
    {
        "inputs": {"question": "配送需要多久？"},
        "outputs": {"answer": "标准配送 3-5 个工作日，加急配送 1-2 个工作日..."}
    },
    {
        "inputs": {"question": "可以修改收货地址吗？"},
        "outputs": {"answer": "订单发货前可以修改，请进入订单详情页..."}
    }
]
 
for example in examples:
    client.create_example(
        inputs=example["inputs"],
        outputs=example["outputs"],
        dataset_id=dataset.id
    )

1. 在 Traces 页面选择高质量的 trace
2. 点击 "Add to Dataset"
3. 选择目标数据集或创建新数据集
4. 确认输入和期望输出
5. 保存

from langsmith import Client, evaluate
 
client = Client()
 
# 定义要评估的目标函数
def my_agent(inputs: dict) -> dict:
    question = inputs["question"]
    # 调用你的 Agent 逻辑
    answer = rag_pipeline(question)
    return {"answer": answer}
 
# 定义评估器
def correctness_evaluator(run, example):
    """评估回答的正确性"""
    predicted = run.outputs.get("answer", "")
    expected = example.outputs.get("answer", "")
    # 简单的关键词匹配（实际项目中可用 LLM-as-Judge）
    score = 1.0 if any(kw in predicted for kw in expected.split("，")) else 0.0
    return {"key": "correctness", "score": score}
 
# 运行评估
results = evaluate(
    my_agent,
    data="customer-service-eval",
    evaluators=[correctness_evaluator],
    experiment_prefix="v2.1-gpt4-mini"
)

from langsmith import evaluate
from langchain_openai import ChatOpenAI
 
# LangSmith 内置的 LLM 评估器
def llm_judge(run, example):
    """使用 LLM 作为评判者"""
    judge = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
    
    prompt = f"""请评估以下回答的质量（0-1 分）：
 
问题：{example.inputs['question']}
期望回答：{example.outputs['answer']}
实际回答：{run.outputs['answer']}
 
评分标准：
- 1.0：完全正确且有帮助
- 0.7：基本正确但有遗漏
- 0.3：部分正确但有误导
- 0.0：完全错误
 
请只返回一个数字分数。"""
    
    response = judge.invoke(prompt)
    score = float(response.content.strip())
    return {"key": "llm_judge_score", "score": score}
 
results = evaluate(
    my_agent,
    data="customer-service-eval",
    evaluators=[llm_judge],
    experiment_prefix="v2.1-llm-judge"
)

1. 进入 Organization Settings
2. 点击 "Workspaces" → "Create Workspace"
3. 输入名称（如 "AI-Team-Dev"）
4. 配置工作区设置

1. 进入 Workspace Settings → Members
2. 点击 "Invite Member"
3. 输入邮箱，选择角色
4. 发送邀请

1. 进入 "Annotation Queues" → "Create Queue"
2. 配置过滤条件（如：只包含低分 trace）
3. 分配标注人员
4. 标注人员逐条审查 trace，给出反馈分数和评论
5. 标注结果自动汇总到评估指标中

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_xxx"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-system"
 
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
 
# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_agent: str
 
# 定义 Agent 节点
llm = ChatOpenAI(model="gpt-4.1-mini")
 
def researcher(state: AgentState) -> AgentState:
    """研究员 Agent：负责信息收集"""
    response = llm.invoke(f"作为研究员，请分析：{state['messages'][-1]}")
    return {"messages": [response.content], "next_agent": "writer"}
 
def writer(state: AgentState) -> AgentState:
    """写作者 Agent：负责内容生成"""
    response = llm.invoke(f"作为写作者，基于以下研究撰写内容：{state['messages'][-1]}")
    return {"messages": [response.content], "next_agent": "reviewer"}
 
def reviewer(state: AgentState) -> AgentState:
    """审查者 Agent：负责质量审查"""
    response = llm.invoke(f"作为审查者，请审查以下内容：{state['messages'][-1]}")
    return {"messages": [response.content], "next_agent": "end"}
 
# 构建图
def route(state: AgentState) -> str:
    return state.get("next_agent", "end")
 
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
 
graph.add_edge(START, "researcher")
graph.add_conditional_edges("researcher", route, {"writer": "writer"})
graph.add_conditional_edges("writer", route, {"reviewer": "reviewer"})
graph.add_conditional_edges("reviewer", route, {"end": END})
 
app = graph.compile()
 
# 执行——所有节点的执行过程自动追踪到 LangSmith
result = app.invoke({
    "messages": ["请写一篇关于 AgentOps 的技术博客"],
    "next_agent": "researcher"
})

📊 Trace: multi-agent-system
├── 🔄 StateGraph (总耗时: 12.3s, 总 Token: 4,521)
│   ├── 🔍 researcher (耗时: 3.2s, Token: 1,203)
│   │   └── 🤖 ChatOpenAI (gpt-4.1-mini)
│   ├── ✍️ writer (耗时: 5.1s, Token: 2,105)
│   │   └── 🤖 ChatOpenAI (gpt-4.1-mini)
│   └── 📋 reviewer (耗时: 4.0s, Token: 1,213)
│       └── 🤖 ChatOpenAI (gpt-4.1-mini)

# 服务 A：调用方
import langsmith as ls
 
@ls.traceable
def orchestrator(query: str):
    # 获取当前 trace 的 headers
    headers = ls.get_current_run_tree().to_headers()
    # 将 headers 传递给下游服务
    response = requests.post(
        "http://agent-service/invoke",
        json={"query": query},
        headers=headers
    )
    return response.json()

import os
import langsmith as ls
from langsmith import Client, evaluate
from langsmith.wrappers import wrap_openai
import openai
 
# ===== 第 1 步：环境配置 =====
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_xxx"
os.environ["LANGSMITH_PROJECT"] = "ecommerce-chatbot-prod"
 
client = Client()
oai_client = wrap_openai(openai.Client())
 
# ===== 第 2 步：创建项目和数据集 =====
# 创建评估数据集
dataset = client.create_dataset(
    "ecommerce-qa-golden",
    description="电商客服黄金测试集"
)
 
golden_examples = [
    {"inputs": {"q": "怎么退货？"}, "outputs": {"a": "7天无理由退货，请在订单页申请"}},
    {"inputs": {"q": "发票怎么开？"}, "outputs": {"a": "订单完成后在订单详情页申请电子发票"}},
    {"inputs": {"q": "会员有什么优惠？"}, "outputs": {"a": "会员享受9折优惠和免运费服务"}},
]
 
for ex in golden_examples:
    client.create_example(inputs=ex["inputs"], outputs=ex["outputs"], dataset_id=dataset.id)
 
# ===== 第 3 步：带追踪的 Agent 实现 =====
@ls.traceable(run_type="tool", name="知识库检索")
def search_knowledge_base(query: str) -> str:
    # 实际项目中调用向量数据库
    knowledge = {
        "退货": "7天无理由退货政策...",
        "发票": "支持电子发票和纸质发票...",
        "会员": "会员等级：银卡、金卡、钻石卡..."
    }
    for key, value in knowledge.items():
        if key in query:
            return value
    return "未找到相关信息"
 
@ls.traceable(
    run_type="chain",
    name="客服问答",
    tags=["production", "ecommerce"],
    metadata={"version": "1.0"}
)
def customer_service(question: str) -> dict:
    context = search_knowledge_base(question)
    response = oai_client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": f"你是电商客服。基于知识库回答：\n{context}"},
            {"role": "user", "content": question}
        ]
    )
    answer = response.choices[0].message.content
    return {"a": answer}
 
# ===== 第 4 步：运行评估 =====
results = evaluate(
    lambda inputs: customer_service(inputs["q"]),
    data="ecommerce-qa-golden",
    evaluators=[
        lambda run, example: {
            "key": "contains_key_info",
            "score": 1.0 if any(
                kw in run.outputs.get("a", "")
                for kw in ["退货", "发票", "会员", "订单"]
            ) else 0.0
        }
    ],
    experiment_prefix="v1.0-baseline"
)
 
print(f"评估完成，请在 LangSmith UI 查看详细结果")