RunnableParallel 操纵输入和输出

https://python.langchain.com.cn/docs/expression_language/how_to/map

一、先搞懂：RunnableParallel到底是啥？

咱们先抛掉复杂定义，用生活化的例子理解：
你要给朋友寄礼物，需要把“贺卡”“零食”“玩具”三个东西一起寄。如果分开寄，朋友要收三次，很麻烦；这时你找个“打包员”（就是RunnableParallel），把三个东西装在一个印着“贺卡、零食、玩具”标签的箱子里，朋友一次就能收到所有东西，还知道哪个是哪个。

在代码里，这个“打包员”的作用就是：
把多个可运行对象（比如检索器、模型链，你暂时理解成“能干活的工具”）的结果，打包成一个「键值对字典」（就像带标签的箱子），方便后面的工具（比如提示词模板）直接用。

二、新手必记：它有个“偷懒”技巧——自动类型转换

新手最怕记复杂代码！RunnableParallel有个超贴心的设计：在LCEL链里，普通字典会自动变成RunnableParallel，不用你写长长的RunnableParallel()。

比如这三种写法，效果完全一样，你记最简单的「字典写法」就行（后续复习时看第一种就够）：

1. 推荐写法（字典）：{"context": 检索器, "question": RunnablePassthrough()}
2. 麻烦写法1：RunnableParallel({"context": 检索器, "question": RunnablePassthrough()})
3. 麻烦写法2：RunnableParallel(context=检索器, question=RunnablePassthrough())

这里提一句RunnablePassthrough()：它就是个“传话筒”，收到啥就原样传出去（比如用户输入的问题，不修改直接传给后面），不用记复杂逻辑。

三、核心用法1：解决“输入对不上”的问题（实战：检索问答链）

这是新手最常用的场景，咱们一步步拆，你跟着走一遍就懂了。

1. 问题背景

咱们要做一个“根据资料回答问题”的链（叫“检索问答链”），但遇到个麻烦：

• 后面的「提示词模板（prompt）」需要两个输入：{context}（资料）和{question}（用户的问题）；
• 但用户只会输入一个东西：比如“harrison在哪里工作？”（只有question，没有context）。

这时就需要RunnableParallel来“补全”输入。

2. 解决步骤（附代码+注释）

