Callbacks 与追踪
Callbacks（回调）和 Tracing（追踪）是 LangChain 中用于监控、调试和分析 Agent 执行过程的重要机制。通过它们可以观察 LLM 调用、工具执行、检索操作等内部细节。
概述
┌─────────────────────────────────────────────────────┐
│ Agent 执行追踪 │
│ │
│ LLM 调用 工具调用 检索操作 链执行 错误处理 │
│ ↓ ↓ ↓ ↓ ↓ │
│ ┌──────────────────────────────────────────────┐ │
│ │ LangSmith 追踪平台 │ │
│ └──────────────────────────────────────────────┘ │
│ │
│ → 可视化执行流程 → 分析 token 消耗 → 调试问题 │
└─────────────────────────────────────────────────────┘LangSmith 追踪
LangSmith 是 LangChain 的官方追踪和调试平台。通过简单的环境变量配置，可以自动追踪所有 Agent 执行。
基本配置
bash
# 设置环境变量
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-api-key"
export LANGSMITH_PROJECT="my-langchain-app"或者在代码中配置：
python
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_PROJECT"] = "my-langchain-app"启用追踪后的自动捕获
配置好环境变量后，所有 LangChain 调用都会自动被追踪：
python
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取城市天气"""
return f"{city} 天气：晴朗，22°C"
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini", streaming=True),
tools=[get_weather],
system_prompt="你是一个天气助手。"
)
# 执行 agent（所有步骤都会被追踪到 LangSmith）
result = agent.invoke({
"messages": [("human", "北京的天气怎么样？")]
})查看追踪信息
在 LangSmith 平台可以查看：
- 执行时间线：每个步骤的耗时
- token 使用：输入/输出的 token 数量
- 参数详情：LLM 参数、工具输入输出
- 错误信息：执行过程中的错误详情
- 成本估算：API 调用成本
Callback 处理器
LangChain 提供了丰富的回调事件，可以在 Agent 执行时触发自定义逻辑。
基础回调示例
python
from langchain_core.callbacks import BaseCallbackHandler
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
class MyCallbackHandler(BaseCallbackHandler):
"""自定义回调处理器"""
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"🤖 LLM 开始处理...")
print(f" 提示词数: {len(prompts)}")
def on_llm_end(self, response, **kwargs):
print(f"✅ LLM 处理完成")
print(f" 生成文本: {response.generations[0][0].text[:50]}...")
def on_llm_error(self, error, **kwargs):
print(f"❌ LLM 错误: {error}")
def on_tool_start(self, serialized, input_str, **kwargs):
print(f"🔧 工具开始: {serialized.get('name', 'unknown')}")
print(f" 输入: {input_str}")
def on_tool_end(self, output, **kwargs):
print(f"✅ 工具完成")
print(f" 输出: {str(output)[:80]}")
def on_tool_error(self, error, **kwargs):
print(f"❌ 工具错误: {error}")
def on_chain_start(self, serialized, inputs, **kwargs):
print(f"⛓️ 链开始执行")
def on_chain_end(self, outputs, **kwargs):
print(f"✅ 链执行完成")
# 使用回调
handler = MyCallbackHandler()
@tool
def search_info(query: str) -> str:
"""搜索信息"""
return f"关于 '{query}' 的搜索结果"
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=[search_info],
callbacks=[handler]
)
result = agent.invoke({
"messages": [("human", "搜索一下人工智能的最新发展")]
})详细回调事件清单
python
from langchain_core.callbacks import BaseCallbackHandler
class DetailedCallbackHandler(BaseCallbackHandler):
"""详细回调处理器"""
# ---- LLM 相关 ----
def on_llm_start(self, serialized, prompts, **kwargs):
"""LLM 开始生成"""
pass
def on_llm_new_token(self, token, **kwargs):
"""LLM 生成新 token（流式模式）"""
pass
def on_llm_end(self, response, **kwargs):
"""LLM 生成完成"""
pass
def on_llm_error(self, error, **kwargs):
"""LLM 生成出错"""
pass
# ---- 链相关 ----
def on_chain_start(self, serialized, inputs, **kwargs):
"""链开始执行"""
pass
def on_chain_end(self, outputs, **kwargs):
"""链执行完成"""
pass
def on_chain_error(self, error, **kwargs):
"""链执行出错"""
pass
# ---- 工具相关 ----
def on_tool_start(self, serialized, input_str, **kwargs):
"""工具开始执行"""
pass
def on_tool_end(self, output, **kwargs):
"""工具执行完成"""
pass
def on_tool_error(self, error, **kwargs):
"""工具执行出错"""
pass
# ---- 检索器相关 ----
def on_retriever_start(self, serialized, query, **kwargs):
"""检索器开始检索"""
pass
def on_retriever_end(self, documents, **kwargs):
"""检索器检索完成"""
pass
def on_retriever_error(self, error, **kwargs):
"""检索器检索出错"""
pass
# ---- 文本相关 ----
def on_text(self, text, **kwargs):
"""文本事件（日志、中间结果等）"""
pass带 LangSmith 追踪的完整示例
python
import os
from langgraph.prebuilt import create_agent
from langgraph.checkpoint import InMemorySaver
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.tools import tool
from langchain_core.callbacks import BaseCallbackHandler
# 配置 LangSmith 追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "rag-demo"
# 构建检索工具
loader = TextLoader("knowledge.txt")
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(docs)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
@tool
def search_database(query: str) -> str:
"""从知识库搜索信息"""
docs = retriever.invoke(query)
return "\n\n".join([
f"[来源: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
for doc in docs
])
# 控制台日志回调
class ConsoleLogger(BaseCallbackHandler):
def on_tool_start(self, serialized, input_str, **kwargs):
print(f"\n📋 工具调用: {serialized.get('name', '?')}")
print(f" 输入: {input_str[:100]}")
def on_tool_end(self, output, **kwargs):
print(f" 输出: {str(output)[:100]}...")
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"\n🤔 LLM 思考中... (提示词: {len(prompts)} 条)")
# 创建 Agent
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True),
tools=[search_database],
checkpointer=InMemorySaver(),
callbacks=[ConsoleLogger()],
system_prompt="使用 search_database 工具回答问题。"
)
# 执行
print("=== 开始 Agent 执行 ===\n")
result = agent.invoke({
"messages": [("human", "什么是 RAG 技术？它有哪些优势？")]
})
print("\n=== 最终回答 ===")
print(result["messages"][-1].content)日志级别追踪
python
from langchain_core.callbacks import BaseCallbackHandler
import logging
class LoggingCallbackHandler(BaseCallbackHandler):
"""日志回调处理器"""
def __init__(self, logger=None, level=logging.INFO):
self.logger = logger or logging.getLogger(__name__)
self.level = level
def on_llm_start(self, serialized, prompts, **kwargs):
self.logger.log(self.level, f"LLM 调用开始")
self.logger.debug(f"提示词: {prompts}")
def on_llm_end(self, response, **kwargs):
usage = response.generations[0][0]
self.logger.log(self.level, f"LLM 调用完成")
self.logger.info(f"Token 使用: 输入/输出")
def on_tool_start(self, serialized, input_str, **kwargs):
self.logger.info(f"工具 {serialized.get('name')} 调用")
self.logger.debug(f"工具输入: {input_str}")
def on_tool_end(self, output, **kwargs):
self.logger.info(f"工具完成: {str(output)[:50]}...")
# 使用
logging.basicConfig(level=logging.INFO)
log_handler = LoggingCallbackHandler()
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
callbacks=[log_handler]
)标记特定运行（Run）
为追踪添加自定义标签和元数据：
python
from langchain_core.callbacks import BaseCallbackHandler
import uuid
class RunTracker(BaseCallbackHandler):
"""运行追踪器"""
def __init__(self):
self.run_id = str(uuid.uuid4())
def on_chain_start(self, serialized, inputs, **kwargs):
parent_run = kwargs.get("parent_run_id")
print(f"运行 ID: {self.run_id}")
print(f"父运行 ID: {parent_run}")调用链追踪参数
python
from langchain_core.tracers import LangChainTracer
from langchain_core.callbacks import CallbackManager
# 手动创建追踪器
tracer = LangChainTracer(
project_name="my-project",
tags=["production", "v1.0"],
metadata={
"environment": "production",
"version": "1.0.0",
"user": "张三"
}
)
callback_manager = CallbackManager([tracer])
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
callbacks=callback_manager
)自定义追踪器
python
from langchain_core.tracers import BaseTracer
from typing import Any, Dict
class ConsoleTracer(BaseTracer):
"""简单的控制台追踪器"""
def __init__(self):
super().__init__()
self.runs = []
def _persist_run(self, run: Dict[str, Any]) -> None:
"""保存运行记录"""
self.runs.append({
"id": run["id"],
"name": run["name"],
"type": run["run_type"],
"inputs": run.get("inputs"),
"outputs": run.get("outputs"),
"start_time": run.get("start_time"),
"end_time": run.get("end_time"),
"error": run.get("error"),
})
# 控制台输出
print(f"[{run['run_type']}] {run['name']}")
if run.get("error"):
print(f" ERROR: {run['error']}")
# 使用自定义追踪器
tracer = ConsoleTracer()
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
callbacks=[tracer]
)
result = agent.invoke({
"messages": [("human", "你好")]
})
# 查看运行记录
for run in tracer.runs:
print(f"{run['type']}: {run['name']}")性能监控
python
import time
from langchain_core.callbacks import BaseCallbackHandler
class PerformanceMonitor(BaseCallbackHandler):
"""性能监控回调"""
def __init__(self):
self.timings = {}
self.start_times = {}
def on_llm_start(self, serialized, prompts, **kwargs):
run_id = kwargs.get("run_id")
self.start_times[run_id] = time.time()
def on_llm_end(self, response, **kwargs):
run_id = kwargs.get("run_id")
if run_id in self.start_times:
elapsed = time.time() - self.start_times[run_id]
self.timings["llm"] = elapsed
print(f"⏱️ LLM 耗时: {elapsed:.2f}s")
def on_tool_start(self, serialized, input_str, **kwargs):
run_id = kwargs.get("run_id")
self.start_times[run_id] = time.time()
print(f"🔧 工具: {serialized.get('name')} 开始")
def on_tool_end(self, output, **kwargs):
run_id = kwargs.get("run_id")
if run_id in self.start_times:
elapsed = time.time() - self.start_times[run_id]
print(f"⏱️ 工具耗时: {elapsed:.2f}s")
def print_summary(self):
print("\n=== 性能摘要 ===")
for key, value in self.timings.items():
print(f"{key}: {value:.2f}s")最佳实践
- 始终启用 LangSmith 追踪：开发环境设置
LANGSMITH_TRACING=true - 使用有意义的项目名：不同项目使用不同的
LANGSMITH_PROJECT - 自定义回调用于日志：生产环境使用自定义回调记录关键事件
- 监控性能：使用回调监控 LLM 和工具的响应时间
- 错误处理：在回调中捕获并记录所有错误
- 标记版本：为不同版本的部署添加标签和元数据
- 隐私保护：不要在追踪中包含敏感信息
下一步
- Streaming 流式输出：结合回调实现流式追踪
- 测试与评估：使用追踪数据评估 Agent 性能
- RAG 应用设计：监控 RAG 应用的执行流程
- Memory 记忆：追踪记忆操作