LangChain RAG实战——向量存储与检索篇：从向量数据库到Chain构建

平台：LangChain + 阿里云百炼（Qwen）+ ChromaDB
上篇LangChain RAG实战——数据预处理篇：从文档加载到嵌入向量化
本篇介绍RAG系统的向量存储、语义检索、RAG Chain构建三大核心模块。
下篇LangChain RAG实战——实战与优化篇：完整案例与系统优化

---

目录

1. 向量数据库存储
2. 语义检索实现
3. RAG Chain构建

---

一、向量数据库存储

1.1 主流向量数据库对比

数据库	特点	适用场景	部署方式
ChromaDB	轻量、易用	原型开发	嵌入式
FAISS	高性能、GPU加速	中等规模	本地
Milvus	云原生、分布式	大规模生产	K8s
Pinecone	托管服务	企业级应用	云

1.2 ChromaDB使用

# 04_vector_store.py
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 创建向量数据库（内存模式）
vectorstore = Chroma.from_documents(
    documents=texts,           # 分割后的文档块
    embedding=embeddings,       # 嵌入模型
    persist_directory="./chroma_db"  # 持久化路径
)

# 添加更多文档
vectorstore.add_documents(new_texts)

# 保存到磁盘
vectorstore.persist()

# 删除向量数据库
vectorstore.delete_collection()

1.3 向量数据库持久化与加载

# 04_vector_store.py

# 加载已存在的向量数据库
loaded_vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# 获取集合信息
print(f"文档总数: {loaded_vectorstore._collection.count()}")

# 删除文档
loaded_vectorstore.delete(ids=["doc_id_1", "doc_id_2"])

1.4 FAISS向量数据库

# 04_vector_store.py
from langchain_community.vectorstores import FAISS

# 创建FAISS索引
faiss_vectorstore = FAISS.from_documents(
    documents=texts,
    embedding=embeddings
)

# 保存索引
faiss_vectorstore.save_local("faiss_index")

# 加载索引
loaded_faiss = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True
)

1.5 向量数据库操作

# 04_vector_store.py

# 添加文档
vectorstore.add_documents(documents)

# 更新文档（通过ID）
vectorstore.update_document(doc_id, new_document)

# 删除文档（通过ID或过滤条件）
vectorstore.delete(ids=["doc_id_1", "doc_id_2"])

# 按元数据过滤
results = vectorstore.similarity_search(
    query="搜索内容",
    filter={"source": "document.pdf"}  # 只搜索特定来源
)

---

二、语义检索实现

2.1 基础相似度检索

# 05_retriever.py
from langchain_community.vectorstores import Chroma

# 初始化向量数据库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# 相似度搜索
results = vectorstore.similarity_search(
    query="RAG的核心优势是什么？",
    k=5  # 返回前5个最相似结果
)

for i, doc in enumerate(results):
    print(f"\n--- 结果 {i+1} ---")
    print(f"内容: {doc.page_content[:200]}...")
    print(f"来源: {doc.metadata.get('source', 'Unknown')}")

2.2 带相似度分数的检索

# 05_retriever.py

# 返回带相似度分数的结果
results_with_scores = vectorstore.similarity_search_with_score(
    query="深度学习的发展历程",
    k=10
)

for doc, score in results_with_scores:
    print(f"分数: {score:.4f} | 来源: {doc.metadata.get('source', 'Unknown')}")
    print(f"内容: {doc.page_content[:100]}...")
    print("-" * 50)

2.3 MMR（最大边际相关）检索

MMR在相关性和多样性之间取得平衡，避免返回重复内容：

# 05_retriever.py

# MMR检索
mmr_results = vectorstore.max_marginal_relevance_search(
    query="机器学习的应用领域",
    k=5,           # 总共返回5个
    fetch_k=20,    # 初始检索20个
    lambda_mult=0.5  # 0=多样性, 1=相关性
)

