当前位置：首页 > news >正文

LangChain组件 vector stores向量存储

news 2026/3/27 1:30:54

------------恢复内容开始------------

Vector Store向量存储

基于LangChain的向量存储，存储嵌入数据，并执行相似性搜索。

这是一个典型的向量存储应用，也即是典型的RAG流程

这部分开发主要涉及到：

如何文本转向量
创建向量存储，基于向量存储完成（LangChain为向量存储提供了统一接口）：
- 存入向量 add_document
- 删除向量 delete
- 向量检索 similarity_search

内置向量存储

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddingsvector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())# 添加文档到向量存储中，并指定ID
vector_store.add_documents(documents=["Hello World!", "你好，世界！"], ids=["doc1", "doc2"])# 删除文档 （通过指定的ID删除）
vector_store.delete(ids=["doc1"])# 相似性搜索 返回最相似的1个文档结果
similar_docs = vector_store.similarity_search("Hello", 1)

add_documents结合文档解析器使用

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoadervector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())csv_loader = CSVLoader(file_path="./data/sample.csv",encoding="utf-8",csv_args={"delimiter": ",","quotechar": '"',"fieldnames": ["id", "name", "age", "hobby"],},source_column="name"
)docs = csv_loader.load()
print("文档个数：", len(docs))
# 向量存储的 新增、删除、检索
vector_store.add_documents(documents=docs,ids=[f"doc_{i}" for i in range(1, len(docs) + 1)],
)# 删除 传入[id, id]
vector_store.delete(["doc_1", "doc_2"])# 检索 传入文本 返回相关文档
result = vector_store.similarity_search("喜欢球类运动的人",k=2   # 返回最相关的文档数量
)print("检索结果：", result)

外部（Chroma）向量存储

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chromavector_store = Chroma(collection_name="example_collection",embedding_function=DashScopeEmbeddings(),persist_directory="./chroma_db"
)

综合应用基于向量检索构建提示词提问

from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.vectorstores import InMemoryVectorStoremodel = ChatTongyi(model="qwen-plus")
prompt = ChatPromptTemplate([("system", "根据已提供的参考资料为主，简洁和专业地回答用户问题。参考资料：{context}"),("human", "{question}"),]
)vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())# 准备资料（向量数据库数据）
vector_store.add_texts(["SFR是用来量化摄像头成像系统清晰度（解析力、锐度）的一个关键指标","Yshading是评估成像系统在均匀光照下，画面从中心到边缘/四角的亮度衰减程度","SFR取代了早期更粗糙的 TV Line / 线对数读数法，现在几乎所有模组厂、手机厂商、第三方评测（如 DXOMARK、Image Engineering）都用 SFR 作为核心清晰度打分项。"
])input_text = "SFR与Yshading区别"result = vector_store.similarity_search(input_text, k=2)
reference_text = "["
for doc in result:reference_text += f"{doc.page_content},"
reference_text += "]"def print_prompt(prompt):print(prompt.to_string())print("="*20)return promptchain = prompt | print_prompt | model | StrOutputParser()res = chain.invoke({"context": reference_text, "question": input_text})print(res)

RunnablePassthrough的使用向量检索构建提示词入链

之前的代码中创建了vector_store的向量库，但是它不是Runnable接口的子类实例，无法入链；

在实际需求中，向量检索入链也是比较常用的方式

LangChain为向量存储对象提供了一个as_retriever方法，可以返回一个Runnable接口子类实例对象，以达到入链的目的

retriever = vector_store.as_retriever(search_kwargs={"k": 2})

其中的参数search_kwargs中的k表示之前的代码中result = vector_store.similarity_search(input_text, k=2)对应的k=2，最大匹配结果

retriever 入链输入输出问题

chain = retriever | prompt | print_prompt | model | StrOutputParser()

对于入链，还要考虑每个组件的输入输出问题

retriever：

输入：用户提问 str
输出：向量库检索结果 list[Document]

prompt：

输入：用户提问 + 向量库的检索结果 dict
输出：完整的提示词 PromptValue

如果这样入链，发现输入输出类型完全对应不上，链条会失败，同时用户提问在经过retriever后被丢弃，仅输出了list[Document]

此时就需要用到RunnablePassthrough()功能类似于占位符，在链中，它会复制一份链输入内容给自己，而prompt需要输入字典，retriever输入为字符串

那么我们可以构造一个字典入链，如下

chain = ({"question": RunnablePassthrough(), "context": retriever | format_func } | prompt | print_prompt | model | StrOutputParser()
)

这样链的输入同时给到了字典中的input与context，retriever | format_func则是一条子链，将retriever输出的list[Document]转为字符串接入字典中

这样就完成了向量检索的入链，关键在于RunnablePassthrough()的使用，下面是完整实现

from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStoremodel = ChatTongyi(model="qwen-plus")
prompt = ChatPromptTemplate([("system", "根据已提供的参考资料为主，简洁和专业地回答用户问题。参考资料：{context}"),("human", "{question}"),]
)vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())# 准备资料（向量数据库数据）
vector_store.add_texts(["SFR是用来量化摄像头成像系统清晰度（解析力、锐度）的一个关键指标","Yshading是评估成像系统在均匀光照下，画面从中心到边缘/四角的亮度衰减程度","SFR取代了早期更粗糙的 TV Line / 线对数读数法，现在几乎所有模组厂、手机厂商、第三方评测（如 DXOMARK、Image Engineering）都用 SFR 作为核心清晰度打分项。"
])input_text = "SFR是什么？"# `as_retriever`方法，可以返回一个Runnable接口子类实例对象，以达到入链的目的
retriever = vector_store.as_retriever(search_kwargs={"k": 2})def print_prompt(prompt):print(prompt.to_string())print("="*20)return promptdef format_func(docs):if not docs:return "没有找到相关资料"reference_text = "["for doc in docs:reference_text += f"{doc.page_content},"reference_text += "]"return reference_text
chain = ({"question": RunnablePassthrough(), "context": retriever | format_func } | prompt | print_prompt | model | StrOutputParser()
)res = chain.invoke(input_text)print(res)

查看全文

http://www.jsqmd.com/news/453226/