Callbacks 是 LangChain 里唯一贯穿所有组件的横切接口。Runnable / ChatModel / Retriever / Tool / Agent 在执行的关键节点（开始、结束、出错、产出新 token）都会回调注册到 handler 上的 on_* 方法。换句话说：你不需要改业务代码，只要挂一个 handler，就能旁路拿到整条链路的事件流——这正是日志、计时、token 计数、追踪、流式推送的统一底座。

一、Callbacks：给链路装上可观测性

1.1 自定义 BaseCallbackHandler 钩子

继承 BaseCallbackHandler，重写你关心的钩子即可。常用钩子按「组件 + 生命周期」命名：on_llm_start / on_llm_new_token（流式逐 token）/ on_llm_end / on_llm_error、on_chain_start/end、on_tool_start/end、on_retriever_start/end、Agent 专属的 on_agent_action / on_agent_finish。下面这个 handler 做了生产里最实用的三件事：计时、统计 token、打印 Agent 的每一步决策。

# 依赖（langchain 0.3.x 已拆包，core 是回调与 Runnable 所在）
pip install -U langchain langchain-core langchain-openai
import time
from typing import Any
from uuid import UUID
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_core.agents import AgentAction, AgentFinish


class TimingTokenHandler(BaseCallbackHandler):
    """统计每次 LLM 调用的耗时与 token，并打印 Agent 决策轨迹。"""

    def __init__(self) -> None:
        self._t0: dict[UUID, float] = {}
        self.total_tokens = 0

    # --- LLM 生命周期 ---
    def on_llm_start(self, serialized, prompts, *, run_id: UUID, **kwargs: Any) -> None:
        self._t0[run_id] = time.perf_counter()
        print(f"[LLM start] prompt 数={len(prompts)}")

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        # 仅在 streaming=True 时触发：拿到逐个吐出的 token
        print(token, end="", flush=True)

    def on_llm_end(self, response: LLMResult, *, run_id: UUID, **kwargs: Any) -> None:
        dt = time.perf_counter() - self._t0.pop(run_id, time.perf_counter())
        # token_usage 在 llm_output 里（不同 provider 字段名可能不同）
        usage = (response.llm_output or {}).get("token_usage", {})
        used = usage.get("total_tokens", 0)
        self.total_tokens += used
        print(f"\n[LLM end] 耗时={dt:.2f}s tokens={used} 累计={self.total_tokens}")

    def on_llm_error(self, error: BaseException, **kwargs: Any) -> None:
        print(f"[LLM error] {type(error).__name__}: {error}")

    # --- Agent 决策轨迹 ---
    def on_agent_action(self, action: AgentAction, **kwargs: Any) -> None:
        print(f"[Agent action] 调用工具={action.tool} 入参={action.tool_input}")

    def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> None:
        print(f"[Agent finish] {finish.return_values}")


# 两种挂载方式：
# (A) 构造期挂载——对该实例的所有调用生效
from langchain_openai import ChatOpenAI
handler = TimingTokenHandler()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, callbacks=[handler])
llm.invoke("用一句话解释什么是回调")

# (B) 运行期挂载——只对这一次调用生效，更灵活，推荐
llm2 = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm2.invoke("再来一句", config={"callbacks": [TimingTokenHandler()]})

1.2 set_verbose / set_debug：零代码快速看内部

不想写 handler 时，LangChain 提供两个全局开关。set_verbose(True) 打印每个组件的格式化进出（人类可读）；set_debug(True) 打印全量原始事件（含 raw prompt、完整响应、嵌套结构），调试链路最强但最吵。二者本质都是往全局回调管理器里装内置 handler。

from langchain.globals import set_verbose, set_debug

# 轻量：只看每步的可读输入输出
set_verbose(True)

# 重量：看到喂给模型的完整 prompt、完整返回、每层嵌套（排查 prompt 没拼对、
# RAG 检索到的上下文不对、Agent 中间步骤异常时用它）
set_debug(True)

# 调试完务必关掉，否则生产日志会被刷爆
set_debug(False)
set_verbose(False)

1.3 LangSmith：生产级追踪（推荐的可观测性方案）

set_debug 适合本地一次性排查，但生产环境你需要可检索、可对比、可分享的追踪。LangSmith 是官方托管追踪平台，开启方式是纯环境变量，不需要改一行业务代码——LangChain 在运行时检测到这些变量就会自动把每条链路上报。

