RunnablePassthrough（libs/core/langchain_core/runnables/passthrough.py:74）是"什么都不做"的管子；它的类方法 assign（passthrough.py:207）才是高频主角：

# libs/core/langchain_core/runnables/passthrough.py:207
@classmethod
def assign(cls, **kwargs) -> RunnableAssign:
    """Merge the Dict input with the output produced by the mapping argument."""
    return RunnableAssign(RunnableParallel[dict[str, Any]](kwargs))  # ★委托给 Assign+Parallel

# libs/core/langchain_core/runnables/passthrough.py:484（RunnableAssign 的核心）
def _invoke(self, value, run_manager, config, **kwargs) -> dict[str, Any]:
    if not isinstance(value, dict):
        msg = "The input to RunnablePassthrough.assign() must be a dict."
        raise ValueError(msg)
    return {
        **value,                                  # ① 原输入 dict 全保留
        **self.mapper.invoke(                     # ② 新字段并行算出来，合并进去
            value,
            patch_config(config, callbacks=run_manager.get_child()),  # 回调树续上（Day17!）
            **kwargs,
        ),
    }

RunnableBranch（libs/core/langchain_core/runnables/branch.py:43）构造时收若干 (条件, 链) 对 + 最后一个默认链（branch.py:75）。看 invoke（branch.py:185）：

# libs/core/langchain_core/runnables/branch.py:185（节选）
def invoke(self, input, config=None, **kwargs) -> Output:
    for idx, branch in enumerate(self.branches):
        condition, runnable = branch
        expression_value = condition.invoke(          # ① 条件本身也是 Runnable！
            input,
            config=patch_config(config,
                callbacks=run_manager.get_child(tag=f"condition:{idx + 1}")),
        )
        if expression_value:                          # ② 第一个为真的分支胜出
            output = runnable.invoke(
                input,
                config=patch_config(config,
                    callbacks=run_manager.get_child(tag=f"branch:{idx + 1}")),
                **kwargs)
            break
    else:                                             # ③ for-else：全不中 → 默认分支
        output = self.default.invoke(
            input,
            config=patch_config(config,
                callbacks=run_manager.get_child(tag="branch:default")),
            **kwargs)

RunnableWithFallbacks（libs/core/langchain_core/runnables/fallbacks.py:37），一般通过 runnable.with_fallbacks([...])（runnables/base.py:2176）创建。核心 invoke（fallbacks.py:165）：

# libs/core/langchain_core/runnables/fallbacks.py:165（节选）
def invoke(self, input, config=None, **kwargs) -> Output:
    first_error = None
    last_error = None
    for runnable in self.runnables:               # ① 主 Runnable + 所有备胎，排队上
        try:
            if self.exception_key and last_error is not None:
                input[self.exception_key] = last_error   # ② ★可把上次的异常喂给备胎
            child_config = patch_config(config, callbacks=run_manager.get_child())
            with set_config_context(child_config) as context:
                output = context.run(runnable.invoke, input, config, **kwargs)
        except self.exceptions_to_handle as e:    # ③ 只接住指定异常类型（默认 Exception）
            if first_error is None:
                first_error = e                   #    记住第一个错
            last_error = e
        except BaseException as e:
            run_manager.on_chain_error(e)         # ④ 不在白名单的错：立刻上报、原样抛
            raise
        else:
            run_manager.on_chain_end(output)
            return output                         # ⑤ 谁成功就用谁，后面的不再试
    run_manager.on_chain_error(first_error)
    raise first_error                             # ⑥ 全军覆没：抛"第一个"错

RunnableRetry（libs/core/langchain_core/runnables/retry.py:48），入口是 runnable.with_retry(...)（runnables/base.py:2089）。它没有自己造轮子，直接站在重试库 tenacity 肩上：

# libs/core/langchain_core/runnables/retry.py:152
def _sync_retrying(self, **kwargs: Any) -> Retrying:
    return Retrying(**self._kwargs_retrying, **kwargs)   # tenacity 的重试器

# libs/core/langchain_core/runnables/retry.py:179
def _invoke(self, input_, run_manager, config, **kwargs) -> Output:
    for attempt in self._sync_retrying(reraise=True):    # ① tenacity 的重试循环
        with attempt:
            result = super().invoke(
                input_,
                self._patch_config(config, run_manager, attempt.retry_state),  # ② 每次尝试打 tag
                **kwargs,
            )
        if attempt.retry_state.outcome and not attempt.retry_state.outcome.failed:
            attempt.retry_state.set_result(result)       # ③ 成功：登记结果，循环结束
    return result

# retry.py:165（_patch_config 内部）
attempt = retry_state.attempt_number
tag = f"retry:attempt:{attempt}" if attempt > 1 else None  # ★第2次起打 retry:attempt:N 标签

最后两件。configurable_fields/alternatives（runnables/base.py:2843/2901）返回的 DynamicRunnable 子类（libs/core/langchain_core/runnables/configurable.py:50）核心是"先按 config 备好零件，再执行"：

# libs/core/langchain_core/runnables/configurable.py:141（DynamicRunnable）
def invoke(self, input, config=None, **kwargs) -> Output:
    runnable, config = self.prepare(config)     # ① ★按 config 现场"组装"出真正的 Runnable
    return runnable.invoke(input, config, **kwargs)

# libs/core/langchain_core/runnables/configurable.py:419（RunnableConfigurableFields._prepare 节选）
def _prepare(self, config=None):
    configurable_fields = {                     # ② 从 config["configurable"] 里挑出声明过的字段
        specs_by_id[k][0]: v
        for k, v in config.get("configurable", {}).items()
        if k in specs_by_id and isinstance(specs_by_id[k][1], ConfigurableField)
    }
    if configurable:
        init_params = {k: v for k, v in self.default.__dict__.items()
                       if k in type(self.default).model_fields}
        return (
            self.default.__class__(**{**init_params, **configurable}),  # ③ ★用新参数重建一个实例
            config,
        )
    return (self.default, config)               # ④ 没配置 → 用默认件，零开销

# libs/core/langchain_core/runnables/router.py:107（RouterRunnable）
def invoke(self, input: RouterInput, config=None, **kwargs) -> Output:
    key = input["key"]                          # 输入自带路由 key
    actual_input = input["input"]
    if key not in self.runnables:
        msg = f"No runnable associated with key '{key}'"
        raise ValueError(msg)
    runnable = self.runnables[key]              # ★查表分发，就这么直白
    return runnable.invoke(actual_input, config)