44. HuggingFaceEmbeddings #

44.1 开源模型 #

• huggingface models
• hf-mirror models
• modelscope models
• all-MiniLM-L6-v2

说明：

• 上述模型均可在 HuggingFace Transformers、sentence-transformers 等生态中直接调用，支持向量检索、语义匹配、聚类等场景。
• 选择模型时请考虑：支持语言、速度、体积以及特定任务（如检索、匹配、分类等）投射效果。

44.2. 44.HuggingFaceEmbedding.py #

44.HuggingFaceEmbedding.py

# 安装 huggingface 相关依赖（如需）
# uv add langchain-huggingface

# 导入 HuggingFaceEmbeddings 类（这里用自定义 smartchain 版本）
#from langchain_huggingface import HuggingFaceEmbeddings
from smartchain.embeddings import HuggingFaceEmbeddings

# 指定本地模型路径
model_path = "C:/Users/Administrator/.cache/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2"

# 创建 HuggingFaceEmbeddings 实例，指定模型路径和设备为 cpu
embeddings = HuggingFaceEmbeddings(
    model_name=model_path,
    model_kwargs={"device": "cpu"}
)

# 定义查询文本（单个句子）
query_text = "什么是人工智能？"

# 计算查询文本的嵌入向量
query_embedding = embeddings.embed_query(query_text)

# 打印查询嵌入向量的前5维
print(f"查询嵌入前5维: {query_embedding[:5]}")

# 定义待嵌入的多个文档（句子列表）
documents = [
    "人工智能是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。",
    "机器学习是人工智能的一个子领域，它使计算机能够从数据中学习而无需明确编程。",
    "深度学习是机器学习的一个子集，使用神经网络来模拟人脑的工作方式。",
]

# 计算每个文档的嵌入向量（批量嵌入）
doc_embeddings = embeddings.embed_documents(documents)

# 打印第一个文档嵌入向量的前5维
print(f"文档嵌入第1个前5维: {doc_embeddings[0][:5]}")

# 导入 numpy 库用于数组运算
import numpy as np

# 定义计算两个向量余弦相似度的函数
def cosine_similarity(vec1, vec2):
    # 转为 numpy 数组
    vec1 = np.array(vec1)
    vec2 = np.array(vec2)
    # 计算余弦相似度
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

# 再次获得查询文本的嵌入向量
query_emb = embeddings.embed_query(query_text)

# 计算查询与每个文档嵌入的相似度
similarities = [cosine_similarity(query_emb, doc_emb) for doc_emb in doc_embeddings]

# 找到相似度最高的文档索引
most_similar_idx = int(np.argmax(similarities))

# 打印最相似文档的编号和相似度分值
print(f"最相似文档索引: {most_similar_idx} 相似度: {similarities[most_similar_idx]:.4f}")

44.3. embeddings.py #

smartchain/embeddings.py

# 导入os模块，用于获取环境变量
import os
# 导入OpenAI模块，用于调用OpenAI嵌入服务
from openai import OpenAI
# 导入ABC模块，用于定义抽象基类
from abc import ABC, abstractmethod
# 导入SentenceTransformer类，用于加载句向量模型
from sentence_transformers import SentenceTransformer
# 导入langchain_huggingface包中的HuggingFaceEmbeddings，并重命名
from langchain_huggingface import (
    HuggingFaceEmbeddings as LangchainHuggingFaceEmbeddings,
)
# 导入requests库，用于接口请求
import requests

# 定义嵌入基类Embedding，继承ABC抽象类
class Embedding(ABC):
    # 抽象方法：嵌入单个查询文本
    @abstractmethod
    def embed_query(self, text):
        pass

    # 抽象方法：嵌入多个文档文本
    @abstractmethod
    def embed_documents(self, texts):
        pass

