第2章 架构总览

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

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

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

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
      ...
  partners/                # Partner 集成包
    openai/                # langchain-openai
    anthropic/             # langchain-anthropic
    chroma/                # langchain-chroma
    ollama/                # langchain-ollama
    ...
  text-splitters/          # langchain-text-splitters
  standard-tests/          # 标准测试套件

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",
    "pydantic>=2.7.4,<3.0.0",
    "typing-extensions>=4.7",
    "packaging>=23.2,<25.0",
]

只有 7 个运行时依赖，且每一个都有明确的理由：pydantic 用于数据建模和校验，langsmith 用于追踪，tenacity 用于重试，jsonpatch 用于流式日志，PyYAML 用于配置，typing-extensions 用于向后兼容的类型注解，packaging 用于版本比较。

2.1.2 langchain：组合层

langchain（在最新版本中命名为 langchain-classic）构建在 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-classic 依赖 langchain-core，但 langchain-core 绝不依赖 langchain-classic。这种严格的单向依赖是架构健康的核心保证。

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-core，不依赖 langchain-classic 或其他 Partner 包。这意味着你可以只安装 langchain-openai 而不拉取 langchain-anthropic 的任何依赖。

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-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

    P1 -.->|不依赖| P2
    P1 -.->|不依赖| LC

    style CORE fill:#4CAF50,color:#fff
    style LC fill:#FF9800,color:#fff
    style P1 fill:#2196F3,color:#fff
    style P2 fill:#2196F3,color:#fff
    style P3 fill:#2196F3,color:#fff
    style P4 fill:#2196F3,color:#fff

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

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

1. 依赖爆炸：安装 langchain 就意味着安装 OpenAI SDK、Anthropic SDK、ChromaDB 等全部依赖
2. 版本耦合：OpenAI SDK 的一次 breaking change 会导致整个 langchain 发版
3. 贡献瓶颈：所有 PR 都汇聚到一个仓库，审核压力巨大

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

• 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
  messages/            # 消息类型系统
  language_models/     # BaseLLM、BaseChatModel 等抽象接口
  prompts/             # PromptTemplate、ChatPromptTemplate 等
  output_parsers/      # StrOutputParser、JsonOutputParser 等
  callbacks/           # CallbackManager、BaseCallbackHandler
  tracers/             # LangSmith 追踪器、ConsoleCallbackHandler
  tools/               # BaseTool 接口
  documents/           # Document 数据类型
  load/                # 序列化/反序列化（Serializable 基类）
  embeddings/          # Embeddings 抽象接口
  vectorstores/        # VectorStore 抽象接口
  indexing/            # 索引 API
  utils/               # 通用工具函数

其中 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) RunnablePick
        +assign(**kwargs) RunnableAssign
        +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，是对这个现实的诚实表达。如果一个 RunnableLambda 恰好可以序列化（例如使用了 @chain 装饰器的命名函数），框架不会阻止你，但也不会承诺这个能力。

为什么 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__ 方法：

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

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

coerce_to_runnable 会将非 Runnable 对象转换为 Runnable。在这里 mul_two 已经是 RunnableLambda，所以直接返回。

RunnableSequence.__init__ 接收可变参数 *steps，将它们展平（如果某个 step 本身是 RunnableSequence，会被解包），然后分配到 first、middle、last：

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

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：

# 源码文件：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 配置管理的核心。它做三件事：

# 源码文件：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 ...})
    return empty

这里的 var_child_runnable_config 是一个 ContextVar，它使得嵌套调用中的子 Runnable 能自动继承父 Runnable 的配置（如 tags、metadata、callbacks），而无需开发者手动传递。

2.4.4 第四步：CallbackManager 与追踪

get_callback_manager_for_config 从配置中提取回调信息，创建一个 CallbackManager。on_chain_start 通知所有注册的回调处理器"一个新的 chain 执行开始了"，并返回一个 RunManager，用于后续的子步骤追踪。

2.4.5 第五步：逐步执行

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

