LangChain 设计与实现

第2章 架构总览

作者 杨艺韬 · 13,164 字 · 发布于 · 更新于

本章基于 langchain-core 1.2.26 / langchain-classic 1.0.3 源码分析。源码路径：libs/ 目录。

理解一个框架的架构，最好的方式不是从概念图开始，而是从一次真实的调用开始。当你写下 chain.invoke("Hello") 并按下回车键时，数据在 LangChain 的哪些层之间流动？哪些对象被创建和销毁？回调事件在什么时机被触发？

本章将回答这些问题。我们首先俯瞰 LangChain 的三层包架构（langchain-core、langchain-classic / langchain、Partners），然后逐目录解析 langchain-core 的内部结构，最后通过跟踪一次完整的 chain.invoke() 调用，将抽象的架构图变成具体的执行流。

本章要点

  • 三层包架构：langchain-core 是基石，langchain-classic 与 langchain 1.x 是组合层，Partners 是集成层，上层依赖下层、langchain-core 不反向依赖任何一层
  • langchain-core 目录结构：17 个公开子包的职责划分，核心抽象的层级关系
  • Runnable 类层次：从 Runnable ABC 到 RunnableSerializable 再到具体组件的继承链
  • chain.invoke() 的完整旅程：从用户调用到 CallbackManager、ContextVar、线程池的底层执行
  • 配置传播机制：RunnableConfig 如何通过 ensure_config、patch_config、set_config_context 在组件间流转

2.1 三层包架构

LangChain 的代码分布在一个 monorepo 中，libs/ 目录下包含多个独立的 Python 包。这种 monorepo + 多包的结构既便于开发时的跨包调试，又保证了发布时的独立性。

libs/
  core/                    # langchain-core 1.2.26
    langchain_core/
      runnables/           # Runnable 协议与 LCEL
      messages/            # 消息类型系统
      language_models/     # LLM/ChatModel 抽象
      prompts/             # Prompt 模板
      output_parsers/      # 输出解析器
      callbacks/           # 回调与追踪
      tools/               # 工具接口
      documents/           # 文档类型
      ...
  langchain/               # langchain-classic 1.0.3
    langchain_classic/
      chains/              # 经典 Chain 实现
      agents/              # Agent 实现
      memory/              # Memory 实现
      retrievers/          # 高级 Retriever
      ...
  langchain_v1/            # langchain 1.2.15（1.0 之后的新 langchain 包）
    langchain/
      agents/              # create_agent 等 1.x Agent 入口
      ...
  partners/                # Partner 集成包（该快照下共 15 个）
    openai/                # langchain-openai
    anthropic/             # langchain-anthropic
    chroma/                # langchain-chroma
    ollama/                # langchain-ollama
    ...
  text-splitters/          # langchain-text-splitters
  standard-tests/          # 标准测试套件
  model-profiles/          # langchain-model-profiles：更新集成包内模型能力档案数据的 CLI 工具

目录名 ≠ 包名

libs/langchain/ 里装的是 langchain-classic，而真正的 langchain 包在 libs/langchain_v1/ 下。这是 1.0 拆包后留下的目录名与包名错位， 第一次进仓库找代码几乎必踩。判断标准只有一个——看该目录 pyproject.toml 的 name 字段。

2.1.1 langchain-core：基石层

langchain-core 是整个生态系统的地基。它的设计目标是：用最少的外部依赖，定义最完整的抽象接口。

查看其 pyproject.toml，运行时依赖极为克制：

# 源码文件：libs/core/pyproject.toml

[project]
name = "langchain-core"
version = "1.2.26"
requires-python = ">=3.10.0,<4.0.0"
dependencies = [
    "langsmith>=0.3.45,<1.0.0",
    "tenacity!=8.4.0,>=8.1.0,<10.0.0",
    "jsonpatch>=1.33.0,<2.0.0",
    "PyYAML>=5.3.0,<7.0.0",
    "typing-extensions>=4.7.0,<5.0.0",
    "packaging>=23.2.0",
    "pydantic>=2.7.4,<3.0.0",
    "uuid-utils>=0.12.0,<1.0",
]

只有 8 个运行时依赖，且每一个都有明确的理由：pydantic 用于数据建模和校验，langsmith 用于追踪，tenacity 用于重试，jsonpatch 用于流式日志（astream_log 的 JSONPatch 增量）和 JSON 输出解析器的流式 diff，PyYAML 用于 prompt、LLM 等对象的 YAML 读写，typing-extensions 用于向后兼容的类型注解，packaging 用于版本比较，uuid-utils 提供 langchain_core/utils/uuid.py 里的 UUIDv7 实现。

2.1.2 langchain：组合层

组合层在该快照里是两个包。0.x 时代 langchain 包里的 Chain、Memory、经典 Agent 等，1.0 起迁入独立发行的 langchain-classic（目录 libs/langchain/），它构建在 langchain-core 之上，提供高级应用模式：

# 源码文件：libs/langchain/pyproject.toml

[project]
name = "langchain-classic"
version = "1.0.3"
dependencies = [
    "langchain-core>=1.2.19,<2.0.0",
    "langchain-text-splitters>=1.1.1,<2.0.0",
    "langsmith>=0.1.17,<1.0.0",
    "pydantic>=2.7.4,<3.0.0",
    "SQLAlchemy>=1.4.0,<3.0.0",
    ...
]

langchain 这个名字则留给了 1.x 的新包（目录 libs/langchain_v1/，该快照为 1.2.15），主打 create_agent 等 Agent 入口，运行时依赖只有三项：

# 源码文件：libs/langchain_v1/pyproject.toml

[project]
name = "langchain"
version = "1.2.15"
dependencies = [
    "langchain-core>=1.2.10,<2.0.0",
    "langgraph>=1.1.5,<1.2.0",
    "pydantic>=2.7.4,<3.0.0",
]

也就是说，1.x 的 langchain 建在 LangGraph 之上，而 langchain-classic 不依赖 LangGraph；两者互不依赖。上面 langchain-classic 依赖里的 langchain-text-splitters（目录 libs/text-splitters/，1.1.1）是文本切分器的独立包（第 9 章），运行时依赖只有 langchain-core。

关键观察：langchain-classic、langchain、langchain-text-splitters 都依赖 langchain-core，但 langchain-core 的运行时依赖里没有其中任何一个。这种单向依赖是架构健康的核心保证。

2.1.3 Partner 包：集成层

libs/partners/ 目录下有 15 个独立的 Partner 包，每个包封装一个第三方服务的集成：

partners/
  openai/        # ChatOpenAI, OpenAIEmbeddings
  anthropic/     # ChatAnthropic
  chroma/        # Chroma vector store
  ollama/        # ChatOllama
  fireworks/     # ChatFireworks
  groq/          # ChatGroq
  mistralai/     # ChatMistralAI
  huggingface/   # HuggingFace 模型和嵌入
  ...

Partner 包在 LangChain 一侧只依赖 langchain-core（外加各自厂商的 SDK），不依赖 langchain-classic 或 langchain。Partner 之间一般也互不依赖，例外是 langchain-deepseek 和 langchain-xai：两者都依赖 langchain-openai（复用其 OpenAI 兼容实现）。所以你可以只安装 langchain-openai 而不拉取 langchain-anthropic 的任何依赖，但装 langchain-deepseek 会连带装上 langchain-openai。

graph TB
    subgraph "应用代码"
        APP["from langchain_openai import ChatOpenAI<br/>from langchain_core.prompts import ChatPromptTemplate"]
    end
    subgraph "Partner 包"
        direction LR
        P1["langchain-openai"]
        P2["langchain-anthropic"]
        P3["langchain-chroma"]
        P4["langchain-ollama"]
    end
    subgraph "langchain-classic 1.0.3"
        LC["chains / agents / memory / retrievers"]
    end
    subgraph "langchain 1.2.15"
        LV1["create_agent 等 1.x Agent 入口"]
    end
    LG["langgraph"]
    subgraph "langchain-core 1.2.26"
        CORE["Runnable / Messages / Prompts / LMs / Callbacks / Tools / Documents"]
    end
    APP --> P1
    APP --> LC
    APP --> CORE
    P1 --> CORE
    P2 --> CORE
    P3 --> CORE
    P4 --> CORE
    LC --> CORE
    LV1 --> LG
    LV1 --> CORE
    LG --> CORE
    P1 -.->|不依赖| P2
    P1 -.->|不依赖| LC
    style CORE fill:#4CAF50,color:#333
    style LC fill:#FF9800,color:#333
    style LV1 fill:#FF9800,color:#333
    style P1 fill:#1976d2,color:#fff
    style P2 fill:#1976d2,color:#fff
    style P3 fill:#1976d2,color:#fff
    style P4 fill:#1976d2,color:#fff

2.1.4 设计决策：为什么分成三层