# 定义OpenAIEmbeddings类，实现Embedding接口
class OpenAIEmbeddings(Embedding):
    # 初始化方法
    def __init__(self, model="text-embedding-3-small", **kwargs):
        # 保存模型名
        self.model = model
        # 从参数或环境变量获取API Key
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有API Key，则抛出异常
        if not self.api_key:
            raise ValueError(f"需要提供 api_key")
        # 收集除api_key外的其他参数作为embedding_kwargs
        self.embedding_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建OpenAI客户端
        self.client = OpenAI(api_key=self.api_key)

    # 嵌入单个查询文本
    def embed_query(self, text):
        # 调用OpenAI接口获取嵌入结果
        response = self.client.embeddings.create(
            model=self.model, input=text, **self.embedding_kwargs
        )
        # 返回第一个嵌入向量
        return response.data[0].embedding

    # 嵌入多个文档文本
    def embed_documents(self, texts):
        # 批量调用OpenAI接口获取嵌入
        response = self.client.embeddings.create(
            model=self.model, input=texts, **self.embedding_kwargs
        )
        # 返回所有文档的嵌入向量列表
        return [item.embedding for item in response.data]

# 定义 HuggingFaceEmbeddings 类，实现 Embedding 接口
+class HuggingFaceEmbeddings(Embedding):
+    # 初始化方法
+    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2", **kwargs):
+        """
+        初始化 HuggingFace 嵌入模型
+
+        Args:
+            model_name: HuggingFace 模型名称，默认为 "sentence-transformers/all-MiniLM-L6-v2"
+            **kwargs: 传递给 langchain_huggingface.HuggingFaceEmbeddings 的其他参数
+        """
+        # 设置模型名称
+        self.model_name = model_name
+        # 使用 langchain_huggingface 的 HuggingFaceEmbeddings 类构建嵌入对象
+        self.embeddings = LangchainHuggingFaceEmbeddings(
+            model_name=model_name, **kwargs
+        )
+
+    # 嵌入单个查询文本
+    def embed_query(self, text):
+        # 返回HuggingFace嵌入结果
+        return self.embeddings.embed_query(text)
+
+    # 嵌入多个文档文本
+    def embed_documents(self, texts):
+        # 批量计算嵌入向量
+        return self.embeddings.embed_documents(texts)

# 定义 SentenceTransformerEmbeddings 类，实现 Embedding 接口
class SentenceTransformerEmbeddings(Embedding):
    # 初始化方法
    def __init__(self, model="all-MiniLM-L6-v2", **kwargs):
        """
        初始化 SentenceTransformer 嵌入模型

        Args:
            model: 模型名称或路径，默认为 "all-MiniLM-L6-v2"
            **kwargs: 传递给 SentenceTransformer 的其他参数
        """
        # 设置模型名称
        self.model_name = model or "all-MiniLM-L6-v2"
        # 加载 SentenceTransformer 模型
        self.model = SentenceTransformer(self.model_name, **kwargs)
        # 是否归一化嵌入向量，默认False
        self.normalize_embeddings = kwargs.get("normalize_embeddings", False)

    # 嵌入单个查询文本
    def embed_query(self, text):
        # 调用模型编码，获取嵌入向量
        embedding = self.model.encode(
            text, normalize_embeddings=self.normalize_embeddings
        )
        # 若有tolist方法则转为list，否则直接返回
        return embedding.tolist() if hasattr(embedding, "tolist") else embedding

    # 嵌入多个文档文本（批量处理）
    def embed_documents(self, texts):
        # 若输入是字符串则转成列表
        if isinstance(texts, str):
            texts = [texts]
        # 批量编码文本
        embeddings = self.model.encode(
            texts, normalize_embeddings=self.normalize_embeddings
        )
        # 转为列表格式返回
        if hasattr(embeddings, "tolist"):
            return embeddings.tolist()
        return embeddings

# 设置火山引擎文本向量API的URL
VOLC_EMBEDDINGS_API_URL = "https://ark.cn-beijing.volces.com/api/v3/embeddings"