1. patch_config：用 run_manager.get_child() 创建一个子回调管理器，确保子步骤的追踪事件能正确嵌套在父步骤之下
2. set_config_context：将当前配置写入 ContextVar，然后拷贝当前上下文（copy_context()），在新的上下文中执行步骤
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：

# 源码文件：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.")

_call_with_config 是所有 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["将配置写入 ContextVar"] --> S2["拷贝当前上下文"]
        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（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(...))——去重 + 确定性排序。关键是排序：如果不排序、同一 config 在两个进程里 merge 出来的 tags 顺序不同、LangSmith trace 的签名就会不一样。sorted 让 trace 可比较、可 diff。

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

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

callbacks 值可以是 None / list[handler] / manager 三种形态、两边合起来就是 3 × 3 - 3（两边相同视为同一）= 6 种不对称配对：

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)

每种情形走不同路径，为什么不统一？因为 callback 的身份要保留——handler 是有状态对象（比如 LangSmith tracer 持一个 run_id 上下文），直接合并列表会让两个 tracer 共用状态导致 trace 错乱。6 种处理确保每个 handler 的 lifecycle 完整、不共享状态。

⑤ 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——不符合直觉。用"值不是默认"作为判定标志是 Python config 合并里的常用 idiom。

⑥ COPIABLE_KEYS 的深拷贝（上一节讲过）。

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

这 7 种策略是 "用单一函数表达 config 合并的所有正确行为" 的范本——换成朴素 update() 在 tags 顺序、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 会根据 config["max_concurrency"] 创建一个 ThreadPoolExecutor。如果 max_concurrency 未指定，使用 Python 默认的线程数（通常是 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 一边被下一步处理，实现真正的端到端流式。

对于不支持原生 transform 的步骤（如 RunnableLambda），默认实现会先累积全部输入再产出输出——这就是为什么 LangChain 文档建议在需要流式的场景中使用 RunnableGenerator 而非 RunnableLambda。

2.9 设计决策总结

TypedDict vs Pydantic Model for Config

LangChain 选择用 TypedDict 而非 Pydantic Model 定义 RunnableConfig。这是因为 TypedDict 就是一个普通的 dict，可以用 dict.update 来合并，性能开销极小。而 Pydantic Model 的实例化和验证在高频调用路径上会带来可感知的性能损失——RunnableConfig 在每次 invoke 中都会被创建和传递多次。

ContextVar 的选择

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

递归限制

默认递归限制 DEFAULT_RECURSION_LIMIT = 25 是为了防止无限递归的 Agent 循环。当一个 Agent 反复调用工具而不收敛时，这个限制会自动终止执行。这是一个务实的安全措施。

2.10 小结

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

首先，我们认识了三层包架构：langchain-core 是最小化依赖的基石层，定义了所有核心抽象；langchain（langchain-classic）是组合层，提供 Chains、Agents 等高级模式；Partner 包是集成层，每个包独立封装一个第三方服务。三者之间的依赖严格单向。

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

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

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

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

6261 行中 Runnable 基类占 40%——剩下 60% 是基于它的 8 个变种——这是"核心抽象 + 多种特化"的 OOP 范式典范。

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

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

base.py:1860-1924 的 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 默认没开、LC 把它改成 True 是生产经验
• exponential_jitter_params——可以控制 initial / max / exp_base——暴露底层 tenacity 的调参能力、但不强制
• stop_after_attempt=3——最多 3 次——不是 10 次——因为连续重试 3 次还不行、基本是上游真挂了、再多也白搭

这 4 个默认值的选择——都是 LangChain 社区多年生产经验的沉淀——你如果从零写一个 retry、大概率会先写出"stop_after_attempt=5 + no jitter + catch all"的"初学者版"——工业级默认值就是"被反复调教过的数字"。

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
• 全部失败 → 抛 exceptions_to_handle

exception_key 参数的巧妙——可以把前一次的 exception 作为参数传给 fallback：

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

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

本书第 20 章《MCP Build Server》§20.27 讨论过**"自愈 Agent"的思路——LangChain 的 with_fallbacks + exception_key 是工业级实现。

2.14 batch_as_completed 的流式返回

