LangSmith可观察性平台技能Skill langsmith-observability

LangSmith是一个用于大型语言模型(LLM)的可观察性平台,提供追踪、评估和监控功能,帮助开发人员调试AI应用、进行系统评估和性能监控。关键词:LLM、可观察性、追踪、评估、监控、调试、AI应用。

AI应用 0 次安装 0 次浏览 更新于 3/21/2026

名称: langsmith-可观察性 描述: LLM可观察性平台,用于追踪、评估和监控。在调试LLM应用、评估模型输出对比数据集、监控生产系统或构建AI应用的系统测试管道时使用。 版本: 1.0.0 作者: Orchestra Research 许可证: MIT 标签: [可观察性, LangSmith, 追踪, 评估, 监控, 调试, 测试, LLM运维, 生产] 依赖项: [langsmith>=0.2.0]

LangSmith - LLM可观察性平台

用于调试、评估和监控语言模型和AI应用的开发平台。

何时使用LangSmith

使用LangSmith当:

  • 调试LLM应用问题(提示、链、代理)
  • 系统性地评估模型输出对比数据集
  • 监控生产LLM系统
  • 构建AI功能的回归测试
  • 分析延迟、令牌使用和成本
  • 协作提示工程

关键功能:

  • 追踪:捕获所有LLM调用的输入、输出、延迟
  • 评估:使用内置和自定义评估器进行系统测试
  • 数据集:从生产追踪或手动创建测试集
  • 监控:在生产中跟踪指标、错误和成本
  • 集成:与OpenAI、Anthropic、LangChain、LlamaIndex配合使用

使用替代方案当:

  • Weights & Biases:深度学习实验跟踪,模型训练
  • MLflow:通用ML生命周期,模型注册重点
  • Arize/WhyLabs:ML监控,数据漂移检测

快速开始

安装

pip install langsmith

# 设置环境变量
export LANGSMITH_API_KEY="your-api-key"
export LANGSMITH_TRACING=true

使用@traceable进行基本追踪

from langsmith import traceable
from openai import OpenAI

client = OpenAI()

@traceable
def generate_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# 自动追踪到LangSmith
result = generate_response("什么是机器学习?")

OpenAI包装器(自动追踪)

from langsmith.wrappers import wrap_openai
from openai import OpenAI

# 包装客户端以进行自动追踪
client = wrap_openai(OpenAI())

# 所有调用自动追踪
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好!"}]
)

核心概念

运行和追踪

运行是单个执行单元(LLM调用、链、工具)。运行形成分层追踪,显示完整执行流程。

from langsmith import traceable

@traceable(run_type="chain")
def process_query(query: str) -> str:
    # 父运行
    context = retrieve_context(query)  # 子运行
    response = generate_answer(query, context)  # 子运行
    return response

@traceable(run_type="retriever")
def retrieve_context(query: str) -> list:
    return vector_store.search(query)

