LangChain 设计与实现

第1章 为什么需要理解 LangChain

作者 杨艺韬 · 8,073 字 · 发布于 · 更新于

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

当我们今天回望 AI 应用开发的演进历程，会发现一个有趣的规律：每一次底层模型能力的跃迁，都会催生出一个新的应用框架浪潮。从最初手写 HTTP 请求调用 OpenAI API，到使用各种轻量封装库，再到如今以 LangChain 为代表的完整应用框架生态——这个过程并非偶然，而是由真实的工程痛点驱动的必然选择。

LangChain 是当前 LLM 应用开发领域最具代表性的框架之一。它的生态覆盖模型调用、Prompt、工具、检索、Agent、回调、追踪和供应商集成。然而，对于大多数开发者而言，LangChain 仍然是一个"会用但不理解"的黑盒。本专栏的使命，就是带领读者打开这个黑盒，从源码层面理解其设计哲学与实现细节。

本章要点

  • AI 应用框架的三个演进阶段：从裸 API 调用到轻量封装，再到声明式框架
  • LangChain 的核心创新：LCEL 表达式语言与 Runnable 统一接口协议
  • 为什么要读源码：超越文档和教程的局限性，理解设计决策背后的权衡
  • 本专栏路线图：从核心抽象到具体实现，系统掌握 LangChain 的每一层

1.1 AI 应用框架的演进

1.1.1 第一阶段：裸 API 调用时代

最初的 LLM 应用开发是简单直接的。开发者直接构造 HTTP 请求，发送给 API 端点，解析返回的 JSON。这种方式的问题很快就暴露了：

# 2022 年前后的典型写法（text-davinci-003 于 2022 年 11 月发布）
import requests
import json

def call_llm(prompt: str) -> str:
    response = requests.post(
        "https://api.openai.com/v1/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"model": "text-davinci-003", "prompt": prompt, "max_tokens": 500}
    )
    return response.json()["choices"][0]["text"]

# 问题1：错误处理？重试逻辑？速率限制？
# 问题2：如何组合多个调用？如何传递上下文？
# 问题3：如何追踪调试？如何衡量性能？
# 问题4：如何在不同提供商之间切换？

每个项目都在重复解决相同的问题：重试逻辑、流式输出处理、Prompt 管理、上下文拼接。更关键的是，当应用复杂度增长时，代码会迅速变成一团意大利面条——各种回调嵌套、异常处理逻辑散落各处、不同 LLM 提供商的 API 差异让代码充满条件分支。

1.1.2 第二阶段：轻量封装时代

随后出现了各种轻量封装库，它们解决了最基本的痛点：统一不同 LLM 提供商的接口、提供重试机制、简化流式输出处理。但这些库大多是"功能的集合"而非"架构的表达"——它们提供了一堆工具函数，但缺乏一个统一的组合范式。

# 轻量封装的典型问题：缺乏统一的组合方式
result = llm.generate(
    prompt_template.format(
        context=retriever.search(query),
        question=query
    )
)
parsed = output_parser.parse(result)

# 每一步都是命令式的，流程硬编码在业务代码中
# 想要添加流式输出？需要重写整个流程
# 想要并行检索？需要手动管理线程池
# 想要添加追踪？需要在每个调用点插入日志

1.1.3 第三阶段：声明式框架时代

LangChain 代表了第三个阶段的到来。需要说明的是，这三个阶段是按理念划分的，并不严格对应时间先后：按 pin 树的 git 历史，LangChain 仓库的首个提交是 2022-10-24；Runnable 协议在 2023-07-26 的「Runnable single protocol」（#7800）中引入，首个包含它的 tag 是 v0.0.245（2023-07-27）；2023-11-20 的 #13577 把核心抽象拆成独立的 langchain-core 包；2025-10-04 的 #33256 把旧包改名为 langchain-classic，langchain-core 与新的 langchain 1.0.0 的 tag 都落在 2025-10-17。它的核心洞察是：LLM 应用本质上是一个数据处理管道，每个组件接收输入、产生输出，组件之间通过统一的接口协议连接。 这个洞察催生了两个关键创新：

  1. Runnable 协议：一个统一的接口，所有组件都实现 invoke/batch/stream 方法
  2. LCEL（LangChain Expression Language）：一种声明式的管道组合语法，用 | 操作符连接组件
# LangChain LCEL 的声明式写法
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 声明式地描述数据流
chain = (
    RunnableParallel(
        context=retriever,
        question=RunnablePassthrough(),
    )
    | prompt
    | model
    | StrOutputParser()
)

# 同步调用、异步调用、批量处理、流式输出——全部自动支持
chain.invoke("什么是 LangChain？")
await chain.ainvoke("什么是 LangChain？")
chain.batch(["问题1", "问题2", "问题3"])
for chunk in chain.stream("什么是 LangChain？"):
    print(chunk, end="")

