LangSmith 监控

LangSmith 是 LangChain 团队推出的开发者平台，用于监控、调试和评估 LLM 应用。它是生产环境中不可或缺的工具。

为什么需要 LangSmith

LLM 应用的调试挑战：

问题描述
黑盒调用不知道 LLM 内部发生了什么
链路追踪多步骤执行难以追踪
性能分析不清楚哪个环节耗时
成本控制Token 消耗不透明
质量评估难以量化输出质量

LangSmith 提供了解决这些问题的完整工具链。

核心功能

1. 可观测性（Tracing）

追踪每次 LLM 调用的完整信息：

  • 输入和输出
  • 执行时间
  • Token 使用量
  • 错误信息
  • 中间步骤

2. 评估（Evaluation）

测试和评分模型表现：

  • 自动化测试
  • 人工标注
  • 对比评估

3. 提示词工程

管理和优化提示词：

  • 版本控制
  • A/B 测试
  • 提示词优化

4. 数据集管理

管理测试数据：

  • 创建数据集
  • 标注样本
  • 回归测试

快速开始

1. 注册账号

访问 smith.langchain.com 注册账号。

2. 获取 API Key

在设置页面获取 API Key。

3. 配置环境变量

# .env 文件
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your-api-key
LANGCHAIN_PROJECT=my-project  # 项目名称

4. 在代码中启用

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载环境变量，自动启用 LangSmith
load_dotenv()

llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke("你好")
print(response.content)

# 访问 LangSmith 控制台查看追踪记录

追踪（Tracing）

自动追踪

设置环境变量后，所有 LangChain 调用自动追踪：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("翻译成英文：{text}")
chain = prompt | llm | StrOutputParser()

# 这次调用会自动追踪
result = chain.invoke({"text": "你好世界"})

手动追踪

使用装饰器追踪自定义函数：

from langsmith import traceable

@traceable(name="my_function")
def process_data(data: str) -> str:
    """处理数据的函数"""
    # 你的逻辑
    return f"处理结果：{data}"

@traceable(name="main_workflow")
def main_workflow(input_text: str):
    """主工作流"""
    step1 = process_data(input_text)
    
    llm = ChatOpenAI()
    response = llm.invoke(step1)
    
    return response.content

# 整个工作流会被追踪
result = main_workflow("测试输入")

添加元数据

from langsmith import traceable

@traceable(
    name="custom_chain",
    metadata={"version": "1.0", "author": "dev"},
    tags=["production", "v1"]
)
def custom_chain(input_text: str):
    # 你的逻辑
    pass

使用上下文管理器

from langsmith import trace

with trace(name="my_operation", metadata={"type": "experiment"}) as rt:
    # 在这个上下文中的所有调用都会被追踪
    llm = ChatOpenAI()
    result = llm.invoke("问题")
    
    # 可以添加自定义输出
    rt.end(outputs={"custom_result": "some value"})

评估（Evaluation）

创建数据集

from langsmith import Client

client = Client()

# 创建数据集
dataset = client.create_dataset(
    dataset_name="qa_dataset",
    description="问答测试数据集"
)

# 添加样本
examples = [
    {
        "input": {"question": "什么是 Python？"},
        "output": {"answer": "Python 是一种高级编程语言"}
    },
    {
        "input": {"question": "什么是 LangChain？"},
        "output": {"answer": "LangChain 是一个 AI 应用开发框架"}
    }
]

for example in examples:
    client.create_example(
        inputs=example["input"],
        outputs=example["output"],
        dataset_id=dataset.id
    )

运行评估

from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI

# 定义要评估的函数
def my_qa_function(inputs: dict) -> dict:
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    response = llm.invoke(inputs["question"])
    return {"answer": response.content}

# 定义评估器
def exact_match(run, example):
    """精确匹配评估"""
    prediction = run.outputs.get("answer", "")
    reference = example.outputs.get("answer", "")
    return {"score": 1 if prediction == reference else 0}