§2.7 讨论了 batch——还有一个更精妙的batch_as_completed（base.py:918）——谁先完成谁先返回：

for idx, result in chain.batch_as_completed(inputs):
    # 按完成顺序返回、而不是按 inputs 顺序
    if isinstance(result, Exception):
        print(f'input[{idx}] 失败: {result}')
    else:
        process(idx, result)

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

这是"asyncio.as_completed"的 Runnable 封装——LangChain 把 asyncio 复杂度隐藏在 Runnable API 下——用户不用学 asyncio 也能写并发代码。

2.15 coerce_to_runnable 的类型魔法

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

def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
    if isinstance(thing, Runnable):
        return thing
    if callable(thing):  # 普通函数
        return RunnableLambda(thing)
    if isinstance(thing, dict):  # 字典 -> RunnableParallel
        return cast(Runnable[Input, Output], RunnableParallel(thing))
    # ... 还有 async callable / generator 等
    raise TypeError(f"Expected a Runnable, callable or dict. Instead got: {type(thing)}")

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

反面对比——如果必须显式写 RunnableLambda(fn) | RunnableParallel({"x": ...})——用户体验大跌——很多竞品框架就是死在这种冗长 API 上。

2.16 chain 装饰器——PEP 612 @overload 的多重载

base.py:6204-6227 的 chain 装饰器——四个 @overload——类型完全不同：

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

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

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

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

为什么要 4 个 overload——因为 func 可能是同步 / 异步 / 生成器 / 异步生成器 4 种——每种返回的 Runnable 类型略不同——IDE 要给出精确的类型提示。

没有 overload 的话——IDE 只能推断成 Runnable[Input, Any]——用户写 .invoke().strip() 时 TypeScript/Python 类型检查器不报错、但运行时可能 Any 是个 Awaitable。

这 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=ConfigurableField(id="llm_model")
)

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

为什么这个设计值得——

• 同一 chain 既能用于"稳定输出场景（temperature=0）"也能用于"创意场景（temperature=1）"
• 不用重新实例化 chain——运行时动态配置
• LangSmith / LangServe 可以自动生成 UI——用户点 slider 改 temperature

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

2.18 astream_events 是 LangGraph 的"观测基石"

base.py:1273 的 astream_events——本章§2.8 没深入讨论——但它是 LangGraph 流式能力的基础。

典型事件——

• on_chain_start / on_chain_end
• on_llm_start / on_llm_stream / on_llm_end
• on_tool_start / on_tool_end
• on_parser_end
• on_retriever_start / on_retriever_end

20+ 种事件——每种都能被上层业务 subscribe——构建"实时 UI / telemetry / debugging"的基石。

本书第 13 章《LangGraph Streaming》详细讨论过 StreamMessagesHandler + TAG_NOSTREAM——那套机制的底层就是 astream_events——没有 Runnable 的 events 流、就没有 LangGraph 的 streaming——Runnable 是 LangGraph 的"运行时事件总线"。

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

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

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

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

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

• Runnable 相当于 JavaScript 的 Promise——都是"可延迟执行的一等公民"、都支持.then() 式组合
• Runnable 相当于 Rust 的 Future——都是"惰性求值 + 组合子"、都靠 runtime（async/executor）真正执行
• Runnable 相当于 Haskell 的 IO Monad——都是"纯函数包裹副作用"、都用>>= / | 组合

这三个类比不是"相等"、是"同根同源"——都是"Composable Effectful Computation"这一理论模型的不同语言落地。

学过任何一个——理解 Runnable 极快——没学过——通过 Runnable 反向理解另外三个。

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

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

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

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

限制 3——动态节点数——Runnable 是编译期固定的 DAG——运行时决定"再加一步"做不到——需要 LangGraph 的动态 graph。

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

本书第 13-18 章详细讨论 LangGraph——你读完会发现"Runnable 是 LangGraph 的构件、LangGraph 是 Runnable 的组合框架"——层次分明。

2.22 架构章的方法论升华

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

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

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

这种思考方式——读完本章后用到你下次读其他框架源码（React / Vue / Tokio / hyper）——你会发现"优秀框架的底层设计哲学是相通的"。