# 在 .env 或 shell 里设置；设好后所有 LCEL / Agent 调用自动上报到 LangSmith
export LANGSMITH_TRACING=true
export LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
export LANGSMITH_API_KEY="lsv2_..."     # 在 smith.langchain.com 申请
export LANGSMITH_PROJECT="my-langchain-app"   # 不设则归到 default 项目

# 注意：旧版本用 LANGCHAIN_ 前缀（LANGCHAIN_TRACING_V2 等），
# 新版本统一为 LANGSMITH_ 前缀，两套目前仍兼容，新项目用 LANGSMITH_
场景用什么为什么
想精确控制某次调用、做自定义计时/计数/推送自定义 BaseCallbackHandler唯一能拿到结构化事件对象、可编程的方案
本地临时看某步输入输出对不对set_verbose(True)一行开关、人类可读、零依赖
本地排查 prompt/上下文/中间步骤的原始内容set_debug(True)全量原始事件，最详细
生产环境长期可观测、团队共享、回归对比LangSmith 环境变量托管、可检索、可视化、不改代码
口诀编程拿事件→Handler；瞄一眼→verbose；挖原始→debug；上生产→LangSmith

二、四类高频生产踩坑

版本兼容：import 突然报错

典型表现
升级后 from langchain.chat_models import ChatOpenAI 报 ImportError 或 deprecation 警告；initialize_agent 等老 API 找不到。
判断标准
import 全部来自正确的拆分包，无 LangChainDeprecationWarning，pip show langchain langchain-core 版本一致段（如都在 0.3.x）。
解决方向
langchain 0.1 起按 provider 拆包：模型类从 langchain_openai 等集成包导入，不再从 langchain.chat_models；核心抽象（Runnable、回调、prompts、output_parsers）从 langchain_core 导入。装包时让 langchain 与 langchain-core 大版本对齐，集成包单独装：pip install -U langchain langchain-core langchain-openai。

token 超限：context length exceeded

典型表现
报 context_length_exceeded 或 maximum context length is N tokens；RAG/长对话/大文档时尤甚。
判断标准
单次请求 input tokens + 预留 output tokens < 模型上下文窗口；长对话有裁剪策略。
解决方向
三管齐下：①RAG 端控制检索条数与分块大小（k 调小、chunk_size 调小）；②对话历史用 trim_messages 按 token 上限裁剪（见上一章 Memory）；③用回调里的 token 统计（1.1 的 handler）提前预警。根因是没给 output 留预算——窗口要减去你期望的回复长度。

流式与结构化输出冲突

典型表现
对 with_structured_output(...) 的链路调用 stream()，要么报错、要么只在最后一次性吐出完整对象，逐 token 流式失效。
判断标准
需要逐 token 的展示用纯文本/字符串解析；需要严格 JSON/Pydantic 的场景接受非增量返回，或改用流式 JSON 解析器。
解决方向
结构化输出底层走 tool calling，必须等参数收齐才能解析成对象，天然与「逐 token 渐进」矛盾。要么：①面向人类的流式展示就用普通文本输出 + StrOutputParser；②确需结构化又想流，改用 JsonOutputParser（支持解析部分 JSON 增量），并接受字段是逐步补全的中间态。不要既要严格 Pydantic 又要逐字流。

Agent 死循环 / 不收敛

典型表现
Agent 反复调用同一个工具、来回试探不给最终答案，直到 token 烧光或超时。
判断标准
Agent 在有限步内终止；设置了迭代与时间上限；observation 信息足够让模型决策。
解决方向
构造 AgentExecutor 时务必设兜底：max_iterations（最大步数）和 max_execution_time（墙钟秒数），并设 early_stopping_method。同时检查工具返回的 observation 是否清晰——模型看不懂工具结果就会反复试。开 set_debug(True) 或看 LangSmith 轨迹定位它卡在哪一步。
# Agent 死循环的兜底配置（真实 AgentExecutor 参数）
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个会用工具的助手，能直接回答就别调用工具。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
tools = []  # 这里放你的工具列表
agent = create_tool_calling_agent(llm, tools, prompt)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=6,           # 最多 6 步，防死循环
    max_execution_time=30,      # 最多跑 30 秒墙钟时间
    early_stopping_method="force",  # 到上限直接停
    handle_parsing_errors=True, # 解析失败时把错误喂回模型而不是崩
    verbose=True,               # 打印中间步骤，便于看它卡哪
)
# executor.invoke({"input": "..."})

三、LangServe：一行把 Runnable 变成 FastAPI 服务

