RunnablePassthroughLangChain 中的“数据直通”组件RunnablePassthrough是 LangChain 表达式语言LCEL中的一个核心组件它的作用就像一个透明的数据管道——它接收输入数据然后原封不动地或经细微处理后把它传递下去同时不改变数据本身。一句话定义RunnablePassthrough是一个“透明代理”它在链中充当一个中间节点允许你插入副作用如日志、调试或配合RunnableParallel实现数据的“同时分流”而不干扰主数据流。为什么要用RunnablePassthrough在 LCEL 链中默认的|管道会依次传递数据前一个Runnable的输出作为下一个Runnable的输入。但有些场景下你希望在不改变当前数据的前提下执行一些额外操作例如打印、记录、缓存或者你需要在一个RunnableParallel中“偷看”并复用原始输入。RunnablePassthrough正是为此而生它提供两种主要形态纯透传RunnablePassthrough()什么也不做直接将输入原样传给下一个组件。带副作用RunnablePassthrough.assign(...)在传递原始输入的同时额外向数据字典中添加或修改某些字段。1.RunnablePassthrough()无副作用的直通最简单的用法就是直接RunnablePassthrough()它相当于一个“空操作”节点。在调试链时你可以把它放在中间便于观察输入输出的变化。fromlangchain_core.runnablesimportRunnablePassthroughfromlangchain_core.runnablesimportRunnableLambda# 定义一个简单的函数仅打印输入deflog_data(x):print(f当前数据:{x})returnx# 构建链输入 - 打印透传 - 后续处理chainRunnablePassthrough()|RunnableLambda(log_data)|lambdax:x.upper()resultchain.invoke(hello)# 输出: 当前数据: hello# result HELLO在这个例子中RunnablePassthrough没有修改数据直接把hello传给了下一个节点。但在实际链中我们通常不会单独用RunnablePassthrough()因为它和直接写|没什么区别。它的真正威力在于与RunnableParallel配合。2.RunnablePassthrough.assign()在并行中“旁路”数据RunnablePassthrough.assign(**kwargs)是更强大的用法它接受一个字典键是新的字段名值是生成该字段的Runnable。它会保留原始输入的所有字段同时用这些Runnable的结果添加或覆盖指定字段形成一个新的字典。这非常适用于RunnableParallel场景比如你需要同时处理原始输入和基于原始输入生成的附加信息然后将它们合并成一个更丰富的上下文。fromlangchain_core.runnablesimportRunnableParallel,RunnablePassthroughfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI modelChatOpenAI()# 定义两个独立的 Promptprompt_1ChatPromptTemplate.from_template(描述一下 {topic})prompt_2ChatPromptTemplate.from_template(为 {topic} 起一个英文名)# 构建并行任务一个用于描述一个用于命名parallel_chainRunnableParallel(descriptionprompt_1|model|StrOutputParser(),english_nameprompt_2|model|StrOutputParser(),)# 我们希望最终输出时不仅包含这两个新生成的字段还要保留原始输入的 topic# 使用 assign 将原始输入保留下来并添加新字段chain(# 先用一个 RunnablePassthrough 透传原始输入字典# 同时用 assign 在传递过程中添加新字段RunnablePassthrough.assign(# 注意这里接收的输入是原始 input 字典# 我们需要把原始 topic 也传给 parallel_chain但 parallel_chain 需要 topic 字段# 解决方法用 lambda 提取 topic 并构建新字典**(lambdax:parallel_chain.invoke({topic:x[topic]}))()))# 但实际上上述写法不对因为 assign 需要每个键对应一个 Runnable# 不能直接传一个函数返回字典。正确写法chainRunnablePassthrough.assign(descriptionlambdax:(prompt_1|model|StrOutputParser()).invoke({topic:x[topic]}),english_namelambdax:(prompt_2|model|StrOutputParser()).invoke({topic:x[topic]}))# 或者用 RunnableParallel 嵌套更简洁chain(RunnablePassthrough()# 先透传原始数据|RunnablePassthrough.assign(descriptionprompt_1|model|StrOutputParser(),english_nameprompt_2|model|StrOutputParser()))# 最终输出字典包含原始 topic 以及 description 和 english_nameresultchain.invoke({topic:人工智能})print(result)输出类似{topic:人工智能,description:人工智能是...,english_name:Artificial Intelligence}关键点assign中的Runnable可以访问到当前完整的输入字典。在上例中prompt_1和prompt_2都能从输入中提取topic。3. 与RunnableLambda的区别特性RunnablePassthroughRunnableLambda核心职责透传原数据可附带添加新字段转换可能改变数据返回值返回原始输入或增加字段后的新字典返回任意值可能改变类型适用场景需要保留原始输入同时执行额外操作如日志、添加衍生物需要对数据进行清洗、映射或转换如果只是需要打日志且不改变输入可以用RunnablePassthrough配合RunnableLambda实现副作用deflog(x):print(x)returnx chainRunnablePassthrough()|RunnableLambda(log)|...4. 典型应用场景多源数据合并在 RAG 中可能需要同时获取检索结果和原始 Query然后把它们合并到一个字典传递给 LLM。调试在链中插入RunnablePassthrough并绑定额外的日志函数不影响数据流。缓存/审计在assign中调用缓存检查将结果合并到当前数据中。条件分支根据输入动态决定是否添加某些字段。5. 面试高频追问QRunnablePassthrough与|直接串联有何区别A|是数据传输前一个Runnable的输出成为下一个的输入。RunnablePassthrough本身不改变数据但它允许你在管道中插入副作用或配合assign在平行处理中保留原始上下文。QRunnablePassthrough.assign()和RunnableParallel有什么区别ARunnableParallel用于并行执行多个独立的Runnable返回一个包含所有结果的字典。而RunnablePassthrough.assign()是在保留原始输入的基础上添加新字段它本身就是一种“并行 合并”的快捷方式。二者可以组合使用。Q如何利用RunnablePassthrough实现条件性添加字段A可以结合RunnableLambda判断如果满足条件则通过assign添加否则仅透传。但较复杂的条件逻辑建议用RunnableBranch。总结RunnablePassthrough是 LCEL 中一个看似简单但极其有用的工具它让数据流变得更加灵活。通过assign它允许你在不破坏原始数据的前提下给数据“附加”新的衍生信息这在构建复杂 RAG 或 Agent 系统时非常关键。理解它能帮助你写出更可读、更健壮的链式代码。