# 定义获取火山引擎嵌入向量的函数
def get_volc_embedding(
    doc_content, model="doubao-embedding-text-240715", api_key=None, api_url=None
):
    """
    获取火山引擎文档向量

    Args:
        doc_content: 文档内容（字符串或字符串列表）
        model: 模型名称，默认为 "doubao-embedding-text-240715"
        api_key: API密钥，如果未提供则从环境变量 VOLC_API_KEY 获取
        api_url: API地址，如果未提供则使用默认地址

    Returns:
        嵌入向量（单个文本返回向量，多个文本返回向量列表）
    """
    # 如果api_key未传入，则从环境变量VOLC_API_KEY获取
    if api_key is None:
        api_key = os.getenv("VOLC_API_KEY")
    # 如果还没有api_key，抛出异常
    if not api_key:
        raise ValueError("需要提供 api_key 或设置环境变量 VOLC_API_KEY")

    # 若未指定api_url，则使用默认URL
    api_url = api_url or VOLC_EMBEDDINGS_API_URL

    # 构造请求头
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}",
    }

    # 构造请求体
    payload = {"model": model, "input": doc_content}

    # 发送POST请求获取嵌入
    try:
        response = requests.post(api_url, json=payload, headers=headers, timeout=30)
        # 检查状态码，非200自动抛异常
        response.raise_for_status()  # 如果状态码不是 200，会抛出异常

        # 解析响应体，转为json格式
        data = response.json()

        # 判断输入单/多条
        if isinstance(doc_content, list):
            # 输入为列表时，返回所有嵌入向量组成的列表
            return [item["embedding"] for item in data.get("data", [])]
        else:
            # 单条文本时，返回第一个嵌入向量
            return data["data"][0]["embedding"]

    # 请求异常处理
    except requests.exceptions.RequestException as e:
        raise Exception(f"火山引擎 Embedding API 请求失败: {str(e)}")
    # 响应解析异常处理
    except (KeyError, IndexError) as e:
        raise Exception(f"火山引擎 Embedding API 响应格式错误: {str(e)}")

# 定义火山引擎 VOLCEmbeddings 类，实现 Embedding 接口
class VOLCEmbeddings(Embedding):
    # 初始化方法
    def __init__(
        self, model="doubao-embedding-text-240715", api_key=None, api_url=None, **kwargs
    ):
        """
        初始化火山引擎嵌入模型

        Args:
            model: 模型名称，默认为 "doubao-embedding-text-240715"
            api_key: API密钥，如果未提供则从环境变量 VOLC_API_KEY 获取
            api_url: API地址，如果未提供则使用默认地址
            **kwargs: 其他参数（保留用于未来扩展）
        """
        # 保存模型名
        self.model = model
        # 保存API key，优先参数，其次环境变量
        self.api_key = api_key or os.getenv("VOLC_API_KEY")
        # 没有API key报错
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置环境变量 VOLC_API_KEY")
        # 保存API URL
        self.api_url = api_url or VOLC_EMBEDDINGS_API_URL
        # 记录其他参数
        self.embedding_kwargs = kwargs

    # 嵌入单个查询文本
    def embed_query(self, text):
        return get_volc_embedding(
            text, model=self.model, api_key=self.api_key, api_url=self.api_url
        )

    # 嵌入多个文档文本
    def embed_documents(self, texts):
        # 如果是字符串转成列表
        if isinstance(texts, str):
            texts = [texts]
        # 批量获取嵌入
        return get_volc_embedding(
            texts, model=self.model, api_key=self.api_key, api_url=self.api_url
        )

45.Chroma #

• chromadb

HuggingFace 本地模型和Chroma 向量数据库，实现本地语义检索的完整流程，包括：

• 如何加载本地的句向量模型（如 all-MiniLM-L6-v2）
• 利用 HuggingFaceEmbeddings 封装统一的文本向量化接口
• 使用 Chroma 向量存储实现文本的批量入库与向量索引
• 检索示例：给定查询语句，返回最相关的文档片段及其得分

代码示例中的 HuggingFaceEmbeddings 类封装了句向量编码，无需联网即可推理；而 Chroma 类支持本地高效的向量化内容存储和相似度搜索。
可以灵活指定待入库的文本和元数据，并按需对中文/英文/多语种文本进行检索，适用于知识问答、文档语义查找、RAG 等场景。

45.1. 45.Chroma.py #

45.Chroma.py

# 导入 HuggingFaceEmbeddings 类，用于句向量编码
from smartchain.embeddings import HuggingFaceEmbeddings
# 导入 Chroma 类，实现本地向量数据库
from smartchain.vectorstores import Chroma

# 指定本地模型路径
model_path = "C:/Users/Administrator/.cache/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2"