LangServe 的核心心智模型只有一句：任何 Runnable 都能被 add_routes(app, runnable, path) 暴露成 REST 接口。因为整个 LangChain 的统一接口就是 Runnable（invoke/stream/batch + 异步版），LangServe 只是把这套接口一一映射成 HTTP 端点：/invoke、/batch、/stream、/stream_events，外加自动生成的输入输出 schema 和一个交互式 /playground。服务端是标准 FastAPI，客户端用 RemoteRunnable 把远端接口当本地 Runnable 用。

# LangServe 服务端依赖
pip install -U "langserve[server]" fastapi "uvicorn[standard]" langchain-openai
# 仅做客户端（RemoteRunnable）时
pip install -U "langserve[client]"
# server.py —— 把一条 LCEL 链暴露为生产接口
from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langserve import add_routes

app = FastAPI(
    title="LangChain 翻译服务",
    version="1.0",
    description="用 LangServe 暴露的 LCEL 链",
)

# 1) 构造任意 Runnable（这里是 prompt | model | parser 三件套）
prompt = ChatPromptTemplate.from_template("把下面这句话翻译成{language}：\n{text}")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | model | StrOutputParser()

# 2) 一行挂载：自动生成 /translate/invoke /translate/stream
#    /translate/batch /translate/playground 以及 /translate/input_schema 等
add_routes(app, chain, path="/translate")

# 3) 也可以同时挂多条链：每条一个 path
add_routes(app, model, path="/raw-model")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# 启动：python server.py
# 浏览器打开 http://localhost:8000/translate/playground/ 即可交互调试
# 接口文档：http://localhost:8000/docs
# client.py —— 用 RemoteRunnable 把远端接口当本地 Runnable 用
from langserve import RemoteRunnable

# 指向 server 端的 path，RemoteRunnable 实现了完整 Runnable 接口
remote_chain = RemoteRunnable("http://localhost:8000/translate/")

# invoke：一次性返回
print(remote_chain.invoke({"language": "英语", "text": "今天天气很好"}))

# stream：逐块流式返回（走服务端 /stream 端点）
for chunk in remote_chain.stream({"language": "日语", "text": "我喜欢编程"}):
    print(chunk, end="", flush=True)

# batch：批量并发
print(remote_chain.batch([
    {"language": "法语", "text": "早上好"},
    {"language": "德语", "text": "晚上好"},
]))

# 关键价值：远端链对客户端完全透明，可以直接拼进本地 LCEL 管道：
# local_pipeline = some_local_runnable | remote_chain | another_runnable

# 也可以不用 RemoteRunnable，直接 HTTP 调用（请求体固定为 {"input": ...}）：
#   curl -X POST http://localhost:8000/translate/invoke \
#     -H 'Content-Type: application/json' \
#     -d '{"input": {"language": "英语", "text": "今天天气很好"}}'
✓推荐做法
  • 用 config={"callbacks": [...]} 运行期挂载 handler，让回调自动沿链路传播
  • 生产开 LangSmith 追踪（环境变量），本地用 set_debug 临时排查后立刻关掉
  • AgentExecutor 一律设 max_iterations + max_execution_time + handle_parsing_errors
  • LangServe 服务端把每条链给一个清晰 path，并锁定 langserve/fastapi/pydantic 版本
✗不推荐
  • 把 set_debug(True) 带到生产——日志会被原始 prompt/响应刷爆
  • 既要 with_structured_output 的严格对象、又要逐 token 流式，二者天然冲突
  • 升级后还从 langchain.chat_models 导入模型类——已迁移到 provider 集成包
  • 用 Agent 时不设步数/时间上限，放任它烧 token 直到超时
⚠常见误区
  • 回调钩子里抛异常会影响主链路——handler 内部要自己 try/except 兜住
  • token_usage 字段名因 provider 而异，跨模型统计要做兼容
  • RemoteRunnable 的 path 末尾斜杠、与服务端 add_routes 的 path 要对齐
  • LANGSMITH_ 新前缀与 LANGCHAIN_ 旧前缀混用可能导致追踪不生效

链路内部可观测（能拿到每步事件/耗时/token），Agent 有限步收敛，Runnable 通过 LangServe 暴露后 /invoke /stream /playground 均可用，且 RemoteRunnable 调用与本地行为一致。

没有验证的完成 = 没完成。能被度量的链路才能被改进——Callbacks 给你度量，LangServe 给你交付。

— 本章小结