当前位置： 首页 > news >正文

[智能体-188]：LangChain Runnable 统一协议 详解

news 2026/6/1 10:32:28

Runnable是 LangChainLCEL 管道、组件编排的底层抽象协议（抽象基类），所有能通过|串联、统一调用的组件（提示词、大模型、解析器、工具、自定义逻辑）都必须遵守该协议。它抹平了不同组件的实现差异，让整套框架拥有统一的调用、组合、运行能力。

一、协议核心定位

  1. 统一接口标准：规定所有组件对外暴露完全一致的方法，调用方式无差别。
  2. 管道串联基础：重载 Python__or__运算符，支撑A | B | C管道语法。
  3. 统一数据流转：约定输入、输出数据格式，保证组件拼接后数据自动传递。
  4. 多运行模式兼容：一套组件同时支持单次调用、流式输出、批量执行。
  5. 可观测与扩展：内置配置、回调、链路追踪能力，适配生产环境运维。

二、协议整体架构

1. 继承关系

plaintext

BaseRunnable（顶层抽象基类） ├─ RunnableSequence 串行组合（A|B|C） ├─ RunnableMap 并行组合（多分支） ├─ RunnableLambda 普通函数适配器 ├─ RunnablePassthrough 数据透传/字段重组 └─ 业务组件（Prompt / ChatModel / OutputParser / Tool 等）

所有子类都必须实现BaseRunnable定义的抽象方法，才算合规的Runnable组件。

2. 协议核心组成

分为四大模块：

  1. 标准执行方法（对外调用入口）
  2. 组合运算符（管道串联能力）
  3. 数据转换 / 适配工具（字段、格式处理）
  4. 配置与回调（监控、追踪、参数绑定）

三、核心一：标准执行方法（必实现）

这是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 current

2.__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_func

2.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统一约定主流数据格式流转规则：

  1. 链路入口：标准 Python 字典{key: value}（匹配 Prompt 变量）
  2. Prompt 组件：字典 → 对话消息列表List[BaseMessage]
  3. LLM 组件：消息列表 →AIMessage模型回复对象
  4. 解析器：AIMessage→ 字符串 / 字典 / Pydantic 对象
  5. 自定义组件：按需转换格式，不兼容时用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轻量适配
可观测性薄弱原生配置、回调、链路追踪

十、速记要点（面试 / 复习）

  1. 本质：Runnable是 LangChain 抽象基类 + 统一组件协议，是 LCEL 的底层基石。
  2. 三大核心方法：invoke(单次)、stream(流式)、batch(批量)。
  3. 管道原理：重载__or__方法，|生成串行RunnableSequence，数据左→右流转。
  4. 常用配套组件：
    • RunnableLambda：包装普通函数
    • RunnablePassthrough：字段透传、新增字段
    • RunnableMap：并行多分支
  5. 工程能力：bind绑定参数、with_config链路监控、统一回调。
  6. 核心思想：接口标准化、组合自由化、运行模式统一化。
http://www.jsqmd.com/news/928605/

相关文章：

  • 从YOLOv5到v8，我如何用PySide6给表情识别系统加个‘脸’？附完整代码与避坑指南
  • 2026年龙虾安全防护平台哪家好？企业级OpenClaw安全管理平台推荐与选型指南 - 品牌2025
  • 武商一卡通回收全攻略：轻松掌握回收技巧与注意事项 - 团团收购物卡回收
  • Qt 高级开发 021：零基础吃透 QVBoxLayout 垂直布局
  • 保姆级教程：用PHPStudy本地环境+Office 2021/2019完美配置MathType（避坑指南）
  • Kafka日志目录(Log Dirs)故障深度解析：从ERROR Shutdown broker到数据安全清理的最佳实践
  • SwanLab vs. TensorBoard/WB：轻量级实验看板的远程监控方案对比与选型指南
  • 彻底搞懂 DHCP：从原理机制到跨网段部署的终极实战指南（附故障排查与避坑手册）
  • 广州黄金回收哪家靠谱？2026各区正规门店地址电话汇总（可免费上门） - 行行星
  • 2018年2月科技复盘：AI产业化、云战争与数据觉醒的转折点
  • 架构腐化：代码是怎么从“小甜甜“变成“牛夫人“的
  • 全国上门名包名表服务机构盘点 按需选择适配方案 - 互联网科技品牌测评
  • 铜川卖金怕被坑？余生黄金回收2026年5月上门回收全攻略来了 - 余生黄金回收
  • 学生信息管理前端页面套件（含成绩图表、响应式个人页与欢迎动画）
  • 星载SAR实测与仿真数据的MATLAB线性调频变标（CS）成像完整实现包
  • 告别双系统！在Ubuntu 22.04上用Katoolin一键安装Kali渗透工具包（附常见问题解决）
  • 2026年哪些安全厂商能做龙虾安全检测？智能体数据安全与防泄露平台推荐 - 品牌2025
  • AI、5G与安全如何重塑移动应用开发：技术融合与实践指南
  • 惠州黄金回收实测：六家机构上门测评与避坑全记录 - 上门黄金回收
  • 衢州黄金回收市场简报：区域需求分化与六大回收机构服务解析 - 上门黄金回收
  • Freepbx搭建内网电话后，如何用软电话（如Zoiper）注册分机并实现互拨？
  • 保姆级教程：在VMware ESXi上从零安装OPNsense防火墙（含网卡避坑指南）
  • 太原黄金回收市场简报：各区域需求分化明显，六大机构实况对比 - 黄金上门回收
  • 如何深度集成 GPT 到 Zotero：5个实用配置技巧提升学术研究效率
  • 广东顺翼机械科技有限公司：以精密涂布技术引领行业，打造靠谱涂布机厂家 - 变量人生001
  • 人类与AGI认知能力对比：从学习推理到社会智能的深度剖析
  • MATLAB版M/N逻辑航迹起始实现：含50与100阈值对比可视化
  • AI建站工具全流程攻略：从零到一搭建企业官网的保姆级指南
  • 免安装QT翻译工具：填百度密钥就能批量译TXT/CSV，结果原格式保存
  • Windows窗口置顶解决方案：AlwaysOnTop 深度解析与实战指南