LangChain Runnable 链式调用深度解析从itemgetter到RunnableLambda本文基于 LangChain 框架深入解析 Runnable 链式调用中的核心机制重点剖析itemgetter、|管道符以及RunnableLambda的用法与设计哲学。一、从一个典型示例说起先看一段典型的 LangChain 链式代码fromoperatorimportitemgetterfromlangchain_core.runnablesimportRunnableLambdafromlangchain.promptsimportChatPromptTemplatefromlangchain_community.chat_modelsimportChatTongyifromlangchain_core.output_parsersimportStrOutputParser# 计算字符串长度的普通函数deflength_function(text):returnlen(text)# Prompt 模板promptChatPromptTemplate.from_template({a} {b} ? 计算结果是多少)# 大模型modelChatTongyi()# 输出解析器outStrOutputParser()# 构建链chain({a:itemgetter(k1)|RunnableLambda(length_function),b:itemgetter(k2)|RunnableLambda(length_function)}|prompt|model|out)# 调用resultchain.invoke({k1:hello,k2:world})# 结果模型回答 10这段代码的核心是将数据提取、函数处理、Prompt 构造、大模型推理串联成一个流水线。下面逐层拆解。二、itemgetter从标准库走来的提取器2.1 基本用法itemgetter来自 Python 标准库operator模块用于从可迭代对象中提取指定索引或键的值fromoperatorimportitemgetter# 从字典提取data{k1:hello,k2:world}getteritemgetter(k1)print(getter(data))# 输出: hello# 从列表提取按索引arr[a,b,c]getteritemgetter(0,2)print(getter(arr))# 输出: (a, c)2.2 本质一个可调用对象itemgetter(k1)返回的是一个函数对象等价于lambdax:x[k1]它本身不是 LangChain 的组件也不懂什么是 Runnable、什么是管道符|。三、|管道符LangChain 的链式编排语法糖3.1 Python 运算符重载机制Python 执行A | B时按以下顺序尝试1. 调用 A.__or__(B) ← 左边对象的或方法 2. 如果返回 NotImplemented 调用 B.__ror__(A) ← 右边对象的反向或方法Python 只负责打电话接不接、怎么接完全由对象自己决定。ror 就是 right or何时返回 NotImplemented一句话总结NotImplemented 是我处理不了让别人试试的信号。在 A | B 中A.__or__(B) 返回 NotImplemented → 尝试 B.__ror__(A)两边都返回 NotImplemented → 抛 TypeError方法根本不存在 → 等同于返回 NotImplemented 直接进入下一步LangChain 的 Runnable.__ror__ 之所以能自动包装左边正是因为 Python 的这个回退机制普通函数没有 __or__ 所以解释器自动去尝试 RunnableLambda.__ror__() LangChain 在那里拦截并包装。3.2 LangChain 的双向策略LangChain 的Runnable基类实现了__or__和__ror__但两边策略截然不同位置处理逻辑原因左边__ror__如果不是 Runnable自动包一层RunnableLambda入口要兼容各种原始数据格式右边__or__如果不是 Runnable直接抛TypeError中间环节必须类型安全# 源码逻辑示意伪代码classRunnable:def__ror__(self,other):# other 在左边ifnotisinstance(other,Runnable):otherRunnableLambda(other)# 自动包装returnRunnableSequence(other,self)def__or__(self,other):# other 在右边ifnotisinstance(other,Runnable):raiseTypeError(fExpected a Runnable, got{type(other)})returnRunnableSequence(self,other)3.3 为什么是左包右不包这是LangChain 的设计选择不是 Python 语法限制左边是数据源需要兼容字典、函数、常量等各种输入形式自动包装降低使用门槛右边是处理环节链的中间节点必须是可控的 Runnable避免隐式转换带来的调试困难显式优于隐式右边想用普通函数请显式声明RunnableLambda(func)四、RunnableLambda让普通函数融入流水线4.1 核心作用RunnableLambda是 LangChain 提供的适配器将普通 Python 函数包装成符合Runnable接口的对象fromlangchain_core.runnablesimportRunnableLambdadefmy_func(x):returnx.upper()# 包装前普通函数my_func(hello)# 直接调用# 包装后Runnable 对象runnableRunnableLambda(my_func)runnable.invoke(hello)# 通过 Runnable 接口调用包装后该函数就具备了 Runnable 的全部能力invoke()—— 同步单条执行batch()—— 批量执行stream()—— 流式输出如果函数支持生成器ainvoke()—— 异步执行可以用|与其他 Runnable 串联4.2 完整用法示例示例 1基础包装与调用fromlangchain_core.runnablesimportRunnableLambdadefadd_one(x:int)-int:returnx1runnableRunnableLambda(add_one)# 各种调用方式print(runnable.invoke(5))# 6print(runnable.batch([1,2,3]))# [2, 3, 4]# 异步调用importasyncioasyncdefmain():resultawaitrunnable.ainvoke(5)print(result)# 6asyncio.run(main())示例 2函数接收字典输入defextract_and_count(data:dict)-int:textdata.get(text,)returnlen(text)runnableRunnableLambda(extract_and_count)print(runnable.invoke({text:hello}))# 5示例 3在链中串联使用fromlangchain_core.runnablesimportRunnableLambda,RunnablePassthrough chain(RunnablePassthrough()# 透传输入|RunnableLambda(lambdax:x*2)# 乘以 2|RunnableLambda(lambdax:x1)# 加 1|RunnableLambda(lambdax:f结果:{x})# 格式化)print(chain.invoke(5))# 结果: 11 5*2111示例 4与 itemgetter 配合使用本文开头示例的变体fromoperatorimportitemgetterfromlangchain_core.runnablesimportRunnableLambda# 原始数据data{user:{name:Alice,age:30}}# 构建链提取嵌套字段 - 格式化defformat_user(name:str)-str:returnf用户名:{name}# itemgetter 提取嵌套值chain(itemgetter(user)# 提取 {name: Alice, age: 30}|itemgetter(name)# 提取 Alice|RunnableLambda(format_user)# 格式化为 用户名: Alice)print(chain.invoke(data))# 用户名: Alice4.3 输入输出类型标注LangChain 支持通过类型提示自动推断输入输出模式fromtypingimportTypedDictclassInput(TypedDict):text:strclassOutput(TypedDict):length:intdefcount_chars(data:Input)-Output:return{length:len(data[text])}runnableRunnableLambda(count_chars)# LangChain 会自动识别 Input/Output 的结构4.4 错误处理与重试RunnableLambda 继承 Runnable 的全部能力包括重试机制fromlangchain_core.runnablesimportRunnableLambdadefflaky_function(x):importrandomifrandom.random()0.5:raiseValueError(随机失败)returnx*2# 配置重试runnableRunnableLambda(flaky_function).with_retry(stop_after_attempt3,wait_exponential_jitterTrue)print(runnable.invoke(5))# 自动重试最多 3 次五、完整数据流回顾以本文开头的示例为例完整数据流如下输入: {k1: hello, k2: world} ↓ ┌─────────────────────────────────────────────┐ │ 并行分支处理字典构造 │ │ ├─ k1 → itemgetter(k1) → hello │ │ │ ↓ │ │ │ RunnableLambda(length_function) │ │ │ ↓ │ │ │ 5 ────────────┐ │ │ │ │ │ │ └─ k2 → itemgetter(k2) → world │ │ ↓ │ │ RunnableLambda(length_function) │ │ ↓ │ │ 5 ─────────────┘ │ │ ↓ │ │ {a: 5, b: 5} │ └─────────────────────────────────────────────┘ ↓ ChatPromptTemplate.from_template({a} {b} ? ...) ↓ 5 5 ? 计算结果是多少 ↓ ChatTongyi() 大模型推理 ↓ StrOutputParser() 解析输出 ↓ 最终结果: 10六、总结概念核心要点itemgetterPython 标准库函数用于提取字典键/列表索引值|管道符LangChain 的链式编排语法依赖 Python 运算符重载左包右不包LangChain 设计选择__ror__自动包装左边__or__要求右边必须是 RunnableRunnableLambda将普通 Python 函数包装为 Runnable使其可接入流水线设计哲学入口宽松兼容原始数据中间严格保证类型安全显式优于隐式RunnableLambda是连接普通 Python 代码与LangChain 流水线的桥梁理解它的用法是掌握 LangChain 链式编排的关键一步。本文基于 LangChain 框架源码及实践总结如有错误欢迎指正。