【第三周】RAG与Agent实战31：RunnablePassthrough的使用 —— 打造自动化 RAG 链

在上一节中手动完成了“检索 - 格式化 - 组装 Prompt”的过程。虽然逻辑清晰但在实际工程中我们希望构建一条全自动化的流水线Chain输入只需要用户的一句话String。处理链内部自动去检索、自动格式化、自动组装。输出直接得到模型的回答。要实现这一点我们需要解决一个核心问题如何让同一个输入用户问题既传给检索器又直接传给 Prompt 的input字段答案就是 LangChain 中的“透传”神器 ——RunnablePassthrough。 核心痛点数据流的分支与合并在构建 RAG 链时数据流向通常是这样的用户输入(query) 进入链条。分支 Aquery-Retriever- 检索结果 (docs) -格式化-context。分支 Bquery-直接传递-input。合并将context和input打包成字典喂给 Prompt。如果没有RunnablePassthrough我们需要写很多中间变量来暂存query。而有了它我们可以用一行优雅的 LCEL 代码完成所有操作。 实战代码构建全自动 RAG 链我们将重构之前的代码利用as_retriever()和RunnablePassthrough实现端到端的自动化。完整代码实现fromlangchain_community.chat_modelsimportChatTongyifromlangchain_core.vectorstoresimportInMemoryVectorStorefromlangchain_community.embeddingsimportDashScopeEmbeddingsfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthrough# ---------------------------------------------------------# 1. 初始化组件# ---------------------------------------------------------modelChatTongyi(modelqwen-plus)# 定义 Prompt 模板# 注意这里需要两个变量 {context} 和 {input}promptChatPromptTemplate.from_messages([(system,以我提供的已知参考资料为主简洁和专业的回答用户问题。参考资料{context}),(user,用户提问{input})])# 初始化向量存储vector_storeInMemoryVectorStore(embeddingDashScopeEmbeddings(modeltext-embedding-v4))# ---------------------------------------------------------# 2. 准备数据# ---------------------------------------------------------vector_store.add_texts(texts[减肥就是要少吃多练,在减脂过程中要注意饮食控制避免吃太多的高热量食物如高热量的食物、高热量的饮料等。,跑步是很好的减肥方法因为它可以帮助你燃烧更多的卡路里从而达到减肥的效果。,],ids[id0,id1,id2])# ---------------------------------------------------------# 3. 创建检索器 (Retriever)# ---------------------------------------------------------# as_retriever() 将 vector_store 转换为一个 Runnable 对象# 它的输入是 str (query)输出是 list[Document]retrievervector_store.as_retriever(search_kwargs{k:2})# ---------------------------------------------------------# 4. 定义辅助函数# ---------------------------------------------------------# 格式化函数将 list[Document] 转为字符串defformat_func(docs):ifnotdocs:return无相关参考资料# 简单拼接实际项目中建议加换行符或分隔符formatted_str[fordocindocs:formatted_strdoc.page_content formatted_str]returnformatted_str# 调试函数打印最终生成的 Promptdefprint_prompt(prompt_value):print( 最终组装的 Prompt:)print(prompt_value.to_string())print(*30)returnprompt_value# ---------------------------------------------------------# 5. 构建 LCEL 链 (核心部分)# ---------------------------------------------------------# 语法解析# {# input: RunnablePassthrough(), -- 分支B原样传递输入字符串# context: retriever | format_func -- 分支A输入 - 检索 - 格式化# }# 这两个分支并行执行结果合并为一个 dict: {input: ..., context: ...}# 然后传递给 promptchain({input:RunnablePassthrough(),context:retriever|format_func}|prompt|print_prompt|model|StrOutputParser())# ---------------------------------------------------------# 6. 执行调用# ---------------------------------------------------------input_text我该如何减肥# 现在只需传入一个字符串不需要手动检索或组装字典reschain.invoke(inputinput_text)print(f 模型回答:\n{res})️ 运行流程深度解析当调用chain.invoke(我该如何减肥)时内部发生了以下神奇的并行处理输入分发字符串我该如何减肥同时进入了两个分支。分支处理Keyinput:RunnablePassthrough()直接返回原字符串我该如何减肥。Keycontext:字符串进入retriever在向量库中搜索返回 2 个Document对象。Document列表进入format_func被拼接成字符串[减肥就是要少吃多练跑步是很好的减肥方法...]。自动合并LangChain 自动将两个分支的结果合并成一个字典{input:我该如何减肥,context:[减肥就是要少吃多练跑步是很好的减肥方法...]}后续流程这个字典被传入prompt模板填充变量生成完整的 PromptValue再经过print_prompt调试最后发给model生成答案。 关键概念详解1.RunnablePassthrough是什么它是一个特殊的 Runnable 组件作用非常简单接收什么输入就原封不动地输出什么。用途在 LCEL 链中当你需要将原始输入保留并传递到下游例如作为 Prompt 的{input}变量而同时其他分支在对输入进行处理时必须使用它。对比如果不使用它原始输入在经过第一个组件后可能会丢失或被转换导致后续无法访问。2.as_retriever()vssimilarity_searchvector_store.similarity_search(query, k): 这是一个普通的 Python 方法输入输出都是普通对象不能直接用|连接。vector_store.as_retriever(): 这是一个工厂方法返回一个BaseRetriever对象。该对象实现了Runnable接口。输入str(query)。输出List[Document]。优势可以直接参与 LCEL 链式编排 (retriever | format_func)。3. 字典语法的并行执行在 LCEL 中当你使用字典语法构建链时{a:chain_a,b:chain_b}LangChain 会并行执行chain_a和chain_b如果它们都依赖同一个输入然后将结果合并。这大大提高了 RAG 的响应速度因为检索和原始输入的处理是同时进行的。⚖️ 为什么要这样重构特性手动组装 (上节课)LCEL 自动化 (本节课)代码量多 (需手动调用 search, 循环拼接)少 (声明式配置)可维护性低 (逻辑分散)高 (逻辑集中在一个 chain 对象中)扩展性难 (添加新步骤需改多处)易 (直接在链中 输入接口需手动构造 dict仅需传入 String异步支持不支持原生支持异步 (ainvoke)这种写法是 LangChain 官方推荐的标准 RAG 模式。它不仅代码整洁而且天然支持异步调用await chain.ainvoke(...)在高并发场景下性能更优。 总结本节课我们掌握了构建生产级 RAG 应用的关键技巧RunnablePassthrough学会了如何“透传”原始输入解决数据流分支合并的难题。as_retriever理解了如何将向量存储转换为可编排的 Retriever 组件。LCEL 字典语法掌握了利用字典结构并行处理数据流并自动合并结果的强大模式。端到端封装成功将检索、格式化、提示词构建封装在一个chain对象中对外暴露极简的 String-in/String-out 接口。现在我们的 RAG 系统已经非常现代化了。但是如果用户的问题很模糊或者检索到的资料不够精准怎么办仅仅靠相似度搜索Similarity Search有时是不够的。