def contains_keyword(run, example):
    """关键词包含评估"""
    prediction = run.outputs.get("answer", "").lower()
    reference = example.outputs.get("answer", "").lower()
    
    # 检查参考答案中的关键词是否出现在预测中
    keywords = reference.split()[:3]  # 取前3个词
    matches = sum(1 for kw in keywords if kw in prediction)
    
    return {"score": matches / len(keywords) if keywords else 0}

# 运行评估
results = evaluate(
    my_qa_function,
    data="qa_dataset",
    evaluators=[exact_match, contains_keyword],
    experiment_prefix="qa_eval"
)

print(f"评估完成，结果ID: {results.experiment_name}")

使用内置评估器

from langsmith.evaluation import evaluate, LangChainStringEvaluator

# 使用 LangChain 内置评估器
evaluators = [
    LangChainStringEvaluator("cot_qa"),  # 思维链问答评估
    LangChainStringEvaluator("criteria", config={"criteria": "conciseness"}),
]

results = evaluate(
    my_qa_function,
    data="qa_dataset",
    evaluators=evaluators
)

提示词管理

在 LangSmith 中创建提示词

from langchain import hub
from langchain_core.prompts import ChatPromptTemplate

# 创建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}"),
    ("human", "{input}")
])

# 推送到 LangSmith Hub
hub.push("my-prompt", prompt, new_repo_is_public=False)

从 Hub 拉取提示词

from langchain import hub

# 拉取提示词
prompt = hub.pull("my-prompt")

# 使用特定版本
prompt_v1 = hub.pull("my-prompt:v1")

提示词版本管理

from langchain import hub

# 推送新版本
prompt_v2 = ChatPromptTemplate.from_messages([
    ("system", "你是一个更专业的{role}，请详细回答"),
    ("human", "{input}")
])

hub.push("my-prompt", prompt_v2)  # 自动创建新版本

监控面板

关键指标

LangSmith 面板显示：

  • 延迟 - 每次调用的响应时间
  • Token 使用 - 输入/输出 Token 数
  • 成功率 - 调用成功/失败比例
  • 成本 - 估算的 API 费用

筛选和搜索

# 使用标签方便筛选
from langsmith import traceable

@traceable(
    tags=["production", "customer_service"],
    metadata={"user_id": "user_123"}
)
def customer_service_bot(message: str):
    # ...
    pass

设置告警

在 LangSmith 控制台可以设置：

  • 延迟阈值告警
  • 错误率告警
  • 成本告警

反馈收集

记录用户反馈

from langsmith import Client

client = Client()

# 获取运行 ID（从追踪中获取）
run_id = "your-run-id"

# 记录反馈
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1.0,  # 0-1 分数
    comment="回答很有帮助"
)

# 记录二元反馈（点赞/点踩）
client.create_feedback(
    run_id=run_id,
    key="thumbs_up",
    score=1  # 1 = 赞，0 = 踩
)

在应用中集成反馈

from langchain_openai import ChatOpenAI
from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

@traceable
def chat_with_feedback(message: str):
    llm = ChatOpenAI()
    response = llm.invoke(message)
    
    # 获取当前运行 ID
    run_tree = get_current_run_tree()
    run_id = run_tree.id if run_tree else None
    
    return {
        "response": response.content,
        "run_id": str(run_id)  # 返回给前端用于反馈
    }

# 使用
result = chat_with_feedback("你好")
print(f"回复: {result['response']}")
print(f"运行ID: {result['run_id']}")  # 用户可以用这个ID提交反馈

完整示例：生产级监控

"""
生产级监控示例 - LangSmith
"""

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langsmith import Client, traceable
from langsmith.run_helpers import get_current_run_tree
from datetime import datetime

# 加载环境变量
load_dotenv()

# 确保启用追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-chatbot"

client = Client()