2.24 RunnableSequence.invoke 里的隐藏细节

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

• run_manager 在 loop 开始前创建——所有 step 共享同一个 run_id——trace 里看到的是一个整体链路、不是 N 个孤立 step
• step loop 里 config = patch_config(...)——每步生成一份新的 config、带 child callback_manager——父子链路自动嵌套
• config = patch_config(config, callbacks=run_manager.get_child(f"seq:step:{idx+1}"))——tag 里带 seq:step:1、seq:step:2——LangSmith UI 用这个命名自动画调用树
• 异常时 run_manager.on_chain_error(e) 必调——确保失败也有 end 事件——不会出现"开始但永远没结束"的僵尸 trace

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

2.25 RunnableParallel 的并发保证

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

with get_executor_for_config(config) as executor:
    steps = {
        key: executor.submit(_invoke_step, step, input, patched_config, key)
        for key, step in self.steps.items()
    }
    # 所有 future 同时启动、然后串行收集结果

线程安全要求——

• 每个 value 必须是"thread-safe Runnable"——大多数 LLM / retrieval 是（因为都用 HTTP client）
• 共享可变状态的自定义 RunnableLambda——必须自己加锁、否则数据竞争
• config 的 tags / metadata——每个分支都拿到 snapshot、不是共享 reference

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

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

2.26 RunnableGenerator 的流式友好

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

• 签名：fn: Callable[[Iterator[Input]], Iterator[Output]]——输入输出都是 Iterator——天然支持流式
• async 版本：afn: Callable[[AsyncIterator[Input]], AsyncIterator[Output]]
• 双模支持：一个类同时有同步和异步 func——按调用 context 自动选
• 中间不累积——和 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()

chain = llm | RunnableGenerator(stream_tokens)
for chunk in chain.stream("hello"):
    print(chunk, end="", flush=True)

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

用 RunnableLambda 写同样逻辑——会等 LLM 全部输出完才开始处理——用户感知延迟 × 10。

所以——任何"流式要求高"的场景、选 Generator 不选 Lambda——LangChain 文档对此反复强调。

2.27 RunnableBinding 的**"洋葱模型"

§2.19 的九件套大多数都返回 RunnableBinding 或其子类——base.py:5530-5931——洋葱一层层包。

chain.with_retry().with_fallbacks([...]).with_config({"tag": "prod"})
# 实际结构：
# RunnableBindingWithConfig(
#     bound=RunnableWithFallbacks(
#         bound=RunnableRetry(
#             bound=chain)))

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

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

LangChain 把它演绎到 Runnable 层——一致性极高——任何学过 middleware 的工程师 5 分钟就能上手。

2.28 Runnable 和 LCEL 的关系

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

Runnable 是 LCEL 的运行时——LCEL 表达式都会 compile 成 Runnable 树——本章讲清了 Runnable、LCEL 也就懂了。

对比其他框架——

• LlamaIndex 有 QueryPipeline + QueryComponent——思路类似、但组合 API 不如 LCEL 简洁
• Haystack 有 Pipeline + Node——YAML 配置为主、不如 LCEL 的代码直观
• DSPy 有 Module + forward——模仿 PyTorch、偏 ML 研究

LCEL 的独特价值——**"像 shell pipe 一样简洁 + 类型安全 + 异步 + 可观测"——四项都做到才是 LangChain 占据生态主导的根本原因。

2.29 本章大结

本章可以用一句话概括——

"Runnable 是一个支持 invoke/batch/stream、能用 | 组合、自动生成 observability 事件、九件套扩展、6261 行源码撑起整个 LangChain 生态的"统一计算原语"。"

一句话太长——分成三块记：

• 统一 API（invoke/batch/stream）
• 组合基元（| / dict / with_xxx）
• 内置 observability（callbacks / astream_events / tags）

三块 = Runnable = LangChain 架构。

2.30 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.26）
• 类型敏感——继承 RunnableSerializable 自定义类、不用 Lambda——类型更清晰

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

2.31 .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.32 .with_alisteners 的异步版本

