欢迎来到啾啾的博客🐱。
记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。
有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。

引言

在使用Document Loaders加载外部数据，并通过Text Splitter将大文本数据分割成更小、更易于管理的文本块chunks后，下一步，就是将这些文本块转换成机器能够理解和比较其语义含义的形式。
即文本转向量，将文本（或其他类型的数据，如图像）转换成一个固定大小的数字列表（即向量）。
这个文本转向量的过程就是嵌入 (Embeddings)。
而转换完成的向量则需要存入向量数据库。

向量数据库

向量数据库是专门设计用于存储、管理和高效检索大量嵌入向量的数据库。
在RAG流程中，用户查询也转换成一个嵌入向量，需要快速地从存储的所有向量中找到与查询向量最相似的 K 个向量（以及它们对应的原始文本块）。传统的数据库不擅长这种基于向量相似度的高维空间搜索。
向量数据库使用特殊的索引结构（如 HNSW, IVF, LSH）来实现高效的近似最近邻 (ANN) 搜索。

• 常见的向量数据库：
  • 本地/内存中:
    • FAISS (Facebook AI Similarity Search): 一个非常流行的开源库，可以在内存中或磁盘上运行。LangChain 集成了它。简单易用，适合中小型数据集。
    • ChromaDB: 一个开源的、为AI应用设计的嵌入数据库。易于上手，可以持久化到磁盘。
    • LanceDB: 另一个开源的、为AI/ML设计的嵌入数据库，专注于性能和易用性。
  • 云托管/生产级:
    • Pinecone: 托管的、高性能的向量数据库服务。
    • Weaviate: 开源的向量搜索引擎，可以自托管或使用云服务。
    • Milvus: 开源的、为大规模向量相似性搜索设计的数据库。
    • Redis: 通过 RediSearch 模块支持向量搜索。
    • PostgreSQL (pgvector): 通过扩展支持向量存储和搜索。

LangChain为向量存储提供了一个统一的抽象类VerctorStore，以及许多具体数据库的实现。

• 常用创建方法:

  • from_documents(cls, documents: List[Document], embedding: Embeddings, **kwargs) -> VectorStore: 这是一个类方法，接收分割好的 Document 对象列表和 Embeddings 对象，它会自动为每个文档的 page_content 生成嵌入，并将文档和嵌入一起存储到向量数据库中。
  • from_texts(cls, texts: List[str], embedding: Embeddings, metadatas: Optional[List[dict]] = None, **kwargs) -> VectorStore: 类似 from_documents，但直接接收文本列表和可选的元数据列表。

• 核心检索方法:

  • similarity_search(self, query: str, k: int = 4, **kwargs) -> List[Document]: 接收一个查询字符串，内部会调用 embedding.embed_query(query) 将其转换为向量，然后在数据库中执行相似性搜索，返回最相似的 k 个 Document 对象。
  • similarity_search_by_vector(self, embedding: List[float], k: int = 4, **kwargs) -> List[Document]: 直接接收一个查询向量进行搜索。
  • as_retriever(**kwargs) -> VectorStoreRetriever: 将向量存储本身转换成一个 LangChain Retriever 对象，这是在 RAG 链中更常用的方式。

Embeddings

LangChain为嵌入提供了统一的抽象类Embeddings(langchain_core.embeddings.Embeddings),许多具体模型的实现（如 OpenAIEmbeddings, HuggingFaceEmbeddings, DeepSeekEmbeddings 等）。

• 主要方法：
  • embed_documents(self, texts: List[str]) -> List[List[float]]: 接收一个文本字符串列表，为每个文本生成一个嵌入向量，并返回一个向量列表。
  • embed_query(self, text: str) -> List[float]: 接收一个单独的查询文本字符串，为其生成一个嵌入向量，并返回该向量。通常，用于文档的嵌入和用于查询的嵌入模型是相同的，但某些模型可能会有细微差别或针对查询进行优化。

嵌入模型由各厂商大模型提供，如OpenAI的嵌入模型text-embedding-ada-002。

文本转向量存储