这个分层不是一开始就有的。早期的 LangChain 是一个单体包，所有代码都在 langchain 中。当生态系统快速增长后，问题开始显现：

  1. 集成代码堆积：各家厂商 SDK 在 0.0.x 的 pyproject.toml 里大多是可选依赖（extras），但所有集成的适配代码都随 langchain 这一个包分发
  2. 版本耦合：任何一个集成要跟进上游 SDK 的 breaking change，都得随整个 langchain 发版

分层架构解决了这些问题：

  • langchain-core 只放抽象接口，提供相对稳定的接口契约
  • Partner 包可以独立发版，跟随各自上游的节奏
  • 开发者只安装需要的包，依赖树干净清晰

2.2 langchain-core 目录结构导航

langchain-core 是我们在本专栏中花费最多时间的地方。让我们逐一认识它的子模块。

langchain_core/
  runnables/           # [核心] Runnable 协议、LCEL 所有组合原语
    base.py            # ~6200 行，Runnable/RunnableSequence/RunnableParallel/RunnableLambda
    config.py          # RunnableConfig、ensure_config、patch_config
    branch.py          # RunnableBranch 条件分支
    passthrough.py     # RunnablePassthrough/RunnableAssign/RunnablePick
    configurable.py    # 运行时可配置的 Runnable
    fallbacks.py       # RunnableWithFallbacks 降级策略
    history.py         # RunnableWithMessageHistory 对话历史
    retry.py           # 重试逻辑
    router.py          # 路由器
    graph.py           # 计算图表示
    graph_mermaid.py   # Mermaid 图生成
    utils.py           # 工具函数、类型定义
    schema.py          # StreamEvent 等 schema
    ...                # graph_ascii.py、graph_png.py 等
  messages/            # 消息类型系统（第 4 章）
  language_models/     # BaseLLM、BaseChatModel 等抽象接口（第 5 章）
  outputs/             # ChatGeneration、LLMResult 等模型输出容器（第 5 章）
  prompts/             # PromptTemplate、ChatPromptTemplate 等（第 6 章）
  example_selectors/   # few-shot 示例选择器（第 6 章）
  output_parsers/      # StrOutputParser、JsonOutputParser 等（第 7 章）
  tools/               # BaseTool 接口（第 8 章）
  documents/           # Document、Blob 数据类型（第 9 章）
  document_loaders/    # BaseLoader、BlobLoader 等加载抽象（第 9 章）
  embeddings/          # Embeddings 抽象接口（第 10 章）
  vectorstores/        # VectorStore 抽象接口（第 10 章）
  indexing/            # 索引 API（第 9 章 §9.12）
  callbacks/           # CallbackManager、BaseCallbackHandler（第 12 章）
  tracers/             # LangSmith 追踪器、ConsoleCallbackHandler（第 12 章）
  load/                # 序列化/反序列化（Serializable 基类，第 16 章）
  utils/               # 通用工具函数
  retrievers.py        # BaseRetriever（单文件模块，不是子包；第 10 章）
  ...                  # 其余单文件模块及 _api/、_security/ 两个私有子包

上面列出的 17 个目录就是 langchain_core 的全部公开子包（另有 _api/、_security/ 两个私有子包）。其中 runnables/ 目录是最核心的，它包含了 LangChain 的"操作系统内核"——所有其他模块中的组件最终都要实现 Runnable 协议。

2.3 核心抽象层级

LangChain 的类层次设计遵循"层层添加能力"的原则。从最抽象的 Runnable 到最具体的 ChatOpenAI，每一层都在前一层的基础上增加特定的能力。

classDiagram
    class Runnable {
        <<ABC>>
        +name: str
        +invoke(input, config) Output
        +batch(inputs, config) list~Output~
        +stream(input, config) Iterator~Output~
        +ainvoke(input, config) Output
        +abatch(inputs, config) list~Output~
        +astream(input, config) AsyncIterator~Output~
        +__or__(other) RunnableSequence
        +__ror__(other) RunnableSequence
        +pipe(*others) RunnableSequence
        +pick(keys) RunnableSequence
        +assign(**kwargs) RunnableSequence
        +with_config(config) RunnableBinding
        +with_retry() RunnableRetry
        +with_fallbacks() RunnableWithFallbacks
        +get_graph() Graph
        +input_schema: BaseModel
        +output_schema: BaseModel
    }
    class Serializable {
        <<ABC>>
        +is_lc_serializable() bool
        +get_lc_namespace() list
        +lc_id() list
        +to_json() dict
    }
    class RunnableSerializable {
        +name: str
        +configurable_fields(**kwargs)
        +configurable_alternatives(which, **kwargs)
    }
    class RunnableSequence {
        +first: Runnable
        +middle: list~Runnable~
        +last: Runnable
        +steps: list~Runnable~
    }
    class RunnableParallel {
        +steps__: Mapping~str, Runnable~
    }
    class RunnableLambda {
        +func: Callable
        +afunc: Callable
    }
    class RunnableBranch {
        +branches: list~tuple~
        +default: Runnable
    }
    Runnable <|-- RunnableLambda
    Runnable <|-- RunnableGenerator
    Serializable <|-- RunnableSerializable
    Runnable <|-- RunnableSerializable
    RunnableSerializable <|-- RunnableSequence
    RunnableSerializable <|-- RunnableParallel
    RunnableSerializable <|-- RunnableBranch
    RunnableSerializable <|-- RunnableBindingBase
    RunnableBindingBase <|-- RunnableBinding

这个类层次体现了几个重要的设计决策：

为什么 RunnableLambda 不继承 RunnableSerializable？ 因为一个 Python 函数（lambda 或普通函数）在一般情况下是无法序列化的。将 RunnableLambda 直接继承自 Runnable 而非 RunnableSerializable，是对这个现实的诚实表达。即便是 @chain 装饰器包装的命名函数，得到的也还是 RunnableLambda，对它调用 dumpd 只会得到 type 为 not_implemented 的占位结构（第 16 章）。

为什么 RunnableSequence 把步骤分成 first、middle、last？ 这是为了类型安全。通过这种分拆，RunnableSequence[Input, Output] 可以确保 first 的输入类型是 Input，last 的输出类型是 Output，而 middle 的类型可以是 Any。如果只用一个 list[Runnable]，就无法在类型层面表达这个约束。

# 源码文件：libs/core/langchain_core/runnables/base.py

class RunnableSequence(RunnableSerializable[Input, Output]):
    first: Runnable[Input, Any]     # 输入类型由此决定
    middle: list[Runnable[Any, Any]] = Field(default_factory=list)
    last: Runnable[Any, Output]     # 输出类型由此决定

    @property
    def steps(self) -> list[Runnable[Any, Any]]:
        return [self.first, *self.middle, self.last]

2.4 跟踪一次完整的 chain.invoke()

现在让我们跟踪一次完整的调用，观察数据如何在 LangChain 的架构中流动。假设我们有如下代码：

from langchain_core.runnables import RunnableLambda

add_one = RunnableLambda(lambda x: x + 1)
mul_two = RunnableLambda(lambda x: x * 2)

chain = add_one | mul_two  # 创建 RunnableSequence
result = chain.invoke(3)   # 期望结果: (3 + 1) * 2 = 8

2.4.1 第一步：构建 RunnableSequence

当 Python 执行 add_one | mul_two 时，调用的是 Runnable.__or__ 方法（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:618）：

# 源码文件：libs/core/langchain_core/runnables/base.py (第618行)

def __or__(self, other):
    return RunnableSequence(self, coerce_to_runnable(other))

coerce_to_runnable（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:6176）会将非 Runnable 对象转换为 Runnable。在这里 mul_two 已经是 RunnableLambda（类定义在同文件 :4399），所以直接返回。

RunnableSequence（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:2817）的 __init__（同文件 :2911）接收可变参数 *steps，将它们展平（如果某个 step 本身是 RunnableSequence，会被解包），然后分配到 first、middle、last：

# 源码文件：libs/core/langchain_core/runnables/base.py (第2911行)
# 简化：省略了 first/middle/last 关键字参数的分支，以及展平后不足 2 步时抛 ValueError 的检查

def __init__(self, *steps: RunnableLike, name: str | None = None, ...) -> None:
    steps_flat: list[Runnable] = []
    for step in steps:
        if isinstance(step, RunnableSequence):
            steps_flat.extend(step.steps)  # 展平嵌套的 Sequence
        else:
            steps_flat.append(coerce_to_runnable(step))
    super().__init__(
        first=steps_flat[0],
        middle=list(steps_flat[1:-1]),
        last=steps_flat[-1],
        name=name,
    )

此时内存中的对象结构：

RunnableSequence
  first: RunnableLambda(lambda x: x + 1)
  middle: []
  last: RunnableLambda(lambda x: x * 2)

2.4.2 第二步：invoke 的入口

调用 chain.invoke(3) 进入 RunnableSequence.invoke（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:3131）：

# 源码文件：libs/core/langchain_core/runnables/base.py (第3131行)

