关于Python的回调函数，其实之前也有用过，本质上就是将一个函数传递给另一个函数，当另一个函数执行完成之后，就会调用传递过来的函数
比如下面就是一个简单的回调函数示例
|
def compute(x, y, callback): result = x + y callback(result)
def print_result(value): print(f”The result is: {value}”)
# 使用print_result作为回调 compute(3, 4, print_result) # 输出: The result is: 7 |
而在LangChain之中，其Callback可以允许开发者在不同阶段进行自定义操作
比如日志记录，监控 数据流处理
这个机制通过CallbackHandler进行实现。
我们可以通过集成BaseCallbackHandler来实现自己的回调处理器
在BaseCallbackHandler之中，提供多种方法，比如on_llm_start/on_chat, on_llm_error等。
LangChain提供了一些内置的处理，比如StdOutCallbackHandler，会将所有的事件记录输出到标准输出。
而想要将CallbackHandler在组件之中使用，则可以有两种方式使用
比如在创建组件的时候传入，或者在调用的时候传入
说了这些基础知识，我们看下集成的一个简单Demo
|
# 创建同步回调处理器 class MyFlowerShopSyncHandler(BaseCallbackHandler): def on_llm_new_token(self, token: str, **kwargs) -> None: print(f”获取花卉数据: token: {token}”)
# 创建异步回调处理器 class MyFlowerShopAsyncHandler(AsyncCallbackHandler):
async def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any ) -> None: print(“正在获取花卉数据…”) await asyncio.sleep(0.5) # 模拟异步操作 print(“花卉数据获取完毕。提供建议…”)
async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None: print(“整理花卉建议…”) await asyncio.sleep(0.5) # 模拟异步操作 print(“祝你今天愉快！”)
# 主要的异步函数 async def main(): flower_shop_chat = ChatOpenAI( max_tokens=100, streaming=True, callbacks=[MyFlowerShopSyncHandler(), MyFlowerShopAsyncHandler()], )
# 异步生成聊天回复 await flower_shop_chat.agenerate([[HumanMessage(content=”哪种花卉最适合生日？只简单说3种，不超过50字”)]])
# 运行主异步函数 asyncio.run(main()) |
这里我们分别使用两个类继承了同步和异步回调函数类
同步回调函数类中，MyFlowerShopSyncHandle中获取最新的鲜花数据
异步回调函数类中，MyFlowerShopAsyncHandler监听了请求的获取操作的开始和结束。
那么我们总结一下，在LangChain之中，回调函数是一个很重要的概念
可以供我们在一些特殊的节点，对LangChain框架进行一些自定义操作的执行。