我们从HuggingFaceEmbeddings开始，中文嵌入模型AI推荐使用 shibing624/text2vec-base-chinese（小巧），或者更强大的 BAAI/bge-small-zh-v1.5 、 BAAI/bge-base-zh-v1.5 等模型。
向量数据库则推荐使用FAISS（一个简单的内存向量数据库）。
windows安装FAISS弯弯绕绕，麻烦，推荐使用Chroma。
Chroma官方文档：https://docs.trychroma.com/
ChromaDB 是一个开源的、为 AI 应用设计的嵌入数据库，它易于上手，并且可以将数据持久化到磁盘，这对于多次运行脚本或在不同会话间保持数据非常有用。

环境配置

pip install langchain-chroma -i https://mirrors.aliyun.com/pypi/simple/

Demo

步骤简单总结：初始化一个Embeddings 模型，使用 VectorStore.from_documents(chunks, embedding_model) 将这些文本块及其嵌入向量存入向量数据库

import os
from langchain_core.documents import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter

# --- 1. 嵌入 (Embeddings) ---
from langchain_huggingface import HuggingFaceEmbeddings

# --- 2. 向量数据库 (Vector Stores) ---
from langchain_chroma import Chroma


# (复用之前的示例文本和文档对象)
long_text_example = """LangChain 是一个用于开发由语言模型驱动的应用程序的框架。
它使得应用程序能够：
- 具有上下文感知能力：将语言模型连接到上下文来源（提示指令、少量示例、需要响应的内容等）。
- 具有推理能力：依赖语言模型进行推理（关于如何根据提供的上下文进行操作、何时进行操作等）。

LangChain 的主要价值主张是：
1. 组件化：LangChain 提供了模块化的抽象，用于构建语言模型应用程序所需的组件。
2. 用例驱动的链：LangChain 还提供了特定用例的链，这些链是预先构建的组件组合。

文本分割是处理长文本时的重要步骤。选择合适的分割器和参数对于后续的嵌入和检索效果至关重要。
嵌入模型将文本转换为数字向量，捕捉其语义。相似的文本具有相似的向量。
向量数据库用于存储这些向量并进行高效的相似性搜索。Chroma 是一个流行的选择。
"""

sample_document = Document(page_content=long_text_example, metadata={"source": "manual_example", "category": "framework_intro", "doc_id": "doc1"})

another_document_content = """DeepSeek 是由深度求索公司开发的一系列大型语言模型。
它包括了代码模型和通用对话模型。DeepSeek Coder 在代码生成任务上表现优异。
DeepSeek-LLM 67B 是一个强大的通用基础模型。
这些模型旨在推动人工智能领域的发展，并为开发者提供强大的工具。
深度求索致力于开源其研究成果，促进社区合作。
"""
another_document = Document(page_content=another_document_content, metadata={"source": "deepseek_info", "category": "llm_provider", "doc_id": "doc2"})

all_documents = [sample_document, another_document]


# --- 步骤 1: 分割文档 ---
def get_document_chunks(documents_to_split):
    print("\n--- 步骤 1: 分割文档 ---")
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=200,
        chunk_overlap=20,
        length_function=len,
    )
    chunks = text_splitter.split_documents(documents_to_split)
    print(f"文档被分割成了 {len(chunks)} 个块。")
    return chunks

# --- 步骤 2: 初始化嵌入模型 ---
def initialize_embedding_model():
    print("\n--- 步骤 2: 初始化嵌入模型 ---")
    model_name = "BAAI/bge-small-zh-v1.5"
    model_kwargs = {'device': 'cuda'}
    encode_kwargs = {'normalize_embeddings': True}

    try:
        embeddings = HuggingFaceEmbeddings(
            model_name=model_name,
            model_kwargs=model_kwargs,
            encode_kwargs=encode_kwargs
        )
        print(f"HuggingFace 嵌入模型 '{model_name}' 初始化成功。")
        return embeddings
    except Exception as e:
        print(f"初始化嵌入模型时出错: {e}")
        return None