for i, doc in enumerate(mmr_results):
    print(f"结果 {i+1}: {doc.page_content[:100]}...")

2.4 按向量检索

# 05_retriever.py

# 将查询转为向量后检索
query_vector = embeddings.embed_query("用户问题")
results = vectorstore.similarity_search_by_vector(
    embedding=query_vector,
    k=5
)

2.5 元数据过滤检索

# 05_retriever.py

# 按时间过滤（假设有date字段）
results = vectorstore.similarity_search(
    query="最新的人工智能技术",
    k=5,
    filter={
        "date": {"$gte": "2024-01-01"}  # 只搜索2024年后的文档
    }
)

# 按来源过滤
results = vectorstore.similarity_search(
    query="技术文档",
    k=5,
    filter={"source": {"$in": ["doc1.pdf", "doc2.pdf"]}}
)

---

三、RAG Chain构建

3.1 LangChain RAG流程

用户输入 → 检索相关文档 → 组装Prompt → LLM生成 → 返回答案

3.2 基础RAG Chain

# 06_rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 初始化LLM
llm = ChatOpenAI(
    model="qwen-plus",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.7
)

# 定义Prompt模板
template = """基于以下参考内容回答用户问题。如果参考内容中没有相关信息，请如实说明。

参考内容:
{context}

用户问题: {question}

回答:"""

prompt = ChatPromptTemplate.from_template(template)

# 格式化检索结果
def format_docs(docs):
    return "\n\n".join([f"文档{i+1}: {doc.page_content}" for i, doc in enumerate(docs)])

# 构建Chain
rag_chain = (
    {"context": vectorstore.as_retriever(search_kwargs={"k": 5}) | format_docs,
     "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 执行查询
result = rag_chain.invoke("RAG技术有什么优势？")
print(result)

3.3 使用LCEL构建RAG

# 06_rag_chain.py
from langchain import hub
from langchain_core.documents import Document

# 从LangChain Hub加载预定义Prompt
rag_prompt = hub.pull("rlm/rag-prompt")

# 自定义RAG Prompt
custom_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的AI助手。请根据提供的参考文档，准确回答用户问题。"),
    ("human", """参考文档:
{context}

用户问题: {question}

请给出详细、准确的回答。""")
])

# 构建完整Chain
rag_chain = (
    {
        "context": vectorstore.as_retriever() | format_docs,
        "question": RunnablePassthrough()
    }
    | custom_prompt
    | llm
    | StrOutputParser()
)

3.4 带源文档返回的Chain

# 06_rag_chain.py

# 返回答案和源文档
from langchain_core.runnables import RunnableParallel

def format_docs_with_source(docs):
    return "\n\n".join([
        f"[文档{i+1}] {doc.page_content}\n来源: {doc.metadata.get('source', 'Unknown')}"
        for i, doc in enumerate(docs)
    ])

# 修改Chain，返回原始文档
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

rag_chain_with_sources = (
    {
        "context": retriever | format_docs_with_source,
        "question": RunnablePassthrough(),
        "docs": retriever  # 同时返回原始文档
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 执行
result = rag_chain_with_sources.invoke("LangChain如何实现RAG？")
print(result)

3.5 检索器配置

# 06_rag_chain.py

# 配置检索器参数
retriever = vectorstore.as_retriever(
    search_type="similarity",  # similarity / mmr
    search_kwargs={
        "k": 5,              # 返回数量
        "score_threshold": 0.5  # 相似度阈值
    }
)

# 使用自定义检索器
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

---

本篇总结

本篇介绍了RAG系统的三大核心模块：

模块	核心组件	作用
向量存储	ChromaDB、FAISS	高效存储和检索向量
语义检索	similarity_search、MMR	精准获取相关文档
RAG Chain	LCEL表达式	端到端流水线

---

下篇预告：《LangChain RAG实战——实战与优化篇》
将介绍完整项目实战、系统优化策略、效果评估等高级内容。