RunnableLambda

在 LangChain 中，RunnableLambda 是一个工具类，用于将普通的 Python 函数或可调用对象（callable） 转换为符合 Runnable 接口的对象。这使得普通函数可以无缝融入 LangChain 的工作流系统，与其他 Runnable 组件（如 LLM 模型、检索器、工具调用等）进行链式组合、并行执行等操作。

为什么需要 RunnableLambda？

LangChain 中的核心工作流（如链式调用、Agent 逻辑、RAG 流程等）依赖于统一的 Runnable 接口。这个接口定义了一套标准方法（如 invoke 同步调用、ainvoke 异步调用、stream 流式输出等），确保不同组件能互相协作。

普通函数本身并不遵循这个接口，无法直接参与这些工作流。而 RunnableLambda 的作用就是给普通函数“套一层壳”，让它具备 Runnable 接口的能力，从而可以和其他 LangChain 组件（如 ChatOpenAI、Retriever 等）通过 | 运算符组合成复杂流程。

结合你的例子理解

你的代码中：

from langchain_core.runnables import RunnableLambda
from langchain_core.documents import Document
from typing import List# 定义一个普通函数：格式化文档列表为字符串
def format_docs(docs: List[Document]) -> str:'''Format the docs.'''return ", ".join([doc.page_content for doc in docs])# 用 RunnableLambda 包装函数，使其成为 Runnable 对象
format_docs = RunnableLambda(format_docs)

包装后，format_docs 就从普通函数变成了 Runnable 对象，获得了以下能力：

1. 支持标准调用方法：

   • 可以用 format_docs.invoke([doc1, doc2]) 调用（替代直接 format_docs([doc1, doc2])）；
   • 支持异步调用 await format_docs.ainvoke([doc1, doc2])；
   • 支持流式输出（如果函数逻辑适合）。

2. 可与其他 Runnable 组合：
   例如在 RAG 流程中，可将“检索文档”→“格式化文档”→“传给 LLM 生成回答”串联起来：

   from langchain.chat_models import ChatOpenAI
   from langchain.retrievers import VectorstoreRetriever# 假设已有检索器和LLM
   retriever = VectorstoreRetriever(...)  # 检索文档的 Runnable
   llm = ChatOpenAI(...)  # 生成回答的 Runnable# 组合成完整流程：检索 → 格式化 → 生成
   rag_chain = retriever | format_docs | llm# 执行流程
   result = rag_chain.invoke("用户的问题")
   

   这里的 | 运算符是 LangChain 中 Runnable 组合的语法，只有实现了 Runnable 接口的对象才能这样使用。

总结

RunnableLambda 的核心作用是**“适配”**：将普通函数转换为 LangChain 工作流体系中的标准组件（Runnable），从而让这些函数能够参与复杂的链式调用、异步处理等高级流程。这在构建结构化的 LLM 应用（如 RAG、Agent）时非常有用，能大幅提升代码的可组合性和扩展性。