这种声明式的方式带来了根本性的改变：开发者描述"数据如何流动"，框架负责"如何高效执行"。

graph TD
    A["裸 API 调用时代"] -->|痛点：重复代码| B["轻量封装时代"]
    B -->|痛点：缺乏组合范式| C["声明式框架时代"]
    A1["手写 HTTP 请求"] --> A
    A2["手动错误处理"] --> A
    A3["硬编码流程"] --> A
    B1["统一 API 封装"] --> B
    B2["基础重试机制"] --> B
    B3["仍然命令式"] --> B
    C1["Runnable 统一协议"] --> C
    C2["LCEL 声明式语法"] --> C
    C3["自动并行/流式/追踪"] --> C
    style A fill:#f9d5d5
    style B fill:#f9ecd5
    style C fill:#d5f9d5

1.2 LangChain 的定位与竞品对比

在深入 LangChain 的核心创新之前，有必要将它放在更广阔的 AI 应用框架版图中审视。当前市场上有多个值得关注的 LLM 应用框架：

框架 核心定位 组合范式
LangChain 通用 LLM 应用框架 LCEL 管道操作符（pin 快照 monorepo 内另有 15 个 partner 包）
LlamaIndex 官方自述为"在你的数据之上构建 LLM 驱动的 Agent"的框架 Query Engine；事件驱动的 Workflows
Haystack 用 LLM 构建搜索等应用的组件化框架 Pipeline：组件构成的有向多重图，允许环
Semantic Kernel 微软的轻量级 AI Agent 开发套件（C#/Python/Java） Kernel + Plugin
LangGraph 有状态 Agent 编排 图状态机

（其他框架一栏按各自官方文档归纳，核对于 2026-09-25；各框架都在快速演进，以对方最新文档为准。）

LangChain 的独特之处在于它选择了最通用的抽象层级。它不仅仅解决检索增强生成（RAG）或 Agent 编排这一个问题，而是提供了一套可以组合任意 LLM 操作的基础设施。这种通用性既是它最大的优势（适用场景极广），也是它最被诟病的地方（初学者容易迷失在抽象层中）。

LangGraph 是 LangChain 团队推出的另一个框架，专注于有状态的、图结构的 Agent 编排。它构建在 langchain-core 之上，复用了 Runnable 协议和 RunnableConfig 机制。这个关系本身就说明了 langchain-core 的抽象设计具有足够的通用性。

1.3 LangChain 的核心创新

1.3.1 LCEL：声明式表达语言

LCEL 并非一种独立的编程语言，而是利用 Python 的运算符重载（__or__ 和 __ror__）实现的领域特定语言（DSL）。当你写出 prompt | model | parser 时，Python 解释器实际上在构建一棵由 RunnableSequence 和 RunnableParallel 组成的执行树。

这种设计的精妙之处在于：它是惰性的。| 操作符不会触发任何实际的计算，它只是构建了一个描述数据流的对象图。真正的计算在调用 invoke/stream 时才会发生。这使得框架可以在执行前就拿到整条管道的结构——比如自动推断输入输出 schema、构建用于可视化的计算图。注意并行不会被自动识别：只有显式写成 RunnableParallel（或 dict）的分支才会并发执行。

# LCEL 的管道操作符本质（简化：省略了类型标注与 docstring）
# 源码文件：libs/core/langchain_core/runnables/base.py

class Runnable(ABC, Generic[Input, Output]):
    def __or__(self, other):
        """a | b 等价于 RunnableSequence(a, b)"""
        return RunnableSequence(self, coerce_to_runnable(other))

    def __ror__(self, other):
        """当左操作数不是 Runnable 时触发"""
        return RunnableSequence(coerce_to_runnable(other), self)

1.3.2 统一接口协议

LangChain 最深层的架构决策是：让所有组件——从简单的字符串处理函数到复杂的 LLM 调用——都遵循同一个接口协议。 这个协议定义在 Runnable 抽象基类中：

graph LR
    subgraph "Runnable 协议"
        direction TB
        I["invoke(input) -> output"]
        B["batch(inputs) -> outputs"]
        S["stream(input) -> Iterator"]
        AI["ainvoke(input) -> output"]
        AB["abatch(inputs) -> outputs"]
        AS["astream(input) -> AsyncIterator"]
    end
    P["PromptTemplate"] -.->|实现| I
    M["ChatModel"] -.->|实现| I
    R["Retriever"] -.->|实现| I
    OP["OutputParser"] -.->|实现| I
    L["RunnableLambda"] -.->|实现| I
    style I fill:#e1f5fe
    style B fill:#e1f5fe
    style S fill:#e1f5fe
    style AI fill:#e1f5fe
    style AB fill:#e1f5fe
    style AS fill:#e1f5fe

这个设计的优势在于组合的一致性。一个 RunnableSequence（由多个 Runnable 组成的管道）本身也是一个 Runnable，它自动拥有 invoke/batch/stream 的全部能力。这意味着你可以将一个复杂的 chain 作为另一个 chain 的组件——组合可以任意嵌套，每一层都对外暴露同一套接口。

