langsmith-observability

LangSmith - 大模型可观测性平台

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

何时使用 LangSmith

在以下情况下使用 LangSmith：

• 调试大模型应用问题（提示词、链、Agent）
• 根据数据集系统地评估模型输出
• 监控生产环境中的大模型系统
• 为 AI 功能构建回归测试
• 分析延迟、令牌用量和成本
• 协作进行提示词工程

主要功能：

• 追踪：捕获所有大模型调用的输入、输出和延迟
• 评估：使用内置和自定义评估器进行系统化测试
• 数据集：从生产追踪中创建测试集，或手动创建
• 监控：跟踪生产环境中的指标、错误和成本
• 集成：支持 OpenAI、Anthropic、LangChain、LlamaIndex

以下情况请使用其他工具：

• Weights & Biases：深度学习实验跟踪、模型训练
• MLflow：通用机器学习生命周期管理，侧重模型注册
• Arize/WhyLabs：机器学习监控、数据漂移检测

快速入门

安装

pip install langsmith

# Set environment variables
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

# Automatically traced to LangSmith
result = generate_response("What is machine learning?")

OpenAI 包装器（自动追踪）

from langsmith.wrappers import wrap_openai
from openai import OpenAI

# Wrap client for automatic tracing
client = wrap_openai(OpenAI())

# All calls automatically traced
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

核心概念

运行和追踪

运行是一个单独的执行单元（大模型调用、链、工具）。运行会形成分层的追踪，展示完整的执行流程。

from langsmith import traceable

@traceable(run_type="chain")
def process_query(query: str) -> str:
    # Parent run
    context = retrieve_context(query)  # Child run
    response = generate_answer(query, context)  # Child run
    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}\n\nQuestion: {query}")

项目

项目用于整理相关运行。通过环境变量或代码进行设置：

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

# Or per-function
@traceable(project_name="my-project")
def my_function():
    pass

客户端 API

from langsmith import Client

client = Client()

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

# Get run details
run = client.read_run(run_id="...")

# Create feedback
client.create_feedback(
    run_id="...",
    key="correctness",
    score=0.9,
    comment="Good answer"
)

数据集和评估

创建数据集

from langsmith import Client

client = Client()

# Create dataset
dataset = client.create_dataset("qa-test-set", description="QA evaluation")

# Add examples
client.create_examples(
    inputs=[
        {"question": "What is Python?"},
        {"question": "What is ML?"}
    ],
    outputs=[
        {"answer": "A programming language"},
        {"answer": "Machine learning"}
    ],
    dataset_id=dataset.id
)

运行评估

from langsmith import evaluate

def my_model(inputs: dict) -> dict:
    # Your model logic
    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"Average score: {results.aggregate_metrics['correctness']}")

内置评估器

from langsmith.evaluation import LangChainStringEvaluator

# Use LangChain evaluators
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"}
):
    # All traceable calls inherit context
    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% sampling

LangChain 集成

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Tracing enabled automatically with LANGSMITH_TRACING=true
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

chain = prompt | llm

# All chain runs traced automatically
response = chain.invoke({"input": "Hello!"})

生产环境监控

Hub 提示词

from langsmith import Client

client = Client()

# Pull prompt from hub
prompt = client.pull_prompt("my-org/qa-prompt")

# Use in application
result = prompt.invoke({"question": "What is 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()

# Collect user feedback
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,  # Normalize to 0-1
        comment=comment
    )

# In your application
record_feedback(run_id="...", user_rating=4, comment="Helpful response")

测试集成

Pytest 集成

from langsmith import test

@test
def test_qa_accuracy():
    result = my_qa_function("What is 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]
    )

    # Fail CI if accuracy drops
    assert results.aggregate_metrics["accuracy"] >= 0.9, \
        f"Accuracy {results.aggregate_metrics['accuracy']} below threshold"

最佳实践

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

常见问题

未显示追踪：

import os
# Ensure tracing is enabled
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-key"

# Verify connection
from langsmith import Client
client = Client()
print(client.list_projects())  # Should work

追踪导致高延迟：

# Enable background batching (default)
from langsmith import Client
client = Client(auto_batch_tracing=True)

# Or use sampling
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1"

负载过大：

# Hide sensitive/large fields
@traceable(
    process_inputs=lambda x: {k: v for k, v in x.items() if k != "large_field"}
)
def my_function(data):
    pass

参考资料

• 高级用法 - 自定义评估器、分布式追踪、Hub 提示词
• 故障排除 - 常见问题、调试、性能

资源

• 文档：https://docs.smith.langchain.com
• Python SDK：https://github.com/langchain-ai/langsmith-sdk
• Web 应用：https://smith.langchain.com
• 版本：0.2.0+
• 许可证：MIT

Agent 操作流程

1. 检查环境。 确认硬件、框架和服务版本、模型格式以及预期负载。
2. 明确输入。 根据数据或用户提供的信息，确认格式、标识符和参数。任何会影响结果的值都不要猜测，应先询问。
3. 先运行小规模版本。 扩大规模前，先服务或记录单个请求，或端到端运行一次。
4. 按照上述说明和参考资料执行完整任务。
5. 验证结果。 对照参考数据测量延迟、吞吐量和输出正确性；检查资源使用情况和成本。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及仍有哪些不确定之处。

如果发生这种情况	这样做
服务无法启动或发生 OOM	检查模型大小与内存是否匹配，以及量化和并行设置。
已安装版本中缺少这些说明提到的函数、标志或端点	查阅已安装版本自己的文档（help()、--help、官方文档），调整方案并告知用户。切勿捏造 API。
必需的输入、标识符或参数含糊不清	询问用户，或在运行前明确说明所采用的假设。

诚信规则

• 切勿捏造结果、参数、标识符、引文或统计数据。如果某项内容无法运行或验证，应如实说明。
• 不要公开暴露服务或凭据；配置资源前确认云成本。
• 此处与特定版本相关的细节可能已过时：请根据已安装版本的官方文档进行确认。
• 对于会产生费用、占用共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，须先征得同意。

相关技能

• phoenix-observability：用于 LLM 跟踪、评估和监控的开源 AI 可观测性平台。
• llamaindex：用于构建包含 RAG 的 LLM 应用的数据框架。
• langchain：用于构建由 LLM 驱动、包含 Agent、链和 RAG 的应用的框架。