RunnablePassthrough：LangChain 中的“数据直通”组件

RunnablePassthrough 是 LangChain 表达式语言（LCEL）中的一个核心组件，它的作用就像一个透明的数据管道——它接收输入数据，然后原封不动地（或经细微处理后）把它传递下去，同时不改变数据本身。

一句话定义：RunnablePassthrough 是一个“透明代理”，它在链中充当一个中间节点，允许你插入副作用（如日志、调试）或配合 RunnableParallel 实现数据的“同时分流”，而不干扰主数据流。

---

为什么要用 RunnablePassthrough？

在 LCEL 链中，默认的 | 管道会依次传递数据：前一个 Runnable 的输出作为下一个 Runnable 的输入。但有些场景下，你希望在不改变当前数据的前提下执行一些额外操作（例如打印、记录、缓存），或者你需要在一个 RunnableParallel 中“偷看”并复用原始输入。

RunnablePassthrough 正是为此而生，它提供两种主要形态：

1. 纯透传（RunnablePassthrough()）：什么也不做，直接将输入原样传给下一个组件。
2. 带副作用（RunnablePassthrough.assign(...)）：在传递原始输入的同时，额外向数据字典中添加或修改某些字段。

---

1. RunnablePassthrough()：无副作用的直通

最简单的用法，就是直接 RunnablePassthrough()，它相当于一个“空操作”节点。在调试链时，你可以把它放在中间，便于观察输入输出的变化。

from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables import RunnableLambda

# 定义一个简单的函数，仅打印输入
def log_data(x):
    print(f"当前数据: {x}")
    return x

# 构建链：输入 -> 打印（透传） -> 后续处理
chain = RunnablePassthrough() | RunnableLambda(log_data) | lambda x: x.upper()

result = chain.invoke("hello")
# 输出: 当前数据: hello
# result == "HELLO"

在这个例子中，RunnablePassthrough 没有修改数据，直接把 "hello" 传给了下一个节点。但在实际链中，我们通常不会单独用 RunnablePassthrough()，因为它和直接写 | 没什么区别。它的真正威力在于与 RunnableParallel 配合。

---

2. RunnablePassthrough.assign()：在并行中“旁路”数据

RunnablePassthrough.assign(**kwargs) 是更强大的用法，它接受一个字典，键是新的字段名，值是生成该字段的 Runnable。它会保留原始输入的所有字段，同时用这些 Runnable 的结果添加或覆盖指定字段，形成一个新的字典。

这非常适用于 RunnableParallel 场景：比如你需要同时处理原始输入和基于原始输入生成的附加信息，然后将它们合并成一个更丰富的上下文。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI()

# 定义两个独立的 Prompt
prompt_1 = ChatPromptTemplate.from_template("描述一下 {topic}")
prompt_2 = ChatPromptTemplate.from_template("为 {topic} 起一个英文名")

# 构建并行任务：一个用于描述，一个用于命名
parallel_chain = RunnableParallel(
    description=prompt_1 | model | StrOutputParser(),
    english_name=prompt_2 | model | StrOutputParser(),
)

# 我们希望最终输出时，不仅包含这两个新生成的字段，还要保留原始输入的 topic
# 使用 assign 将原始输入保留下来，并添加新字段
chain = (
    # 先用一个 RunnablePassthrough 透传原始输入（字典）
    # 同时用 assign 在传递过程中添加新字段
    RunnablePassthrough.assign(
        # 注意：这里接收的输入是原始 input 字典
        # 我们需要把原始 topic 也传给 parallel_chain，但 parallel_chain 需要 topic 字段
        # 解决方法：用 lambda 提取 topic 并构建新字典
        **(lambda x: parallel_chain.invoke({"topic": x["topic"]}))()
    )
)

# 但实际上上述写法不对，因为 assign 需要每个键对应一个 Runnable，
# 不能直接传一个函数返回字典。正确写法：
chain = RunnablePassthrough.assign(
    description=lambda x: (prompt_1 | model | StrOutputParser()).invoke({"topic": x["topic"]}),
    english_name=lambda x: (prompt_2 | model | StrOutputParser()).invoke({"topic": x["topic"]})
)

# 或者用 RunnableParallel 嵌套，更简洁：
chain = (
    RunnablePassthrough()  # 先透传原始数据
    | RunnablePassthrough.assign(
        description=prompt_1 | model | StrOutputParser(),
        english_name=prompt_2 | model | StrOutputParser()
    )
)

# 最终输出字典包含原始 topic 以及 description 和 english_name
result = chain.invoke({"topic": "人工智能"})
print(result)

输出类似：

{
  "topic": "人工智能",
  "description": "人工智能是...",
  "english_name": "Artificial Intelligence"
}

关键点：assign 中的 Runnable 可以访问到当前完整的输入字典。在上例中，prompt_1 和 prompt_2 都能从输入中提取 topic。

---

3. 与 RunnableLambda 的区别

特性	RunnablePassthrough	RunnableLambda
核心职责	透传原数据，可附带添加新字段	转换（可能改变）数据
返回值	返回原始输入（或增加字段后的新字典）	返回任意值（可能改变类型）
适用场景	需要保留原始输入，同时执行额外操作（如日志、添加衍生物）	需要对数据进行清洗、映射或转换

如果只是需要打日志且不改变输入，可以用 RunnablePassthrough 配合 RunnableLambda 实现副作用：

def log(x):
    print(x)
    return x

chain = RunnablePassthrough() | RunnableLambda(log) | ...

---

4. 典型应用场景

• 多源数据合并：在 RAG 中，可能需要同时获取检索结果和原始 Query，然后把它们合并到一个字典传递给 LLM。
• 调试：在链中插入 RunnablePassthrough 并绑定额外的日志函数，不影响数据流。
• 缓存/审计：在 assign 中调用缓存检查，将结果合并到当前数据中。
• 条件分支：根据输入动态决定是否添加某些字段。

---

5. 面试高频追问

Q：RunnablePassthrough 与 | 直接串联有何区别？
A：| 是数据传输，前一个 Runnable 的输出成为下一个的输入。RunnablePassthrough 本身不改变数据，但它允许你在管道中插入副作用或配合 assign 在平行处理中保留原始上下文。

Q：RunnablePassthrough.assign() 和 RunnableParallel 有什么区别？
A：RunnableParallel 用于并行执行多个独立的 Runnable，返回一个包含所有结果的字典。而 RunnablePassthrough.assign() 是在保留原始输入的基础上添加新字段，它本身就是一种“并行 + 合并”的快捷方式。二者可以组合使用。

Q：如何利用 RunnablePassthrough 实现条件性添加字段？
A：可以结合 RunnableLambda 判断，如果满足条件则通过 assign 添加，否则仅透传。但较复杂的条件逻辑建议用 RunnableBranch。

---

总结

RunnablePassthrough 是 LCEL 中一个看似简单但极其有用的工具，它让数据流变得更加灵活。通过 assign，它允许你在不破坏原始数据的前提下，给数据“附加”新的衍生信息，这在构建复杂 RAG 或 Agent 系统时非常关键。理解它能帮助你写出更可读、更健壮的链式代码。