def invoke(self, input: Input, config: RunnableConfig | None = None, **kwargs) -> Output:
    # 1. 初始化配置
    config = ensure_config(config)
    # 2. 配置回调管理器
    callback_manager = get_callback_manager_for_config(config)
    # 3. 启动根级追踪
    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

    # 4. 依次执行每个步骤
    try:
        for i, step in enumerate(self.steps):
            config = patch_config(
                config,
                callbacks=run_manager.get_child(f"seq:step:{i + 1}")
            )
            with set_config_context(config) as context:
                if i == 0:
                    input_ = context.run(step.invoke, input_, config, **kwargs)
                else:
                    input_ = context.run(step.invoke, input_, config)
    except BaseException as e:
        run_manager.on_chain_error(e)  # 5a. 错误上报
        raise
    else:
        run_manager.on_chain_end(input_)  # 5b. 成功上报
        return cast("Output", input_)

2.4.3 第三步：ensure_config 的配置初始化

ensure_config（langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:225）是 LangChain 配置管理的核心。它的主干是三件事（下面是简化版）：

# 源码文件：libs/core/langchain_core/runnables/config.py (第225行)

def ensure_config(config: RunnableConfig | None = None) -> RunnableConfig:
    # 1. 创建默认配置
    empty = RunnableConfig(
        tags=[], metadata={}, callbacks=None,
        recursion_limit=DEFAULT_RECURSION_LIMIT,  # 25
        configurable={},
    )
    # 2. 从 ContextVar 继承父级配置
    if var_config := var_child_runnable_config.get():
        empty.update({k: v.copy() if k in COPIABLE_KEYS else v ...})
    # 3. 用显式传入的配置覆盖
    if config is not None:
        empty.update({k: v ...})
    # （省略）不在 CONFIG_KEYS 里的键移入 configurable；
    # configurable 中的 str/int/float/bool 值（除 api_key 和 __ 开头的键）抄进 metadata
    return empty

注意第 2、3 步都是 dict.update，显式传入的 tags 会整体覆盖从 ContextVar 继承来的 tags，而不是拼接；需要合并语义时用的是 §2.5.4 的 merge_configs。

DEFAULT_RECURSION_LIMIT 的 25 定义在 langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:141。这里的 var_child_runnable_config（同文件 :144）是一个 ContextVar，它使得嵌套调用中的子 Runnable 能自动继承父 Runnable 的配置（如 tags、metadata、callbacks），而无需开发者手动传递。

2.4.4 第四步：CallbackManager 与追踪

get_callback_manager_for_config（langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:498）从配置中提取回调信息，创建一个 CallbackManager。on_chain_start 通知所有注册的回调处理器"一个新的 chain 执行开始了"，并返回一个 RunManager，用于后续的子步骤追踪。

2.4.5 第五步：逐步执行

对于每个步骤，框架做了三件关键的事：

  1. patch_config（langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:324）：用 run_manager.get_child() 创建一个子回调管理器，确保子步骤的追踪事件能正确嵌套在父步骤之下
  2. set_config_context（langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:194）：先拷贝当前上下文（copy_context()），再在这份副本里把配置写入 ContextVar，交给调用方在副本中执行步骤
  3. context.run(step.invoke, ...)：在隔离的上下文中执行步骤的 invoke
sequenceDiagram
    participant User as 用户代码
    participant Seq as RunnableSequence
    participant Config as ensure_config
    participant CB as CallbackManager
    participant Step1 as Step 1 (add_one)
    participant Step2 as Step 2 (mul_two)
    participant Ctx as ContextVar
    User->>Seq: chain.invoke(3)
    Seq->>Config: ensure_config(None)
    Config->>Ctx: var_child_runnable_config.get()
    Config-->>Seq: config{tags:[], metadata:{}, ...}
    Seq->>CB: on_chain_start(input=3)
    CB-->>Seq: run_manager
    Seq->>Seq: patch_config(callbacks=child)
    Seq->>Ctx: set_config_context(config)
    Seq->>Step1: step.invoke(3, config)
    Step1-->>Seq: 4
    Seq->>Seq: patch_config(callbacks=child)
    Seq->>Ctx: set_config_context(config)
    Seq->>Step2: step.invoke(4, config)
    Step2-->>Seq: 8
    Seq->>CB: on_chain_end(8)
    Seq-->>User: 8

2.4.6 第六步：RunnableLambda.invoke 的内部

当执行到 step.invoke(3, config) 时，进入 RunnableLambda.invoke（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:4997）：

# 源码文件：libs/core/langchain_core/runnables/base.py (第4997行)

def invoke(self, input, config=None, **kwargs):
    if hasattr(self, "func"):
        return self._call_with_config(
            self._invoke,
            input,
            ensure_config(config),
            **kwargs,
        )
    raise TypeError("Cannot invoke a coroutine function synchronously.Use `ainvoke` instead.")

_call_with_config（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:2027）是所有 Runnable 共享的模板方法，它负责：

  • 启动子级的回调追踪（on_chain_start）
  • 在正确的上下文中调用实际的函数
  • 处理错误并上报（on_chain_error）
  • 上报成功结果（on_chain_end）

最终，self.func(3) 被调用，返回 4。

2.5 配置传播机制深入

RunnableConfig 是 LangChain 的"中枢神经系统"。理解它的传播机制对于掌握整个框架至关重要。

2.5.1 RunnableConfig 的结构

# 源码文件：libs/core/langchain_core/runnables/config.py (第49行)

class RunnableConfig(TypedDict, total=False):
    tags: list[str]            # 标签，用于过滤和追踪
    metadata: dict[str, Any]   # 元数据，传递给回调处理器
    callbacks: Callbacks       # 回调处理器链
    run_name: str              # 当前运行的名称
    max_concurrency: int | None  # 最大并发数
    recursion_limit: int       # 递归深度限制（默认25）
    configurable: dict[str, Any]  # 运行时可配置字段
    run_id: uuid.UUID | None   # 唯一运行标识

total=False 意味着所有字段都是可选的。这使得配置可以被"部分创建、逐步合并"——一个组件可以只设置 tags，另一个组件可以只设置 metadata，merge_configs 会将它们正确合并。

2.5.2 三种配置操作

LangChain 提供了三个核心的配置操作函数：

graph TD
    subgraph "ensure_config"
        E1["创建默认配置"] --> E2["从 ContextVar 继承"]
        E2 --> E3["用显式参数覆盖"]
        E3 --> E4["返回完整配置"]
    end
    subgraph "patch_config"
        P1["接收现有配置"] --> P2["替换指定字段<br/>(callbacks, recursion_limit等)"]
        P2 --> P3["返回新配置"]
    end
    subgraph "set_config_context"
        S1["拷贝当前上下文"] --> S2["在副本中写入 ContextVar"]
        S2 --> S3["yield 上下文副本"]
        S3 --> S4["清理 ContextVar"]
    end
  • ensure_config：确保配置完整。如果传入 None，返回默认配置；同时从 ContextVar 继承父级配置
  • patch_config：修补配置。在现有配置基础上替换特定字段（如替换 callbacks 为子级回调管理器）
  • set_config_context：设置上下文。拷贝一份上下文并在其中写入 ContextVar，形成隔离的执行上下文

2.5.3 COPIABLE_KEYS 的深意

# 源码文件：libs/core/langchain_core/runnables/config.py

COPIABLE_KEYS = ["tags", "metadata", "callbacks", "configurable"]

当配置从 ContextVar 或显式参数中继承时，COPIABLE_KEYS 中的字段会被 copy() 而非直接引用。这是为了防止"共享引用"问题——如果子 Runnable 向 tags 列表中添加元素，不应该影响父 Runnable 的 tags。这是一个容易被忽视但极其重要的细节。

2.5.4 merge_configs 的 7 种字段特化策略

merge_configs（libs/core/langchain_core/runnables/config.py:366）不是简单的 dict.update——它对每个已知字段都有特化的合并语义。读完这 60 行你会明白为什么两个 runnable 组合时它们的 config "自然就对了"。

def merge_configs(*configs: RunnableConfig | None) -> RunnableConfig:
    base: RunnableConfig = {}
    for config in (ensure_config(c) for c in configs if c is not None):
        for key in config:
            if key == "metadata":                           # ① 字典合并
                base["metadata"] = {
                    **base.get("metadata", {}),
                    **(config.get("metadata") or {}),
                }
            elif key == "tags":                             # ② 去重并排序
                base["tags"] = sorted(
                    set(base.get("tags", []) + (config.get("tags") or [])),
                )
            elif key == "configurable":                     # ③ 字典合并
                base["configurable"] = {
                    **base.get("configurable", {}),
                    **(config.get("configurable") or {}),
                }
            elif key == "callbacks":
                # callbacks 有 6 种合并情况（见下）
                ...
            elif key == "recursion_limit":                  # ⑤ 只在非默认时覆盖
                if config["recursion_limit"] != DEFAULT_RECURSION_LIMIT:
                    base["recursion_limit"] = config["recursion_limit"]
            elif key in COPIABLE_KEYS and config[key] is not None:  # ⑥ 拷贝
                base[key] = config[key].copy()
            else:                                           # ⑦ 后来非空覆盖
                base[key] = config[key] or base.get(key)
    return base