1.3.3 自动化的能力推导

统一接口带来了一个深远的好处：能力的自动传播。当你用 | 组合两个 Runnable 时，框架会自动为组合体生成：

  • 同步执行能力（invoke）：依次调用每个步骤
  • 异步执行能力（ainvoke）：Runnable 的默认实现通过线程池执行同步版本，子类可覆写为原生异步（RunnableSequence 自己就覆写了，逐步 await 每个步骤的 ainvoke）
  • 批量处理能力（batch）：默认并行调用 invoke，使用线程池执行器
  • 流式输出能力（stream）：通过 transform 方法将前一步的流式输出喂给下一步
  • Schema 推断能力：自动从第一步推断输入类型，从最后一步推断输出类型
  • 可视化能力：自动生成 Mermaid 图表描述数据流

这意味着开发者用 LCEL 拼出的管道，不写额外代码就具备这些能力；各能力的实际效果仍取决于每个步骤的实现（例如某一步不支持流式，整条链的流式输出就会在那里被阻塞）。

graph TD
    subgraph "能力自动传播"
        C["chain = a | b | c"]
        C -->|自动获得| F1["chain.invoke()"]
        C -->|自动获得| F2["chain.ainvoke()"]
        C -->|自动获得| F3["chain.batch()"]
        C -->|自动获得| F4["chain.stream()"]
        C -->|自动获得| F5["chain.get_graph()"]
        C -->|自动获得| F6["chain.input_schema"]
    end
    subgraph "stream 的传播机制"
        S1["输入流"] --> ST1["a.transform()"]
        ST1 --> ST2["b.transform()"]
        ST2 --> ST3["c.transform()"]
        ST3 --> S2["输出流"]
    end
    style C fill:#fff3e0

1.4 为什么要读源码

1.4.1 文档的局限性

LangChain 的官方文档是以用例驱动的——它告诉你"如何使用"，但很少解释"为什么这样设计"以及"底层如何实现"。这在简单场景下不是问题，但当你遇到以下情况时，文档就力不从心了：

  • 性能调优：为什么我的 chain 比预期慢？batch 的默认并发度是多少？流式输出的缓冲策略是什么？
  • 调试困难：回调链（callbacks）是如何传播的？为什么我的自定义回调没有收到预期的事件？
  • 扩展定制：如何正确地子类化一个 Runnable？_call_with_config 和直接调用 invoke 的区别是什么？
  • 架构理解：langchain-core 和 langchain 的边界在哪里？为什么 Partner 包要独立出去？

1.4.2 源码揭示真实的设计权衡

每一行源码背后都是一个设计决策，每一个设计决策背后都是一组权衡。通过阅读源码，你会发现：

ensure_config 为什么要检查 ContextVar？ 因为 LangChain 使用 var_child_runnable_config 这个 ContextVar 来实现配置的自动传播——父 Runnable 的配置会自动继承给子 Runnable，无需显式传递。这个设计让 LCEL 的嵌套调用变得简洁，但也带来了隐式状态的复杂性。

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

var_child_runnable_config: ContextVar[RunnableConfig | None] = ContextVar(
    "child_runnable_config", default=None
)

def ensure_config(config: RunnableConfig | None = None) -> RunnableConfig:
    empty = RunnableConfig(tags=[], metadata={}, callbacks=None, ...)
    # 先从 ContextVar 继承父级配置
    if var_config := var_child_runnable_config.get():
        empty.update({k: v.copy() if k in COPIABLE_KEYS else v ...})
    # 再用显式传入的配置覆盖
    if config is not None:
        empty.update({k: v ...})
    return empty

coerce_to_runnable 为什么要区分生成器函数和普通函数？ 因为生成器函数天然支持流式输出（transform），将其包装为 RunnableGenerator 可以保留这个能力；而普通函数只能包装为 RunnableLambda，它在流式场景下必须先累积全部输入再产出输出。

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

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)       # 保留流式能力
    if callable(thing):
        return RunnableLambda(cast(..., thing))  # 无原生流式能力
    if isinstance(thing, dict):
        return RunnableParallel(thing)          # 字典 -> 并行执行
    raise TypeError(...)

1.4.3 源码是最好的学习材料

LangChain 的源码有几个特点使其非常适合阅读：

  1. 类型标注完善：几乎所有函数都有完整的类型标注，泛型参数使用清晰
  2. 文档字符串详尽：公开的类和方法大多有详细的 docstring，不少带使用示例
  3. 层次分明：langchain-core 约 6200 行的 base.py 虽然体量大，但类之间的继承关系清晰
  4. 设计模式丰富：策略模式、组合模式、装饰器模式等在源码中很常见

1.5 LangChain 的分层架构概览