@traceable(run_type="llm")
def generate_answer(query: str, context: list) -> str:
    return llm.invoke(f"Context: {context}

Question: {query}")

项目

项目组织相关运行。通过环境或代码设置:

import os
os.environ["LANGSMITH_PROJECT"] = "my-project"

# 或按函数设置
@traceable(project_name="my-project")
def my_function():
    pass

客户端API

from langsmith import Client

client = Client()

# 列出运行
runs = list(client.list_runs(
    project_name="my-project",
    filter='eq(status, "success")',
    limit=100
))

# 获取运行详情
run = client.read_run(run_id="...")

# 创建反馈
client.create_feedback(
    run_id="...",
    key="correctness",
    score=0.9,
    comment="好答案"
)

数据集和评估

创建数据集

from langsmith import Client

client = Client()

# 创建数据集
dataset = client.create_dataset("qa-test-set", description="QA评估")

# 添加示例
client.create_examples(
    inputs=[
        {"question": "什么是Python?"},
        {"question": "什么是ML?"}
    ],
    outputs=[
        {"answer": "一种编程语言"},
        {"answer": "机器学习"}
    ],
    dataset_id=dataset.id
)

运行评估

from langsmith import evaluate

def my_model(inputs: dict) -> dict:
    # 您的模型逻辑
    return {"answer": generate_answer(inputs["question"])}

def correctness_evaluator(run, example):
    prediction = run.outputs["answer"]
    reference = example.outputs["answer"]
    score = 1.0 if reference.lower() in prediction.lower() else 0.0
    return {"key": "correctness", "score": score}

results = evaluate(
    my_model,
    data="qa-test-set",
    evaluators=[correctness_evaluator],
    experiment_prefix="v1"
)

print(f"平均分数: {results.aggregate_metrics['correctness']}")

内置评估器

from langsmith.evaluation import LangChainStringEvaluator

# 使用LangChain评估器
results = evaluate(
    my_model,
    data="qa-test-set",
    evaluators=[
        LangChainStringEvaluator("qa"),
        LangChainStringEvaluator("cot_qa")
    ]
)

高级追踪

追踪上下文

from langsmith import tracing_context

with tracing_context(
    project_name="experiment-1",
    tags=["production", "v2"],
    metadata={"version": "2.0"}
):
    # 所有可追踪调用继承上下文
    result = my_function()

手动运行

from langsmith import trace

with trace(
    name="custom_operation",
    run_type="tool",
    inputs={"query": "test"}
) as run:
    result = do_something()
    run.end(outputs={"result": result})

处理输入/输出

def sanitize_inputs(inputs: dict) -> dict:
    if "password" in inputs:
        inputs["password"] = "***"
    return inputs

@traceable(process_inputs=sanitize_inputs)
def login(username: str, password: str):
    return authenticate(username, password)

采样

import os
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"  # 10% 采样

LangChain集成

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 通过LANGSMITH_TRACING=true自动启用追踪
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    ("user", "{input}")
])

chain = prompt | llm

# 所有链运行自动追踪
response = chain.invoke({"input": "你好!"})

生产监控

Hub提示

from langsmith import Client

client = Client()

# 从Hub拉取提示
prompt = client.pull_prompt("my-org/qa-prompt")

# 在应用中使用
result = prompt.invoke({"question": "什么是AI?"})

异步客户端

from langsmith import AsyncClient

async def main():
    client = AsyncClient()

    runs = []
    async for run in client.list_runs(project_name="my-project"):
        runs.append(run)

    return runs

反馈收集

from langsmith import Client

client = Client()

# 收集用户反馈
def record_feedback(run_id: str, user_rating: int, comment: str = None):
    client.create_feedback(
        run_id=run_id,
        key="user_rating",
        score=user_rating / 5.0,  # 归一化到0-1
        comment=comment
    )

# 在您的应用中
record_feedback(run_id="...", user_rating=4, comment="有帮助的回复")

测试集成

Pytest集成

from langsmith import test

@test
def test_qa_accuracy():
    result = my_qa_function("什么是Python?")
    assert "programming" in result.lower()

CI/CD中的评估

from langsmith import evaluate

def run_evaluation():
    results = evaluate(
        my_model,
        data="regression-test-set",
        evaluators=[accuracy_evaluator]
    )

    # 如果准确率下降则CI失败
    assert results.aggregate_metrics["accuracy"] >= 0.9, \
        f"准确率 {results.aggregate_metrics['accuracy']} 低于阈值"

最佳实践

  1. 结构化命名 - 使用一致的项目/运行命名约定
  2. 添加元数据 - 包括版本、环境、用户信息
  3. 生产采样 - 使用采样率控制量
  4. 创建数据集 - 从有趣的生产案例构建测试集
  5. 自动化评估 - 在CI/CD管道中运行评估
  6. 监控成本 - 跟踪令牌使用和延迟趋势

常见问题

追踪不出现:

import os
# 确保追踪已启用
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-key"

# 验证连接
from langsmith import Client
client = Client()
print(client.list_projects())  # 应该工作

追踪导致高延迟:

# 启用后台批处理(默认)
from langsmith import Client
client = Client(auto_batch_tracing=True)

# 或使用采样
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"

大型负载:

# 隐藏敏感/大型字段
@traceable(
    process_inputs=lambda x: {k: v for k, v in x.items() if k != "large_field"}
)
def my_function(data):
    pass

参考

资源