7 种字段各自的合并逻辑：

① metadata 后写入的键覆盖（Python dict 合并的默认语义）——元数据层层附加、父子都能贡献自己的信息。

② tags 用 sorted(set(...))——去重 + 确定性排序。set 对字符串的迭代顺序受哈希随机化影响，不排序的话同一组 tags 在不同进程里 merge 出来的顺序可能不同；sorted 让结果顺序确定、可比较、可 diff。

③ configurable 和 metadata 一样——字典合并。

④ callbacks 有 6 种合并情况（line 394-421，最复杂）：

callbacks 值可以是 None / list[handler] / manager 三种形态，两边组合共 3 × 3 = 9 种；新值为 None 时直接跳过不合并，去掉这 3 种，剩下 6 种真正要处理的配对（源码注释原话就是 "merging two callbacks values has 6 cases"）：

base=None,    new=list     → new.copy()
base=None,    new=manager  → new.copy()
base=list,    new=list     → base + new
base=list,    new=manager  → 创建 new.copy()、把 base 里的 handler 加进去
base=manager, new=list     → base.copy() + add_handler(h) 循环
base=manager, new=manager  → base.merge(new)

每种情形走不同路径，是因为两种形态携带的信息不同：list 只是一串 handler，而 manager 除了 handler 还带着 parent_run_id、可继承的 tags / metadata 等上下文。所以只要有一方是 manager，结果就一定是 manager（copy() 后再 add_handler(..., inherit=True)，或 merge），不会退化成丢掉这些上下文的纯列表；只有两边都是 list 时才直接拼接。

⑤ recursion_limit 的"非默认时才覆盖"（line 422-424）：

if config["recursion_limit"] != DEFAULT_RECURSION_LIMIT:
    base["recursion_limit"] = config["recursion_limit"]

DEFAULT_RECURSION_LIMIT = 25（line 141）。config 里如果 recursion_limit 是 25（默认）、merge 时不覆盖 base——理解为"用户没显式设就不动"。这避免了默认值无意间压掉上层明确设置的值：如果用户上层 config["recursion_limit"] = 50、下层传了 {}（经 ensure_config 填默认 25）、朴素 merge 会让 25 覆盖 50——不符合直觉（实跑 merge_configs({"recursion_limit": 50}, {}) 得到的仍是 50）。代价是：下层无法用显式的 25 把上层的 50 改回默认值。

⑥ COPIABLE_KEYS 的拷贝（上一节讲过，.copy() 是浅拷贝）。不过 COPIABLE_KEYS 里的四个键在前面 ①–④ 分支都已先行处理，这个分支在该快照里实际走不到，属于防御性写法。

⑦ 其他字段"后来非空覆盖"——base[key] = config[key] or base.get(key)。如果后来的 config 里这个字段是 None / 空、保留 base 的；否则用后来的。Python 的 or 短路 + truthy 检查。

这 7 个分支是 "用单一函数表达 config 合并语义" 的范本——换成朴素 update()，tags 会被整体覆盖而不是合并、manager 形态的 callbacks 会被列表直接替换、默认的 recursion_limit 会压掉上层的显式值。这些情况全都在 60 多行代码里被处理。

2.6 Runnable 的通用方法体系

Runnable 基类不仅定义了核心的 invoke/batch/stream 协议，还提供了一整套用于修饰和增强的"方法修饰器"。这些方法遵循一个统一的模式：它们不修改原 Runnable，而是返回一个新的包装 Runnable。

# 所有修饰方法都返回新的 Runnable，不修改原始对象

chain = prompt | model | parser

# with_retry: 返回 RunnableRetry 包装
chain_with_retry = chain.with_retry(stop_after_attempt=3)

# with_fallbacks: 返回 RunnableWithFallbacks 包装
chain_with_fallback = chain.with_fallbacks([fallback_chain])

# with_config: 返回 RunnableBinding 包装
chain_with_config = chain.with_config({"tags": ["production"]})

# configurable_fields: 返回 RunnableConfigurableFields 包装
chain_configurable = model.configurable_fields(
    temperature=ConfigurableField(id="temp")
)
graph LR
    R["原始 Runnable"]
    R -->|".with_retry()"| R1["RunnableRetry<br/>包装原始 Runnable"]
    R -->|".with_fallbacks()"| R2["RunnableWithFallbacks<br/>包装原始 Runnable"]
    R -->|".with_config()"| R3["RunnableBinding<br/>包装原始 Runnable"]
    R -->|".configurable_fields()"| R4["RunnableConfigurableFields<br/>包装原始 Runnable"]
    R -->|".pick(keys)"| R5["原始 Runnable | RunnablePick"]
    R -->|".assign(**kw)"| R6["原始 Runnable | RunnableAssign"]
    style R fill:#e8f5e9
    style R1 fill:#fff3e0
    style R2 fill:#fff3e0
    style R3 fill:#fff3e0
    style R4 fill:#fff3e0
    style R5 fill:#e3f2fd
    style R6 fill:#e3f2fd

这种"不可变包装"的设计模式（装饰器模式）使得：

  • 原始 Runnable 不受影响，可以被多次修饰生成不同变体
  • 每个包装层都可以独立测试
  • 包装是可组合的：chain.with_retry().with_fallbacks([...]) 是合法的

2.7 batch 与并行执行

Runnable 基类提供了 batch 的默认实现，它使用线程池并行执行多个 invoke：

# 源码文件：libs/core/langchain_core/runnables/base.py (第867行)

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):
        if return_exceptions:
            try:
                return self.invoke(input_, config, **kwargs)
            except Exception as e:
                return e
        else:
            return self.invoke(input_, config, **kwargs)

    # 单个输入时不使用线程池
    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))

get_executor_for_config（langchain-core-1.2.26/libs/core/langchain_core/runnables/config.py:589）会把 config["max_concurrency"] 原样传给 ContextThreadPoolExecutor(max_workers=...)。如果 max_concurrency 未指定（即 None），就落回 Python ThreadPoolExecutor 的默认线程数 min(32, CPU 核数 + 4)。

而 RunnableSequence 覆写了 batch，它的实现更加精妙——它对序列中的每个步骤调用 batch，而不是对整个序列调用多次 invoke。这意味着如果某个步骤（如 LLM 调用）有原生的批量 API，它可以利用这个 API 来提高效率。

2.8 stream 与流式执行

流式执行是 LangChain 最复杂也最精妙的部分之一。RunnableSequence 的流式执行依赖于 transform 方法：

# 源码文件：libs/core/langchain_core/runnables/base.py (第3465行)

def _transform(self, inputs, run_manager, config, **kwargs):
    steps = [self.first, *self.middle, self.last]
    # 将每个步骤的 transform 串联成管道
    final_pipeline = cast("Iterator[Output]", inputs)
    for idx, step in enumerate(steps):
        config = patch_config(
            config, callbacks=run_manager.get_child(f"seq:step:{idx + 1}")
        )
        if idx == 0:
            final_pipeline = step.transform(final_pipeline, config, **kwargs)
        else:
            final_pipeline = step.transform(final_pipeline, config)
    yield from final_pipeline

核心思想是：不是等前一步完全执行完再开始下一步，而是将前一步的输出流直接接入下一步的输入流。 这使得支持 transform 的步骤（如 LLM 的流式输出）可以一边产生 token 一边被下一步处理，实现真正的端到端流式。

对于不能原生转换输入流的步骤（如 RunnableLambda），会先把输入缓冲到全部到齐再开始产出（_transform 里的源码注释原话是 "buffer input in memory until all available"）——所以需要边收边处理输入流时，应使用 RunnableGenerator 而非 RunnableLambda。

2.9 三个关键设计决策

TypedDict vs Pydantic Model for Config——LangChain 选择用 TypedDict 而非 Pydantic Model 定义 RunnableConfig。TypedDict 在运行时就是一个普通的 dict，可以直接用 dict.update、.copy() 来合并和复制，没有实例化与校验的开销；而 RunnableConfig 在每次 invoke 中都会被创建和传递多次（每个步骤都要经过 ensure_config / patch_config）。

ContextVar 的选择——使用 ContextVar 而非线程本地存储（threading.local）是一个面向 async 友好的决策。ContextVar 在 asyncio.Task 之间正确传播，而 threading.local 不会。这使得 LangChain 在异步场景下的配置传播无缝工作。

递归限制——默认递归限制 DEFAULT_RECURSION_LIMIT = 25 用来防止无限递归。在 langchain-core 里，它只约束 RunnableLambda 的函数返回另一个 Runnable 时的嵌套调用：每嵌套一层减 1，减到 0 时抛 RecursionError（base.py:4880-4885）。经典 AgentExecutor 的循环上限另由 max_iterations（默认 15）控制；LangGraph 则沿用 recursion_limit 这个配置键来限制超步数。

2.10 主线小结

本章从三个层面建立了对 LangChain 架构的全面理解。