在深入任何细节之前，我们先鸟瞰 LangChain 的整体分层结构。这将帮助你在后续章节中始终保持全局视角。

graph TB
    subgraph "应用层"
        APP["你的 AI 应用"]
    end
    subgraph "langchain-classic 1.0.3"
        LC_CHAINS["Chains / Agents"]
        LC_MEMORY["Memory"]
        LC_RETRIEVERS["Retrievers"]
        LC_LOADERS["Document Loaders"]
    end
    subgraph "Partner 包"
        P_OPENAI["langchain-openai"]
        P_ANTHROPIC["langchain-anthropic"]
        P_CHROMA["langchain-chroma"]
        P_MORE["langchain-ollama ..."]
    end
    subgraph "langchain-core 1.2.26"
        CORE_RUNNABLE["Runnable / LCEL"]
        CORE_MSG["Messages"]
        CORE_PROMPT["Prompts"]
        CORE_LM["Language Models"]
        CORE_PARSER["Output Parsers"]
        CORE_TOOLS["Tools"]
        CORE_CB["Callbacks / Tracers"]
        CORE_DOC["Documents"]
    end
    APP --> LC_CHAINS
    APP --> P_OPENAI
    APP --> CORE_RUNNABLE
    LC_CHAINS --> CORE_RUNNABLE
    LC_CHAINS --> CORE_MSG
    LC_MEMORY --> CORE_MSG
    LC_RETRIEVERS --> CORE_DOC
    P_OPENAI --> CORE_LM
    P_ANTHROPIC --> CORE_LM
    P_CHROMA --> CORE_DOC
    CORE_LM --> CORE_RUNNABLE
    CORE_PROMPT --> CORE_RUNNABLE
    CORE_PARSER --> CORE_RUNNABLE
    CORE_TOOLS --> CORE_RUNNABLE
    CORE_CB --> CORE_RUNNABLE
    style CORE_RUNNABLE fill:#4CAF50,color:#333
    style APP fill:#1976d2,color:#fff

这个架构有三层清晰的分界：

  • langchain-core：最底层，定义了所有核心抽象。Runnable 协议、消息类型、Prompt 模板、语言模型接口、输出解析器、工具接口、回调系统——这些都在这里。这一层的设计原则是"最小化依赖、最大化抽象"。
  • Partner 包：中间层的一部分，主流的 LLM 提供商、向量数据库等集成各有独立的包。它们实现 langchain-core 定义的接口，但彼此完全独立。这种设计避免了"安装 LangChain 就要拉取所有依赖"的问题。
  • langchain-classic（经典包，目录仍叫 libs/langchain/）：提供高级抽象，如 Chains、Agents、Memory。它组合 langchain-core 的基础组件，构建常见的应用模式。

图中没有画出的还有 1.0 之后的新 langchain 包（目录 libs/langchain_v1/，该快照为 1.2.15），它以基于 LangGraph 的 create_agent（libs/langchain_v1/langchain/agents/factory.py）为核心，本专栏只少量引用。

1.6 本专栏路线图

本专栏按照"从核心到外围、从抽象到具体"的路线展开，每一章都建立在前一章的基础之上。

基础篇（第 1-5 章）

  • 第 1 章（本章）：为什么需要理解 LangChain——建立学习动机和全局视角
  • 第 2 章 架构总览：langchain-core、langchain-classic 与 langchain 1.x、Partners 的关系，chain.invoke() 的完整旅程
  • 第 3 章 Runnable 与 LCEL：深入 Runnable 协议，理解 | 操作符、RunnableSequence、RunnableParallel 等核心组合原语
  • 第 4 章 消息系统：BaseMessage 家族、消息的序列化与类型安全
  • 第 5 章 语言模型抽象：BaseLLM、BaseChatModel 的设计，缓存与速率限制

进阶篇（第 6-11 章）

  • 第 6 章 Prompt 工程：模板系统的设计与实现，Few-shot 选择器
  • 第 7 章 输出解析器：从字符串到结构化数据的桥梁
  • 第 8 章 工具系统：BaseTool 的设计，工具调用协议
  • 第 9 章 文档加载与文本分割：Document 抽象、文档加载器与切分策略
  • 第 10 章 向量存储与检索器：VectorStore 与 Retriever 的实现与优化策略
  • 第 11 章 Chains：经典 Chain 的设计与 LCEL 的关系

高级篇（第 12-18 章）

  • 第 12 章 回调与可观测性：Callback 系统的观察者模式实现，LangSmith 集成
  • 第 13 章 记忆与会话管理：对话记忆的多种策略
  • 第 14 章 Agent 架构：从 ReAct 到工具调用 Agent 的演进
  • 第 15 章 工具调用 Agent：现代 Agent 的实现细节
  • 第 16 章 序列化：Serializable 体系，JSON 序列化与反序列化
  • 第 17 章 Partner 集成架构：如何构建 LangChain 集成包
  • 第 18 章 设计模式总结：贯穿 LangChain 的架构智慧
