RunnablePassthrough 两种写法对比：.assign() 的参数到底要不要包 RunnableLambda？

一、两种调用方式对比

LangChain 提供了两种调用 assign 的方式，功能等价但写法不同。

1.1 写法一：实例方法 .assign()

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

chain = RunnableParallel(
    passed=RunnablePassthrough().assign(
        modified=lambda x: x["k1"] + "!!!"
    ),
)

1.2 写法二：类方法 assign()

from langchain_core.runnables import RunnablePassthrough


def get_user_age(data):
    return 25


chain = RunnablePassthrough.assign(
    age=get_user_age
)

二、核心对比

对比项	写法一：RunnablePassthrough().assign(...)	写法二：RunnablePassthrough.assign(...)
调用方式	实例方法（先创建实例再调用）	类方法（直接通过类调用）
返回值	Runnable 实例	Runnable 实例
功能效果	保留原字段 + 添加新字段	保留原字段 + 添加新字段
使用场景	通常配合 RunnableParallel 使用	通常作为独立步骤使用
本质区别	先实例化再配置	类级别的工厂方法

结论：两种写法功能完全等价，选择取决于代码上下文和个人习惯。

三、参数要不要包 RunnableLambda？

这是最容易混淆的地方。答案是：在 .assign() 里不需要；在管道运算符 \| 里，只要相邻位置有 Runnable，也不需要。

3.1 不需要包的场景

以下方法内部会自动把函数或 Lambda 包装成 RunnableLambda：

from langchain_core.runnables import RunnablePassthrough, RunnableParallel


def get_user_age(data):
    return 25


# ✅ RunnablePassthrough.assign() —— 自动转换
chain1 = RunnablePassthrough.assign(age=get_user_age)

# ✅ RunnableParallel() —— 自动转换
chain2 = RunnableParallel({
    "result": lambda x: x["name"].upper()
})

# ✅ 直接传 lambda —— 自动转换
chain3 = RunnablePassthrough().assign(
    modified=lambda x: x["k1"] + "!!!"
)

3.2 管道运算符 \| 的真相

管道运算符 \| 并非"只接受 Runnable 对象，不会自动转换"。准确的说法是：

管道中相邻的两个对象，只要至少有一个是 Runnable，就能正常工作。 Runnable 的 __or__ 和 __ror__ 方法会自动将旁边的 callable 包装为 RunnableLambda。

from langchain_core.runnables import RunnableLambda, RunnableParallel

step1 = RunnableLambda(lambda x: x)

def get_user_age(x):
    return x["age"]

# ✅ 正确：step1 是 Runnable，get_user_age 自动转换
chain_good = step1 | get_user_age

# ❌ 错误：两边都不是 Runnable，无法启动链条
# chain_bad = get_user_age | another_func

只有当整个链条全是普通对象（没有一个是 Runnable）时，才需要手动包装：

# ✅ 修复：至少把一个转成 Runnable，链条就能自动运转
chain_fixed = RunnableLambda(get_user_age) | another_func

四、四种等价写法实测

from langchain_core.runnables import RunnablePassthrough, RunnableLambda


def get_user_age(data):
    return 25


# 写法1：直接传函数（最简洁，推荐）
r1 = RunnablePassthrough.assign(age=get_user_age)

# 写法2：传 lambda（也简洁）
r2 = RunnablePassthrough.assign(age=lambda x: 25)

# 写法3：手动包 RunnableLambda（啰嗦，但效果一样）
r3 = RunnablePassthrough.assign(age=RunnableLambda(get_user_age))

# 写法4：实例方式调用
r4 = RunnablePassthrough().assign(age=get_user_age)

# 测试
for r in [r1, r2, r3, r4]:
    result = r.invoke({"name": "Alice"})
    print(result)

输出：

{'name': 'Alice', 'age': 25}
{'name': 'Alice', 'age': 25}
{'name': 'Alice', 'age': 25}
{'name': 'Alice', 'age': 25}

五、自动转换方法汇总

方法/类	自动转换函数为 RunnableLambda？	说明
RunnablePassthrough.assign()	✅ 是	内部自动包装
RunnableParallel()	✅ 是	内部自动包装
RunnableLambda() 构造函数	✅ 是（当然）	本身就是包装器
管道运算符 |（有 Runnable 相邻）	✅ 是	__or__ / __ror__ 自动转换
管道运算符 |（无 Runnable 相邻）	❌ 否	两边都不是 Runnable，无法启动
RunnableSequence()	❌ 否	构造函数不自动转换

六、可行性组合总表

写法	是否可行	实际调用	说明
Runnable | callable	✅	Runnable.__or__	自动包装 callable
callable | Runnable	✅	Runnable.__ror__	自动包装 callable
Runnable | Runnable	✅	Runnable.__or__	直接串联
Runnable | dict	✅	Runnable.__or__	dict → RunnableParallel
callable | callable	❌	—	两边都没有 __or__ / __ror__
dict | callable	❌	—	dict 和 callable 都没有
list | Runnable	✅	Runnable.__ror__	list → RunnableSequence

七、最佳实践建议

场景	推荐写法	原因
.assign() 里传函数	age=get_user_age	最简洁，可读性最好
.assign() 里简单逻辑	age=lambda x: ...	无需定义额外函数
管道 | 里传函数（旁边有 Runnable）	直接写 func 或 lambda	自动转换，无需手动包
管道 | 里传函数（链条无 Runnable）	RunnableLambda(func)	必须手动包，启动链条
配合 RunnableParallel	RunnablePassthrough().assign(...)	实例方式更自然

八、一句话总结

.assign() 和 RunnableParallel 是"友好"的 API，看到函数会自动帮你包成 RunnableLambda；管道运算符 \| 也是"友好"的——只要旁边有 Runnable，它会通过 __or__ / __ror__ 自动吸收并包装 callable。

真正需要手动包 RunnableLambda 的只有一种情况：你的链条里没有任何 Runnable 对象，全是普通函数/字典，需要手动点燃第一把火。

在 .assign() 里直接写函数或 Lambda 就行，别画蛇添足包 RunnableLambda；在管道里，确保至少有一个 Runnable 节点，其余交给自动转换。