Runnable是 LangChainLCEL 管道、组件编排的底层抽象协议（抽象基类），所有能通过|串联、统一调用的组件（提示词、大模型、解析器、工具、自定义逻辑）都必须遵守该协议。它抹平了不同组件的实现差异，让整套框架拥有统一的调用、组合、运行能力。
一、协议核心定位
- 统一接口标准：规定所有组件对外暴露完全一致的方法，调用方式无差别。
- 管道串联基础：重载 Python
__or__运算符，支撑A | B | C管道语法。 - 统一数据流转：约定输入、输出数据格式，保证组件拼接后数据自动传递。
- 多运行模式兼容：一套组件同时支持单次调用、流式输出、批量执行。
- 可观测与扩展：内置配置、回调、链路追踪能力，适配生产环境运维。
二、协议整体架构
1. 继承关系
plaintext
BaseRunnable（顶层抽象基类） ├─ RunnableSequence 串行组合（A|B|C） ├─ RunnableMap 并行组合（多分支） ├─ RunnableLambda 普通函数适配器 ├─ RunnablePassthrough 数据透传/字段重组 └─ 业务组件（Prompt / ChatModel / OutputParser / Tool 等）所有子类都必须实现BaseRunnable定义的抽象方法，才算合规的Runnable组件。
2. 协议核心组成
分为四大模块：
- 标准执行方法（对外调用入口）
- 组合运算符（管道串联能力）
- 数据转换 / 适配工具（字段、格式处理）
- 配置与回调（监控、追踪、参数绑定）
三、核心一：标准执行方法（必实现）
这是Runnable协议最核心的部分，所有组件强制实现，对外提供三套统一调用入口。
1.invoke(input, config=None)
单次同步调用，完整执行后一次性返回结果。
- 用途：常规问答、接口请求、单次任务处理
- 入参：通用输入（大多为字典
dict）+ 可选运行配置 - 出参：组件处理后的结果
伪代码逻辑
python
运行
def invoke(self, input_data, config=None): # 组件自身业务逻辑 return result示例（通用调用）
python
运行
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(temperature=0) prompt = ChatPromptTemplate.from_messages([("user", "你好：{name}")]) chain = prompt | llm | StrOutputParser() # 统一 invoke 调用 res = chain.invoke({"name": "小明"}) print(res)2.stream(input, config=None)
流式迭代输出，按片段逐一生成并返回，返回迭代器。
- 用途：对话打字机效果、长文本实时输出、前端流式交互
- 特点：链路中每个组件都需支持流式，数据分片逐级传递
调用示例
python
运行
# 流式调用，逐块打印 for chunk in chain.stream({"name": "小明"}): print(chunk, end="")3.batch(inputs, config=None)
批量执行，接收一组输入列表，批量并行处理并返回结果列表。
- 用途：离线批量推理、文档批量处理、数据批任务
- 特点：框架内部做调度，无需手动循环
调用示例
python
运行
batch_input = [{"name": "张三"}, {"name": "李四"}] batch_res = chain.batch(batch_input) print(batch_res)关键结论：同一条 Runnable 链路，三套调用方法完全复用，无需修改代码。
四、核心二：组合运算符（管道串联能力）
协议重载 Python 魔法方法，实现组件拼接，是 LCEL|语法的底层。
1.__or__(self, other)重载|运算符
语法：r1 | r2
- 触发逻辑：
r1.__or__(r2) - 返回值：新的
RunnableSequence串行组合对象 - 执行规则：运行时
r1 输出 → r2 输入，数据从左向右流转
极简底层模拟
python
运行
class BaseRunnable: def __or__(self, other): # 两个 Runnable 组合为串行链 return RunnableSequence([self, other]) class RunnableSequence(BaseRunnable): def __init__(self, runnables): self.runnables = runnables def invoke(self, input_data): current = input_data # 从左到右依次执行 for run in self.runnables: current = run.invoke(current) return current2.__ror__(self, other)反向兼容
当左侧对象不是Runnable、只有右侧是Runnable时，Python 会调用__ror__，保证拼接不报错，提升兼容性。
3. 其他组合能力（拓展）
协议还支持pipe()方法，等价于|，语义更直观：
python
运行
# 两种写法完全等价 chain1 = r1 | r2 | r3 chain2 = r1.pipe(r2).pipe(r3)五、核心三：内置辅助 Runnable（协议配套工具）
LangChain 基于Runnable协议实现了一批通用工具组件，用于数据适配、分支、透传，是工程开发高频用法。
1.RunnableLambda函数适配器
- 作用：将普通 Python 函数包装为标准
Runnable，接入管道。 - 适用：自定义数据处理、格式转换、接口调用等逻辑。
python
运行
from langchain_core.runnables import RunnableLambda def add_prefix(text: str) -> str: return "【回答】" + text # 包装为 Runnable run_func = RunnableLambda(add_prefix) # 正常参与管道串联 chain = prompt | llm | StrOutputParser() | run_func2.RunnablePassthrough数据透传
- 作用：原样传递输入数据，配合
.assign()新增字段，实现多字段共存。 - 场景：保留原文 + 新增摘要、关键词等派生数据。
python
运行
from langchain_core.runnables import RunnablePassthrough chain = ( RunnablePassthrough.assign( summary = prompt | llm | StrOutputParser() ) ) # 结果同时包含原 content 和新 summary res = chain.invoke({"content": "Runnable 协议讲解"})3.RunnableMap并行分支
- 作用：同一输入分发到多个子链路并行执行，最后汇总结果。
- 场景：同时做总结、关键词、情感分析。
python
运行
from langchain_core.runnables import RunnableMap chain1 = prompt1 | llm | StrOutputParser() chain2 = prompt2 | llm | StrOutputParser() parallel_chain = RunnableMap({ "总结": chain1, "关键词": chain2 })六、核心四：配置与扩展能力（工程化必备）
Runnable协议提供统一配置接口，用于参数绑定、链路标记、回调、追踪，对接 LangSmith 监控。
1.bind(**kwargs)预绑定参数
提前给组件固定参数，调用时无需重复传入。
python
运行
# 固定温度、超时等参数 fixed_llm = llm.bind(temperature=0.1, timeout=10) chain = prompt | fixed_llm | StrOutputParser()2.with_config(**config)附加运行配置
添加链路名称、标签、回调、追踪配置，用于日志与监控。
python
运行
chain = prompt | llm | StrOutputParser() # 标记链路名称与标签，LangSmith 可识别 chain_trace = chain.with_config(run_name="问答链", tags=["demo", "lcel"]) chain_trace.invoke({"name": "测试"})3. 回调 Callbacks
协议统一挂载回调函数，可在组件执行前 / 后 / 异常触发自定义逻辑（日志、埋点、告警）。
七、数据流转规范（协议约定）
为保证不同组件无缝拼接，Runnable统一约定主流数据格式流转规则：
- 链路入口：标准 Python 字典
{key: value}（匹配 Prompt 变量） - Prompt 组件：字典 → 对话消息列表
List[BaseMessage] - LLM 组件：消息列表 →
AIMessage模型回复对象 - 解析器：
AIMessage→ 字符串 / 字典 / Pydantic 对象 - 自定义组件：按需转换格式，不兼容时用
RunnableLambda适配
格式不匹配 = 管道报错，这是调试 LCEL 的常见切入点。
八、自定义实现一个合规 Runnable（手写协议）
基于抽象基类，实现一个自定义组件，直观理解协议要求：
python
运行
from langchain_core.runnables import Runnable from typing import Iterator, List # 自定义 Runnable 组件：文本转大写 class UpperRunnable(Runnable[str, str]): # 实现单次调用 def invoke(self, input: str, config=None) -> str: return input.upper() # 实现流式调用 def stream(self, input: str, config=None) -> Iterator[str]: for char in input.upper(): yield char # 实现批量调用 def batch(self, inputs: List[str], config=None) -> List[str]: return [s.upper() for s in inputs] # 使用管道串联 run_upper = UpperRunnable() chain = prompt | llm | StrOutputParser() | run_upper res = chain.invoke({"name": "test"}) print(res)只要实现invoke/stream/batch，就是合法Runnable，支持|串联。
九、Runnable 协议价值总结
1. 对开发者
- 学习成本低：一套接口学一次，所有组件用法一致；
- 代码极简：
|管道直观表达流程，告别老式 Chain 冗余写法； - 灵活扩展：普通函数、第三方逻辑通过适配器快速接入。
2. 对框架本身
- 解耦：组件之间只依赖协议，不依赖具体实现，可单独替换；
- 统一生态：Prompt、LLM、工具、Agent 全部基于同一协议，生态打通；
- 工程化：原生支持流式、批量、监控、回调，直接适配线上项目。
3. 与传统 LLMChain 对比
| 特性 | 传统 LLMChain | 基于 Runnable 协议的 LCEL | |
|---|---|---|---|
| 接口统一性 | 弱，各组件调用方式不一 | 强，invoke/stream/batch全局统一 | |
| 组合能力 | 手动传参，组合繁琐 | 原生 ` | `、并行、嵌套，自由组合 |
| 流式 / 批量 | 额外开发 | 原生支持 | |
| 扩展自定义逻辑 | 侵入性高 | RunnableLambda轻量适配 | |
| 可观测性 | 薄弱 | 原生配置、回调、链路追踪 |
十、速记要点（面试 / 复习）
- 本质：
Runnable是 LangChain 抽象基类 + 统一组件协议，是 LCEL 的底层基石。 - 三大核心方法：
invoke(单次)、stream(流式)、batch(批量)。 - 管道原理：重载
__or__方法，|生成串行RunnableSequence，数据左→右流转。 - 常用配套组件：
RunnableLambda：包装普通函数RunnablePassthrough：字段透传、新增字段RunnableMap：并行多分支
- 工程能力：
bind绑定参数、with_config链路监控、统一回调。 - 核心思想：接口标准化、组合自由化、运行模式统一化。