graph LR
    subgraph "基础篇"
        C1["1. 为什么"] --> C2["2. 架构"]
        C2 --> C3["3. Runnable"]
        C3 --> C4["4. 消息"]
        C4 --> C5["5. 模型"]
    end
    subgraph "进阶篇"
        C5 --> C6["6. Prompt"]
        C6 --> C7["7. 解析器"]
        C7 --> C8["8. 工具"]
        C8 --> C9["9. 文档"]
        C9 --> C10["10. 检索"]
        C10 --> C11["11. Chains"]
    end
    subgraph "高级篇"
        C11 --> C12["12. 回调"]
        C12 --> C13["13. Memory"]
        C13 --> C14["14. Agent"]
        C14 --> C15["15. 工具Agent"]
        C15 --> C16["16. 序列化"]
        C16 --> C17["17. Partner"]
        C17 --> C18["18. 总结"]
    end
    style C1 fill:#e3f2fd
    style C2 fill:#e3f2fd
    style C3 fill:#e3f2fd
    style C4 fill:#e3f2fd
    style C5 fill:#e3f2fd

1.7 阅读源码的方法论

在开始阅读 LangChain 源码之前，这里提供几个实用的方法论建议。

1.7.1 从 base.py 入手

LangChain-core 的 libs/core/langchain_core/runnables/base.py 是整个框架的心脏，约 6200 行代码。不要被它的体量吓到。这个文件中的类（外加定义在 branch.py 的 RunnableBranch）遵循清晰的层次：

Runnable (ABC)              # 最底层的抽象，定义协议
  |
  +-- RunnableSerializable  # 加入序列化能力
  |     |
  |     +-- RunnableSequence    # 管道组合
  |     +-- RunnableParallel    # 并行组合
  |     +-- RunnableBranch      # 条件分支（branch.py）
  |     +-- RunnableBindingBase # 参数绑定
  |
  +-- RunnableLambda         # 函数包装
  +-- RunnableGenerator      # 生成器包装

建议从 Runnable.__or__ 开始，跟踪 | 操作符如何创建 RunnableSequence；然后阅读 RunnableSequence.invoke，理解管道的执行流程。

1.7.2 跟踪一次完整调用

理解 LangChain 最有效的方式是跟踪一次完整的 chain.invoke() 调用。从用户的 chain.invoke(input) 开始，观察数据如何流经：

  1. ensure_config 初始化配置（合并 ContextVar 中的父级配置）
  2. CallbackManager 启动追踪（触发 on_chain_start 事件）
  3. 数据依次流过每个 step.invoke（每一步都 patch_config 传递子回调）
  4. 最终结果通过 on_chain_end 报告给追踪系统

1.7.3 关注 config 的流转

RunnableConfig 是 LangChain 的"血液循环系统"。几乎每一个方法都接收一个可选的 config 参数。理解 config 如何在组件之间传递、合并、转换，是理解整个框架的关键。

# RunnableConfig 的核心字段
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   # 唯一运行标识

1.7.4 使用调试工具辅助阅读

LangChain 内置了丰富的调试和可视化能力，这些工具在阅读源码时是极好的辅助：

from langchain_core.globals import set_debug

# 开启全局调试，所有 Runnable 调用都会打印详细的输入/输出
set_debug(True)

# 可视化 chain 的结构（print_ascii 需另装 grandalf，否则抛 ImportError）
chain.get_graph().print_ascii()

# 生成 Mermaid 图
print(chain.get_graph().draw_mermaid())

# 检查 chain 的输入/输出 schema
print(chain.input_schema.model_json_schema())
print(chain.output_schema.model_json_schema())

# 使用 ConsoleCallbackHandler 追踪执行过程
from langchain_core.tracers import ConsoleCallbackHandler
chain.invoke("什么是 LangChain？", config={"callbacks": [ConsoleCallbackHandler()]})

这些工具不仅在生产调试中有用，在阅读源码时也能帮助你快速验证你对某段代码行为的理解是否正确。

1.7.5 推荐的阅读顺序

基于我们对 LangChain 源码的分析经验，推荐以下阅读顺序：

  1. libs/core/langchain_core/runnables/config.py（641 行）：先理解 RunnableConfig 的结构和 ensure_config/patch_config 的逻辑
  2. libs/core/langchain_core/runnables/base.py 中的 Runnable 类（base.py:124-2585，约 2460 行）：理解 invoke/batch/stream 的默认实现和 __or__/__ror__ 操作符
  3. libs/core/langchain_core/runnables/base.py 中的 RunnableSequence（base.py:2817-3564，约 750 行）：理解管道的核心执行逻辑
  4. libs/core/langchain_core/runnables/base.py 中的 RunnableLambda（libs/core/langchain_core/runnables/base.py:4399-5271，约 870 行）：理解函数包装和类型推断
  5. libs/core/langchain_core/callbacks/manager.py：理解回调系统如何驱动追踪
  6. libs/core/langchain_core/language_models/chat_models.py：理解 LLM 如何实现 Runnable 协议