# 实例化 HuggingFaceEmbeddings，指定模型路径和设备为 CPU
embeddings = HuggingFaceEmbeddings(
    model_name=model_path,
    model_kwargs={"device": "cpu"}
)

# 创建 Chroma 数据库实例，指定持久化目录、嵌入函数、集合名称和集合元数据（指定使用余弦距离）
chroma_db = Chroma(
    persist_directory="chroma_db",
    embedding_function=embeddings,
    collection_name="test",
    collection_metadata={"hnsw:space": "cosine"}
)

# 定义待入库的文本列表
texts = [
    "Hello, world!",
    "Artificial Intelligence is fascinating.",
    "机器学习是人工智能的重要领域。",
    "Deep learning simulates the human brain using neural networks.",
    "欢迎使用Chroma向量数据库。",
]

# 定义对应文本的元数据列表
metadatas = [
    {"lang": "en", "category": "greeting"},
    {"lang": "en", "category": "tech"},
    {"lang": "zh", "category": "tech"},
    {"lang": "en", "category": "tech"},
    {"lang": "zh", "category": "demo"},
]

# 向 Chroma 数据库中批量添加文本和元数据信息
chroma_db.add_texts(texts, metadatas)

# 使用英语查询，检索最相关的2条文本
results = chroma_db.similarity_search("What is artificial intelligence?", k=2)
# 输出检索结果
for i, doc in enumerate(results):
    print(f"Result {i}: {doc}")

# 使用中文查询，检索最相关的2条文本，并返回得分
results_with_score = chroma_db.similarity_search_with_score("人工智能的主要领域有哪些？", k=2)
# 输出检索结果及其相似度分数
for i, (doc, score) in enumerate(results_with_score):
    print(f"Result {i}: Score={score:.4f} Doc={doc}")

45.2. vectorstores.py #

smartchain/vectorstores.py

# 导入os模块
import os
# 导入numpy用于数值计算
import numpy as np
# 导入抽象基类相关模块
from abc import ABC, abstractmethod
# 导入faiss库用于向量检索
import faiss
# 导入uuid用于生成唯一ID
+import uuid
+import chromadb
# 定义余弦相似度计算函数
def cosine_similarity(from_vec, to_vecs):
    """
    计算一个向量与多个向量的余弦相似度

    参数:
        from_vec: 单个向量（1D或2D数组或列表）
        to_vecs: 多个向量（2D或3D数组，每一行为一个向量）

    返回:
        一维数组，包含单个向量与每个多个向量的相似度

    示例:
        sims = cosine_similarity(from_vec, to_vecs)
    """
    # 将from_vec转换为NumPy数组并设置为float类型
    from_vec = np.array(from_vec, dtype=float)
    # 将to_vecs转换为NumPy数组并设置为float类型
    to_vecs = np.array(to_vecs, dtype=float)
    # 计算from_vec的范数
    norm1 = np.linalg.norm(from_vec)
    # 如果from_vec为零向量，直接返回全0相似度数组
    if norm1 == 0:
        return np.zeros(len(to_vecs))
    # 初始化相似度结果列表
    similarities = []
    # 遍历每一个目标向量to_vec
    for to_vec in to_vecs:
        # 计算from_vec和to_vec的点积
        dot_product = np.sum(from_vec * to_vec)
        # 计算to_vec的范数
        norm_vec = np.linalg.norm(to_vec)
        # 如果to_vec为零向量，相似度记为0
        if norm_vec == 0:
            similarities.append(0.0)
        else:
            # 计算余弦相似度
            similarity = dot_product / (norm1 * norm_vec)
            similarities.append(similarity)
    # 返回相似度结果的NumPy数组
    return np.array(similarities)

