with_listeners 运行流程与解析

一、前言

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 RootListenersTracer

    return RunnableBinding(
        bound=self,
        config_factories=[
            lambda config: {
                "callbacks": [
                    RootListenersTracer(
                        config=config,
                        on_start=on_start,
                        on_end=on_end,
                        on_error=on_error,
                    )
                ],
            }
        ],
    )