1.8 设计决策：为什么 LangChain 选择了这条路

为什么选择运算符重载而非方法链？——LangChain 主推用 | 来组合 Runnable，但方法链并没有缺席：Runnable.pipe(*others)（base.py:660）就是方法形式，docstring 写明它等价于 RunnableSequence(self, *others) 或 self | others[0] | ...。两者背后都是 RunnableSequence，| 只是更短的写法；它在视觉上暗示了"管道"的概念，与 Unix 管道和函数式编程中的管道操作符（|>）形成呼应。

为什么不用 DAG 框架如 Airflow？——LangChain 的目标场景与 Airflow 等 DAG 框架有根本区别。Airflow 针对的是长时间运行的批处理任务，强调调度和持久化；LangChain 针对的是实时交互，需要低延迟和实时的流式输出。此外，LLM 应用的数据流结构通常是在运行时动态确定的（例如 Agent 根据模型输出决定下一步调用什么工具），这与 Airflow 的静态 DAG 定义截然不同。

为什么 langchain-core 要独立出来？——这是一个关于依赖管理的决策。如果所有代码都在一个包中，那么安装 langchain-openai 就意味着安装 langchain 的全部依赖。通过将核心抽象独立为 langchain-core（依赖极少），Partner 包只需依赖 langchain-core 而非整个 langchain。这减少了依赖冲突的可能性。

1.9 一个完整的 LCEL 示例：从概念到源码

在结束本章之前，让我们通过一个完整的示例，将前面讨论的所有概念串联起来。这个示例展示了一个典型的 RAG（Retrieval-Augmented Generation）管道：

from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 假设 retriever 是一个已经配置好的向量检索器
prompt = ChatPromptTemplate.from_template(
    "根据以下上下文回答问题：\n\n上下文：{context}\n\n问题：{question}"
)
model = ChatOpenAI(model="gpt-4o-mini")

# LCEL 声明式管道
chain = (
    RunnableParallel(
        context=retriever,
        question=RunnablePassthrough(),
    )
    | prompt
    | model
    | StrOutputParser()
)

让我们逐行分析这段代码在源码层面发生了什么（对应 libs/core/langchain_core/runnables/base.py）：

第一行 RunnableParallel(...)：创建了一个并行执行体。retriever 如果不是 Runnable，会被 coerce_to_runnable 转换。RunnablePassthrough() 创建了一个透传原始输入的 Runnable。

第二行 | prompt：触发 Runnable.__or__（base.py:618）。实现只有一行：

# base.py:637
return RunnableSequence(self, coerce_to_runnable(other))

coerce_to_runnable 会把裸函数、dict、list 等按规则转成 Runnable——所以 | lambda x: ... 也能用。这一步产出的是一个两元素的 RunnableSequence(parallel, prompt)。

第三行 | model：触发 RunnableSequence.__or__（base.py:3077，带 @override 装饰器）。它不是直接再包一层，而是做一次"扁平合并"：

# base.py:3095
return RunnableSequence(
    self.first,
    *self.middle,
    self.last,
    coerce_to_runnable(other),
    name=self.name,
)

无论左边积累了多少步骤，新步骤都被展平拼到同一层，避免嵌套产生 Sequence(Sequence(Sequence(...))) 这种难以调试的套娃。如果右边 other 本身也是一个 RunnableSequence，base.py:3085-3094 会把两边的 first/middle/last 都拆开合并——这让 a | b | (c | d) 这种风格的表达式也得到平坦结果。

第四行 | StrOutputParser()：同理再走一次 RunnableSequence.__or__，StrOutputParser 被追加到步骤列表。最终 chain 是一个 4 元素的 RunnableSequence（first=parallel，middle=[prompt, model]，last=parser）。

为什么是 first / middle / last 三字段，不是一个列表？

读到这里有个很自然的疑问：既然是"步骤序列"，最直接的存法是 steps: list[Runnable]——为什么 RunnableSequence 要拆成三个字段？libs/core/langchain_core/runnables/base.py:2901 的注释直接给了答案：

The steps are broken into first, middle and last, solely for type checking purposes. It allows specifying the Input on the first type, the Output of the last type.

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

这是让泛型系统表达"链首输入 = 整体输入、链尾输出 = 整体输出"的直接办法。如果用 list[Runnable[Any, Any]]，静态类型检查器就丢失了"第一个元素的 Input 和整条链的 Input 相关联"这个信息——使用者拿到 RunnableSequence[Input, Output] 推导不出来 invoke 接受什么、返回什么。拆成 first: Runnable[Input, Any] 和 last: Runnable[Any, Output] 让 Input 和 Output 两个类型变量绑在具体字段上，类型推断自然工作。