# 定义最大边际相关性（MMR）算法函数
def mmr_select(query_vector, doc_vectors, k=3, lambda_mult=0.5):
    """
    使用最大边际相关性（MMR）算法选择文档

    参数:
        query_vector: 查询向量（1D数组）
        doc_vectors: 文档向量集合（2D数组，每行一个文档向量）
        k: 要选择的文档数量（默认3）
        lambda_mult: λ参数，平衡相关性与多样性（默认0.5）
                     - λ=1: 只看相关性
                     - λ=0: 只看多样性
                     - λ=0.5: 平衡相关性和多样性

    返回:
        selected: 选中的文档索引列表（从0开始）

    示例:
        selected = mmr_select(query_vector, doc_vectors, k=3, lambda_mult=0.5)
    """
    # 计算所有文档与查询向量的余弦相似度
    query_similarities = cosine_similarity(query_vector, doc_vectors)
    # 打印每个文档与查询的相关性分数
    print("文档与查询的相关性分数：")
    for i, sim in enumerate(query_similarities):
        # 打印文档编号及对应相似度
        print(f"文档{i+1}: {sim:.4f}")
    # 选中相关性最高的文档索引，作为第一个已选文档
    selected = [int(np.argmax(query_similarities))]
    # 直到已选择k个文档之前循环
    while len(selected) < k:
        # 初始化MMR分数列表
        mmr_scores = []
        # 遍历所有文档索引
        for i in range(len(doc_vectors)):
            # 如果该文档还未被选中
            if i not in selected:
                # 当前文档与查询的相关性分数
                relevance = query_similarities[i]
                # 获取已选文档的向量集合
                selected_vecs = doc_vectors[selected]
                # 计算当前文档与已选各文档的余弦相似度
                sims = cosine_similarity(doc_vectors[i], selected_vecs)
                # 获取与已选文档中最大相似度（最不多样的）
                max_sim = np.max(sims)
                # 按MMR公式计算分数
                mmr_score = lambda_mult * relevance - (1 - lambda_mult) * max_sim
                # 记录文档编号和对应的MMR分数
                mmr_scores.append((i, mmr_score))
        # 如果没有可选文档就跳出循环
        if not mmr_scores:
            break
        # 选取MMR分数最高的文档编号
        best_idx, best_score = max(mmr_scores, key=lambda x: x[1])
        # 将该文档编号加入已选列表
        selected.append(best_idx)
    # 返回已选中文档的索引
    return selected

# 定义Document文档对象类
class Document:
    # 文档类，存储内容、元数据和嵌入向量
    def __init__(self, page_content: str, metadata=None, , embedding_value=None):
        # 初始化文档内容
        self.page_content = page_content
        # 初始化元数据，默认空字典
        self.metadata = metadata or {}
        # 初始化嵌入向量
        self.embedding_value = embedding_value

+   def __repr__(self):
+       """返回文档的字符串表示"""
+       metadata_str = f", metadata={self.metadata}" if self.metadata else ""
+       content_preview = self.page_content[:50] + "..." if len(self.page_content) > 50 else self.page_content
+       return f"Document(page_content='{content_preview}'{metadata_str})"

# 定义向量存储抽象基类
class VectorStore(ABC):
    # 向量存储抽象基类

    # 抽象方法，添加文本到向量存储
    @abstractmethod
    def add_texts(
        self,
        texts,
        metadatas=None
    ):
        # 添加文本方法为抽象方法，由子类实现
        pass

    # 抽象方法，最大边际相关性检索
    @abstractmethod
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = 20
    ):
        # 最大边际相关性检索方法为抽象方法，由子类实现
        pass
    @abstractmethod
    def similarity_search(
        self,
        query: str,
        k: int = 4
    ):
        """相似度搜索"""
        pass
    # 抽象类方法，从文本批量构造向量存储
    @classmethod
    @abstractmethod
    def from_texts(
        cls,
        texts,
        embeddings,
        metadatas=None
    ):
        # 批量通过文本构建向量存储的抽象方法，由子类实现
        pass