class ProductionChatbot:
    """生产级聊天机器人"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的客服助手。
请友好、准确地回答用户问题。
如果不确定，请诚实说明。"""),
            ("human", "{input}")
        ])
        
        self.chain = self.prompt | self.llm | StrOutputParser()
    
    @traceable(
        name="chat",
        tags=["production", "customer_service"],
        metadata={"version": "1.0"}
    )
    def chat(self, message: str, user_id: str = None) -> dict:
        """处理用户消息"""
        start_time = datetime.now()
        
        try:
            response = self.chain.invoke({"input": message})
            
            # 获取运行信息
            run_tree = get_current_run_tree()
            run_id = str(run_tree.id) if run_tree else None
            
            end_time = datetime.now()
            latency = (end_time - start_time).total_seconds()
            
            return {
                "success": True,
                "response": response,
                "run_id": run_id,
                "latency": latency,
                "user_id": user_id
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "user_id": user_id
            }
    
    def submit_feedback(self, run_id: str, rating: int, comment: str = None):
        """提交用户反馈"""
        if not run_id:
            return False
        
        try:
            # 提交评分
            client.create_feedback(
                run_id=run_id,
                key="user_rating",
                score=rating / 5.0,  # 转换为 0-1
                comment=comment
            )
            
            # 提交二元反馈
            client.create_feedback(
                run_id=run_id,
                key="helpful",
                score=1 if rating >= 4 else 0
            )
            
            return True
        except Exception as e:
            print(f"反馈提交失败: {e}")
            return False
    
    def get_metrics(self, hours: int = 24) -> dict:
        """获取监控指标"""
        # 这里是模拟数据，实际应该从 LangSmith API 获取
        return {
            "total_requests": 1000,
            "success_rate": 0.98,
            "avg_latency": 1.2,
            "total_tokens": 50000,
            "estimated_cost": 0.15
        }


def main():
    bot = ProductionChatbot()
    
    print("=" * 50)
    print("  生产级聊天机器人（LangSmith 监控）")
    print("  输入 'quit' 退出")
    print("  输入 'metrics' 查看监控指标")
    print("  输入 'feedback' 提交反馈")
    print("=" * 50)
    
    last_run_id = None
    
    while True:
        user_input = input("\n你：").strip()
        
        if not user_input:
            continue
        
        if user_input.lower() == "quit":
            print("再见！")
            break
        
        if user_input.lower() == "metrics":
            metrics = bot.get_metrics()
            print("\n--- 监控指标（最近24小时）---")
            print(f"总请求数: {metrics['total_requests']}")
            print(f"成功率: {metrics['success_rate']*100:.1f}%")
            print(f"平均延迟: {metrics['avg_latency']:.2f}s")
            print(f"总 Token: {metrics['total_tokens']}")
            print(f"估算成本: ${metrics['estimated_cost']:.2f}")
            continue
        
        if user_input.lower() == "feedback":
            if not last_run_id:
                print("没有可反馈的对话")
                continue
            
            try:
                rating = int(input("评分 (1-5): "))
                comment = input("评论（可选）: ").strip() or None
                
                if bot.submit_feedback(last_run_id, rating, comment):
                    print("感谢您的反馈！")
                else:
                    print("反馈提交失败")
            except ValueError:
                print("无效的评分")
            continue
        
        # 正常对话
        result = bot.chat(user_input, user_id="demo_user")
        
        if result["success"]:
            print(f"\n助手：{result['response']}")
            print(f"[延迟: {result['latency']:.2f}s]")
            last_run_id = result["run_id"]
        else:
            print(f"\n错误：{result['error']}")


if __name__ == "__main__":
    main()

小结

本章介绍了：

✅ LangSmith 的核心功能
✅ 配置和启用追踪
✅ 手动追踪和元数据
✅ 创建和运行评估
✅ 提示词版本管理
✅ 监控面板使用
✅ 用户反馈收集
✅ 生产级监控示例

下一步

掌握 LangSmith 后，让我们学习部署与优化，将你的 AI 应用部署到生产环境。

练习

  1. 在 LangSmith 中创建一个项目并追踪调用
  2. 创建测试数据集并运行评估
  3. 设置自定义评估器
  4. 实现用户反馈收集功能