分类信息网站的建设维护视频网站开发视频

一、前言

with_listeners() 方法的底层实现机制包含两个关键点：

​​1. 回调注入逻辑​​
通过以下方式将回调函数整合到配置中：

config["callbacks"] = [on_start, on_end, on_error]

（实际实现会合并而非覆盖已有回调）

​​2. 事件监听原理​​
本质上是通过内置的 CallbackHandler 类实现生命周期事件的监听处理，其工作流程为：

• ✅ 注册事件钩子
• ✅ 标准化事件格式
• ✅ 分发回调执行

二、运行流程：

三、资料推荐

• 🔗 官方文档参考
• 💡大模型中转API推荐
• ✨中转使用教程

四、核心源码

# langchain_core/runnables/base.py -> Runnable
def with_listeners(self,*,on_start: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,on_end: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,on_error: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,
) -> Runnable[Input, Output]:from langchain_core.tracers.root_listeners import RootListenersTracerreturn RunnableBinding(bound=self,config_factories=[lambda config: {"callbacks": [RootListenersTracer(config=config,on_start=on_start,on_end=on_end,on_error=on_error,)],}],)