middle: list[Runnable[Any, Any]] 里的类型是 Any——中间步骤的类型检查被故意放弃了。这是妥协：Python 的类型系统没有"异构列表的变长类型链"概念（TypeScript 的 tuple 泛型 [A, B, C, D] 可以做到，Python 还不行），所以中间步骤只能统一降为 Any。这会让工具在识别 chain.middle[1] 的 I/O 类型时失效——是个已知的表达力限制，不是 bug。

源码里还有个 steps 属性（libs/core/langchain_core/runnables/base.py:2963），把三字段重新拼回一个列表方便运行时遍历：

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

所以"静态类型看三字段、运行时遍历看 steps"——两种视角共存。这个拆分背后是 Python 类型系统当前能力边界的一个现实折中，也是理解 LangChain 源码里"形式看似啰嗦的结构其实都是类型系统逼出来的"这条规律的绝佳入门案例。

当调用 chain.invoke("什么是向量数据库？") 时：

sequenceDiagram
    participant U as 用户
    participant S as RunnableSequence
    participant P as RunnableParallel
    participant R as Retriever
    participant PT as RunnablePassthrough
    participant PR as ChatPromptTemplate
    participant M as ChatOpenAI
    participant OP as StrOutputParser
    U->>S: invoke("什么是向量数据库？")
    S->>P: invoke("什么是向量数据库？")
    par 并行执行
        P->>R: invoke("什么是向量数据库？")
        R-->>P: [doc1, doc2, doc3]
    and
        P->>PT: invoke("什么是向量数据库？")
        PT-->>P: "什么是向量数据库？"
    end
    P-->>S: {context: [doc1, doc2, doc3], question: "什么是向量数据库？"}
    S->>PR: invoke({context: [...], question: "..."})
    PR-->>S: ChatPromptValue(messages=[...])
    S->>M: invoke(ChatPromptValue)
    M-->>S: AIMessage(content="向量数据库是...")
    S->>OP: invoke(AIMessage)
    OP-->>S: "向量数据库是..."
    S-->>U: "向量数据库是..."

这个例子展示了 LangChain 的核心价值：用一段声明式代码，构建了一个具备并行检索、Prompt 格式化、LLM 调用、输出解析，并天然支持回调追踪、异步、批量处理和流式输出的管道。

1.9.1 实测：LangChain monorepo 规模——7 个包共 20 万行 Python

把 §1.5 提到的"分层架构"对应到基准快照（langchain-core==1.2.26，commit 0a1d290a）的实测。统计口径：只数各包自己的 Python 包目录（libs/core/langchain_core/、libs/langchain/langchain_classic/、libs/partners/*/langchain_*/ 等），不含 tests/，也不含 scripts/ 下的 CI 辅助脚本——即"随 wheel 发出去的库代码"。本专栏的子模块行数（runnables/ 13730、prompts/ 4359……）都用同一把尺子，可以直接相加。复现命令：

