回调与监控

构建生产级 LLM 应用需要完善的可观测性能力——追踪每次调用、调试异常行为、评估输出质量。LangChain 通过 LangSmith 平台和回调系统提供了完整的解决方案。

LangSmith 简介

LangSmith 是 LangChain 官方提供的开发者平台，专注于 LLM 应用的全生命周期管理。它不仅仅是一个监控工具，更是一个集调试、测试、评估、部署于一体的综合性平台。

核心功能

追踪（Tracing）：可视化展示每次 LLM 调用的完整链路，包括输入、输出、耗时、Token 消耗等
调试（Debugging）：快速定位 Agent 决策、工具调用、提示词构建等环节的问题
评估（Evaluation）：在测试数据集上评估应用表现，支持多种评估指标
监控（Monitoring）：生产环境实时监控请求量、延迟、错误率等指标
部署（Deployment）：一键部署 LangGraph 应用，支持自动扩展

为什么需要 LangSmith？

LLM 应用的调试与传统软件有很大不同：

输出不确定性：同样的输入可能产生不同输出，需要记录每次调用
多步骤链路：一个请求可能涉及多次 LLM 调用、工具调用、检索等，需要完整追踪
成本敏感：Token 消耗直接影响成本，需要精确统计
质量评估：答案好坏难以用代码断言，需要人工或 AI 辅助评估

LangSmith 针对这些痛点提供了完整的解决方案。

快速开始

1. 注册和配置

首先在 smith.langchain.com 注册账号，获取 API Key：

# 安装依赖
pip install langsmith

# 配置环境变量
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=ls-xxxxx  # 从 LangSmith 获取

# 可选配置
export LANGSMITH_PROJECT=my-project  # 项目名称，默认为 "default"
export LANGSMITH_ENDPOINT=https://api.smith.langchain.com

2. 基础追踪

配置完成后，所有 LangChain 调用会自动追踪：

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-xxxxx"

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser

# 初始化
model = init_chat_model("openai:gpt-4o-mini")

# 构建链
chain = (
    ChatPromptTemplate.from_template("解释{topic}")
    | model
    | StrOutputParser()
)

# 执行会自动记录到 LangSmith
result = chain.invoke({"topic": "Python装饰器"})

# 访问 smith.langchain.com 查看追踪详情

3. 查看追踪结果

在 LangSmith Dashboard 中可以看到：

输入输出：每个组件的输入和输出内容
执行时间：每个步骤的耗时分析
Token 统计：LLM 调用的 Token 消耗
调用链路：完整的调用树结构
错误信息：如有错误会显示详细堆栈

追踪配置详解

环境变量配置

# 必需配置
LANGSMITH_TRACING=true           # 启用追踪
LANGSMITH_API_KEY=ls-xxxxx       # API 密钥

# 可选配置
LANGSMITH_PROJECT=my-project     # 项目名称
LANGSMITH_ENDPOINT=https://api.smith.langchain.com  # API 端点

# 性能优化
LANGCHAIN_CALLBACKS_BACKGROUND=true   # 后台异步发送追踪（默认）
# 服务端环境建议设置为 false，确保追踪完成

选择性追踪

有时只想追踪特定调用，而不是全部：

方式一：使用 LangChainTracer 回调

from langsmith import Client
from langchain.callbacks import LangChainTracer

# 创建追踪器
tracer = LangChainTracer(
    project_name="my-project",
    client=Client(api_key="ls-xxxxx")
)

# 只追踪特定调用
result = chain.invoke(
    {"topic": "Python装饰器"},
    config={"callbacks": [tracer]}
)

方式二：使用 tracing_context 上下文管理器

from langsmith import tracing_context

# 只在上下文内追踪
with tracing_context(project_name="my-project"):
    result = chain.invoke({"topic": "Python装饰器"})

# 上下文外的调用不会被追踪
result2 = chain.invoke({"topic": "其他问题"})

动态项目名称

根据环境动态设置项目：

import os

# 根据环境设置项目
if os.environ.get("ENV") == "production":
    os.environ["LANGSMITH_PROJECT"] = "production-app"
elif os.environ.get("ENV") == "staging":
    os.environ["LANGSMITH_PROJECT"] = "staging-app"
else:
    os.environ["LANGSMITH_PROJECT"] = "development"

添加元数据和标签

为追踪添加额外信息，方便后续查询和分析：

result = chain.invoke(
    {"topic": "Python装饰器"},
    config={
        "metadata": {
            "user_id": "user_123",
            "environment": "production",
            "version": "1.0.0"
        },
        "tags": ["customer-support", "faq"]
    }
)

元数据会继承到所有子调用，方便按用户、环境等维度分析。