对应的 with_alisteners(on_start, on_end, on_error) 用 async callback——不阻塞主流程：

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

async_chain = chain.with_alisteners(on_end=log_to_datadog)

为什么要单独的异步版——

• 埋点 sink 是远程服务（Datadog / LangSmith / BQ）——同步发送会阻塞 main chain 几十 ms 到几秒
• 异步 callback 用 asyncio.create_task fire-and-forget——main chain 不等
• 失败（sink 挂了）——不影响 main chain——分离故障域

这和本书第 19 章§19.20 Claude Code 的 Datadog integration 的"不阻塞主流程" 原则完全一致——任何第三方 telemetry 都必须异步 + 解耦。

2.34 常见问题的快速答案

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

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

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

Q3——ConfigurableField 和 ConfigurableAlternatives 区别？——ConfigurableField 改"一个字段"、ConfigurableAlternatives 换"整个 Runnable 的实现"。

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

Q5——为什么 chain.invoke 慢、chain.batch 也慢？——首次 chain 构造会做 Pydantic 验证、后续每次 invoke 还会 validate input——参考base.py:3140 _call_with_config 的 input validation 开销。

Q6——如何把 Runnable 当 FastAPI route？——LangServe（langchain/langserve 包）自动把任意 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 返回新字典、原 config 不变——immutable semantics——线程安全的保证。

2.36 和 LangGraph 的架构对照

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

两者互补、不是替代——

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

LangGraph 的内部机制（pregel superstep、channel-based state、checkpoint 后端）在本丛书的《LangGraph》分册中有独立章节。

2.37 附：LangChain 版本演进简史

本章讨论的是 2025 Q4 / 2026 Q1 的版本——简要回顾它的演进：

• v0.0.x（2022-10 发布）——原始设计、Chain 为中心、无 Runnable 概念
• v0.1.x（2024-01）——三层包拆分（core / community / partner）——重大架构变革
• v0.2.x（2024-05）——Runnable + LCEL 成为正式 API——老 Chain 标记 deprecated
• v0.3.x（2024-10）——Pydantic 2 迁移完成、性能提升
• v1.0 LTS（2025）——API 稳定、官方保证 3 年兼容

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

读者选版——生产选 v1.x LTS、尝鲜可 follow v1.x latest——v0.x 已终止维护、不要用。

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

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

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

from unittest.mock import patch

# 只 mock model、保留 prompt / parser 真实运行
with patch.object(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):
    def on_chain_start(self, serialized, inputs, **kwargs):
        assert inputs["question"] == "what is 2+2"

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

三种范式合起来——chain 的单元测试覆盖率能做到 80%+——每一步行为都可断言。

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

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. 不要把 RunnableSequence 序列化到 DB（用 LangServe 的 schema）
7. 不要忽略 RunnableConfig 的 callbacks（失去 trace）
8. 不要在 chain 里 await（用 ainvoke 全链路 async）
9. 不要把 API key 放 chain（用 env + .bind）
10. 不要在 on_chain_end 里做重活（用 .with_alisteners）

2.42 源码**"必读"**五个文件

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

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

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

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

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

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                # 20+ events
  │     ├── .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 composition)            Chains, Agents, legacy
  langchain-*     (partner packages)               OpenAI, Anthropic, Pinecone

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

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

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

理由 2：observability 内置——LangSmith / LangFuse / Langtrace 都原生支持 LangChain callback——零配置 trace。

理由 3：生态完善——Partner 包覆盖 100+ vendor——新 LLM / 新 vector db 快速接入。

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

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

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

2.48 附赠一段性能 tips

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

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

2. batch 而不是循环 invoke——N 个请求、batch 能并发 10-100 倍——大多数 LLM provider 的 API 天然支持。

3. 开启 Prompt Caching（本书第 20 章§20.4 详述）——system prompt 和 tool 定义加 cache_control——长对话成本降 90%。

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

5. max_concurrency 配好——batch 时的并发上限——太高打垮下游、太低浪费——通常 10-50 是甜点。

这 5 条做到——chain 性能接近"手写 SDK"代码——但同时享受 LangChain 的全部工程化福利。