文章目录【63.PythonAI】LangChain Callbacks在AI流水线里插桩监控一切导入语1 ~ Callbacks 的运行机制1.1 什么是 Callback1.2 回调的生命周期1.3 核心回调事件一览2 ~ 案例一Token 消耗实时统计器2.1 为什么要统计 Token2.2 实现自定义 Callback3 ~ 案例二敏感词拦截过滤器3.1 场景需求3.2 实现敏感词拦截3.3 单层拦截 vs 双层拦截4 ~ 案例三全链路耗时分析器4.1 需求4.2 实现耗时追踪4.3 解读耗时报告5 ~ 多个 Callback 的组合5.1 场景5.2 Callback 的执行顺序思考 总结结尾【63.PythonAI】LangChain Callbacks在AI流水线里插桩监控一切文章简介本文系统讲解LangChain的Callbacks回调机制——这是深入观察和控制AI流水线行为的核心工具。文章从BaseCallbackHandler基类入手逐步拆解on_llm_start/on_llm_end/on_tool_start等生命周期回调的执行时序并通过三个实战案例Token消耗实时统计器、敏感词拦截过滤器、全链路耗时分析器展示Custom Callback的编写模式。配有Mermaid时序图展示一条Chain中回调的触发顺序适合正在构建生产级LLM应用、需要对每个环节进行可观测性监控的开发者。 个人主页源码骑士❄专栏传送门《Android开发基础》《python基础课程》⭐️热衷从源码视角拆解技术底层原理将复杂架构讲得通俗易懂 源码骑士的简介5年Android Framework系统开发经验曾主导多项系统级性能优化专项技术栈覆盖Android系统全链路Binder/Handler/AMS/WMS/启动流程及Java后端全家桶Spring MyBatis Redis Oracle累计产出原创技术文章100篇文章以流程图为特色被读者评价为看一篇胜过啃一周源码导入语你的LLM应用上线了用户反馈响应太慢。但你不知道是慢在LLM调用、还是慢在检索、还是慢在工具调用——因为你写的Chain是一个黑盒一步进去一步出来中间发生了什么完全看不到。这就是 Callbacks 要解决的问题在AI流水线的每个关键节点插桩打点观察每一步的输入、输出、耗时和Token消耗。看完这篇文章你写的每一个Chain都能做到全链路可观测。1 ~ Callbacks 的运行机制1.1 什么是 CallbackCallback回调是一个观察者——它不参与业务逻辑但会在特定事件发生时被自动调用。在 LangChain 中每当 LLM 开始调用、Chain 开始执行、Tool 被触发时框架都会触发对应的回调事件。1.2 回调的生命周期CallbackToolLLMChainUserCallbackToolLLMChainUserinvoke()on_chain_start()调用大模型on_llm_start()返回结果on_llm_end()调用工具on_tool_start()工具返回on_tool_end()最终结果on_chain_end()1.3 核心回调事件一览事件触发时机关键参数on_llm_startLLM开始推理prompts, run_idon_llm_endLLM推理完成response含token用量on_llm_errorLLM调用出错erroron_chain_startChain开始执行inputs, run_idon_chain_endChain执行完成outputson_tool_startTool开始执行tool_input, run_idon_tool_endTool执行完成tool_outputon_retriever_start检索器开始检索queryon_retriever_end检索完成documents2 ~ 案例一Token 消耗实时统计器2.1 为什么要统计 Token调用 GPT-4 的 API 时每 1000 token 都是真金白银。但 LangChain 默认不会在任何地方展示这轮对话到底烧了多少 Token。2.2 实现自定义 Callbackfromlangchain.callbacks.baseimportBaseCallbackHandlerfromcollectionsimportdefaultdictclassTokenCounterCallback(BaseCallbackHandler):实时统计每个模型的Token消耗def__init__(self):self.token_usagedefaultdict(lambda:{prompt:0,completion:0})self.total_cost0.0# GPT-4 价格$ / 1K tokensself.price_map{gpt-4:(0.03,0.06),# (prompt, completion)gpt-3.5-turbo:(0.0015,0.002),}defon_llm_end(self,response,**kwargs):LLM调用结束时从响应中提取token统计model_nameresponse.llm_output.get(model_name,unknown)# 提取token使用量token_usageresponse.llm_output.get(token_usage,{})prompt_tokenstoken_usage.get(prompt_tokens,0)completion_tokenstoken_usage.get(completion_tokens,0)# 累加self.token_usage[model_name][prompt]prompt_tokens self.token_usage[model_name][completion]completion_tokens# 计算费用ifmodel_nameinself.price_map:p_price,c_priceself.price_map[model_name]cost(prompt_tokens/1000)*p_price(completion_tokens/1000)*c_price self.total_costcostprint(f [Token]{model_name}: {prompt_tokens}prompt, {completion_tokens}completion)defreport(self):打印统计报告print(\n Token 消耗报告 )formodel,usageinself.token_usage.items():print(f 模型:{model})print(f 输入Token:{usage[prompt]:,})print(f 输出Token:{usage[completion]:,})print(f 合计:{usage[prompt]usage[completion]:,})print(f\n 预估总费用: ${self.total_cost:.4f})print(\n)# 使用counterTokenCounterCallback()fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4,callbacks[counter])llm.invoke(用100字解释什么是机器学习)counter.report()3 ~ 案例二敏感词拦截过滤器3.1 场景需求你的AI客服应用绝对不能输出竞品名称、政治敏感词、脏话。与其在每个地方都加一层过滤不如在 Callback 里统一拦截。3.2 实现敏感词拦截classSensitiveWordFilter(BaseCallbackHandler):敏感词拦截——在LLM返回结果时检查并拦截SENSITIVE_WORDS[敏感词A,敏感词B,竞品名X,竞品名Y]def__init__(self,fallback_message抱歉我无法回答这个问题。):self.fallback_messagefallback_message self.blocked_count0defon_llm_end(self,response,**kwargs):在LLM返回后、交给用户前检查输出内容generationsresponse.generationsforgen_listingenerations:forgeningen_list:textgen.textforwordinself.SENSITIVE_WORDS:ifwordintext:self.blocked_count1# 替换为安全回复gen.textself.fallback_messageprint(f [拦截] 检测到敏感词:{word})breakdefon_llm_start(self,prompts,**kwargs):也可以拦截用户输入中的敏感词pass# 根据业务需求实现# 使用filter_callbackSensitiveWordFilter()llmChatOpenAI(callbacks[filter_callback])3.3 单层拦截 vs 双层拦截含敏感词安全含敏感词安全用户输入输入过滤检查敏感词直接拒绝LLM处理输出过滤检查敏感词替换为安全回复返回给用户双层拦截输入输出是生产环境的标准做法防止漏网之鱼从LLM生成的结果中溜出去。4 ~ 案例三全链路耗时分析器4.1 需求你的Chain包含检索→LLM推理→工具调用→LLM二次推理四个环节。用户说慢你无从下手——因为不知道哪个环节耗时最长。4.2 实现耗时追踪importtimeclassLatencyTracker(BaseCallbackHandler):跟踪每个环节的耗时def__init__(self):self.start_times{}self.latenciesdefaultdict(list)def_get_time(self):returntime.perf_counter()defon_chain_start(self,chain,inputs,run_id,**kwargs):self.start_times[run_id]self._get_time()defon_chain_end(self,outputs,run_id,**kwargs):ifrun_idinself.start_times:elapsedself._get_time()-self.start_times.pop(run_id)chain_namechainself.latencies[chain_name].append(elapsed)print(f⏱ [Chain] 耗时:{elapsed:.2f}s)defon_llm_start(self,prompts,run_id,**kwargs):self.start_times[run_id]self._get_time()defon_llm_end(self,response,run_id,**kwargs):ifrun_idinself.start_times:elapsedself._get_time()-self.start_times.pop(run_id)modelresponse.llm_output.get(model_name,llm)self.latencies[model].append(elapsed)print(f⏱ [LLM/{model}] 耗时:{elapsed:.2f}s)defon_tool_start(self,tool,input_str,run_id,**kwargs):self.start_times[run_id]self._get_time()defon_tool_end(self,output,run_id,**kwargs):ifrun_idinself.start_times:elapsedself._get_time()-self.start_times.pop(run_id)self.latencies[tool].append(elapsed)print(f⏱ [Tool] 耗时:{elapsed:.2f}s)defreport(self):生成耗时报告print(\n 耗时分析报告 )forname,timesinself.latencies.items():avgsum(times)/len(times)totalsum(times)print(f{name}:)print(f 调用次数:{len(times)})print(f 平均耗时:{avg:.2f}s)print(f 总耗时:{total:.2f}s)print(\n)# 使用trackerLatencyTracker()fromlangchain.chainsimportRetrievalQA qa_chainRetrievalQA.from_chain_type(llmllm,retrieverretriever,callbacks[tracker])qa_chain.invoke({query:什么是量子计算})tracker.report()4.3 解读耗时报告拿到报告后用二八法则——先解决耗时占比最高的环节。如果 80% 的时间花在 LLM 调用上你就该考虑换更快的模型或加缓存如果花在检索上就该优化向量数据库。5 ~ 多个 Callback 的组合5.1 场景你既想监控 Token、又想追踪耗时、还想拦截敏感词——三个Callback一起用。# 组合多个 Callbackfromlangchain.callbacksimportCallbackManager managerCallbackManager([TokenCounterCallback(),LatencyTracker(),SensitiveWordFilter()])llmChatOpenAI(callbacksmanager)# 所有 Callback 按顺序依次执行responsellm.invoke(什么是微服务)# 分别打印报告manager.handlers[0].report()# Token统计manager.handlers[2].report()# 耗时分析5.2 Callback 的执行顺序多个 Callback 按照注册顺序依次调用。如果中间的 Callback 抛异常后续的不会被执行——所以把过滤器放在最后把监控放在前面。思考 总结Callbacks 是 LangChain 可观测性的核心基础设施框架内部几乎所有关键操作都会触发回调事件。掌握了 Callback就掌握了对 Chain 内部行为的透视能力。三个经典案例覆盖 80% 的生产需求Token统计成本控制、敏感词拦截安全合规、耗时追踪性能优化——这三个 Case 实现了你的LLM应用就具备了基本的可观测性。组合使用比单独使用更强大用CallbackManager把多个 Callback 组合在一起一条 Chain 就能同时被多个观察者监控互不干扰。注意异步场景如果是AsyncCallbackHandler需要继承AsyncCallbackHandler并使用async def方法。不要在生产环境开 verboseTrueverbose 模式会把所有中间结果打印到控制台生产环境应该用 Custom Callback 按需记录并输出到日志文件或监控平台。Callback 就像给 AI 流水线装上了仪表盘——之前你是蒙眼开车之后每个环节的转速、油量、温度一目了然。结尾各位小伙伴本文的内容到这里就全部结束了源码骑士在这里再次感谢您的阅读源码骑士 — Android Framework 全栈开发关注跟博主一起从源码视角深耕底层原理见证每一次成长❤️点赞让优质内容被更多人看见让知识传递更有力量⭐收藏把核心知识点存好在需要时随时查、随时用评论分享你的经验或疑问评论区一起交流避坑一键四连不要忘记给博主一键四连哦️寄语技术之路难免有困惑但同行的人会让前进更有方向结语AI应用上线后最大的恐惧不是它答错了而是我不知道它为什么这样答。Callbacks 让你从黑盒调用变成全链路透视出了任何问题都能第一时间定位到具体环节。不要忘记给博主一键四连哦
)