# 1. 先准备工具：检索器（存了资料“harrison worked at kensho”）
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
vectorstore = FAISS.from_texts(["harrison worked at kensho"], embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()  # 检索器：能根据问题找资料# 2. 准备提示词模板（需要{context}和{question}两个输入）
from langchain_core.prompts import ChatPromptTemplate
template = """Answer the question based only on the following context:
{context}  # 需要检索器提供的资料
Question: {question}  # 需要用户输入的问题
"""
prompt = ChatPromptTemplate.from_template(template)# 3. 准备模型和输出解析器（暂时不用深懂，知道是“生成回答”和“整理回答格式”就行）
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI()
output_parser = StrOutputParser()# 4. 用RunnableParallel补全输入，搭完整链
retrieval_chain = (# 关键：用字典（自动转RunnableParallel）打包两个输入{"context": retriever,  # 让检索器根据问题找资料，填到{context}"question": RunnablePassthrough()}  # 把用户的问题原样传给{question}| prompt  # 把打包好的输入传给提示词模板| model  # 模型根据提示词生成回答| output_parser  # 整理回答格式
)# 5. 测试：用户只输入问题
result = retrieval_chain.invoke("where did harrison work?")
print(result)  # 输出：harrison worked at kensho.

3. 复习重点

• 这个场景的核心：用RunnableParallel把“缺的输入（context）”补上，让后面的prompt能正常工作；
• 记住RunnablePassthrough()的作用：用户输入啥，就传啥，不用改。

四、核心用法2：用itemgetter简化“拿数据”（新手进阶技巧）

如果你的输入本身是个「字典」（比如用户同时给了“问题”和“回答语言”），要从中提取指定内容，用itemgetter比写复杂代码简单——它就像“一把钥匙”，专开字典里指定的“抽屉”。

1. 实战例子：指定语言回答

比如用户输入{"question": "harrison在哪工作？", "language": "意大利语"}，要让回答用意大利语，步骤如下：

# 1. 先导入itemgetter（从operator模块里拿）
from operator import itemgetter# 2. 其他工具（检索器、prompt、模型）和之前一样，只改“打包输入”的部分
chain = ({# 用itemgetter("question")：从用户输入字典里，拿出“question”的值，传给检索器找资料"context": itemgetter("question") | retriever,# 用itemgetter("question")：拿出“question”的值，传给prompt的{question}"question": itemgetter("question"),# 用itemgetter("language")：拿出“language”的值，传给prompt的{language}"language": itemgetter("language"),}| prompt| model| output_parser
)# 3. 测试：输入带两个键的字典
result = chain.invoke({"question": "where did harrison work", "language": "italian"})
print(result)  # 输出意大利语的回答：Harrison ha lavorato a Kensho.

2. 复习重点

• itemgetter(键名)：从字典里“一键提取”对应值，不用写lambda x: x["键名"]这种复杂代码；
• 搭配RunnableParallel用：能快速从用户输入字典里，拆出后面需要的所有内容。

五、核心用法3：让任务“并行跑”，变快！

“并行”就是“同时做多个事”，比如你早上“同时煮咖啡和烤面包”，比“先煮咖啡再烤面包”快。RunnableParallel就能让多个工具“同时干活”，节省时间。

1. 实战例子：同时生成笑话和诗歌

# 1. 定义两个独立的“小链”
model = ChatOpenAI()
# 小链1：根据主题生成笑话
joke_chain = ChatPromptTemplate.from_template("讲个关于{topic}的笑话") | model
# 小链2：根据主题生成两行诗
poem_chain = ChatPromptTemplate.from_template("写一首关于{topic}的两行诗") | model# 2. 用RunnableParallel让两个小链“同时跑”
map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)# 3. 测试：输入主题“熊”，同时得到笑话和诗歌
result = map_chain.invoke({"topic": "bear"})
print(result["joke"])  # 输出笑话
print(result["poem"])  # 输出诗歌

2. 并行的好处：真的变快了！

文档里用%%timeit（计时工具）测过：

• 单独跑笑话链：约0.5秒；
• 单独跑诗歌链：约0.5秒；
• 用map_chain同时跑两个：还是约0.5秒！

这就是并行的魔力——不用等一个做完再做另一个，时间差不多减半。

3. 复习重点

• 什么时候用并行：多个任务之间没关系（比如生成笑话和诗歌互不影响），就能用RunnableParallel让它们同时跑；
• 记住：RunnableParallel的输出是字典，键是你给每个任务起的名字（比如joke“poem”），值是任务结果。

六、复习总结：一张表记全核心

为了方便你后续复习，把这节课的重点整理成表：

核心工具	作用（新手理解）	关键技巧/场景	记忆点
RunnableParallel（RunnableMap）	打包多个任务结果成字典，适配输入	1. 补全输入（检索问答链） 2. 并行跑任务	字典写法最简洁，自动类型转换
itemgetter	从字典里快速提取指定值	输入是字典时，搭配RunnableParallel用	像“钥匙”，拿值不用写复杂代码
并行执行	同时做多个无关任务，节省时间	生成笑话+诗歌、多数据源检索等	时间和单个任务差不多，效率翻倍

最后：复习小建议

下次复习时，你可以先看“复习总结表”回忆核心，再对着每个“实战例子”的代码看：

1. 先看代码里的RunnableParallel部分（字典写法），想它在打包什么；
2. 再看后面的工具（prompt、模型），想打包的结果是不是刚好满足需求；
3. 遇到itemgetter，想它在从哪个字典里拿什么值。

这样一步步对照，很快就能回忆起所有内容啦！如果某个点没懂，随时回来翻对应的“实战例子”，咱们的例子都是最基础的，没有多余复杂代码~