说明下面是参考 LangChain 源码提炼的最简可运行模拟实现不是直接复制官方源码用于看懂整个回调流转全貌BaseCallbackHandler基类定义全部生命周期回调钩子模拟简易版LangSmithTracer模仿真实LangChainTracer模拟 LLM、Tool、Chain、LangGraph‑like 节点在各个业务节点内部主动触发回调看清楚是谁、在什么时机调用on_xxx_start / on_xxx_end / on_xxx_errorfromtypingimportAny,Optional,List,Dictimporttimefromdataclassesimportdataclass# # 1. 回调处理器基类 BaseCallbackHandler# 所有回调钩子定义在这里和 langchain 接口对齐# dataclassclassBaseCallbackHandler:LangChain 回调处理器基类全部生命周期事件钩子defon_chain_start(self,serialized:Dict[str,Any],inputs:Dict[str,Any],**kwargs:Any)-None:chain/图节点开始执行...defon_chain_end(self,outputs:Dict[str,Any],**kwargs:Any)-None:chain/图节点正常结束...defon_chain_error(self,error:BaseException,**kwargs:Any)-None:chain/图节点异常崩溃...defon_llm_start(self,serialized:Dict[str,Any],prompts:List[str],**kwargs:Any)-None:开始调用 LLM...defon_llm_end(self,response:Any,**kwargs:Any)-None:LLM 返回成功结果...defon_llm_error(self,error:BaseException,**kwargs:Any)-None:LLM 调用报错...defon_tool_start(self,serialized:Dict[str,Any],input_str:str,**kwargs:Any)-None:工具调用开始...defon_tool_end(self,output:str,**kwargs:Any)-None:工具调用成功结束...defon_tool_error(self,error:BaseException,**kwargs:Any)-None:工具调用报错...# # 2. 模拟 LangSmithTracer实现上面回调采集trace信息# 真实 LangChainTracer 内部会构建 Run 对象异步上报 LangSmith# classMockLangSmithTracer(BaseCallbackHandler):模拟 LangSmith 的追踪回调处理器defon_chain_start(self,serialized,inputs,**kwargs):print(f[LangSmithTrace] CHAIN_START | name{serialized.get(name)}, inputs{inputs})defon_chain_end(self,outputs,**kwargs):print(f[LangSmithTrace] CHAIN_END | outputs{outputs})defon_chain_error(self,error,**kwargs):print(f[LangSmithTrace] CHAIN_ERROR | error{str(error)})defon_llm_start(self,serialized,prompts,**kwargs):print(f[LangSmithTrace] LLM_START | prompts{prompts})defon_llm_end(self,response,**kwargs):print(f[LangSmithTrace] LLM_END | response{response})defon_llm_error(self,error,**kwargs):print(f[LangSmithTrace] LLM_ERROR | error{str(error)})defon_tool_start(self,serialized,input_str,**kwargs):print(f[LangSmithTrace] TOOL_START | tool{serialized.get(name)}, input{input_str})defon_tool_end(self,output,**kwargs):print(f[LangSmithTrace] TOOL_END | output{output})defon_tool_error(self,error,**kwargs):print(f[LangSmithTrace] TOOL_ERROR | error{str(error)})# # 3. 模拟业务组件LLM、Tool、Chain、GraphNode# ✨重点组件内部持有 callbacks业务执行前后主动调用回调方法# classMockLLM:def__init__(self,callbacks:Optional[List[BaseCallbackHandler]]None):self.callbackscallbacksor[]definvoke(self,prompt:str):# -------- 触发 on_llm_start 回调 --------forcbinself.callbacks:cb.on_llm_start(serialized{name:MockLLM},prompts[prompt])try:time.sleep(0.2)respfllm_result:{prompt}# -------- 触发 on_llm_end 回调 --------forcbinself.callbacks:cb.on_llm_end(responseresp)returnrespexceptExceptionase:forcbinself.callbacks:cb.on_llm_error(errore)raiseeclassMockTool:def__init__(self,callbacks:Optional[List[BaseCallbackHandler]]None):self.callbackscallbacksor[]defrun(self,query:str):# -------- 触发 on_tool_start 回调 --------forcbinself.callbacks:cb.on_tool_start(serialized{name:SearchTool},input_strquery)try:time.sleep(0.2)outftool_search_result({query})# -------- 触发 on_tool_end 回调 --------forcbinself.callbacks:cb.on_tool_end(outputout)returnoutexceptExceptionase:forcbinself.callbacks:cb.on_tool_error(errore)raiseeclassMockChain:模拟普通 Chain / LangGraph Nodedef__init__(self,llm:MockLLM,tool:MockTool,callbacks:Optional[List[BaseCallbackHandler]]None):self.llmllm self.tooltool self.callbackscallbacksor[]definvoke(self,user_input:str):# -------- chain/node 开始on_chain_start --------forcbinself.callbacks:cb.on_chain_start(serialized{name:MockAgentChain},inputs{user_input:user_input})try:# chain内部逻辑先调用tool再调用llmtool_resultself.tool.run(user_input)llm_resultself.llm.invoke(fprocess:{tool_result})outputs{answer:llm_result}# -------- chain/node 正常结束on_chain_end --------forcbinself.callbacks:cb.on_chain_end(outputsoutputs)returnoutputsexceptExceptionase:# -------- chain/node 异常on_chain_error --------forcbinself.callbacks:cb.on_chain_error(errore)raisee# # 4. 运行演示# if__name____main__:tracerMockLangSmithTracer()llmMockLLM(callbacks[tracer])toolMockTool(callbacks[tracer])chainMockChain(llmllm,tooltool,callbacks[tracer])print( 执行 Agent Chain )reschain.invoke(帮我查天气)print(f\n最终返回:{res})输出示例 执行 Agent Chain [LangSmithTrace] CHAIN_START | nameMockAgentChain, inputs{user_input: 帮我查天气} [LangSmithTrace] TOOL_START | toolSearchTool, input帮我查天气 [LangSmithTrace] TOOL_END | outputtool_search_result(帮我查天气) [LangSmithTrace] LLM_START | prompts[process:tool_search_result(帮我查天气)] [LangSmithTrace] LLM_END | responsellm_result: process:tool_search_result(帮我查天气) [LangSmithTrace] CHAIN_END | outputs{answer: llm_result: process:tool_search_result(帮我查天气)} 最终返回: {answer: llm_result: process:tool_search_result(帮我查天气)}和真实 LangGraph / LangSmith 的对应关系MockLangSmithTracer对应真实langchain.tracers.langchain.LangChainTracer真实内部回调被触发时组装Run对象丢进后台上报队列发给 LangSmith API。MockChain.invoke对应 LangGraph 的图执行逻辑每一个节点执行前后框架内部遍历注册的callbacks挨个调用对应回调钩子。callbacks[tracer]就是把回调处理器实例传给组件组件内部自己决定什么时候调用回调方法。关键问题用户点停止生成async task cancel如果在MockLLM.invoke执行过程中协程被task.cancel()抛出CancelledErroron_llm_end不会执行on_llm_error也不会执行cancel 是特殊异常业务的try‑except捕获不到→ 回调没有被触发Run 没有end事件LangSmith trace 残缺。这就是停止生成 trace 断的根源。真实源码入口你可以去看BaseCallbackHandlerlangchain_core/callbacks/base.pyLangChainTracerlangchain/tracers/langchain.pyLangGraph 在langgraph/pregel/_loop.pypregel 循环内部每个step 触发回调。