# --- 步骤 3 & 4: 使用 Chroma 创建向量数据库并进行相似性搜索 ---
def create_chroma_vector_store_and_search(document_chunks, embedding_model, persist_directory="./chroma_db_store"):
    if not document_chunks or not embedding_model:
        print("文档块或嵌入模型未准备好，跳过向量数据库创建和搜索。")
        return None

    print("\n--- 步骤 3: 从文档块创建 Chroma 向量数据库 ---")
    # 如果 persist_directory 存在且包含数据，Chroma 会尝试加载它。
    # 否则，它会创建一个新的数据库并持久化到该目录。
    # 如果只想在内存中运行（不持久化），可以不指定 persist_directory。
    # vector_store = Chroma.from_documents(documents=document_chunks, embedding=embedding_model) # 内存模式

    try:
        # 确保目录存在，如果需要
        if persist_directory and not os.path.exists(persist_directory):
            os.makedirs(persist_directory)
            print(f"创建持久化目录: {persist_directory}")

        vector_store = Chroma.from_documents(
            documents=document_chunks, # 文档块列表
            embedding=embedding_model,   # 嵌入函数
            persist_directory=persist_directory, # 持久化存储的目录
            collection_name="my_knowledge_base" # （可选）给集合一个名字
        )
        print(f"Chroma 向量数据库创建/加载成功，并持久化到 '{persist_directory}'。")
        # 调用 persist() 确保数据被写入磁盘 (对于某些版本的Chroma可能是必要的，或者在特定操作后)
        # vector_store.persist() # 在最新版本中，from_documents 时指定 persist_directory 通常会自动处理

    except Exception as e:
        print(f"创建 Chroma 向量数据库时出错: {e}")
        return None

    print("\n--- 步骤 4: 进行相似性搜索 (使用 Chroma) ---")
    query1 = "LangChain的价值主张是什么？"
    print(f"\n查询 1: '{query1}'")
    results1 = vector_store.similarity_search(query1, k=2)
    for i, doc in enumerate(results1):
        print(f"  结果 {i+1} (来自: {doc.metadata.get('source', 'N/A')}, 元数据: {doc.metadata}):")
        print(f"    内容: '{doc.page_content[:150]}...'")

    query2 = "DeepSeek 是什么？"
    print(f"\n查询 2: '{query2}'")
    results2 = vector_store.similarity_search_with_score(query2, k=2)
    for i, (doc, score) in enumerate(results2):
        # Chroma 的分数通常是距离 (如 L2 距离或余弦距离的相反数)，越小越相似
        # 或者如果配置为余弦相似度，则越大越相似。默认情况下，LangChain 的 Chroma 包装器
        # 返回的 score 是距离，所以越小越好。
        print(f"  结果 {i+1} (来自: {doc.metadata.get('source', 'N/A')}, 得分: {score:.4f}, 元数据: {doc.metadata}):")
        print(f"    内容: '{doc.page_content[:150]}...'")

    # 使用元数据过滤进行搜索
    query_filtered = "LangChain的组件有哪些？"
    print(f"\n带元数据过滤的查询: '{query_filtered}' (只在 'framework_intro' 类别中搜索)")
    try:
        # Chroma 的过滤语法是使用 `where` 子句，类似于 MongoDB
        # 对于简单的等值匹配，可以这样写：
        results_filtered = vector_store.similarity_search(
            query_filtered,
            k=2,
            filter={"category": "framework_intro"} # LangChain 抽象层面的 filter
            # 或者使用 Chroma 原生的 where (更灵活，但可能需要通过 search_kwargs 传递)
            # search_kwargs={'where': {'category': 'framework_intro'}}
        )
        if results_filtered:
            for i, doc in enumerate(results_filtered):
                print(f"  过滤结果 {i+1} (元数据: {doc.metadata}):")
                print(f"    内容: '{doc.page_content[:150]}...'")
        else:
            print("  没有找到符合过滤条件的文档。")
    except Exception as e:
        print(f"  带元数据过滤的搜索失败: {e}")


    # 将 VectorStore 转换为 Retriever
    retriever = vector_store.as_retriever(
        search_kwargs={"k": 3, "filter": {"category": "llm_provider"}}
    )
    print(f"\n查询 (使用 Retriever, 过滤类别 'llm_provider'): '{query2}'") # 用 query2 测试
    retrieved_docs = retriever.invoke(query2)
    for i, doc in enumerate(retrieved_docs):
        print(f"  Retrieved Doc {i+1} (元数据: {doc.metadata}): '{doc.page_content[:100]}...'")

    return vector_store

