听众基类 BaseCallbackHandler（libs/core/langchain_core/callbacks/base.py:496）——注意它是靠一堆 Mixin 拼出来的：

# libs/core/langchain_core/callbacks/base.py:496
class BaseCallbackHandler(
    LLMManagerMixin,        # 提供 on_llm_new_token / on_llm_end / on_llm_error
    ChainManagerMixin,      # 提供 on_chain_end / on_chain_error / on_agent_action...
    ToolManagerMixin,       # 提供 on_tool_end / on_tool_error
    RetrieverManagerMixin,  # 提供 on_retriever_end / on_retriever_error
    CallbackManagerMixin,   # 提供各种 on_*_start
    RunManagerMixin,        # 提供 on_text / on_retry / on_custom_event
):
    """Base callback handler."""

    raise_error: bool = False   # 听众自己抛异常时，要不要拖垮主流程？默认不要
    run_inline: bool = False    # 是否在主线程同步执行（追踪器需要顺序保证时置 True）

    @property
    def ignore_llm(self) -> bool:
        """Whether to ignore LLM callbacks."""
        return False            # 子类可声明"我不关心 LLM 事件"，广播时会跳过它

广播方 CallbackManager（libs/core/langchain_core/callbacks/manager.py:1377）。看最常走的 on_chat_model_start（manager.py:1431）：

# libs/core/langchain_core/callbacks/manager.py:1431
def on_chat_model_start(self, serialized, messages, run_id=None, **kwargs
) -> list[CallbackManagerForLLMRun]:
    managers = []
    for message_list in messages:                 # 每批消息一个独立的 run
        run_id_ = run_id or uuid7()               # ① 发一个 run_id（追踪树的节点编号）
        handle_event(                             # ② ★向所有 handler 广播"模型开始了"
            self.handlers, "on_chat_model_start", "ignore_chat_model",
            serialized, [message_list],
            run_id=run_id_, parent_run_id=self.parent_run_id,
            tags=self.tags, metadata=self.metadata, **kwargs,
        )
        managers.append(CallbackManagerForLLMRun( # ③ 返回"本次运行专属"的子管理器
            run_id=run_id_, handlers=self.handlers,
            inheritable_handlers=self.inheritable_handlers,
            parent_run_id=self.parent_run_id, ...))
    return managers

广播的底层是个通用函数 handle_event（manager.py:285）：

# libs/core/langchain_core/callbacks/manager.py:285
def handle_event(handlers, event_name, ignore_condition_name, *args, **kwargs):
    for handler in handlers:
        try:
            if ignore_condition_name is None or not getattr(
                handler, ignore_condition_name          # ① 尊重 ignore_chat_model 等开关
            ):
                event = getattr(handler, event_name)(*args, **kwargs)  # ② 反射调钩子
                if asyncio.iscoroutine(event):
                    coros.append(event)                 #    异步钩子攒起来统一跑
        except NotImplementedError as e:
            if event_name == "on_chat_model_start":     # ③ ★老 handler 不认识 chat 事件？
                handle_event([handler], "on_llm_start", # 降级成 on_llm_start 再喂一次
                             "ignore_llm", args[0], message_strings, ...)

你可能好奇：我只在 config 里塞了个 handler，谁把它装进 manager 的？每个组件执行前都会调 CallbackManager.configure（libs/core/langchain_core/callbacks/manager.py:1683），它委托给模块级的 _configure（manager.py:2390）。核心逻辑三步：

先看一个反直觉的事实——Runnable.stream 的默认实现（libs/core/langchain_core/runnables/base.py:1182）：

# libs/core/langchain_core/runnables/base.py:1182
def stream(self, input, config=None, **kwargs) -> Iterator[Output]:
    """Default implementation of stream, which calls invoke.
    Subclasses must override this method if they support streaming output."""
    yield self.invoke(input, config, **kwargs)   # ★默认：憋完一整个结果，一次 yield

stream 只能拿到最终输出的碎片；想要"检索完成了、模型开始了"这类中间事件，用 astream_events（libs/core/langchain_core/runnables/base.py:1356）。它的 v2 实现在 tracers/event_stream.py:1008——原理妙极了，就是"自己给自己装个听众"：

# libs/core/langchain_core/tracers/event_stream.py:1008
async def _astream_events_implementation_v2(runnable, value, config=None, ...):
    event_streamer = _AstreamEventsCallbackHandler(   # ① ★造一个特殊听众（tracers/event_stream.py:101）
        include_names=include_names, include_types=include_types, ...)

    config = ensure_config(config)
    callbacks = config.get("callbacks")
    if callbacks is None:
        config["callbacks"] = [event_streamer]        # ② 把这个听众塞进 config
    elif isinstance(callbacks, list):
        config["callbacks"] = [*callbacks, event_streamer]

    async def consume_astream() -> None:
        try:      # ③ 后台任务：正常跑 astream，输出也 tap 进事件流
            async with aclosing(runnable.astream(value, config, **kwargs)) as stream:
                async for _ in event_streamer.tap_output_aiter(run_id, stream):
                    pass                              # 内容都被听众截走了
        finally:
            await event_streamer.send_stream.aclose()

    task = asyncio.create_task(consume_astream())     # ④ 链在后台跑
    async for event in event_streamer:                # ⑤ ★前台：从队列里逐条吐事件
        ...
        yield event