# 单个包
find libs/core/langchain_core -name '*.py' | xargs cat | wc -l   # 64050
find libs/core/langchain_core -name '*.py' | wc -l               # 176 个文件
# partners 全体
find libs/partners/*/langchain_* -name '*.py' | xargs cat | wc -l  # 42379

（换成"libs/<pkg>/ 下排除 tests/"这把尺子，各包会各自多出几十行 CI 脚手架——core 多出的 89 行来自 langchain-core-1.2.26/libs/core/scripts/check_imports.py 与 langchain-core-1.2.26/libs/core/scripts/check_version.py，classic 多出的 33 行来自 langchain-core-1.2.26/libs/langchain/scripts/check_imports.py——于是 core 64139、classic 70682、partners 42715。本专栏一律不采用这个口径。）

包 包目录 库代码行 角色
langchain-core libs/core/langchain_core 64050（176 个 .py） 核心抽象——本专栏主角；包含 Runnable/Message/Prompt/Tool/Tracer 等所有基础协议
langchain-classic libs/langchain/langchain_classic 70649（1321 个 .py） 经典层（包目录是 langchain_classic）——legacy Chain / Agent / Memory 等被 LCEL 替代的旧抽象
langchain libs/langchain_v1/langchain 12384 1.0 重构后的新版本入口（该快照为 1.2.15）
partners/ libs/partners//langchain_ 42379 15 个集成包，含模型供应商与 chroma / qdrant 等向量库、exa 搜索（anthropic / openai / huggingface / chroma / deepseek / fireworks / groq / mistralai / nomic / ollama / openrouter / perplexity / qdrant / xai / exa）
text-splitters libs/text-splitters/langchain_text_splitters 3531 ch09 §9.11.3 已实测的 13 文件
standard-tests libs/standard-tests/langchain_tests 7726 让第三方 partner 包跑同一套测试合约
model-profiles libs/model-profiles/langchain_model_profiles 403 各模型的能力描述（context window / supports tools 等元数据）
合计 — 201,122 —

langchain-core 内部大小分布——

子模块 行 角色
runnables/ 13730 LCEL 全部组合原语（§3.12.1 实测）
messages/ 9282 BaseMessage 家族（System/Human/AI/Tool/Function 等）
tracers/ 5057 可观测性（ch12 §12.9.5）
language_models/ 4923 BaseLLM / BaseChatModel ABC + streaming + tokenizer 抽象
callbacks/ 4677 Mixin + Manager + RunManager（ch12 §12.9.5）
prompts/ 4359 PromptTemplate / ChatPromptTemplate / FewShotPromptTemplate
tools/ 2793 BaseTool / @tool 装饰器（ch08 §8.5.4）
output_parsers/ 2227 StrOutputParser / JsonOutputParser 等（ch07 §7.12.1）
documents/ 555 Document / Blob 数据类（ch09 §9.1、§9.2）
其余 余下 utils / load / vectorstores / embeddings / outputs / indexing 等

两条值得记住的物理事实——

  1. langchain_classic (70649) > langchain_core (64050)——传统印象里"core 是核心、langchain 是壳"——实测经典包反而更大——因为 langchain_classic 里塞满了 legacy Chain / Agent / Memory / Hub / Cache，外加成批的 deprecation 重定向 shim（ch13 §13.11.4 实测的 memory/chat_message_histories/ 二十来个二十余行的重定向文件就在这）—— legacy 代码量比新核心还重：框架的演进必须背包袱。
  2. partners/ 15 个集成包 42379 行——平均每个包 ~2825 行——是 §1.3.2 "统一接口协议"的物理代价：每个 partner 包把厂商特定的 SDK / API 字段 / 流式协议 / function calling 格式翻译成 LangChain 的 BaseChatModel 接口；其中 openai 最大（10822 行）、anthropic（5564）与 openrouter（4951）次之，nomic（154）和 exa（386）最小——M+N 架构里，N 端的集成代码总量已达 core 的三分之二左右（42379 对 64050），而这还只是 monorepo 内的 15 个包。

1.9.2 版本边界：本专栏为什么先读 core 再读 classic

本专栏的基准快照是 langchain-core==1.2.26 这个 tag（commit 0a1d290a），版本边界可以直接从 pyproject.toml 验证：libs/core/pyproject.toml:6 的包名是 langchain-core，libs/core/pyproject.toml:24 的版本是 1.2.26；libs/langchain/pyproject.toml:6 的包名是 langchain-classic，libs/langchain/pyproject.toml:23 的版本是 1.0.3，并在 libs/langchain/pyproject.toml:26 依赖 langchain-core>=1.2.19,<2.0.0。

这组版本关系决定了阅读顺序：

层 先读还是后读 原因
langchain-core 先读 Runnable、Message、Tool、Callback、OutputParser 都在这里定义协议
Partner 包 中间穿插读 它们把厂商 API 翻译成 core 协议，是 M+N 的 N 端
langchain-classic 后读 这里保留 Chain、Agent、Memory 等历史抽象，很多代码是兼容层

如果反过来从 classic 的旧 Chain / Agent 入手，读者很容易把兼容包袱当成 LangChain 的核心设计。正确路径是先把 core 的协议读清楚，再看 classic 如何迁移和包裹这些协议。后续第 3 章讲 Runnable，第 8 章讲 Tool，第 12 章讲 Callback，都是 core 层；第 13、15 章再进入 classic 的 Memory 和 Agent，才能看清哪些是新架构，哪些是历史债务。

这也是判断框架质量的基本方法：不要先看示例多不多，而要看底层协议是否稳定、边界是否清楚、历史兼容是否被隔离。协议稳定，生态才能扩展；边界清楚，问题才能定位；兼容层隔离，旧代码才不会拖垮新架构。

这条阅读顺序会贯穿整个专栏，也会避免把历史兼容误读成核心设计。

1.10 小结

本章建立了阅读本专栏的动机和全局视角。我们了解了 AI 应用框架从裸 API 调用到声明式表达语言的演进历程，理解了 LangChain 的两个核心创新——Runnable 统一协议和 LCEL 表达式语言——如何从根本上改变了 LLM 应用的开发方式。

更重要的是，我们明确了为什么要读源码：官方文档告诉你"怎么用"，但只有源码能告诉你"为什么这样设计"以及"底层如何实现"。当你遇到性能问题、调试困难或需要深度定制时，源码级别的理解将是你最强大的武器。

物理事实：LangChain monorepo 7 个包共 201122 行库代码（口径见 §1.9.1）——langchain_core 64050（runnables 13730 + messages 9282 + tracers 5057 + callbacks 4677 等）、langchain_classic 70649（含 legacy 反而比 core 大，印证"框架演进必须背包袱"）、partners 15 个集成包 42379（平均 ~2825 行/包，M+N 的 N 端集成代码已达 core 的三分之二左右）。