# --- 如何加载已持久化的 Chroma 数据库 ---
def load_persisted_chroma_db(embedding_model, persist_directory="./chroma_db_store"):
    if not embedding_model:
        print("嵌入模型未初始化，无法加载 Chroma 数据库。")
        return None
    if not os.path.exists(persist_directory):
        print(f"持久化目录 '{persist_directory}' 不存在，无法加载。")
        return None

    print(f"\n--- 尝试从 '{persist_directory}' 加载已存在的 Chroma 数据库 ---")
    try:
        loaded_vector_store = Chroma(
            persist_directory=persist_directory,
            embedding_function=embedding_model, # 必须提供相同的嵌入函数
            collection_name="my_knowledge_base" # 如果创建时指定了，加载时也需要
        )
        print("Chroma 数据库加载成功。")

        # 测试一下加载的数据库
        query_test = "什么是 LangChain 组件化？"
        print(f"\n在加载的数据库中测试查询: '{query_test}'")
        results_test = loaded_vector_store.similarity_search(query_test, k=1)
        if results_test:
            print(f"  结果 (元数据: {results_test[0].metadata}): '{results_test[0].page_content[:100]}...'")
        else:
            print("  未找到结果。")
        return loaded_vector_store
    except Exception as e:
        print(f"加载 Chroma 数据库时出错: {e}")
        print("确保持久化目录存在且包含有效的 Chroma 数据，并且嵌入函数与创建时一致。")
        return None

if __name__ == '__main__':
    doc_chunks = get_document_chunks(all_documents)
    embeddings_model = initialize_embedding_model()

    if embeddings_model:
        # 首次运行时会创建并持久化
        chroma_db = create_chroma_vector_store_and_search(doc_chunks, embeddings_model, persist_directory="./my_chroma_data")

        if chroma_db:
            print("\nChromaDB 创建和检索演示完成。")
            # 你可以删除 ./my_chroma_data 目录再运行，它会重新创建
            # 或者注释掉 create_chroma_vector_store_and_search，然后取消下面的注释来测试加载

        # --- 测试加载持久化的数据库 ---
        # (确保上面的 persist_directory 和这里的路径一致，比如都用 "./my_chroma_data")
        print("\n--- 第二次运行或在不同脚本中，可以尝试加载 ---")
        loaded_db = load_persisted_chroma_db(embeddings_model, persist_directory="./my_chroma_data")
        if loaded_db:
            # 你现在可以使用 loaded_db 进行更多查询了
            query_after_load = "深度求索公司是做什么的？"
            print(f"\n用加载的DB查询: '{query_after_load}'")
            search_results = loaded_db.similarity_search(query_after_load, k=1)
            if search_results:
                 print(f"  结果 (元数据: {search_results[0].metadata}): '{search_results[0].page_content[:100]}...'")

    else:
        print("由于嵌入模型初始化失败，无法继续进行向量数据库操作。")

API解释

Embeddings

Embeddings (例如 HuggingFaceEmbeddings, OpenAIEmbeddings)

• 用途: 将文本转换为数字向量。
• 构造函数参数 (常见):
  • model_name (HuggingFace): 指定 Hugging Face Hub 上的模型名称 (例如 “BAAI/bge-small-zh-v1.5”)。
  • model_kwargs (HuggingFace): 传递给底层 sentence-transformers 模型的额外参数，如 {‘device’: ‘cpu’}。
  • encode_kwargs (HuggingFace): 传递给嵌入编码函数的额外参数，如 {‘normalize_embeddings’: True}。
  • api_key / openai_api_key (OpenAI): 你的 API 密钥。
  • model (OpenAI): 指定 OpenAI 的嵌入模型名称 (例如 “text-embedding-ada-002”)。
• 核心方法:
  • embed_documents(self, texts: List[str]) -> List[List[float]]:
    • 输入: texts - 一个字符串列表，每个字符串是一个文档块的内容。
    • 输出: 一个向量列表，每个向量是对应输入文本的嵌入表示。
    • 何时使用: 当你需要为一批文档块生成嵌入时，例如在创建向量数据库的初始阶段。
  • embed_query(self, text: str) -> List[float]:
    • 输入: text - 一个字符串，通常是用户的查询。
    • 输出: 该查询文本的嵌入向量。
    • 何时使用: 当用户提出问题，你需要将其转换为向量以便在向量数据库中进行搜索时。

VectorStore

VectorStore (例如 FAISS, Chroma)

• 用途: 存储文档的嵌入向量，并提供高效的相似性搜索功能。