首先，我们认识了三层包架构：langchain-core 是最小化依赖的基石层，定义了所有核心抽象；langchain-classic 与 1.x 的 langchain 是组合层，前者提供 Chains、经典 Agent 等高级模式，后者基于 LangGraph 提供 create_agent；Partner 包是集成层，每个包独立封装一个第三方服务。上层依赖下层，langchain-core 不反向依赖任何一层。

然后，我们深入了 langchain-core 的目录结构和类层次，理解了 Runnable -> RunnableSerializable -> 具体组件的继承链，以及为什么 RunnableLambda 和 RunnableGenerator 直接继承自 Runnable 而非 RunnableSerializable。

最后，我们跟踪了一次完整的 chain.invoke() 调用，从 ensure_config 的配置初始化，到 CallbackManager 的追踪启动，到 patch_config + set_config_context 的上下文隔离，再到每个步骤的实际执行。这个过程揭示了 LangChain 如何将简洁的用户 API（一行 invoke 调用）转化为复杂的内部编排。

以下为源码实证附录。 到上一节为止是本章主线，读完已可动手。从这里开始的各节是针对langchain-core 关键文件的具体实现的独立深挖，彼此不构成阅读顺序，按需查阅即可。

2.11 base.py 6261 行——Runnable 的主场

本章讨论了 Runnable 的概念——真实的 libs/core/langchain_core/runnables/base.py 有 6261 行。这一节把它的结构一张地图画清：

范围 行号 定义
Runnable[Input, Output] 抽象基类 124-2585 超过 2400 行——LangChain 最重的一个类
RunnableSerializable 2586-2731 引入 Pydantic / LangSmith 序列化
模块级 schema 辅助函数 2733-2816 _seq_input_schema / _seq_output_schema 等
RunnableSequence 2817-3564 | 操作符链接的核心实现
RunnableParallel 3565-4095 字典式并行
RunnableGenerator 4096-4398 async iterator 友好
RunnableLambda 4399-5271 wrap 普通函数
RunnableEachBase / RunnableEach 5272-5529 .map() 的实现
RunnableBindingBase / RunnableBinding 5530-6139 bind / with_config / with_types / with_listeners 的返回类型（with_retry 返回的 RunnableRetry 也继承 RunnableBindingBase，但定义在 retry.py；with_fallbacks 返回的 RunnableWithFallbacks 在 fallbacks.py）
Protocols + coerce 6140-6260 类型守卫、自动类型转换

6261 行中 Runnable 基类占约 40%（2462/6261）——剩下 60% 是建在它上面的各种特化实现——这是"核心抽象 + 多种特化"的 OOP 范式。

本章前面引用的行号都来自这个文件——记住这张表，下次查源码不用 grep -n。

2.12 with_retry 源码——一条"为什么 LangChain 值得看"的证据

libs/core/langchain_core/runnables/base.py:1860-1923 的 with_retry 签名——

def with_retry(
    self, *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]:

四个参数的设计都值得拆——

  • retry_if_exception_type=(Exception,)——默认捕获所有——但生产应该改成(RateLimitError, TimeoutError, APIConnectionError) 这类明确可重试的——不重试 LogicError
  • wait_exponential_jitter=True——默认开 jitter——防止多 Client 同时重试打垮上游（thundering herd）——tenacity 本身不设默认等待策略，要不要退避、要不要抖动由调用方决定，LangChain 在这里替你选了"指数退避 + 抖动"
  • exponential_jitter_params——可以控制 initial / max / exp_base——暴露底层 tenacity 的调参能力、但不强制
  • stop_after_attempt=3——最多尝试 3 次（含第一次调用）——连续 3 次都失败时，上游多半不是瞬时抖动，继续重试收益有限

这 4 个默认值的取舍——默认捕获面宽、默认带退避抖动、默认次数少——生产环境里通常只需要把第一个参数收窄到可重试的异常类型。

2.13 with_fallbacks——LangChain 的"降级优雅"

base.py:1947- 的 with_fallbacks 是 Runnable 的另一把利器——主调用失败时自动尝试备选：

gpt4_chain.with_fallbacks([sonnet_chain, haiku_chain])

行为——

  • 先跑 gpt4_chain
  • 失败（Exception）→ 跑 sonnet_chain
  • 还失败 → 跑 haiku_chain
  • 全部失败 → 抛出第一次的异常（runnables/fallbacks.py:213 的 raise first_error），不是最后一次

exception_key 参数的巧妙——可以把前一次的 exception 作为输入字段传给 fallback（写进输入字典的该键，所以输入必须是 dict，第 3 章 §3.10.1）：

# fallback 可以接收上一次的 exception 做 'recovery prompt'
recovery_chain = prompt_with_error_context | model
main.with_fallbacks([recovery_chain], exception_key="error")

这就是"让 AI 自己纠错"的实现路径——上一次失败的信息被显式塞进下一次调用的 prompt——model 能根据错误调整策略——比单纯 retry 更聪明。

《MCP 协议设计与实现》专栏第 18 章讨论过"失败—引导—重试"赋予工具调用的自愈能力——LangChain 的 with_fallbacks + exception_key 是同一思路在 Runnable 层的实现。

2.14 batch_as_completed 的流式返回

§2.7 讨论了 batch——还有一个更精妙的batch_as_completed（实现在 langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:937，前面 917-936 是两个 @overload 签名）——谁先完成谁先返回：

# 不传 return_exceptions=True 时，任一输入失败会直接抛出
for idx, result in chain.batch_as_completed(inputs, return_exceptions=True):
    # 按完成顺序返回、而不是按 inputs 顺序
    if isinstance(result, Exception):
        print(f'input[{idx}] 失败: {result}')
    else:
        process(idx, result)

典型用法——并发 10 个查询、哪个先回哪个先处理——比"等 10 个全完成"用户感知延迟低很多。

同步版用线程池 + concurrent.futures.wait(..., return_when=FIRST_COMPLETED) 实现，异步版 abatch_as_completed 用的是 asyncio.as_completed——LangChain 把并发细节隐藏在 Runnable API 下——用户不用自己管理线程池或事件循环也能写并发代码。

2.15 coerce_to_runnable 的类型魔法

libs/core/langchain_core/runnables/base.py:6176 的 coerce_to_runnable 函数——把几乎所有"看起来像 callable"的东西转成 Runnable：

def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
    if isinstance(thing, Runnable):
        return thing
    if is_async_generator(thing) or inspect.isgeneratorfunction(thing):
        return RunnableGenerator(thing)      # 生成器优先于普通 callable
    if callable(thing):                      # 普通函数
        return RunnableLambda(cast("Callable[[Input], Output]", thing))
    if isinstance(thing, dict):              # 字典 -> RunnableParallel
        return cast("Runnable[Input, Output]", RunnableParallel(thing))
    msg = (
        f"Expected a Runnable, callable or dict."
        f"Instead got an unsupported type: {type(thing)}"
    )
    raise TypeError(msg)

用户用| 拼链时、a | b | c | {"x": d, "y": e}——字典被自动识别为 RunnableParallel、函数被自动识别为 RunnableLambda——"一切皆可 Runnable"——这种类型强制的自动化让 API 极简。

反面对比——如果必须显式写 RunnableLambda(fn) | RunnableParallel({"x": ...})——API 会冗长得多。

2.16 chain 装饰器——@overload（PEP 484）的多重载

libs/core/langchain_core/runnables/base.py:6203-6227 的 chain 装饰器——四个 @overload——入参的 callable 形态完全不同、返回类型却一致：

@overload
def chain(func: Callable[[Input], Coroutine[Any, Any, Output]]) -> Runnable[Input, Output]: ...

@overload
def chain(func: Callable[[Input], Iterator[Output]]) -> Runnable[Input, Output]: ...

@overload
def chain(func: Callable[[Input], AsyncIterator[Output]]) -> Runnable[Input, Output]: ...

@overload
def chain(func: Callable[[Input], Output]) -> Runnable[Input, Output]: ...

为什么要 4 个 overload——因为 func 可能是同步 / 协程 / 生成器 / 异步生成器 4 种形态——四个重载的返回类型其实都是 Runnable[Input, Output]，重载的作用是让类型检查器从每一种 callable 形态里正确解出 Input 和 Output 两个类型变量：协程要剥掉 Coroutine[Any, Any, ...] 一层，生成器要剥掉 Iterator[...] 一层。

没有 overload 的话——单一签名只能写成四种 callable 类型的联合，类型检查器无法把 Output 和具体分支对上，推断结果会退化成 Runnable[Input, Any]——IDE 补全和静态检查都失效。

这 4 个 overload = 强类型 + 好开发体验的代价——写框架作者多几行、用户多一份安全。

2.17 configurable_fields 的动态 config 注入

Runnable 还有一个不常被讨论但超有用的能力——.configurable_fields(...)：

model = ChatOpenAI(model="gpt-4", temperature=0.7).configurable_fields(
    temperature=ConfigurableField(id="llm_temperature"),
    model_name=ConfigurableField(id="llm_model"),  # 用字段名，不能用别名 model
)

# 在 invoke 时动态切温度 / 模型
result = model.invoke("hello", config={
    "configurable": {"llm_temperature": 0.0, "llm_model": "gpt-4o"}
})

