今天的主菜，一行不删（runnables/base.py:3418-3451）：

# libs/core/langchain_core/runnables/base.py:3418（完整正文）
def invoke(self, input, config=None, **kwargs):
    # setup callbacks and context
    config = ensure_config(config)                                # ① 补齐工单
    callback_manager = get_callback_manager_for_config(config)    # ② 组装监理团
    # start the root run
    run_manager = callback_manager.on_chain_start(                # ③ 总监理喊开工
        None, input,
        name=config.get("run_name") or self.get_name(),
        run_id=config.pop("run_id", None),
    )
    input_ = input

    # invoke all steps in sequence
    try:
        for i, step in enumerate(self.steps):                     # ④ ★逐节放水
            # mark each step as a child run
            config = patch_config(                                # ⑤ 给这节盖章
                config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
            )
            with set_config_context(config) as context:           # ⑥ 存进 contextvar
                if i == 0:
                    input_ = context.run(step.invoke, input_, config, **kwargs)
                else:
                    input_ = context.run(step.invoke, input_, config)  # ⑦ 出水=下节进水
    # finish the root run
    except BaseException as e:
        run_manager.on_chain_error(e)                             # ⑧ 出事上报
        raise
    else:
        run_manager.on_chain_end(input_)                          # ⑨ 完工上报
        return cast("Output", input_)

两位工单管理员都住在 runnables/config.py。先看补齐员（config.py:255-292）：

# libs/core/langchain_core/runnables/config.py:255（裁剪）
def ensure_config(config: RunnableConfig | None = None) -> RunnableConfig:
    """Ensure that a config is a dict with all keys present."""
    empty = RunnableConfig(
        tags=[], metadata={}, callbacks=None,
        recursion_limit=DEFAULT_RECURSION_LIMIT, configurable={},
    )                                                   # ① 先造一份空白工单
    if var_config := var_child_runnable_config.get():   # ② ★从 contextvar 捞"祖传工单"
        empty.update({k: v.copy() if k in COPIABLE_KEYS else v
                      for k, v in var_config.items() if v is not None})
    if config is not None:                              # ③ 调用方亲手传的，优先级最高
        empty.update({k: ... for k, v in config.items()
                      if v is not None and k in CONFIG_KEYS})
    ...
    return empty

再看盖章员（config.py:357-397）：

# libs/core/langchain_core/runnables/config.py:357（裁剪）
def patch_config(config, *, callbacks=None, recursion_limit=None,
                 max_concurrency=None, run_name=None, configurable=None):
    config = ensure_config(config)
    if callbacks is not None:
        # If we're replacing callbacks, we need to unset run_name
        # As that should apply only to the same run as the original callbacks
        config["callbacks"] = callbacks        # ① 换上"这一节专属监理"
        if "run_name" in config: del config["run_name"]   # ② 名字不跟着下传
        if "run_id" in config:   del config["run_id"]     #    id 也不跟
    if recursion_limit is not None: config["recursion_limit"] = recursion_limit
    ...
    if configurable is not None:
        config["configurable"] = {**config.get("configurable", {}), **configurable}
    return config

主循环调 step.invoke(...) 后，每一节自己还要办"进出手续"。看第一节 prompt 的 invoke（prompts/base.py:210-236）：

# libs/core/langchain_core/prompts/base.py:210（完整正文）
def invoke(self, input, config=None, **kwargs) -> PromptValue:
    config = ensure_config(config)                       # 又 ensure 一次（幂等，白捡兜底）
    if self.metadata:
        config["metadata"] = {**config["metadata"], **self.metadata}
    if self.tags:
        config["tags"] += self.tags                      # 组件自带的标签也上工单
    return self._call_with_config(                       # ★交给基类的"标准过场"
        self._format_prompt_with_error_handling,         #   真正的业务：填模板
        input, config,
        run_type="prompt",                               #   在追踪里标记为 prompt 类型
        serialized=self._serialized,
    )

把 L02-L04 拼起来，一次 chain.invoke({"topic": "猫"}) 的完整时序：