介绍
RunnablePassthrough 是 LangChain 中 Runnable 接口的实现类，核心作用是：
- 透传数据：不修改输入，直接将输入原样传递到下一个环节；
- 增强数据：在透传的同时，新增/修改部分字段（比如补充上下文、格式化数据）；
- 适配链式调用：兼容 LangChain 的
invoke()/stream()/batch()等核心方法，无缝融入链式流程。
它常和 RunnableParallel、RunnableSequence 配合使用，解决链式调用中“数据传递/重组”的问题。
常用方法
| 方法 | 作用 |
|---|---|
invoke(input) |
同步调用，接收输入数据，返回透传/增强后的数据 |
stream(input) |
流式调用，返回数据迭代器（适合大模型流式输出场景） |
assign(**kwargs) |
新增字段，kwargs 的值可以是固定值或函数（函数接收输入数据并返回结果） |
pick(*keys) |
只透传指定的字段（比如只传递”question”，过滤其他无关字段） |
透传数据
from langchain_core.runnables import RunnablePassthrough
# 初始化透传组件
passthrough = RunnablePassthrough()
# 调用invoke，输入会被原样返回
result = passthrough.invoke({"question": "什么是RunnablePassthrough？"})
print(result)
执行结果
{'question': '什么是RunnablePassthrough？'}
增强数据（透传+新增字段）
通过 assign() 方法可以在透传原有数据的同时，新增字段：
from langchain_core.runnables import RunnablePassthrough
# 透传原有输入，并新增"context"字段
enhanced_passthrough = RunnablePassthrough.assign(
context=lambda x: f"用户问题：{x['question']}，回答需简洁易懂"
)
# 调用测试
result = enhanced_passthrough.invoke({"question": "什么是RunnablePassthrough？"})
print(result)
执行结果
# 输出：
{
'question': '什么是RunnablePassthrough？',
'context': '用户问题：什么是RunnablePassthrough？，回答需简洁易懂'
}
链式调用
最常见的场景是“检索+生成”（RAG）中，透传问题并拼接检索到的上下文：
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
data = {
"quest": RunnablePassthrough(), # 相当于占位符
}
# 定义模板
prompt = ChatPromptTemplate.from_messages([
("human", "用户提问：{quest}")
])
# 组成链
chain = data | prompt
question = "25年报告概述"
# 把参数request替换掉 RunnablePassthrough()
res = chain.invoke(question)
print("结果：", res)
执行结果
结果： messages=[HumanMessage(content='用户提问：25年报告概述', additional_kwargs={}, response_metadata={})]