在 LangChain LCEL (LangChain Expression Language) 的世界里数据像水流一样在管道Pipe|中流动。通常一个组件会处理输入并产生新的输出传递给下一个组件。但是有时候我们需要保留原始输入或者将输入原样传递给后续步骤。这时RunnablePassthrough就登场了。1. 什么是 RunnablePassthroughRunnablePassthrough是一个极其简单的 Runnable它的作用就是透传。输入x输出x简单来说它就像一个占位符或连接器不做任何修改直接把拿到的数据递给后面。2. 为什么需要它你可能会问如果它什么都不做为什么需要它核心应用场景有两个数据多路复用Forking在并行处理RunnableParallel中我们需要把同一个输入同时传给多个分支。其中一个分支可能需要处理数据比如检索而另一个分支需要保留原始数据比如填入 Prompt。构建字典在构建 Prompt 输入时我们通常需要一个字典例如{context: ..., question: ...}。RunnablePassthrough允许我们在构建字典时引用“整个输入”。3. 核心应用场景RAG (检索增强生成)这是RunnablePassthrough最经典的使用场景。3.1 场景描述在 RAG 中我们需要做两件事拿用户的问题去检索文档 - 生成context。把用户的问题直接填入 Prompt - 生成question。3.2 代码实现详解让我们看一段模拟代码完整代码见src/examples/chains/demo_runnable_passthrough.py# 模拟一个检索器 (Retriever)# 在真实场景中这通常是 vector_store.as_retriever()deffake_retriever(query:str):logger.info(fRetrieving documents for:{query})returnf这里是关于 {query} 的一些背景知识...defmain():llmget_gemini_llm()# -----------------------------------------------------------# 场景 1: 基础用法 - 原样透传# -----------------------------------------------------------logger.info(--- Demo 1: Basic Passthrough ---)# 这里的 RunnablePassthrough() 就像一个占位符它把 invoke 传入的 Hello 原封不动地传给下一步# 虽然在这个简单的例子里看起来没用但在复杂的字典构造中非常关键chainRunnablePassthrough()resultchain.invoke(Hello World)logger.info(fResult:{result})# Output: Hello World# -----------------------------------------------------------# 场景 2: RAG (检索增强生成) - 最经典用法# -----------------------------------------------------------logger.info(\n--- Demo 2: RAG Scenario ---)promptChatPromptTemplate.from_template(基于以下上下文回答问题:\n\n上下文: {context}\n\n问题: {question})# 我们构建一个并行运行的 Map (RunnableParallel)# 1. context 键把用户输入传给 retriever获取上下文# 2. question 键我们需要把用户原始输入填到这里。# 如果不加 RunnablePassthrough()我们就没法在这里引用“原始输入”了。# 提示这里的 RunnableLambda(fake_retriever) 其实可以简化为直接写 fake_retriever# LangChain 会自动把函数转换为 RunnableLambda。rag_chain({context:fake_retriever,# --- 自动隐式转换为 RunnableLambda(fake_retriever)question:RunnablePassthrough()}|prompt|llm)# 调用链# 用户输入 什么是 LangChain?# 1. fake_retriever(什么是 LangChain?) - 填充 context# 2. RunnablePassthrough() 接收 什么是 LangChain? 并原样返回 - 填充 questionresponserag_chain.invoke(什么是 LangChain?)logger.info(fAI Response:{response.content})3.3 数据流深度解析 (参数是如何传递的)很多同学会问retriever比如fake_retriever的参数是从哪来的当您执行rag_chain.invoke(什么是 LangChain?)时数据流向如下广播 (Broadcasting)最外层的字典结构隐式RunnableParallel接收到输入什么是 LangChain?。它会将这一份输入同时复制并通过invoke()方法传递给字典中的每一个 Value。并行执行分支 A (“context”)调用retriever.invoke(什么是 LangChain?)。因此retriever函数或 Runnable接收到的参数就是这个输入字符串。分支 B (“question”)调用RunnablePassthrough().invoke(什么是 LangChain?)。它不做任何处理直接返回什么是 LangChain?。聚合 (Aggregation)RunnableParallel等待两个分支都执行完毕。它将结果打包成一个新的字典{context: ..., question: ...}。这个字典随后被传给下一个环节prompt。3.4 Mermaid 图解并行分发机制上下文文本LangChain 是什么用户输入: LangChain 是什么调用 retriever(LangChain 是什么)调用 Passthrough(LangChain 是什么)ChatPromptTemplateGemini/GPT如果没有RunnablePassthrough我们就无法在RunnableParallel那个字典结构中引用原始的输入字符串。4. 进阶用法RunnablePassthrough.assign()除了原样透传它还有一个非常有用的静态方法.assign()。它用于在不丢失原始字典数据的情况下添加新的字段。场景假设上一步的输出是{num: 10}你需要计算它的平方但同时保留num字段。代码对比不使用 assign:# 你需要手动构造整个字典很容易丢失旧数据chainRunnableLambda(lambdax:{num:x[num],squared:x[num]**2})使用 assign (优雅):# 自动合并新旧数据chainRunnablePassthrough.assign(squaredlambdax:x[num]**2)输出:{num: 10, squared: 100}这在长链条处理中非常有用可以像“滚雪球”一样不断给数据流增加新的上下文而不会丢弃之前的信息。5. 总结RunnablePassthrough(): 恒等函数f(x) x。用于在并行分支中保留原始输入。RunnablePassthrough.assign(…): 增量更新函数f(dict) dict new_keys。用于给字典添加新字段。它是 LCEL 胶水代码中不可或缺的一部分尤其是在构建 RAG 和复杂数据流时。