自定义运行名称

为追踪设置有意义的名称：

result = chain.invoke(
    {"topic": "Python装饰器"},
    config={
        "run_name": "FAQ-Answer-Generation",
        "tags": ["faq", "python"]
    }
)

在 LangSmith 列表中会显示这个名称，便于识别。

自定义回调处理器

除了 LangSmith，还可以自定义回调处理器实现特定逻辑：

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult

class LoggingCallbackHandler(BaseCallbackHandler):
    """将 LLM 调用记录到日志"""
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        """LLM 开始调用"""
        self.start_time = time.time()
        print(f"[LLM 开始] 提示词: {prompts[0][:100]}...")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        """LLM 结束调用"""
        elapsed = time.time() - self.start_time
        token_usage = response.llm_output.get("token_usage", {})
        print(f"[LLM 结束] 耗时: {elapsed:.2f}s")
        print(f"[Token 消耗] 输入: {token_usage.get('prompt_tokens')}, "
              f"输出: {token_usage.get('completion_tokens')}")
    
    def on_llm_error(self, error, **kwargs):
        """LLM 调用出错"""
        print(f"[LLM 错误] {error}")

# 使用
result = chain.invoke(
    {"topic": "Python装饰器"},
    config={"callbacks": [LoggingCallbackHandler()]}
)

回调事件类型

BaseCallbackHandler 支持多种事件：

事件方法	触发时机
`on_llm_start`	LLM 开始生成
`on_llm_end`	LLM 生成结束
`on_llm_error`	LLM 调用出错
`on_chain_start`	Chain 开始执行
`on_chain_end`	Chain 执行结束
`on_tool_start`	工具开始执行
`on_tool_end`	工具执行结束
`on_retriever_start`	检索开始
`on_retriever_end`	检索结束

组合多个回调

from langchain.callbacks import LangChainTracer
from langchain.callbacks.stdout import StdOutCallbackHandler

callbacks = [
    LangChainTracer(project_name="my-project"),
    StdOutCallbackHandler(),  # 控制台输出
    LoggingCallbackHandler()  # 自定义处理
]

result = chain.invoke(
    {"topic": "Python装饰器"},
    config={"callbacks": callbacks}
)

调试技巧

查看完整调用链

在 LangSmith 中，每次调用形成一棵追踪树：

FAQ-Answer-Generation
├── ChatPromptTemplate
│   └── 格式化提示词
├── ChatOpenAI (gpt-4o-mini)
│   ├── 输入: "解释Python装饰器"
│   └── 输出: "装饰器是..."
└── StrOutputParser
    └── 解析输出

每个节点的详情包括：

输入内容：传入该组件的完整数据
输出内容：组件产生的结果
执行时间：耗时分析
Token 统计：LLM 调用的成本

定位 Agent 问题

Agent 不调用工具是常见问题，LangSmith 可以帮助诊断：

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search_docs(query: str) -> str:
    """搜索文档库"""
    return f"搜索结果: {query}"

agent = create_agent(
    model=model,
    tools=[search_docs],
    system_prompt="你是一个助手"
)

# 执行并查看 LangSmith
result = agent.invoke({
    "messages": [{"role": "user", "content": "帮我查一下 Python 教程"}]
})

在 LangSmith 中检查：

LLM 输出是否包含工具调用意图
工具描述是否足够清晰
系统提示是否引导正确

输出格式调试

使用输出解析器时，解析失败是常见问题：

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel

class Answer(BaseModel):
    summary: str
    details: list[str]

parser = PydanticOutputParser(pydantic_object=Answer)

# 在 LangSmith 中可以看到：
# 1. LLM 原始输出
# 2. 解析器错误信息
# 3. 期望的格式说明

评估系统

创建评估数据集

from langsmith import Client

client = Client()

# 创建数据集
dataset = client.create_dataset(
    dataset_name="faq-evaluation",
    description="FAQ 回答质量评估"
)

# 添加测试用例
client.create_example(
    inputs={"question": "Python 如何定义函数？"},
    outputs={"answer": "使用 def 关键字定义函数..."},
    dataset_id=dataset.id
)

client.create_example(
    inputs={"question": "如何处理异常？"},
    outputs={"answer": "使用 try-except 语句..."},
    dataset_id=dataset.id
)

运行评估

from langsmith.evaluation import evaluate

def target_function(inputs: dict) -> dict:
    """被评估的函数"""
    result = chain.invoke({"topic": inputs["question"]})
    return {"answer": result}

# 运行评估
evaluate(
    target_function,
    data="faq-evaluation",
    evaluators=[
        # 可以添加自定义评估器
    ]
)

自定义评估器