为什么这个设计值得——

  • 同一 chain 既能用于"稳定输出场景（temperature=0）"也能用于"创意场景（temperature=1）"
  • 不用重新实例化 chain——运行时动态配置
  • 可配置项可以被导出——config_schema() 会把它们列进 schema，上层工具可以据此生成配置界面

configurable_fields 的键必须是模型的字段名：它按 type(self).model_fields 校验，而 ChatOpenAI 的 model 只是字段 model_name 的别名，写 model=ConfigurableField(...) 会直接抛 ValueError: Configuration key model not found ...（用带别名字段的假模型实跑）。

这是 Runnable 的"运行时参数化"能力——在没有它的框架里你得写 if-else 创建两个 chain——重复代码。

2.18 astream_events：Runnable 的"事件流"

langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:1273 的 astream_events——本章 §2.8 没深入讨论（详见第 12 章 §12.7）。签名里有个容易忽略的 version: Literal["v1", "v2"] = "v2"（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:1278）——该快照默认就是 v2，v1 只是保留的旧 schema（v1 不上报 parent_ids，也不透出自定义事件）。

事件名不是硬编码的枚举，而是由 run_type 拼出来的——langchain-core-1.2.26/libs/core/langchain_core/tracers/event_stream.py:586 的 f"on_{run_type_}_start"、:610 的 f"on_{run_type}_end"。docstring 里的参考表（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:1317-1334）列了 15 种标准事件：

  • on_chat_model_start / on_chat_model_stream / on_chat_model_end（聊天模型走的是这一组，不是 on_llm_*）
  • on_llm_start / on_llm_stream / on_llm_end（文本补全 BaseLLM 才用）
  • on_chain_start / on_chain_stream / on_chain_end
  • on_tool_start / on_tool_end
  • on_retriever_start / on_retriever_end
  • on_prompt_start / on_prompt_end

外加 on_custom_event（langchain-core-1.2.26/libs/core/langchain_core/tracers/event_stream.py:417，只在 v2 下出现）。每种都能被上层业务 subscribe——构建"实时 UI / telemetry / debugging"的基石。

《LangGraph 设计与实现》第 13 章《流式输出与调试》详细讨论过 StreamMessagesHandler + TAG_NOSTREAM——LangGraph 的 messages 流模式并不经过 astream_events，而是把 StreamMessagesHandler 这个回调处理器挂到回调管理器上；astream_events v2 同样是往回调里注册一个处理器（libs/core/langchain_core/tracers/event_stream.py 的 _AstreamEventsCallbackHandler）。两者共用的底座是 langchain-core 的回调系统（第 12 章）。

2.19 Runnable 能力的"九件套"一张表

Runnable 暴露的九个复合能力——每个都是一行 .xxx() 返回新 Runnable——函数式组合的典范：

方法 作用 典型用途
.bind(**kwargs) 固定某参数 固定 stop 序列
.with_config({...}) 预设 config 固定 tag / metadata
.with_retry(...) 自动重试 §2.12
.with_fallbacks([...]) 降级备选 §2.13
.with_types(input_type, output_type) 覆盖推断类型 泛型窄化
.with_listeners(on_start, on_end, on_error) 生命周期钩子 埋点
.map() 单输入 → 列表输入 batch 简化
.configurable_fields(...) 字段可配置 §2.17
.configurable_alternatives(...) 整体切换实现 dev/prod

九件套都返回 Runnable——可以链式嵌套——model.with_retry(...).with_fallbacks([...]).bind(stop=["\n"])。

2.20 一张比喻图：Runnable 相当于什么

为了让读者快速建立直觉——三个类比：

  • Runnable 有点像 JavaScript 的 Promise——都支持.then() 式组合；但 Promise 一创建就开始执行、只产出一次结果，Runnable 则是可反复 invoke 的计算描述
  • Runnable 有点像 Rust 的 Future——都是"惰性求值 + 组合子"、都靠 runtime（async/executor）真正执行
  • Runnable 有点像 Haskell 的 IO 动作——都把带副作用的计算当作值来传递、组合，Haskell 用 >>=，LCEL 用 |

这三个类比都不是"相等"——它们只是在"把计算当作可组合的值"这一点上相通，求值时机、错误处理、类型保证各不相同。

2.21 Runnable 的限制：不是万能钥匙

本章正面讲了 Runnable 多好——也要讲它的边界：

限制 1——不适合状态机——Runnable 是"pure function + config"模型、多步状态流转（会话、长任务）需要 LangGraph 的"StateGraph"——Runnable 没 checkpoint 概念。

限制 2——条件分支不优雅——RunnableBranch 存在、但复杂多分支用起来很啰嗦——LangGraph 的 add_conditional_edges 更合适。

限制 3——动态节点数——Runnable 组合的结构在构造时就固定了——运行时决定"再加一步"只能靠 RunnableLambda 返回另一个 Runnable 这种动态嵌套（受 recursion_limit 约束，§2.9）——不如 LangGraph 的循环与条件边自然。

所以——Runnable 适合"已知流程的线性 / 并行组合"、LangGraph 适合"状态 + 分支 + 循环"的复杂流程——两者都是 LangChain 体系的一部分、各有定位。

LangGraph 本身不在本专栏范围内，《LangGraph 设计与实现》专栏有系统讲解——读完你会发现"Runnable 是 LangGraph 的构件、LangGraph 是 Runnable 的组合框架"——层次分明。

2.22 架构章的方法论升华

本章从三层包结构开始、到Runnable 抽象、到6261 行源码导航——技术本身讲完。

最后一节——想和读者聊聊"架构思维"的培养：

  • 读源码先读目录结构——§2.1 的三层包、§2.11 的 6261 行 9 个类——先看"骨架"再抠"肌肉"
  • 抽象不是凭空想出来的——从实际需求反推抽象、比先定抽象再套需求可靠
  • 好框架让"简单的事简单、复杂的事可能"——三行写个 chain、几百行写个 Agent——LangChain 实现了这个目标
  • 读一个框架源码要"问它为什么不这样做"——本章多处解释了"为什么 TypedDict 不是 Pydantic"、"为什么 ContextVar 不是 threading.local"、"为什么要有 recursion_limit"——每个"为什么"背后都是一次取舍

2.23 RunnableSequence.invoke 里的隐藏细节

§2.4.2 提到了 RunnableSequence.invoke——真实源码 base.py:3131-3163 还有几个细节值得看：

  • run_manager 在 loop 开始前创建——整条序列只有一个父 run——每个 step 是它下面的子 run（各有自己的 run_id）——trace 里看到的是一个整体链路、不是 N 个孤立 step
  • step loop 里 config = patch_config(...)——每步生成一份新的 config、带 child callback_manager——父子链路自动嵌套
  • run_manager.get_child(f"seq:step:{i + 1}")——给第 i+1 个子 run 打上 seq:step:1、seq:step:2 这样的标签（不向更深层继承）——在 trace 里一眼能看出某个子 run 是序列的第几步
  • 异常时 run_manager.on_chain_error(e) 必调——确保失败也有收尾事件——不会出现"开始但永远没结束"的僵尸 trace

这四条细节——让chain.invoke() 的每次调用在 LangSmith 上都"可追溯"——这是 Runnable 框架天然支持 observability 的根源。

2.24 RunnableParallel 的并发保证

libs/core/langchain_core/runnables/base.py:3834-3885 的 RunnableParallel.invoke 用 ThreadPoolExecutor 并行执行字典里的每个 value——但有一条微妙的线程安全约定：

# copy to avoid issues from the caller mutating the steps during invoke()
steps = dict(self.steps__)
with get_executor_for_config(config) as executor:
    futures = [
        executor.submit(_invoke_step, step, input, config, key)
        for key, step in steps.items()
    ]
    # 一次提交全部分支（实际并发数受 max_concurrency 限制）、然后按 key 顺序收集结果
    output = {key: future.result() for key, future in zip(steps, futures)}

线程安全要求——

  • 每个 value 必须是"thread-safe Runnable"——框架不替你检查
  • 共享可变状态的自定义 RunnableLambda——必须自己加锁、否则数据竞争
  • config 的 tags / metadata——每个分支经 patch_config → ensure_config 拿到各自的浅拷贝、不是共享 reference

这条约定不强制在接口上——依赖开发者自觉——典型症状：同一字典的某两个 value 共享了 DB 连接、并发时串 session。

工程建议——RunnableParallel 的每个 value 最好是"无状态"的——有状态的话各持一份、不共享。

2.25 RunnableGenerator 的流式友好

§2.8 提到 RunnableGenerator——base.py:4096-4398 的完整实现有几个亮点：

  • 签名：transform: Callable[[Iterator[Input]], Iterator[Output]]——输入输出都是 Iterator——天然支持流式
  • async 版本：atransform: Callable[[AsyncIterator[Input]], AsyncIterator[Output]]
  • 双模支持：一个实例可以同时持有同步和异步两个函数——stream 走同步版、astream 走异步版
  • 中间不累积——和 RunnableLambda 的对比：Lambda 在 chain 里 transform 等价于"先消化完整输入、再一次性输出"、Generator 真正流式

