Langchain学习（7）：向量存储实战指南-从内存测试到持久化RAG链构建

Jackylw6

476人浏览 · 2026-02-18 00:30:26

Jackylw6 · 2026-02-18 00:30:26 发布

在RAG（检索增强生成）系统中，向量存储是连接“知识”与“智能”的核心枢纽。它将文本转化为可计算的向量，并通过高效检索为大模型提供精准上下文。本文将深度解析向量存储的全链路实践：从内存测试、持久化部署到RAG链构建，每行代码皆有深意，每个参数皆有讲究。

向量存储核心流程全景图

流程本质：
索引阶段：将知识“编码”为向量并持久化存储（离线）
检索阶段：将问题“编码”后匹配最相关知识（在线）
二者通过同一嵌入模型保证向量空间一致性

环境准备

pip install chroma
# 或
pip install langchain-chroma # LangChain ≥ 0.2.9

版本陷阱：

LangChain ≥ 0.2.9 时，from langchain_community.vectorstores import Chroma 已失效
须使用 from langchain_chroma import Chroma（新包独立维护）

一、内存向量存储

为什么用 InMemoryVectorStore？

场景	优势	局限
原型开发	无需磁盘IO，秒级启动	进程结束即销毁
单元测试	隔离环境，结果可复现	不适合生产
小数据验证	直观观察检索效果	内存占用随数据增长

数据准备：`python_opinion.csv`

name,text
张三,Python语言设计得很友好，学起来很轻松。
李四,Python入门门槛低，是编程学习的好选择。
王五,Python语言设计得很友好，学起来很轻松。
赵六,Python入门门槛低，是编程学习的好选择。
钱七,我觉得Python真的不难，上手很快。
孙八,Python想要精通很难，需要花很多时间学习。
周九,Python的异步编程和装饰器这些概念真的很难。
吴十,Python很难，有太多概念需要理解。
郑一,Python的异步编程和装饰器这些概念真的很难。
王二,Python想要精通很难，需要花很多时间学习。

完整代码

from langchain_community.document_loaders import CSVLoader
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings()
)

loader = CSVLoader(
    file_path="csv_data/python_opinion.csv",
    encoding="utf-8",
    source_column="name",  # 指定源列   
)

documents = loader.load()

# 向量存储的 新增、删除、检索
# 新增
vector_store.add_documents(
    documents,  # 文档列表
    ids=["id" + str(i) for i in range(1, len(documents) + 1)]  # 指定文档id
)

# 删除
vector_store.delete(
    ids=["id1", "id2"]
)

# 检索
res = vector_store.similarity_search(
    query="python学起来很简单",  # 查询
    k=1  # 返回的文档数量
)

print(res)

关键参数解析

操作	参数	作用	为什么重要
加载CSV	`source_column="name"`	将"name"列值存入metadata	检索结果可显示"来源：钱七"，增强可信度
添加文档	`ids=[...]`	为每条文档指定唯一ID	后续精准删除/更新的基础（如删除"id1"）
检索	`k=1`	返回最相似的1条结果	控制上下文长度，避免噪声干扰

输出解读：

[Document(id='id5', metadata={'source': '钱七', 'row': 4}, page_content='name: 钱七\ntext: 我觉得Python真的不难，上手很快。')]

id='id5'：我们指定的唯一标识
metadata：自动携带来源人名+原始行号
page_content：包含"name"和"text"两列（CSVLoader默认行为）

二、持久化向量存储

为什么必须持久化？

问题	内存存储	Chroma持久化
服务重启	数据全丢	秒级恢复
多进程共享	无法共享	同一目录可多服务读取
大数据量	内存爆炸	磁盘存储+索引优化

无缝迁移：仅替换向量存储初始化代码

# 原内存存储
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())

# ↓ 替换为 ↓

# 持久化存储
from langchain_chroma import Chroma
vector_store = Chroma(
    collection_name="test",  # 向量集合名称
    embedding_function=DashScopeEmbeddings(),  # 嵌入模型
    persist_directory="./chroma_db/"  # 持久化目录
)

关键细节：

embedding → embedding_function（Chroma参数命名差异！）
persist_directory 需提前存在（代码中用 os.makedirs 确保）
首次运行自动创建数据库；后续运行直接加载已有向量

完整代码

import os

from langchain_community.document_loaders import CSVLoader
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_chroma import Chroma

persist_directory = "./chroma_storage"
os.makedirs(persist_directory, exist_ok=True)  # 确保目录存在