from langsmith.schemas import Run, Example

def accuracy_evaluator(run: Run, example: Example) -> dict:
    """自定义评估器"""
    prediction = run.outputs.get("answer", "")
    reference = example.outputs.get("answer", "")
    
    # 计算相似度或其他指标
    score = calculate_similarity(prediction, reference)
    
    return {
        "key": "accuracy",
        "score": score,
        "comment": f"预测与参考的相似度: {score:.2f}"
    }

# 使用自定义评估器
evaluate(
    target_function,
    data="faq-evaluation",
    evaluators=[accuracy_evaluator]
)

生产环境监控

监控指标

LangSmith Dashboard 提供关键业务指标：

请求量：每分钟/小时请求数，识别流量高峰
延迟分布：P50、P95、P99 延迟，发现性能瓶颈
错误率：失败请求比例，及时发现问题
Token 消耗：按时间段统计成本
成功率：成功完成的请求比例

告警配置

设置告警规则，异常时自动通知：

# 在 LangSmith UI 中配置：
# 1. 错误率 > 5% 触发告警
# 2. 延迟 P95 > 10s 触发告警
# 3. Token 消耗异常增长触发告警

采样策略

高流量场景下，可以只追踪部分请求：

import random

def should_trace() -> bool:
    """决定是否追踪"""
    # 错误总是追踪
    # 10% 正常请求追踪
    return random.random() < 0.1

# 或在环境中配置
# LANGSMITH_TRACING_SAMPLES=0.1  # 10% 采样

LangSmith 与 LangSmith SDK 互操作

如果同时使用 LangChain 和 LangSmith SDK：

from langsmith import traceable

@traceable(project_name="my-project")
def my_function(input_data):
    """使用 @traceable 装饰的函数"""
    # LangChain 调用会自动成为子追踪
    result = chain.invoke({"topic": input_data})
    return result

# 调用
result = my_function("Python装饰器")

LangChain 对象在 @traceable 函数内调用时，会自动成为该追踪的子节点。

最佳实践

开发环境配置

import os

# 开发环境：详细追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "development"
# 可以添加更多调试信息

生产环境配置

import os

# 生产环境：平衡追踪和性能
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "production"

# 服务端环境：同步发送确保完整
os.environ["LANGCHAIN_CALLBACKS_BACKGROUND"] = "false"

团队协作

# 使用项目区分不同团队或模块
projects = {
    "customer-service": "客服 AI",
    "content-generation": "内容生成",
    "data-analysis": "数据分析"
}

# 每个模块使用独立项目
os.environ["LANGSMITH_PROJECT"] = projects["customer-service"]

确保追踪提交

在脚本或 Lambda 中，确保追踪完成：

from langsmith import ensure_tags, get_client

def main():
    # 执行应用逻辑
    result = chain.invoke({"topic": "Python"})
    
    # 确保追踪提交
    client = get_client()
    client.flush()  # 等待所有追踪发送完成

if __name__ == "__main__":
    main()

常见问题

1. 追踪不显示

检查清单：

LANGSMITH_TRACING=true 是否设置
LANGSMITH_API_KEY 是否正确
网络是否能访问 api.smith.langchain.com
查看控制台是否有错误日志

2. 追踪延迟

设置 LANGCHAIN_CALLBACKS_BACKGROUND=true 异步发送
检查网络延迟
考虑批量发送

3. 敏感数据处理

# 不记录敏感信息的提示词
def sanitize_prompt(prompt: str) -> str:
    """脱敏处理"""
    # 移除或替换敏感信息
    return prompt.replace("API_KEY", "*****")

# 或在追踪时排除
os.environ["LANGSMITH_TRACING"] = "false"  # 敏感操作时关闭

下一步

现在你已经掌握了回调和监控的使用，接下来学习：

速查表 - 常用 API 和配置速查

LangSmith 简介​

核心功能​

为什么需要 LangSmith？​

快速开始​

1. 注册和配置​

2. 基础追踪​

3. 查看追踪结果​

追踪配置详解​

环境变量配置​

选择性追踪​

动态项目名称​

添加元数据和标签​

自定义运行名称​

自定义回调处理器​

回调事件类型​

组合多个回调​

调试技巧​

查看完整调用链​

定位 Agent 问题​

输出格式调试​

评估系统​

创建评估数据集​

运行评估​

自定义评估器​

生产环境监控​

监控指标​

告警配置​

采样策略​

LangSmith 与 LangSmith SDK 互操作​

最佳实践​

开发环境配置​

生产环境配置​

团队协作​

确保追踪提交​

常见问题​

1. 追踪不显示​

2. 追踪延迟​

3. 敏感数据处理​

下一步​