典型用法——

def stream_tokens(tokens: Iterator[str]) -> Iterator[str]:
    for tok in tokens:
        if "STOP" in tok:
            break
        yield tok.upper()

# 聊天模型吐出的是 AIMessageChunk，先经 StrOutputParser 转成字符串
chain = llm | StrOutputParser() | RunnableGenerator(stream_tokens)
for chunk in chain.stream("hello"):
    print(chunk, end="", flush=True)

效果——LLM 每吐一个 token、立即被 upcase、立即打印——真正端到端流式。

用 RunnableLambda 写同样逻辑——会等 LLM 全部输出完才开始处理——首字节延迟从「第一个 token」退化成「整段生成完」。

所以——任何"流式要求高"的场景、选 Generator 不选 Lambda。

2.26 RunnableBinding 的"洋葱模型"

§2.19 的九件套里，bind / with_config / with_types / with_listeners 返回 RunnableBinding，with_retry 返回同属 RunnableBindingBase（libs/core/langchain_core/runnables/base.py:5530-5931）子类的 RunnableRetry，with_fallbacks 返回的 RunnableWithFallbacks 则用 runnable 字段包住原对象——洋葱一层层包。

chain.with_retry().with_fallbacks([...]).with_config({"tags": ["prod"]})
# 实际结构（实跑打印的类型）：
# RunnableBinding(
#     bound=RunnableWithFallbacks(
#         runnable=RunnableRetry(
#             bound=chain),
#         fallbacks=[...]),
#     config={"tags": ["prod"]})

每层包装——把自己的 logic 插入invoke / stream / batch 调用——外层能看到内层、内层看不到外层——严格单向。

这就是 middleware / decorator 模式的函数式实现——和 Express 中间件、Koa onion model、Redux enhancer 同源。

LangChain 把它演绎到 Runnable 层——学过 middleware 的工程师会觉得很眼熟。

2.27 Runnable 和 LCEL 的关系

LCEL = LangChain Expression Language = | + .with_xxx() + {} + RunnableLambda——一套"在 Python 里写"函数式 DSL"。

Runnable 是 LCEL 的运行时——LCEL 表达式在构造时就直接得到一棵 Runnable 对象树，没有单独的编译步骤——本章讲清了 Runnable、LCEL 也就懂了。

对比其他框架——

  • LlamaIndex 走的是 Workflow + @step 的事件驱动路线（llamaindex-0.14.24/llama-index-core/llama_index/core/workflow/__init__.py）——和 LCEL 更像的 QueryPipeline 已被官方标为弃用并移出 0.14 的 core（llamaindex-0.14.24/docs/src/content/docs/framework/changes/deprecated_terms.md 的 "QueryPipeline (and related classes)" 一节）——范式不同，不是表达式式组合
  • Haystack 走的是有向图式 Pipeline 组合组件——配置化程度更高、不如 LCEL 的代码直观
  • DSPy 有 Module + forward——模仿 PyTorch、偏 ML 研究

LCEL 的主要卖点——"像 shell pipe 一样简洁 + 泛型类型标注 + 同步异步同一套接口 + 可观测"。

2.28 RunnableLambda 的"透明"成本

§2.15 提到 coerce_to_runnable 会自动把函数包成 RunnableLambda——看似零成本、实际有隐性开销：

  • 每次 invoke——额外一层函数调用栈 + callback 事件触发（on_chain_start / on_chain_end）
  • stream 场景——Lambda 默认 transform 会"先收集完整输入"——破坏流式——上游 LLM 的流式被断掉
  • 类型推断——RunnableLambda 的 Input/Output 类型靠运行时推断——有时不准——IDE 提示不完美

工程建议——

  • 高频热路径——避免用 Lambda wrap 小函数——开销可能超过函数本身
  • 需要流式——用 RunnableGenerator、不是 RunnableLambda（§2.25）
  • 类型敏感——继承 RunnableSerializable 自定义类、不用 Lambda——类型更清晰

Lambda 是"入门友好、生产谨慎"的工具——用得多不是错、但要知道代价。

2.29 .with_listeners 的观察性妙用

§2.19 的九件套里有一个被低估的.with_listeners(on_start, on_end, on_error)：

def log_start(run: Run, config: RunnableConfig): ...
def log_end(run: Run, config: RunnableConfig): ...

chain = (prompt | model | parser).with_listeners(on_start=log_start, on_end=log_end)

三个 listener 之于 Runnable——相当于 async hook 之于 transaction——整个链路的"开始 / 结束 / 失败"都能被你注入代码。

典型用途——

  • 埋点——每次 chain 执行记一条 trace
  • metrics——统计 latency / token / error rate
  • 缓存预热——on_start 时 async prefetch 下游资源
  • 断路器——on_error 累计、达阈值后暂停调用

三个 callback 只 5 行代码——却是 Runnable 的"观测性基石"——没有它你要自己 wrap 整个 chain。

2.30 .with_alisteners 的异步版本

对应的 with_alisteners(on_start, on_end, on_error) 用 async callback，listener 里可以 await 而不占住事件循环线程：

async def log_to_datadog(run: Run, config): ...

async_chain = chain.with_alisteners(on_end=log_to_datadog)

但它不是 fire-and-forget——

  • AsyncRootListenersTracer 会 await 每个 listener（libs/core/langchain_core/tracers/root_listeners.py:120-130）——实跑：on_end 里 await asyncio.sleep(1)，ainvoke 就多等 1 秒
  • listener 抛异常不会打断主链路——同步版、异步版都一样，异常只被记一条 Error in ...RootListenersTracer... 日志，invoke / ainvoke 照常返回结果（实跑）
  • 真要不等 sink——在 listener 里自己 asyncio.create_task(...) 把发送丢到后台

这和《Harness Engineering》专栏第 19 章讲的"telemetry 不能拖慢主链路" 原则一致——with_alisteners 解决的是"不阻塞事件循环"，"不拖慢这次调用"还得靠自己把发送异步化。

2.31 常见问题的快速答案

读者学 LangChain 架构时常问的 10 个问题——这里一次性回答：

Q1——invoke / ainvoke / batch / stream 何时选？——同步脚本用 invoke、async 服务用 ainvoke、并发多请求用 batch、用户 UI 流式用 stream。

Q2——RunnableSequence 和 RunnablePassthrough().assign(...) 有何不同？——Sequence 是纯链、assign 是在 dict 上加新字段、保留原有。

Q3——configurable_fields() 和 configurable_alternatives() 区别？——前者改"一个字段"、后者换"整个 Runnable 的实现"。两者都是 RunnableSerializable 上的方法，分别返回 RunnableConfigurableFields 和 RunnableConfigurableAlternatives；ConfigurableField 则是描述"哪个字段可配置"的声明对象，别把它和方法名混为一谈。

Q4——RunnableParallel 的字典 value 能嵌套吗？——能、任意深——最终还是 DAG。

Q5——chain.invoke 的框架开销来自哪里？——RunnableSequence.invoke（base.py:3131）每一步都要 patch_config 造一份新 config、run_manager.get_child() 建子 callback manager、set_config_context 拷贝一次 contextvars 上下文。这些是每步固定的常数开销；真正的耗时几乎总是 LLM 的网络往返，先用 astream_events 或 LangSmith trace 定位到具体步骤再谈优化。

Q6——如何把 Runnable 当 FastAPI route？——LangServe（独立发行的 langserve 包，不在本快照的 monorepo 里）把任意 Runnable 暴露为 REST endpoint + OpenAPI 文档 + Playground UI。

Q7——Runnable 可以序列化为 JSON 吗？——只有 RunnableSerializable 子类可以——RunnableLambda 不能（因为 Lambda 里是任意 Python 函数）——所以"可序列化的 chain" 必须只用 Serializable 组件。

Q8——callback 和 events 的关系？——Callback 是低级接口、events 是高级——astream_events 底层就是注册 callback。

Q9——| 链里错误如何传播？——任何一步抛 exception、整个 chain 抛——除非在这一步之后加 .with_fallbacks(...) 或 .with_retry(...)。

Q10——Runnable 里的 config 字典会被改吗？——每次 patch_config 都经 ensure_config 返回新字典、原 config 不变——但只是浅拷贝：tags / metadata 等会各拷一层，callback manager、configurable 里的对象仍是共享引用。

2.32 和 LangGraph 的架构对照

LangChain 和 LangGraph 是同一团队的两个产品——但设计目标不同：

维度 LangChain LangGraph
核心抽象 Runnable StateGraph + Node
组合方式 | / {} add_edge / add_conditional_edges
状态管理 每次 invoke 独立 channel-based state 跨 step 共享
循环 不支持（会 RecursionError） 天然支持（pregel superstep）
Checkpoint 无 内置（pg / sqlite / memory）
典型场景 线性/并行 chain 长时间运行 agent

