这是LangChain进阶教程的最后一篇，Let's get it!!!01 Callback介绍在LangChain中，Callback 是一种非常重要的机制，它允许用户监听和处理在执行链式任务 (Chain) 过程中的各种事件。这包括但不限于开始执行、结束执行、异常处理等。Callback 可以帮助开发者追踪执行过程、调试问题、监控性能指标或者进行日志记录等。基本概念Callback 是一个可以接收事件通知的对象。Callback 可以在不同的执行阶段被触发，例如在语言模型 (LLM) 开始运行、结束运行、生成文本等时刻。Callback 可以是同步或异步的。核心组件BaseCallbackHandler: 所有回调处理器的基础类。CallbackManager: 负责管理多个 CallbackHandler 实例，控制它们的触发顺序和方式。（已标记废弃，不建议使用这个）常用回调处理器StreamingStdOutCallbackHandler: 将流式输出直接打印到标准输出。FileCallbackHandler: 将回调信息写入文件。CustomCallbackHandler: 用户可以自定义自己的回调处理器。如何使用 Callbacks通常通过 CallbackManager 来添加和管理 CallbackHandler 实例。（已标记废弃，不建议使用这个）当创建 Chain 或者其他组件时，可以通过 callbacks 参数传入一个 CallbackManager 实例。02 基础使用这里以 StdOutCallbackHandler 为例：from langchain_core.callbacks import StdOutCallbackHandlerfrom langchain.chains import LLMChainfrom langchain_core.prompts import PromptTemplatehandler = StdOutCallbackHandler()prompt = PromptTemplate.from_template("1 + {number} = ")chain = LLMChain(llm=llm_model, prompt=prompt, callbacks=[handler])chain.invoke({"number":2})"""> Entering new LLMChain chain...Prompt after formatting:1 + 2 = > Finished chain."""上面代码中，我们使用的是 invoke 来执行 chain 的调用，如果我们使用 ainvoke，则建议使用 AsyncCallbackHandler 来实现回调。Custom callback handlers要创建自定义回调处理程序，我们需要确定我们希望回调处理程序处理的事件，以及我们希望回调处理程序在触发事件时执行的操作。然后，我们需要做的就是将回调处理程序附加到对象上，作为构造器回调或请求回调。示例代码如下：import asynciofrom typing import Any, Dict, List, Optionalfrom langchain.callbacks.base import AsyncCallbackHandler, BaseCallbackHandlerfrom langchain_core.messages import HumanMessagefrom langchain_core.outputs import LLMResultclass MyCustomSyncHandler(BaseCallbackHandler): def on_llm_new_token(self, token: str, kwargs) -> None: print(f"Sync handler being called in a thread_pool_executor: token: {token}")class MyCustomAsyncHandler(AsyncCallbackHandler): """Async callback handler that can be used to handle callbacks from langchain.""" async def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], kwargs: Any ) -> None: """Run when chain starts running.""" print("zzzz....") await asyncio.sleep(0.3) class_name = serialized["name"] print("Hi! I just woke up. Your llm is starting") async def on_llm_end(self, response: LLMResult, kwargs: Any) -> None: """Run when chain ends running.""" print("zzzz....") await asyncio.sleep(0.3) print("Hi! I just woke up. Your llm is ending")chat_model = ChatTongyi(dashscope_api_key='sk-da184735f2454123ab213cea8d39e9ce',streaming=True,callbacks=[MyCustomSyncHandler(), MyCustomAsyncHandler()],)await chat_model.agenerate([[HumanMessage(content="Tell me a joke")]])File loggingfrom langchain_core.callbacks import FileCallbackHandler, StdOutCallbackHandlerfrom langchain_core.prompts import PromptTemplatefrom loguru import loggerlogfile = "output.log"logger.add(logfile, colorize=True, enqueue=True)handler_1 = FileCallbackHandler(logfile)handler_2 = StdOutCallbackHandler()prompt = PromptTemplate.from_template("1 + {number} = ")chain = prompt | llm_modelresponse = chain.invoke({"number": 2}, {"callbacks": [handler_1, handler_2]})logger.info(response)我们的日志文件打开后是乱码的，需要用第三方工具做一次转换转换代码如下：from ansi2html import Ansi2HTMLConverterfrom IPython.display import HTML, displaywith open("output.log", "r") as f: content = f.read()conv = Ansi2HTMLConverter()html = conv.convert(content, full=True)display(HTML(html))用 display 查看，最终效果如下（线上可用）：Tags我们可以通过向 call()/run()/apply() 方法传递 tags 参数来向回调添加标签。这对于过滤日志很有用，例如，如果我们想记录对特定 LLMChain 发出的所有请求，可以添加一个标签，然后按该标签过滤您的日志。示例代码如下：from langchain.callbacks.base import AsyncCallbackHandler, BaseCallbackHandlerfrom typing import Any, Dict, Listfrom langchain.chains.llm import LLMChainfrom langchain_community.llms.tongyi import Tongyifrom langchain_core.callbacks import BaseCallbackHandlerfrom langchain_core.prompts import PromptTemplateclass MyCustomHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], kwargs: Any ) -> Any: print(f"自定义回调，on_llm_start") def on_llm_end( self, serialized: Dict[str, Any], kwargs: Any ) -> Any: print(f"自定义回调，on_llm_end") def on_llm_new_token( self, token: str, kwargs: Any ) -> Any: print(f"自定义回调，on_llm_new_token") def on_chain_start( self, serialized: Dict[str, Any], inputs: Dict[str, Any], kwargs: Any, ) -> Any: print(f"自定义回调，on_chain_start, {kwargs}") def on_chain_end( self, outputs: Dict[str, Any], kwargs: Any, ) -> Any: print(f"自定义回调，on_chain_end, {kwargs}") handler = MyCustomHandler()prompt = PromptTemplate.from_template("1 + {number} = ")chain = LLMChain(llm=llm_model, prompt=prompt,verbose=False, callbacks=[handler],tags=['custom-------'])chain.invoke({"number":2})"""自定义回调，on_chain_start, {'run_id': UUID('91086ade-0121-4565-946e-8fadbb870f27'), 'parent_run_id': None, 'tags': ['custom-------'], 'metadata': {}, 'name': 'LLMChain'}自定义回调，on_chain_end, {'run_id': UUID('91086ade-0121-4565-946e-8fadbb870f27'), 'parent_run_id': None, 'tags': ['custom-------']}"""Token counting如果我们正在使用openai的相关模型，我们就能使用这个令牌计数的功能。示例代码如下：import asynciofrom langchain_community.callbacks import get_openai_callbackfrom langchain_openai import OpenAIllm = OpenAI(temperature=0)with get_openai_callback() as cb: llm.invoke("What is the square root of 4?")total_tokens = cb.total_tokensassert total_tokens > 0with get_openai_callback() as cb: llm.invoke("What is the square root of 4?") llm.invoke("What is the square root of 4?")assert cb.total_tokens == total_tokens 2with get_openai_callback() as cb: await asyncio.gather( [llm.agenerate(["What is the square root of 4?"]) for _ in range(3)] )assert cb.total_tokens == total_tokens * 3task = asyncio.create_task(llm.agenerate(["What is the square root of 4?"]))with get_openai_callback() as cb: await llm.agenerate(["What is the square root of 4?"])await taskassert cb.total_tokens == total_tokens03 总结与回顾从2024年3月3日第一篇 LangChain 相关文章的发布，到现在 LangChain 最后一篇文章的发表，期间跨度已近半年，总共输出22篇文章。从初级使用教程（7篇），到进阶理解教程（15篇），我们从知道 LangChain 到使用 LangChain 再到理解 LangChain，一步一步循序渐进的走来。最初写这些文章的目的是为了“武装自己”，但是在看到各大平台上的读者留言后，渐渐的想把这些文章写好写透，因为我能从这些留言上看到“以前自己的影子”。这生活已经很难了，难能可贵的是您能在百忙之中抽空阅读这些文章。如果能给到您一点小小的帮助，也是我非常喜闻乐见的?WX搜索公众号关注以上内容依据官方文档编写，官方地址：python.langchain.com/docs/module…这真的是“进阶系列”的最后一篇文章了哦，我们下个系列再见。Love & Peace~