# 定义FAISS向量存储类，继承自VectorStore
class FAISS(VectorStore):
    # FAISS向量存储实现
    def __init__(
        self,
        embeddings, # 嵌入模型
    ):
        # 保存嵌入模型
        self.embeddings = embeddings
        # 初始化FAISS索引为空
        self.index = None
        # 初始化文档字典，键为文档id，值为Document对象
        self.documents_by_id = {}

    # 添加文本到向量存储
    def add_texts(
        self,
        texts,
        metadatas=None
    ):
        # 如果未传入元数据，则使用空字典列表
        if metadatas is None:
            metadatas = [{}] * len(texts)
        # 利用嵌入模型生成文本的嵌入向量
        embedding_values = self.embeddings.embed_documents(texts)
        # 转换成float32类型的NumPy数组
        embedding_values = np.array(embeddings, dtype=np.float32)
        # 若还未建立FAISS索引，则新建之
        if self.index is None:
            dimension = len(embedding_values[0])
            self.index = faiss.IndexFlatL2(dimension)
        # 添加嵌入向量到FAISS索引
        self.index.add(embedding_values)
        # 获取当前已有文档数量，用于新文档编号
        start_idx = len(self.documents_by_id)
        # 遍历插入的每组文本、元数据、嵌入向量
        for i, (text, metadata, embedding_value) in enumerate(zip(texts, metadatas, embedding_values)):
            # 构造文档id
            doc_id = str(start_idx + i)
            # 构造Document对象
            doc = Document(page_content=text, metadata=metadata, embedding_value=embedding_value)
            # 保存进字典
            self.documents_by_id[doc_id] = doc

    # 最大边际相关性检索方法
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = 20,
        lambda_mult: float = 0.5,
    ):
        # 获取查询文本的嵌入向量
        query_embedding = self.embeddings.embed_query(query)
        # 转为二维NumPy数组
        query_vector = np.array([query_embedding], dtype=np.float32)  # (1, dimension)
        # 用FAISS索引检索出fetch_k个候选文档（距离最近）
        if isinstance(self.index, faiss.Index):
            # 执行检索，返回索引及距离
            _, indices = self.index.search(query_vector, fetch_k)
            # 解析获得候选文档的索引列表
            candidate_indices = indices[0]
        else:
            # 非法索引则抛出异常
            raise RuntimeError("FAISS index is not available.")
        # 如果候选文档不足k个，直接返回这些文档
        if len(candidate_indices) <= k:
            docs = []
            for idx in candidate_indices:
                doc_id = str(idx)
                # 存在于字典才添加
                if doc_id in self.documents_by_id:
                    docs.append(self.documents_by_id[doc_id])
            return docs
        # 从字典中提取候选文档的嵌入向量
        candidate_vectors = np.array([self.documents_by_id[str(i)].embedding_value for i in candidate_indices], dtype=np.float32)
        # 通过mmr_select获取MMR选出的下标
        selected_indices = mmr_select(query_embedding, candidate_vectors, k=k, lambda_mult=lambda_mult)
        # 根据选出下标获得最终文档对象
        docs = []
        for idx in selected_indices:
            doc_id = str(candidate_indices[idx])
            if doc_id in self.documents_by_id:
                docs.append(self.documents_by_id[doc_id])
        return docs

    # 定义相似度检索方法，返回与查询最近的k个文档
    def similarity_search(
        self,
        query: str,
        k: int = 4
    ):
        """
        相似度搜索

        Args:
            query: 查询文本
            k: 返回的文档数量

        Returns:
            List[Document]: 最相似的文档列表
        """
        # 获取查询文本的嵌入向量
        query_embedding = self.embeddings.embed_query(query)
        # 将嵌入向量转换为NumPy二维数组（形状为1行，d维）
        query_vector = np.array([query_embedding], dtype=np.float32)
        # 用FAISS索引执行k近邻检索，得到距离最近的k个索引
        _, indices = self.index.search(query_vector, k)
        # 创建用于存放检索到文档对象的列表
        docs = []
        # 遍历返回的每个文档索引
        for idx in indices[0]:
            # 把数字索引转为字符串形式的文档id
            doc_id = str(idx)
            # 只有字典中存在这个id的文档才加入最终结果
            if doc_id in self.documents_by_id:
                docs.append(self.documents_by_id[doc_id])
        # 返回最终的相似文档列表
        return docs

    # 类方法：通过文本批量创建FAISS向量存储实例
    @classmethod
    def from_texts(
        cls,
        texts,
        embeddings,
        metadatas=None
    ):
        # 创建FAISS实例
        instance = cls(embeddings=embeddings)
        # 添加全部文本及元数据
        instance.add_texts(texts=texts, metadatas=metadatas)
        # 返回构建好的实例
        return instance

# 定义Chroma类，继承自VectorStore
+class Chroma(VectorStore):
    # 设置默认集合名称为"langchain"
+   _LANGCHAIN_DEFAULT_COLLECTION_NAME = "langchain"

    # 构造函数，初始化Chroma实例