两者互补、不是替代——

  • 简单 prompt-LLM-parser 的 chain——用 LangChain
  • 多 agent 协作、长对话、人工审批流——用 LangGraph
  • 既要 chain 又要状态机——LangGraph 的 node 内部用 LangChain 的 Runnable——层次嵌套

LangGraph 的内部机制（pregel superstep、channel-based state、checkpoint 后端）见《LangGraph 设计与实现》专栏。

2.33 附：LangChain 版本演进简史

本章讨论的是 2026 年春的快照（langchain-core 1.2.26 发布于 2026-04-03）——简要回顾它的演进：

  • v0.0.x（仓库首个 commit 在 2022-10-24）——原始设计、Chain 为中心、无 Runnable 概念
  • v0.0.2xx（2023-07-26，PR #7800 "Runnable single protocol" 引入 Runnable 协议，随 v0.0.245 发布）——Runnable 与 LCEL 首次落地；2023-08 PR #9016 把它拆成 schema/runnable/ 目录；同年 11 月 PR #13577 把核心抽象整体拆出为 langchain-core
  • v0.1.0（2024-01-05，tag v0.1.0）——多包拆分（core / community / partner）——重大架构变革
  • v0.2.0（2024-05-17，tag langchain==0.2.0）——LCEL 成为推荐写法、老 Chain 陆续标记 deprecated（如 LLMChain 的 @deprecated 标注是 since="0.1.17"，0.2 前夕已打上）
  • v0.3.0（2024-09-13，tag langchain==0.3.0）——Pydantic 2 迁移完成
  • v1.0.0（2025-10-17，tag langchain-core==1.0.0 与 langchain==1.0.0 同日，langchain-classic==1.0.0 也在当天打 tag）——首个 1.0 正式版、0.x 的 Chain / Agent / Memory 迁入独立发行的 langchain-classic

版本演进的主线——"逐步把隐式约定变成显式抽象"——Chain 是隐式、Runnable 是显式——后者更可组合、更可观测。

读者选版——新项目直接用 1.x。

2.34 让 Runnable 可测试——单元测试范式

生产 LangChain 代码必须可单元测试——但 LLM 调用的 non-determinism 让传统测试难用——Runnable 框架提供了解决方案：

范式 1——Mock 整条 chain 的任一步

from unittest.mock import patch

# 只 mock model、保留 prompt / parser 真实运行
# 注意 patch 的是类：模型是 Pydantic 对象，patch.object(my_model, "invoke", ...)
# 在实例上打补丁会抛 AttributeError；patch 类会影响该类所有实例，只在 with 块内生效
with patch.object(type(my_model), "invoke", return_value=mocked_response):
    result = chain.invoke({"input": "test"})
    assert result == expected

范式 2——FakeListChatModel 预设回复

from langchain_core.language_models import FakeListChatModel

# 不调真 LLM、按列表顺序返回预设回复
fake = FakeListChatModel(responses=["resp1", "resp2", "resp3"])
chain = prompt | fake | parser

范式 3——callback 验证内部行为

from langchain_core.callbacks import BaseCallbackHandler

class AssertCallback(BaseCallbackHandler):
    raise_error = True  # 默认 False：回调里的异常只打日志、不会让测试失败

    def on_chain_start(self, serialized, inputs, **kwargs):
        # 子步骤（prompt、parser）也会触发 on_chain_start，只检查根 run
        if kwargs.get("parent_run_id") is None:
            assert inputs["question"] == "what is 2+2"

chain.invoke({"question": "what is 2+2"}, config={"callbacks": [AssertCallback()]})

三种范式合起来——chain 的每一步行为都可断言——不需要联网、不需要真 API key，也能把组合逻辑测干净。

2.35 给新手的一张"防踩坑"清单

10 条 LangChain 新手常栽的坑——

  1. 不要在 RunnableLambda 里 print（用 .with_listeners 埋点）
  2. 不要在 chain 外 wrap retry（用 .with_retry）
  3. 不要自己 thread.Thread 并发（用 .batch）
  4. 不要 while True 重调（用 LangGraph 的 recursion_limit）
  5. 不要硬编码模型名到 chain（用 configurable_fields）
  6. 不要把序列化的 chain 当成通用存储格式（含 RunnableLambda 的部分只能存成 not_implemented 占位；load 默认只放行 core 里的对象，第 16 章）
  7. 不要忽略 RunnableConfig 的 callbacks（失去 trace）
  8. 不要在 chain 里 await（用 ainvoke 全链路 async）
  9. 不要把 API key 硬编码进 chain（用环境变量）
  10. 不要在 on_chain_end 里做重活（listener 会被等待，重活丢到后台任务，§2.30）

2.36 源码"必读"五个文件

给重度读者——LangChain 源码里最值得精读的 5 个文件：

  1. libs/core/langchain_core/runnables/base.py 6261 行——Runnable 家族（本章主场）
  2. libs/core/langchain_core/runnables/config.py 641 行——RunnableConfig 的所有操作
  3. libs/core/langchain_core/callbacks/manager.py 2697 行——CallbackManager 启动 + 传播
  4. libs/core/langchain_core/runnables/graph.py 739 行——任意 Runnable 的"可视化图"
  5. libs/core/langchain_core/language_models/chat_models.py 1870 行——ChatModel 基类的 invoke/stream 实现

五个文件合计 12208 行（6261 + 641 + 2697 + 739 + 1870）——覆盖了本章讨论的所有抽象的实际实现。顺着 __or__ 到 RunnableSequence.invoke 再到 _call_with_config 到 CallbackManager.configure 这条调用链走一遍，可以把宏观概念和行号级实现对齐。

2.37 最后一张"架构速记图"

把本章全部内容压缩成一张文本图——

LangChain Architecture (3 tiers)

  langchain-core  (foundation, minimal deps)
  ├── Runnable [Input, Output]   <-- ABSTRACT CORE, base.py 6261 lines
  │     ├── invoke / ainvoke              # single call
  │     ├── batch / batch_as_completed    # parallel
  │     ├── stream / astream              # yielding
  │     ├── astream_events                # 15 standard events (+custom)
  │     ├── .bind() / .with_config()      # partial application
  │     ├── .with_retry() / .with_fallbacks()  # resilience
  │     ├── .configurable_fields() / .configurable_alternatives()
  │     ├── .map()                        # batch helper
  │     └── .with_listeners()             # observability hooks
  │
  ├── RunnableSequence      (a | b | c | d)       compose via __or__
  ├── RunnableParallel      ({"x": a, "y": b})    dict-to-parallel
  ├── RunnableLambda        (f: Input -> Output)  wrap functions
  ├── RunnableGenerator     (f: Iter -> Iter)     true streaming
  ├── RunnableBinding       (wraps + adds config) onion layers
  ├── Callbacks / Events                          observability spine
  └── RunnableConfig        (TypedDict)           per-call context (ContextVar)

  langchain-classic (classic composition)          Chains, Agents, legacy
  langchain-*     (partner packages)               OpenAI, Anthropic, Chroma, ...

2.38 一点额外的"为什么选 LangChain" 思考

有同行问：为什么不直接用 Anthropic / OpenAI SDK、而要套 LangChain？——四个理由：

理由 1：provider-agnostic——业务代码写成 chain = prompt | model | parser、切 OpenAI 到 Anthropic 只改一行 model = ChatAnthropic(...)——vendor lock-in 风险低。

理由 2：observability 内置——LangSmith 只需设环境变量即可开启追踪，其他观测平台接入 LangChain 也都走回调这一个口子——业务代码不用改。

理由 3：生态完善——monorepo 内 15 个 partner 包，社区维护的集成包数量更多——新 LLM / 新 vector db 快速接入。

理由 4：LangGraph 一脉相承——从简单 chain 升级到复杂 agent 时、基础代码不用重写。

反面——轻量场景（单次调 GPT + 直接 parse）——直接用 SDK 更直接——不需要为了 LangChain 的复杂度买单。

判据——如果你的场景需要"换 provider / 流式 / 重试 / fallback / observability / batch"中的多项——就值得用 LangChain。

2.39 附赠一段性能 tips

让 LangChain chain 跑得快——五条性能建议：

1. 在 async 服务里用 ainvoke 而不是 invoke——即使单次调用——async 的 event loop 不会被阻塞、上层 FastAPI 的 async handler 能更好地调度。

2. batch 而不是循环 invoke——Runnable.batch 默认用线程池并发跑 N 个 invoke（langchain-core-1.2.26/libs/core/langchain_core/runnables/base.py:867）——串行循环会把 N 次网络往返累加。

3. 开启 Prompt Caching（《Harness Engineering》专栏第 20 章详述）——system prompt 和 tool 定义加 cache_control——命中缓存的前缀部分按更低价计费，长对话省得最多。

4. ContextVar 而不是 threading.local——LangChain 默认用 ContextVar、别自己乱改——异步友好。

5. max_concurrency 配好——batch 时的并发上限（config["max_concurrency"] 决定 ThreadPoolExecutor 的 max_workers）——太高打垮下游、太低浪费——按下游的限流额度反推。