回调与监控，用Callbacks追踪Agent的每一步执行过程

上一篇把LCEL表达式语言讲完了，管道符一路接到底，链跑得也挺顺。但有个事一直憋着没说，链跑起来之后里面到底发生了什么，哪一步调了模型，哪一步调了工具，token花在哪，耗时多少，出问题怎么定位到具体那一段。靠肉眼盯控制台打印是盯不出来的，得靠回调机制。这篇就聊回调与监控，把Agent的执行过程扒开看个明白。

为什么需要回调

Agent跑起来就像个黑盒。你invoke一下，等几秒，结果出来了，中间发生了什么一无所知。模型被调了几次，工具被调了几次，每次的输入输出是什么，全藏在底层。

调试的时候这事特别烦。我之前做一个带工具调用的Agent，线上偶发返回空结果，本地又复现不了。日志里只看到最终输出是空的，至于哪一步出了问题，是模型没生成工具调用，还是工具执行报错被吞了，根本看不出来。后来加了回调，把每一步的入参出参都打出来，才发现是工具偶尔超时，异常没被上层接住，链静默吞掉继续跑，最后就吐了个空。

回调机制就是干这个的。它在链的每个关键节点埋了钩子，节点开始时触发对应的回调方法，结束时再触发一次。你写一个处理器把这些钩子接住，就能拿到整条链的执行轨迹。有了这东西，Agent对你来说就不再是个黑盒。

BaseHandler，自定义回调处理器

LangChain的回调机制核心是BaseCallbackHandler，在langchain_core.callbacks模块下。继承它，重写需要的方法，就能接管各个节点的通知。

fromlangchain_core.callbacksimportBaseCallbackHandlerclassTraceHandler(BaseCallbackHandler):defon_llm_start(self,serialized,prompts,**kwargs):print("[LLM开始] 准备调用模型")defon_llm_end(self,response,**kwargs):print("[LLM结束] 模型返回了")defon_chain_start(self,serialized,inputs,**kwargs):print("[Chain开始] 链开始执行")defon_chain_end(self,outputs,**kwargs):print("[Chain结束] 链执行完毕")defon_tool_start(self,serialized,input_str,**kwargs):print(f"[Tool开始] 工具被调用，输入是{input_str}")defon_tool_end(self,output,**kwargs):print(f"[Tool结束] 工具返回{output}")

写好处理器之后，塞进invoke的config参数就行。

result=chain.invoke({"question":"LCEL是什么"},config={"callbacks":[TraceHandler()]},)

也可以构造链时就绑定回调，这样每次调用自动带上。

chain=chain.with_config({"callbacks":[TraceHandler()]})

几个核心回调方法

BaseCallbackHandler里方法不少，常用的就那几个，对应链的不同节点。

on_llm_start和on_llm_end，模型调用的开始和结束。on_llm_start能拿到序列化的模型信息和提示词，on_llm_end能拿到模型返回的完整响应。统计模型调用次数就看这俩。

on_chain_start和on_chain_end，链的开始和结束。一条LCEL链里可能有多个子链，每一段都会触发一次。嵌套调用时这俩层层触发，靠run_id能理清父子关系。

on_tool_start和on_tool_end，工具调用的开始和结束。Agent调搜索、调数据库、调自定义工具，都走这俩。排查工具相关问题盯这里最直接。

每个方法都有对应的error版本，on_llm_error、on_chain_error、on_tool_error，节点抛异常时触发。生产环境里这几个特别重要，能帮你把错误逮个正着。

还有一个on_llm_new_token，流式输出时每生成一个token触发一次。下一篇聊流式输出会重点用到它，这里先按下不表。

打印执行日志

把处理器用起来，跑一个带工具的Agent看看效果。

fromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttoolfromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.agentsimportcreate_tool_calling_agent,AgentExecutor@tooldefsearch(query:str)->str:"""搜索网络内容"""returnf"关于{query}的搜索结果"model=ChatOpenAI(model="gpt-4o-mini")prompt=ChatPromptTemplate.from_messages([("system","你是一个搜索助手"),("human","{input}"),MessagesPlaceholder("agent_scratchpad"),])agent=create_tool_calling_agent(model,[search],prompt)executor=AgentExecutor(agent=agent,tools=[search],verbose=False)executor.invoke({"input":"LangChain是什么"},config={"callbacks":[TraceHandler()]},)

跑起来你会看到一串日志，LLM开始、Tool开始、Tool结束、LLM开始、LLM结束，按顺序排下来。模型先决定调工具，工具执行完返回结果，模型再拿到结果生成最终回答。整个执行流程清清楚楚，哪一步慢、哪一步出错，瞄一眼日志就知道。

统计Token消耗

光看流程还不够，token花在哪里也得心里有谱。模型调用的response里带着token用量信息，在on_llm_end里捞出来累加就行。

classTokenCounter(BaseCallbackHandler):def__init__(self):self.prompt_tokens=0self.completion_tokens=0self.call_count=0defon_llm_end(self,response,**kwargs):usage=(response.llm_outputor{}).get("token_usage",{})self.prompt_tokens+=usage.get("prompt_tokens",0)self.completion_tokens+=usage.get("completion_tokens",0)self.call_count+=1defreport(self):total=self.prompt_tokens+self.completion_tokensprint(f"模型调用{self.call_count}次，总token{total}")print(f"输入{self.prompt_tokens}，输出{self.completion_tokens}")

用法和前面一样，塞进callbacks里。跑完链调用report方法，一次执行花了多少token清清楚楚。这个数对成本控制特别要紧，链一长token烧得快，哪一步最费钱定位到了，优化才有方向。有些模型不返回token_usage，这种情况下计数器读到的是0，接入前先确认你用的模型支持。

集成LangSmith

自己写回调够用，但要做认真的监控，还是得上专业工具。LangSmith是LangChain官方的可观测平台，和LangChain原生集成，配置几行就能用。

最简单的方式是设环境变量。

importos os.environ["LANGCHAIN_TRACING_V2"]="true"os.environ["LANGCHAIN_API_KEY"]="ls__你的key"os.environ["LANGCHAIN_PROJECT"]="my-agent"

设完之后，所有LangChain调用会自动上报到LangSmith，业务代码一行不用改。打开网页端，每次执行都是一条完整的trace，树状结构展开，每一步的输入输出、耗时、token、错误信息全都有。嵌套调用画成调用树，哪层调了哪层一目了然。

我第一次接入LangSmith的时候挺惊讶，之前自己写回调只能看个大概，LangSmith直接把调用关系可视化，线上问题排查从大海捞针变成顺藤摸瓜。

生产环境监控建议

回调和LangSmith用熟了，生产环境还有几点要叮嘱。

第一，别在生产环境开verbose打印。verbose=True会把所有中间步骤打到控制台，调试方便，生产环境里日志量爆炸，还可能泄露用户输入。用LangSmith这种结构化追踪代替裸打印。

第二，回调里别做重活。回调跑在主流程线程里，里面写文件、调网络会拖慢整条链。需要持久化的话扔到队列里异步处理。

第三，错误回调一定要接。on_tool_error特别容易漏，工具抛异常默认会往上冒，但你可能想在异常时返回默认值让流程继续。在on_tool_error里记日志，配合AgentExecutor的handle_tool_error参数做兜底。

第四，监控要带上下游标识。多用户并发时，每次调用带一个user_id或session_id，LangSmith里按这个过滤，定位某个用户的问题才快。

这篇聊了回调与监控，从为什么需要回调，到BaseCallbackHandler怎么用，几个核心回调方法的职责，打印执行日志、统计token、集成LangSmith，最后给了几条生产环境建议。Agent跑起来不再是黑盒，每一步都看得见。

下一篇聊流式输出，看看怎么让模型一个字一个字往外吐，回调里的on_llm_new_token正好派上用场。