跳到主要内容

回调与监控

LangChain 提供了完善的回调和监控功能，帮助开发者调试和优化应用。本章将介绍如何使用 LangSmith 和回调系统。

LangSmith 简介​

LangSmith 是 LangChain 官方提供的调试、测试和监控平台，可以帮助你：

  • 可视化 Chain 的执行过程
  • 调试 Agent 行为
  • 评估输出质量
  • 团队协作

为什么使用 LangSmith？​

  • 可视化执行：直观看到每个步骤的输入输出
  • 调试：快速定位问题
  • 评估：测试和评估应用效果
  • 监控：生产环境实时监控

快速开始​

环境配置​

# 安装 langsmith
pip install langsmith
# 设置环境变量
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-xxxxx"  # 从 langsmith.ai 获取

基础追踪​

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate

# 开启追踪
os.environ["LANGSMITH_TRACING"] = "true"

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

prompt = ChatPromptTemplate.from_template("{question}")
chain = prompt | model

# 执行会自动记录到 LangSmith
result = chain.invoke({"question": "Python 是什么?"})

# 访问 https://smith.langchain.com 查看追踪结果

追踪配置​

环境变量​

# 必需
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=ls-xxxxx

# 可选
export LANGSMITH_PROJECT=my-project        # 项目名称
export LANGSMITH_ENDPOINT=https://api.smith.langchain.com  # 端点
export LANGSMITH_TRACING_V2=true            # 使用 v2 API

代码配置​

from langsmith import Client

# 创建客户端
client = Client(
    api_key="ls-xxxxx",
    project_name="my-project"
)

# 配置特定链
from langchain.callbacks import LangChainTracer

tracer = LangChainTracer(
    project_name="my-project",
    client=client
)

在不同组件中使用​

Chain 追踪​

from langchain.chat_models import init_chat_model
from langchain.chains import LLMChain

# 追踪 LLMChain
chain = LLMChain(
    llm=model,
    prompt=prompt,
    callbacks=[tracer]
)

result = chain.invoke({"input": "test"})

Agent 追踪​

from langchain.agents import create_agent

# 追踪 Agent
agent = create_agent(
    model=model,
    tools=[tool1, tool2],
    tracing_allowed_tools=["tool1", "tool2"]  # 指定要追踪的工具
)

自定义回调​

from langchain.callbacks.base import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs):
        print(f"Chain 开始: {inputs}")
    
    def on_chain_end(self, outputs):
        print(f"Chain 结束: {outputs}")
    
    def on_llm_start(self, serialized, prompts):
        print(f"LLM 开始: {prompts}")
    
    def on_tool_start(self, serialized, input_str):
        print(f"工具开始: {input_str}")
    
    def on_tool_end(self, output):
        print(f"工具结束: {output}")

# 使用
chain = LLMChain(llm=model, prompt=prompt, callbacks=[MyCallbackHandler()])

调试技巧​

查看执行详情​

在 LangSmith Dashboard 中可以看到：

  • 输入输出：每个步骤的输入和输出
  • Token 消耗：LLM 使用的 token 数量
  • 执行时间：每个步骤的耗时
  • 错误信息：如有错误会显示

常见问题排查​

1. Agent 不调用工具​

# 在 LangSmith 中检查：
# 1. 查看 LLM 的输出是否包含工具调用
# 2. 检查工具描述是否清晰
# 3. 查看系统提示是否正确

2. 输出格式错误​

# 查看：
# 1. LLM 输出内容
# 2. Output Parser 解析过程
# 3. 任何解析错误

3. 性能问题​

# 查看执行时间：
# 1. 哪个步骤最耗时
# 2. LLM 调用次数
# 3. 网络延迟

评估系统​

创建评估数据集​

from langsmith import Client

client = Client()

# 添加示例
client.create_example(
    inputs={"question": "Python 装饰器是什么?"},
    outputs={"answer": "装饰器是..."},
    dataset_name="python-tutor"
)

运行评估​

from langsmith.evaluation import evaluate

def evaluate_chain(example):
    result = chain.invoke({"question": example["question"]})
    return {"prediction": result["answer"]}

# 评估
evaluate(
    "my-chain",
    evaluate_chain,
    data="python-tutor"
)

生产监控​

设置生产追踪​

import os

# 生产环境配置
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = os.environ.get("LANGSMITH_API_KEY")

# 添加项目标签
from langsmith import traceable

@traceable(project_name="production")
def my_chain(input_data):
    return chain.invoke(input_data)

监控指标​

在 LangSmith Dashboard 可以监控：

  • 请求量：每分钟/小时请求数
  • 延迟：平均响应时间
  • 错误率：失败请求比例
  • Token 消耗：Token 使用量
  • 成本：估计花费

回调进阶​

自定义回调处理器​

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish

class DebugHandler(BaseCallbackHandler):
    def on_agent_action(self, action: AgentAction, **kwargs):
        print(f"Agent 行动: {action.tool}")
        print(f"输入: {action.tool_input}")
    
    def on_agent_finish(self, finish: AgentFinish, **kwargs):
        print(f"Agent 完成: {finish.return_values}")

多个回调​

from langchain.callbacks import get_based_callbacks

callbacks = [
    LangChainTracer(),
    DebugHandler(),
    # 添加更多回调
]

chain = LLMChain(
    llm=model,
    prompt=prompt,
    callbacks=callbacks
)

日志记录​

结构化日志​

from langchain.schema import HumanMessage
from langchain.callbacks.stdout import StdOutCallbackHandler

# 打印到 stdout
chain = LLMChain(
    llm=model,
    prompt=prompt,
    verbose=True  # 或使用 callbacks
)

文件日志​

from langchain.callbacks.file import FileCallbackHandler
import logging

logging.basicConfig(filename='app.log', level=logging.INFO)

file_handler = FileCallbackHandler("app.log")

chain = LLMChain(
    llm=model,
    prompt=prompt,
    callbacks=[file_handler]
)

最佳实践​

1. 开发环境​

# 启用详细追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_VERBOSE"] = "true"

2. 生产环境​

# 只追踪错误
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_TRACING_SAMPLES"] = "false"  # 只追踪错误

3. 团队协作​

# 使用项目区分不同环境
os.environ["LANGSMITH_PROJECT"] = "production-v1"

常见问题​

1. 追踪不显示​

  • 检查 API Key 是否正确
  • 确认 LANGSMITH_TRACING=true
  • 查看是否有网络问题

2. 性能影响​

  • 异步追踪减少延迟
  • 生产环境可以采样追踪
  • 关闭不需要的详细追踪

3. 数据隐私​

  • LangSmith 会记录输入输出
  • 如需敏感数据处理，考虑本地追踪
  • 使用数据脱敏

下一步​

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