• 常用类方法 (创建 VectorStore):

  • from_documents(cls, documents: List[Document], embedding: Embeddings, **kwargs) -> VectorStore:

    • 输入:
      • documents: 分割后的 Document 对象列表。
      • embedding: 你初始化的 Embeddings 对象 (如 HuggingFaceEmbeddings 实例)。
      • **kwargs: 特定向量数据库的额外参数 (例如，对于 FAISS，你可能不需要太多额外参数进行基本创建)。
    • 过程: 遍历 documents，对每个 doc.page_content 调用 embedding.embed_documents() (内部通常是批量处理)，然后将原始文档内容、元数据和其嵌入向量一起存入向量数据库。
    • 输出: 一个配置好的 VectorStore 实例。
    • 何时使用: 这是最常用的创建和填充向量数据库的方式。

  • from_texts(cls, texts: List[str], embedding: Embeddings, metadatas: Optional[List[dict]] = None, **kwargs) -> VectorStore:

    • 输入:
      • texts: 纯文本字符串列表。
      • embedding: Embeddings 对象。
      • metadatas (可选): 与 texts 列表对应的元数据字典列表。
    • 输出: VectorStore 实例。
    • 何时使用: 如果你只有文本和元数据，而不是 Document 对象，可以使用这个方法。

• 核心实例方法 (检索):

  • similarity_search(self, query: str, k: int = 4, filter: Optional[dict] = None, **kwargs) -> List[Document]:
    • 输入:
      • query: 用户的查询字符串。
      • k: 希望返回的最相似文档数量。
      • filter (可选): 一个字典，用于根据元数据过滤搜索结果 (例如 {“source”: “manual_example”} 只搜索来源是 manual_example 的文档块)。支持情况和语法因向量数据库而异。
      • **kwargs: 特定向量数据库的额外搜索参数。
    • 过程:
      1. 内部调用 self.embedding_function.embed_query(query) (这里的 embedding_function 是创建 VectorStore 时传入的 embedding 对象) 得到查询向量。
      2. 在数据库中执行与该查询向量的相似性搜索。
    • 输出: 一个包含 k 个最相似 Document 对象的列表，这些对象是根据与查询的语义相似度排序的。
    • 何时使用: 直接进行一次性的相似性搜索。
  • similarity_search_with_score(self, query: str, k: int = 4, filter: Optional[dict] = None, **kwargs) -> List[Tuple[Document, float]]:
    • 与 similarity_search 类似，但返回的是 (Document, score) 的元组列表，其中 score 表示相似度得分。
    • 注意: 得分的含义因向量数据库而异。对于 FAISS (使用 L2 距离)，得分越小越相似。对于使用余弦相似度的，得分越大越相似（通常在 -1 到 1 或 0 到 1 之间）。
    • 何时使用: 当你需要了解相似度得分本身时。
  • similarity_search_by_vector(self, embedding: List[float], k: int = 4, filter: Optional[dict] = None, **kwargs) -> List[Document]:
    • 输入:
      • embedding: 一个已经计算好的查询嵌入向量。
    • 其他参数与 similarity_search 类似。
    • 何时使用: 如果你已经有了查询向量，不想让 VectorStore 再次计算它。
  • add_documents(self, documents: List[Document], **kwargs) -> List[str]:
    • 输入: documents - 一个新的 Document 对象列表。
    • 过程: 为这些新文档生成嵌入，并将它们添加到现有的向量数据库中。
    • 输出: 通常是这些新添加文档的 ID 列表。
    • 何时使用: 当你需要向一个已经存在的向量数据库中增量添加新数据时。

• 转换为 Retriever:

  • as_retriever(self, search_type: str = “similarity”, search_kwargs: Optional[dict] = None) -> VectorStoreRetriever:
    • 输入:
      • search_type: 检索类型，默认为 “similarity”。其他类型可能包括 “mmr” (Maximal Marginal Relevance，最大边际相关性，用于多样化结果) 或 “similarity_score_threshold” (基于相似度得分阈值过滤)。
      • search_kwargs: 传递给底层相似性搜索方法的参数字典，例如 {“k”: 5, “filter”: {“category”: “llm_provider”}}。
    • 输出: 一个 VectorStoreRetriever 对象。Retriever 是 LangChain 中一个更通用的概念，代表任何可以根据查询检索文档的东西。VectorStoreRetriever 是 VectorStore 的一个适配器，使其可以无缝地集成到 LangChain 的链 (Chains) 中，特别是 RAG 链。
    • 何时使用: 这是在构建 RAG 链时推荐的方式。 它将检索逻辑封装起来，使得链的定义更简洁。