开门见山，runnables/base.py:133，类的 docstring 就是官方设计说明书（裁剪）：

# libs/core/langchain_core/runnables/base.py:133
class Runnable(ABC, Generic[Input, Output]):
    """A unit of work that can be invoked, batched, streamed, transformed and composed.

    Key Methods
    ===========
    - invoke/ainvoke: Transforms a single input into an output.
    - batch/abatch:   Efficiently transforms multiple inputs into outputs.
    - stream/astream: Streams output from a single input as it's produced.

    Built-in optimizations:
    - Batch: By default, batch runs invoke() in parallel using a thread pool ...
    - Async: Methods with 'a' prefix are asynchronous ...

    The main composition primitives are RunnableSequence and RunnableParallel.
    ...
    """

# base.py:874  唯一的抽象方法——子类必须实现的就它一个
def invoke(self, input: Input, config: RunnableConfig | None = None,
           **kwargs: Any) -> Output:
    """Transform a single input into an output."""

Python 规定：写 a | b 时，解释器实际调用 a.__or__(b)；如果 a 不会算，再试 b.__ror__(a)（reverse-or，反向兜底）。Runnable 把这两个钩子都实现了（base.py:648-667 与 base.py:691，前面 :628-646 是几个类型重载声明）：

# libs/core/langchain_core/runnables/base.py:648（真身，裁掉类型标注）
def __or__(self, other):
    """Runnable "or" operator.
    Compose this Runnable with another object to create a RunnableSequence."""
    return RunnableSequence(self, coerce_to_runnable(other))   # base.py:667

# base.py:691  反向版：other | self（other 不是 Runnable 时轮到我兜底）
def __ror__(self, other):
    return RunnableSequence(coerce_to_runnable(other), self)

# base.py:712  不喜欢运算符？还有等价的普通方法
def pipe(self, *others, name=None):
    """Equivalent to RunnableSequence(self, *others) or self | others[0] | ..."""
    return RunnableSequence(self, *others, name=name)

文件末尾的守门员（base.py:6628-6652）：

# libs/core/langchain_core/runnables/base.py:6628
def coerce_to_runnable(thing: RunnableLike) -> Runnable:
    """Coerce a Runnable-like object into a Runnable."""
    if isinstance(thing, Runnable):
        return thing                                  # ① 已是标准管件 → 直接放行
    if is_async_generator(thing) or inspect.isgeneratorfunction(thing):
        return RunnableGenerator(thing)               # ② 生成器函数 → 流式管件
    if callable(thing):
        return RunnableLambda(cast(..., thing))       # ③ 普通函数 → Lambda 管件
    if isinstance(thing, dict):
        return RunnableParallel(thing)                # ④ 字典 → 并行管件
    msg = (f"Expected a Runnable, callable or dict."
           f"Instead got an unsupported type: {type(thing)}")
    raise TypeError(msg)                              # ⑤ 其他 → 拒收，报清楚类型

竖线拧出来的东西长什么样？base.py:3063 类定义 + base.py:3150-3154 三个字段 + base.py:3157-3196 构造器：

# libs/core/langchain_core/runnables/base.py:3063
class RunnableSequence(RunnableSerializable[Input, Output]):
    """Sequence of Runnables, where the output of one is the input of the next.
    RunnableSequence is the most important composition operator in LangChain
    as it is used in virtually every chain."""     # ←官方原话：最重要的组合子

    first: Runnable[Input, Any]                       # base.py:3150
    middle: list[Runnable[Any, Any]] = Field(default_factory=list)  # :3152
    last: Runnable[Any, Output]                       # base.py:3154

    def __init__(self, *steps, name=None, first=None, middle=None, last=None):
        steps_flat: list[Runnable] = []
        for step in steps:
            if isinstance(step, RunnableSequence):
                steps_flat.extend(step.steps)         # ★套娃拆平：管中管 → 一排管
            else:
                steps_flat.append(coerce_to_runnable(step))
        if len(steps_flat) < 2:
            raise ValueError("RunnableSequence must have at least 2 steps, ...")
        super().__init__(first=steps_flat[0],
                         middle=list(steps_flat[1:-1]),
                         last=steps_flat[-1], name=name)

    @property
    def steps(self):                                  # base.py:3209
        """All the Runnables that make up the sequence in order."""
        return [self.first, *self.middle, self.last]

L02 说"实现 invoke 就白送 stream/batch"，白送的东西什么成色？看基类默认实现（base.py:1182-1201 与 base.py:919-966）：

# base.py:1182  默认 stream：其实是"假流式"
def stream(self, input, config=None, **kwargs):
    """Default implementation of stream, which calls invoke.
    Subclasses must override this method if they support streaming output."""
    yield self.invoke(input, config, **kwargs)     # base.py:1201 一次全吐

# base.py:919  默认 batch：线程池并发跑 invoke
def batch(self, inputs, config=None, *, return_exceptions=False, **kwargs):
    if not inputs:
        return []
    configs = get_config_list(config, len(inputs))
    def invoke(input_, config): ...                # 内部小函数：调 self.invoke
    if len(inputs) == 1:
        return [invoke(inputs[0], configs[0])]     # 单个输入不开线程池
    with get_executor_for_config(configs[0]) as executor:
        return list(executor.map(invoke, inputs, configs))  # base.py:966 ★并发