+   def __init__(
+       self,
+       collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,  # 集合名称
+       embedding_function=None,    # 嵌入函数
+       persist_directory = None,   # 持久化目录
+       collection_metadata = None, # 集合元数据
+       **kwargs
+   ):
        # 保存嵌入函数
+       self._embedding_function = embedding_function
        # 保存集合名称
+       self._collection_name = collection_name
        # 保存集合元数据，若未提供则为空字典
+       self._collection_metadata = collection_metadata or {}

        # 如果指定了持久化目录，使用持久化客户端
+       if persist_directory is not None:
+           self._client = chromadb.PersistentClient(path=persist_directory)
        # 否则使用内存客户端
+       else:
+           self._client = chromadb.Client()

        # 获取或新建Chroma集合
+       self._collection = self._client.get_or_create_collection(
+           name=collection_name,
+           metadata=self._collection_metadata
+       )

    # 文档批量添加方法
+   def add_texts(
+       self,
+       texts,                    # 文本列表
+       metadatas = None,         # 元数据列表
+       ids = None,               # 文档id列表
+       **kwargs
+   ):
        # 如果没有指定ids，则自动为每个文本生成UUID
+       if ids is None:
+           ids = [str(uuid.uuid4()) for _ in texts]
        # 如果提供了ids，确保全部转为字符串并为None的补充UUID
+       else:
+           ids = [str(id_) if id_ is not None else str(uuid.uuid4()) for id_ in ids]

        # 初始状态的嵌入向量为空
+       embeddings = None
        # 如果有嵌入函数，使用其对文本批量生成嵌入
+       if self._embedding_function is not None:
+           embeddings = self._embedding_function.embed_documents(texts)

        # 如果没有提供元数据，为每个文本默认一个空字典
+       if metadatas is None:
+           metadatas = [{}] * len(texts)
+       else:
            # 若元数据数量少于文本数，进行补全
+           length_diff = len(texts) - len(metadatas)
+           if length_diff > 0:
+               metadatas = metadatas + [{}] * length_diff

        # 执行upsert，将文本、元数据、嵌入、id录入到Chroma集合
+       self._collection.upsert(
+           ids=ids,
+           documents=texts,
+           embeddings=embeddings,
+           metadatas=metadatas
+       )

        # 返回录入的ids
+       return ids

    # 相似度搜索，返回与查询最近的k个文档
+   def similarity_search(
+       self,
+       query: str,      # 查询文本
+       k: int = 4,      # 检索数量
+       filter= None,    # 过滤器
+       **kwargs
+   ):
        # 调用带分数的检索方法
+       docs_and_scores = self.similarity_search_with_score(
+           query=query,
+           k=k,
+           filter=filter,
+           **kwargs
+       )
        # 只返回文档，不返回分数
+       return [doc for doc, _ in docs_and_scores]

    # 相似度检索并返回文档及相似度分数
+   def similarity_search_with_score(
+       self,
+       query: str,         # 查询文本
+       k: int = 4,         # 检索数量
+       filter = None,      # 过滤条件
+       **kwargs
+   ):
        # 查询嵌入向量默认为None
+       query_embedding = None
        # 如果有嵌入函数，获取查询的嵌入向量
+       if self._embedding_function is not None:
+           query_embedding = self._embedding_function.embed_query(query)

        # 如果获得了嵌入向量，按嵌入进行检索
+       if query_embedding is not None:
+           results = self._collection.query(
+               query_embeddings=[query_embedding],
+               n_results=k,
+               where=filter
+           )
        # 否则直接用原始文本检索
+       else:
+           results = self._collection.query(
+               query_texts=[query],
+               n_results=k,
+               where=filter
+           )

        # 格式化返回文档和分数
+       return self._results_to_docs_and_scores(results)

    # 处理Chroma检索结果，转为(doc, score)元组列表
+   def _results_to_docs_and_scores(
+       self,
+       results
+   ):
        # 初始化结果列表
+       docs_and_scores = []

        # 如果没有结果直接返回空列表
+       if not results.get("ids") or not results["ids"][0]:
+           return docs_and_scores

        # 取出文档id、内容、元数据、距离分数
        # 取出文档id
