26. LangChain组件——RunnablePassthrough的使用
1. 目的：将向量检索加入链
2. 代码实战
"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from typing import List
def print_prompt(prompt):
print(prompt.to_string())
print("=" * 20)
return prompt
model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
[
("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
("user", "用户提问：{input}")
]
)
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))
# 准备一下资料（向量库的数据）
# add_texts 传入一个 list
vector_store.add_texts(
["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])
input_text = "怎么减肥？"
# langchain中向量存储对象，有一个方法：as_retriever，可以返回一个Runnable接口的子类实例对象
retriever = vector_store.as_retriever(search_kwargs={"k": 2})
def format_func(docs: List):
if not docs:
return "无相关参考资料"
formatted_str = "["
for doc in docs:
formatted_str += doc.page_content
formatted_str += "]"
return formatted_str
# chain
chain = (
{"input": RunnablePassthrough(), "context": retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)
res = chain.invoke(input_text)
print(res)
"""
retriever:
- 输入：用户的提问 str
- 输出：向量库的检索结果list
prompt:
- 输入：用户的提问 + 向量库的检索结果 dict
- 输出：完整的提示词 PromptValue
"""
26. LangChain组件——RunnablePassthrough的使用
1.jpg1. 目的：
将向量检索加入链
2. 代码实战
"""
提示词：用户的提问 + 向量库中检索到的参考资料
"""
RunnablePassthrough 是 LangChain 中的一个组件，它主要用于在链中传递输入数据而不做任何修改。这允许你将原始输入与其他处理步骤的结果组合起来，例如将用户问题与向量检索的上下文结合。
在以下代码中，我们使用 RunnablePassthrough 来传递用户的输入问题，同时通过向量检索获取相关参考资料，然后将两者合并到提示词中，最后生成回答。
以下是完整的代码示例，填充了缺失的部分，以便你可以直接运行：
from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from typing import List
def print_prompt(prompt):
print(prompt.to_string())
print("=" * 20)
return prompt
# 初始化模型
model = ChatTongyi(model="qwen3-max")
# 定义提示词模板，结合用户输入和检索上下文
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有用的助手。请根据以下参考资料回答问题。如果参考资料不相关，请基于你的知识回答。"),
("human", "问题：{input}\n参考资料：{context}")
])
# 初始化向量存储
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))
# 向向量存储中添加示例文本（参考资料）
vector_store.add_texts([
"减肥需要均衡饮食和定期运动。",
"建议每天摄入少于2000卡路里，并每周运动3次。",
"多喝水有助于新陈代谢。"
])
input_text = "怎么减肥？"
# 创建检索器，设置返回前2个最相关结果
retriever = vector_store.as_retriever(search_kwargs={"k": 2})
# 格式化检索到的文档为字符串
def format_func(docs: List):
if not docs:
return "无相关参考资料"
formatted_str = "\n".join()
return formatted_str
# 构建链：使用RunnablePassthrough传递输入，同时检索上下文
chain = (
{"input": RunnablePassthrough(), "context": retriever | format_func}
| prompt
| print_prompt
| model
| StrOutputParser()
)
# 调用链并输出结果
res = chain.invoke(input_text)
print(res)
代码解释：
- RunnablePassthrough() 直接将用户的输入（字符串）传递到链中，作为字典的 "input" 键的值。
- retriever | format_func 部分执行向量检索，并格式化结果为字符串，作为字典的 "context" 键的值。
- 链将这些组合成一个字典，传递给提示模板，然后经过模型和输出解析器生成最终回答。
关键点：
- RunnablePassthrough 常用于在链中保留原始输入，避免数据丢失。
- 这种方式使得链更灵活，可以轻松集成多个数据源。
如果你有其他问题，请随时提问！
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。
页:
[1]