vector_store = Chroma(
    collection_name="test",  # 向量存储的名字
    embedding_function=DashScopeEmbeddings(),  # 嵌入模型
    persist_directory=persist_directory  # 向量存储的目录
)

loader = CSVLoader(
    file_path="csv_data/python_opinion.csv",
    encoding="utf-8",
    source_column="name",  # 指定源列
)

documents = loader.load()

# 向量存储的 新增、删除、检索
# 新增
vector_store.add_documents(
    documents,  # 文档列表
    ids=["id" + str(i) for i in range(1, len(documents) + 1)]  # 指定文档id
)

# 删除
vector_store.delete(
    ids=["id1", "id2"]
)

# 检索
res = vector_store.similarity_search(
    query="python学起来很简单",  # 查询
    k=1  # 返回的文档数量
)

print(res)

二次查询验证（无需重新插入）

from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from langchain_chroma import Chroma

persist_directory = "./chroma_storage"

vector_store = Chroma(
    collection_name="test",
    embedding_function=DashScopeEmbeddings(),
    persist_directory=persist_directory
)

res = vector_store.similarity_search(query="python学起来很简单", k=1)
print(res)

效果验证：

删除的 id1, id2 不再出现
检索结果与首次插入后一致
全程无文件读写操作——向量已持久化在 ./chroma_storage 中

三、向量检索构建提示词：RAG链雏形

核心思想

将检索结果作为系统提示词的一部分，强制模型基于事实回答，避免“幻觉”。

为什么这样设计提示词？

("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。")

“以…为主”：明确约束模型行为
“参考资料”：建立用户对答案来源的信任
{context}占位符：动态注入检索结果（关键！）

完整代码

from langchain_community.chat_models import ChatTongyi
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

model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问: {input}")
    ]
)

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥?"

result = vector_store.similarity_search(input_text, k=2)
reference_text = "["
for doc in result:
    reference_text += doc.page_content + "\n"
reference_text += "]"


def print_prompt(prompt):
    print(f"{prompt.to_string()}")
    print("=" * 30)
    return prompt


chain = prompt | print_prompt | model | StrOutputParser()

res = chain.invoke({"context": reference_text, "input": input_text})
print(f"res: {res}")

输出：

System: 以我提供的已知参考资料为主...参考资料:[减肥就是要少吃多练
在减脂期间吃东西很重要...]
==============================
res: 减肥的关键在于“少吃多练”...

打印提示词：调试RAG链（确认context是否注入）

四、RunnablePassthrough：新范式RAG链

为什么升级到 RunnablePassthrough？

传统方式	RunnablePassthrough方式	优势
手动检索+拼接context	声明式链构建	代码简洁，逻辑清晰
难以复用	可组合、可测试	符合LangChain 0.2+最佳实践
调试困难	中间结果可观察	通过`print_prompt`精准定位问题

核心组件解析

{"input": RunnablePassthrough(), "context": retriever | format_func}

RunnablePassthrough()：原样传递用户输入到后续节点
retriever | format_func：并行执行检索+格式化，结果注入context

完整代码

from langchain_community.chat_models import ChatTongyi
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 langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

model = ChatTongyi(model="qwen3-max-preview")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问: {input}")
    ]
)

vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))
vector_store.add_texts(
    ["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥?"

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

def print_prompt(prompt):
    print(f"{prompt.to_string()}")
    print("=" * 30)
    return prompt

def format_func(docs: list[Document]):
    if not docs:
        return "无相关参考资料"
    formatted_str = "["
    for doc in docs:
        formatted_str += doc.page_content
    formatted_str += "]"
    return formatted_str

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

res = chain.invoke(input_text)
print(res)

设计亮点：

as_retriever(search_kwargs={"k": 2})：将向量存储转为标准检索器（解耦设计）
format_func：空结果友好处理（避免context为空导致模型困惑）
链式调用：| 操作符直观表达数据流（输入→检索→提示→模型→解析）
调试友好：print_prompt 位置精准，可观察最终注入模型的提示词

选型指南与最佳实践

场景	推荐方案	关键配置	避坑提醒
快速验证	`InMemoryVectorStore`	`source_column`注入元数据	仅限开发调试
生产部署	`Chroma` + 持久化	`persist_directory`路径权限	首次插入后勿重复add（会重复存储）
RAG链构建	`RunnablePassthrough`范式	retriever	format_func
中文优化	嵌入模型选`text-embedding-v4`	separators含中文标点	避免用英文模型处理中文