+       ids = results["ids"][0]
        # 取出文档内容
+       documents = results["documents"][0]
        # 取出文档元数据
+       metadatas = results.get("metadatas", [[]])[0]
        # 取出文档距离分数
+       distances = results.get("distances", [[]])[0]

        # 遍历每个检索到的文档
+       for i, doc_id in enumerate(ids):
            # 排除内容为None的文档
+           if documents[i] is not None:
                # 构建Document对象
+               doc = Document(
+                   page_content=documents[i],
+                   metadata=metadatas[i] if i < len(metadatas) and metadatas[i] else {}
+               )
                # 获取距离分数，若没有则为0.0
+               score = distances[i] if i < len(distances) else 0.0
                # 添加到结果列表
+               docs_and_scores.append((doc, score))

        # 返回(doc, score)列表
+       return docs_and_scores

    # 最大边际相关性（MMR）检索
+   def max_marginal_relevance_search(
+       self,
+       query: str,               # 查询文本
+       k: int = 4,               # 最终筛选的数量
+       fetch_k: int = 20,        # 最初候选数量
+       lambda_mult: float = 0.5, # lambda参数，相关性与多样性平衡
+       **kwargs
+   ):
        # 获取查询文本的嵌入向量
+       query_embedding = self._embedding_function.embed_query(query)

        # 先从Chroma查询fetch_k数量的候选文档及嵌入
+       results = self._collection.query(
+           query_embeddings=[query_embedding],
+           n_results=fetch_k
+       )

        # 初始化候选文档、嵌入、分数列表
+       candidate_docs = []
+       candidate_embeddings = []
+       candidate_scores = []

        # 如果检索有结果且至少有一个文档
+       if results.get("ids") and results["ids"][0]:
            # 取出文档id
+           ids = results["ids"][0]
            # 取出文档内容
+           documents = results["documents"][0]
            # 取出文档元数据
+           metadatas = results.get("metadatas", [[]])[0]
            # 取出文档距离分数
+           distances = results.get("distances", [[]])[0]
            # 取出文档嵌入向量
+           embeddings = results.get("embeddings", [[]])[0]

            # 遍历候选文档
+           for i, doc_id in enumerate(ids):
                # 跳过内容为None的文档
+               if documents[i] is not None:
                    # 构建Document对象
+                   doc = Document(
+                       page_content=documents[i],
+                       metadata=metadatas[i] if i < len(metadatas) and metadatas[i] else {}
+                   )
                    # 加入候选文档
+                   candidate_docs.append(doc)
+                   candidate_embeddings.append(embeddings[i])
                    # 添加候选的距离分数
+                   candidate_scores.append(distances[i] if i < len(distances) else 0.0)

        # 如果候选文档数少于等于k, 直接返回
+       if len(candidate_docs) <= k:
+           return candidate_docs
        # 转换为NumPy数组
+       candidate_vectors = np.array(candidate_embeddings, dtype=np.float32)

        # 最大边际相关性筛选k个文档，返回其索引
+       selected_indices = mmr_select(
+           query_embedding,
+           candidate_vectors,
+           k=k,
+           lambda_mult=lambda_mult
+       )

        # 返回被选中的候选文档
+       return [candidate_docs[idx] for idx in selected_indices if idx < len(candidate_docs)]

    # 类方法：批量通过文本创建Chroma实例
+   @classmethod
+   def from_texts(
+       cls,
+       texts,                            # 文本列表
+       embeddings,                        # 嵌入函数
+       metadatas = None,                 # 元数据列表
+       collection_name = _LANGCHAIN_DEFAULT_COLLECTION_NAME,  # 集合名称
+       persist_directory = None,         # 持久化目录
+       collection_metadata = None,       # 集合元数据
+       **kwargs
+   ):
        # 实例化Chroma对象
+       instance = cls(
+           collection_name=collection_name,#集合名称
+           embedding_function=embeddings,#嵌入函数
+           persist_directory=persist_directory,#持久化目录
+           collection_metadata=collection_metadata,#集合元数据
+           **kwargs#其他参数
+       )
        # 添加文本和元数据
+       instance.add_texts(texts=texts, metadatas=metadatas)
        # 返回实例
+       return instance