为了实现你描述的功能，我们可以将该项目中与向量检索相关的部分抽取出来，并进行简化，使其能够独立运行。重点步骤包括：

  1. 初始化和管理FAISS索引：负责创建FAISS索引、添加向量数据以及保存/加载索引。
  2. 数据存储：将文档嵌入向量并存储在FAISS中，以便于后续检索。
  3. 输入问题匹配查询：根据用户输入的查询向量化后，在FAISS中进行相似度搜索，返回最相似的文档以及匹配分数。

下面是一个基本的代码框架，展示如何使用FAISS完成这些步骤：

第一步：安装依赖

确保已安装FAISS和文本向量化工具，如sentence-transformers用于嵌入生成：

pip install faiss-cpu sentence-transformers

第二步：实现代码框架

我们可以将FAISS相关操作封装在一个类中，比如VectorStore。以下是一个简化的代码示例：

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorStore:
    def __init__(self, embedding_dim: int = 768):
        # 初始化FAISS索引
        self.index = faiss.IndexFlatL2(embedding_dim)
        self.model = SentenceTransformer('paraphrase-MiniLM-L6-v2')  # 使用适合的小型模型生成嵌入

    def add_documents(self, docs: list):
        """
        将文档列表添加到FAISS索引中
        """
        embeddings = self.model.encode(docs)  # 转换文档为嵌入向量
        self.index.add(np.array(embeddings, dtype=np.float32))  # 添加向量到索引

    def search(self, query: str, top_k: int = 5, score_threshold: float = 0.5):
        """
        在索引中搜索与查询最匹配的文档
        """
        query_embedding = self.model.encode([query])  # 将查询转换为嵌入向量
        distances, indices = self.index.search(np.array(query_embedding, dtype=np.float32), top_k)
        
        # 将分数过滤，返回符合阈值的文档及其分数
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            score = 1 / (1 + dist)  # 将距离转换为相似度分数
            if score >= score_threshold:
                results.append({'doc_index': idx, 'score': score})
        return results

    def save_index(self, file_path: str):
        """
        保存FAISS索引到文件
        """
        faiss.write_index(self.index, file_path)

    def load_index(self, file_path: str):
        """
        从文件加载FAISS索引
        """
        self.index = faiss.read_index(file_path)

# 使用示例
if __name__ == "__main__":
    docs = ["你好，这是一个样本文档。", "今天的天气很好。", "这是关于AI的介绍。", "机器学习是人工智能的一个分支。"]
    query = "什么是人工智能？"
    
    # 初始化向量库
    vector_store = VectorStore()
    
    # 添加文档
    vector_store.add_documents(docs)
    
    # 查询检索
    results = vector_store.search(query, top_k=3, score_threshold=0.5)
    for res in results:
        print(f"匹配文档索引: {res['doc_index']}, 匹配得分: {res['score']}")
    
    # 保存索引
    vector_store.save_index("faiss_index.bin")
    
    # 加载索引
    # vector_store.load_index("faiss_index.bin")

代码说明

  1. 初始化：VectorStore类在初始化时，创建了FAISS索引并加载了SentenceTransformer模型，用于生成向量嵌入。
  2. 添加文档：add_documents方法将文档转为向量，并添加到FAISS索引中。
  3. 查询检索：search方法根据输入的查询，生成查询嵌入后在FAISS中搜索最相似的文档。返回的结果会根据score_threshold进行过滤。
  4. 分数计算：检索时FAISS默认返回的是欧式距离，将距离通过公式score = 1 / (1 + distance)转换为相似度分数，使其更直观。
  5. 索引保存和加载：save_index和load_index方法用于存储和加载FAISS索引文件，便于数据持久化。

扩展说明

  • 如果需要实时更新数据，可以在文档变更后更新索引并重新保存。
  • 在大规模数据时，可以考虑使用IndexIVFFlat等更高效的索引类型，并进行训练（适用于更复杂的检索场景）。

通过这种方式，能够独立运行FAISS向量检索模块并进行匹配查询。

在这个简化示例中，score_threshold 的定义和之前代码的定义 相似但不完全相同。在上面的回答中，score_threshold 是基于相似度分数的阈值，而在之前的代码中，score_threshold 实际上是基于匹配结果的 相似度距离（或得分） 来过滤的。下面是两者的详细对比与解释：

1. 原始代码中的 score_threshold 定义

在原始代码中，score_threshold 是一个 相关度阈值，表示匹配结果的最低要求。其定义为 0-1 之间的浮点值，具体作用如下：

  • 得分越低，相关性越高。即得分接近 0 时，表示匹配度很高；而得分接近 1 时，表示几乎无匹配。
  • 过滤标准：匹配结果的得分需要小于等于 score_threshold 才会被认为是有效匹配。因此，如果 score_threshold=0.5，那么只会返回得分小于或等于 0.5 的结果。

2. 简化示例中的 score_threshold 定义

在简化示例中，score_threshold 也是用于过滤相似度的阈值，但它是在 相似度分数 上定义的，且分数越高表示匹配度越高（即得分和原始定义的取值趋势相反）。其主要区别如下：

  • 得分越高，相关性越高。即得分接近 1 时，表示匹配度非常高；而得分接近 0 时，表示几乎无匹配。
  • 过滤标准：在简化示例中，FAISS 返回的是欧式距离，我们将其转换为相似度分数（score = 1 / (1 + distance)），所以这个 score_threshold 实际上是相似度分数的最低要求。只有相似度分数大于等于 score_threshold 的结果才会被返回。

如何调整以保持一致性

要让简化示例中的 score_threshold 和原始代码保持一致，可以将 过滤标准和定义方向保持一致：

  1. 一致性定义：在简化示例中，调整 score_threshold 的定义为 0 到 1，且保持“分数越低相关性越高”。
  2. 直接使用距离过滤：也可以直接使用 distance <= score_threshold 来过滤结果，而不是将其转换为相似度分数。

统一实现示例

下面是调整后、基于欧式距离定义的 score_threshold 实现：

def search(self, query: str, top_k: int = 5, score_threshold: float = 0.5):
    """
    在索引中搜索与查询最匹配的文档
    """
    query_embedding = self.model.encode([query])  # 将查询转换为嵌入向量
    distances, indices = self.index.search(np.array(query_embedding, dtype=np.float32), top_k)
    
    # 将距离过滤，返回符合阈值的文档及其原始距离
    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if dist <= score_threshold:  # 保持距离小于等于 score_threshold
            results.append({'doc_index': idx, 'distance': dist})
    return results

在这种实现中，score_threshold 直接表示最大允许的欧式距离，并和原始代码的定义保持一致（即越小表示匹配度越高）。

FAISS.PY

在 faiss.py 文件中，确实有 score_threshold 的相关实现，尤其是在 similarity_search_with_score_by_vector 方法中。下面是其具体实现方式和作用：

关键代码片段：similarity_search_with_score_by_vector

def similarity_search_with_score_by_vector(
    self,
    embedding: List[float],
    k: int = 4,
    filter: Optional[Union[Callable, Dict[str, Any]]] = None,
    fetch_k: int = 20,
    **kwargs: Any,
) -> List[Tuple[Document, float]]:
    ...
    score_threshold = kwargs.get("score_threshold")
    if score_threshold is not None:
        cmp = (
            operator.ge
            if self.distance_strategy
            in (DistanceStrategy.MAX_INNER_PRODUCT, DistanceStrategy.JACCARD)
            else operator.le
        )
        docs = [
            (doc, similarity)
            for doc, similarity in docs
            if cmp(similarity, score_threshold)
        ]
    return docs[:k]

score_threshold 的作用

  1. 获取阈值：score_threshold 是通过 kwargs.get("score_threshold") 提供的，可选浮点值。通过 kwargs 参数传入，该值用于限制匹配结果的最小相关度得分或最大允许距离。

  2. 条件过滤：根据 distance_strategy 的类型，score_threshold 被定义为相似度的下限或距离的上限：

    • 如果 distance_strategy 是 最大内积（MAX_INNER_PRODUCT） 或 Jaccard 相似度，则 score_threshold 表示相似度的下限，过滤掉低于该分数的结果。
    • 对于其他距离度量（例如欧氏距离），score_threshold 表示距离的上限，过滤掉高于该距离的结果。

  3. 返回筛选结果：通过 if cmp(similarity, score_threshold) 进行过滤，确保只返回满足 score_threshold 条件的文档及其分数。

总结

在 similarity_search_with_score_by_vector 方法中，score_threshold 定义了文档的最低相似度要求或最高允许距离。

实现

为了优化该代码，使 score_threshold 和 score 的实现与前面的 similarity_search_with_score_by_vector 方法一致，我们将 score_threshold 用于 直接过滤距离或相似度，并根据距离度量策略（例如内积或欧氏距离）进行条件判断。以下是优化后的代码：

优化后的代码

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorStore:
    def __init__(self, embedding_dim: int = 768, distance_metric: str = "l2"):
        """
        初始化向量存储，选择距离度量方式。
        :param embedding_dim: 嵌入维度
        :param distance_metric: 距离度量方式，支持 "l2" (欧氏距离) 或 "ip" (内积)
        """
        if distance_metric == "ip":
            self.index = faiss.IndexFlatIP(embedding_dim)  # 内积度量
        else:
            self.index = faiss.IndexFlatL2(embedding_dim)  # 默认为欧氏距离

        self.model = SentenceTransformer('paraphrase-MiniLM-L6-v2')  # 使用小型模型生成嵌入
        self.distance_metric = distance_metric

    def add_documents(self, docs: list):
        """
        将文档列表添加到FAISS索引中
        """
        embeddings = self.model.encode(docs)  # 转换文档为嵌入向量
        if self.distance_metric == "l2":
            faiss.normalize_L2(np.array(embeddings, dtype=np.float32))  # 如果是 L2，进行归一化
        self.index.add(np.array(embeddings, dtype=np.float32))  # 添加向量到索引

    def search(self, query: str, top_k: int = 5, score_threshold: float = 0.5):
        """
        在索引中搜索与查询最匹配的文档
        """
        query_embedding = self.model.encode([query])  # 将查询转换为嵌入向量
        if self.distance_metric == "l2":
            faiss.normalize_L2(np.array(query_embedding, dtype=np.float32))

        distances, indices = self.index.search(np.array(query_embedding, dtype=np.float32), top_k)

        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if self.distance_metric == "ip":
                score = dist  # 对于内积，分数直接是距离
                is_valid = score >= score_threshold  # 内积的分数是相似度，分数越高越相似
            else:
                score = 1 / (1 + dist)  # 欧氏距离转换为相似度分数
                is_valid = score >= score_threshold  # 欧氏距离的分数越低越相似
            
            if is_valid:
                results.append({'doc_index': idx, 'score': score})
        return results

    def save_index(self, file_path: str):
        """
        保存FAISS索引到文件
        """
        faiss.write_index(self.index, file_path)

    def load_index(self, file_path: str):
        """
        从文件加载FAISS索引
        """
        self.index = faiss.read_index(file_path)

# 使用示例
if __name__ == "__main__":
    docs = ["你好，这是一个样本文档。", "今天的天气很好。", "这是关于AI的介绍。", "机器学习是人工智能的一个分支。"]
    query = "什么是人工智能？"
    
    # 初始化向量库
    vector_store = VectorStore(distance_metric="ip")  # 选择内积或欧氏距离

    # 添加文档
    vector_store.add_documents(docs)

    # 查询检索
    results = vector_store.search(query, top_k=3, score_threshold=0.5)
    for res in results:
        print(f"匹配文档索引: {res['doc_index']}, 匹配得分: {res['score']}")
    
    # 保存索引
    vector_store.save_index("faiss_index.bin")
    
    # 加载索引
    # vector_store.load_index("faiss_index.bin")

优化说明

  1. 选择距离度量方式：

    • distance_metric 参数支持 “l2”（欧氏距离） 和 “ip”（内积），并且在初始化时根据此参数选择不同的 FAISS 索引类型。
    • 内积模式适合内积（点积）相似度场景，而欧氏距离模式将嵌入进行归一化以符合相似度查询要求。

  2. 过滤逻辑：

    • 如果使用内积距离度量（ip），score 直接使用距离值（dist），并根据 score_threshold 进行相似度过滤。
    • 如果使用欧氏距离（l2），则将 dist 转换为相似度分数 score = 1 / (1 + dist)，并对 score 进行过滤。
    • 这样在不同距离度量下，score_threshold 的逻辑与 similarity_search_with_score_by_vector 保持一致。

通过此优化代码，现在 score_threshold 的行为和 similarity_search_with_score_by_vector 的实现方式一致，可以支持不同距离度量下的相似度过滤。

优化

为了实现您的需求，我会优化代码，使其不仅能够打印出匹配成功的文档和相应的得分，还能打印与所有文档的相似度得分。这将包括以下改进：

  1. 打印查询内容。
  2. 打印所有文档的相似度得分，无论它们是否满足 score_threshold。
  3. 将匹配成功的结果与相似度得分分别打印。

以下是优化后的代码：

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorStore:
    def __init__(self, embedding_dim: int = 768, distance_metric: str = "l2"):
        """
        初始化向量存储，选择距离度量方式。
        :param embedding_dim: 嵌入维度
        :param distance_metric: 距离度量方式，支持 "l2" (欧氏距离) 或 "ip" (内积)
        """
        if distance_metric == "ip":
            self.index = faiss.IndexFlatIP(embedding_dim)  # 内积度量
        else:
            self.index = faiss.IndexFlatL2(embedding_dim)  # 默认为欧氏距离

        self.model = SentenceTransformer('paraphrase-MiniLM-L6-v2')  # 使用小型模型生成嵌入
        self.distance_metric = distance_metric

    def add_documents(self, docs: list):
        """
        将文档列表添加到FAISS索引中
        """
        embeddings = self.model.encode(docs)  # 转换文档为嵌入向量
        if self.distance_metric == "l2":
            embeddings = np.array(embeddings, dtype=np.float32)
            faiss.normalize_L2(embeddings)  # 如果是 L2，进行归一化
        self.index.add(embeddings)  # 添加向量到索引
        self.docs = docs  # 保存文档内容以便打印

    def search(self, query: str, top_k: int = 5, score_threshold: float = 0.5):
        """
        在索引中搜索与查询最匹配的文档，并打印所有文档的相似度得分
        """
        print(f"\n查询内容: {query}")
        
        query_embedding = self.model.encode([query])  # 将查询转换为嵌入向量
        if self.distance_metric == "l2":
            query_embedding = np.array(query_embedding, dtype=np.float32)
            faiss.normalize_L2(query_embedding)

        # 执行检索，获取所有文档的距离和索引
        distances, indices = self.index.search(query_embedding, len(self.docs))

        results = []
        print("\n所有文档的相似度得分：")
        for dist, idx in zip(distances[0], indices[0]):
            if self.distance_metric == "ip":
                score = dist  # 对于内积，分数直接是距离
                is_valid = score >= score_threshold  # 内积的分数是相似度，分数越高越相似
            else:
                score = 1 / (1 + dist)  # 欧氏距离转换为相似度分数
                is_valid = score >= score_threshold  # 欧氏距离的分数越低越相似
            
            # 打印所有文档的得分
            print(f"文档: {self.docs[idx]}, 相似度得分: {score}")
            
            if is_valid:
                results.append({'doc_index': idx, 'doc': self.docs[idx], 'score': score})

        print("\n匹配成功的文档（符合 score_threshold 的文档）：")
        for res in results:
            print(f"匹配文档索引: {res['doc_index']}, 内容: {res['doc']}, 匹配得分: {res['score']}")
        return results

    def save_index(self, file_path: str):
        """
        保存FAISS索引到文件
        """
        faiss.write_index(self.index, file_path)

    def load_index(self, file_path: str):
        """
        从文件加载FAISS索引
        """
        self.index = faiss.read_index(file_path)

# 使用示例
if __name__ == "__main__":
    docs = ["你好，这是一个样本文档。", "今天的天气很好。", "这是关于AI的介绍。", "机器学习是人工智能的一个分支。"]
    query = "什么是人工智能？"
    
    # 初始化向量库
    vector_store = VectorStore(distance_metric="ip")  # 选择内积或欧氏距离

    # 添加文档
    vector_store.add_documents(docs)

    # 查询检索
    results = vector_store.search(query, top_k=3, score_threshold=0.5)

    # 保存索引
    vector_store.save_index("faiss_index.bin")
    
    # 加载索引
    # vector_store.load_index("faiss_index.bin")

优化细节说明

  1. 所有文档的相似度得分：在搜索过程中，程序会打印每个文档的相似度得分，无论它是否满足 score_threshold。
  2. 符合条件的匹配结果：在 score_threshold 过滤后，程序会打印匹配成功的文档以及其得分。
  3. 清晰的输出格式：为了方便区分，代码将查询内容、所有文档的相似度得分和匹配成功的文档分为三个部分打印。

讲解

好的，下面我将详细讲解上面代码的每个部分，以帮助您理解其工作原理和逻辑。

1. 初始化 VectorStore 类

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorStore:
    def __init__(self, embedding_dim: int = 768, distance_metric: str = "l2"):
        """
        初始化向量存储，选择距离度量方式。
        :param embedding_dim: 嵌入维度
        :param distance_metric: 距离度量方式，支持 "l2" (欧氏距离) 或 "ip" (内积)
        """
        if distance_metric == "ip":
            self.index = faiss.IndexFlatIP(embedding_dim)  # 内积度量
        else:
            self.index = faiss.IndexFlatL2(embedding_dim)  # 默认为欧氏距离

        self.model = SentenceTransformer('paraphrase-MiniLM-L6-v2')  # 使用小型模型生成嵌入
        self.distance_metric = distance_metric
  • faiss 和 numpy：faiss 是 Facebook 提供的快速相似性搜索库，用于索引和检索；numpy 处理向量和矩阵运算。
  • SentenceTransformer：用于将文本转化为向量嵌入，基于 paraphrase-MiniLM-L6-v2 模型。
  • embedding_dim：嵌入的向量维度。在 MiniLM 模型中，这个维度通常是 768。
  • distance_metric：选择距离度量方式，l2 表示欧氏距离，ip 表示内积（即点积）。
    • faiss.IndexFlatIP 是基于内积的索引，适合度量两个向量之间的相似性（分数越高表示越相似）。
    • faiss.IndexFlatL2 是基于欧氏距离的索引，通常用于度量两个向量之间的距离（分数越低表示越相似）。

2. 添加文档 add_documents

    def add_documents(self, docs: list):
        """
        将文档列表添加到FAISS索引中
        """
        embeddings = self.model.encode(docs)  # 转换文档为嵌入向量
        if self.distance_metric == "l2":
            embeddings = np.array(embeddings, dtype=np.float32)
            faiss.normalize_L2(embeddings)  # 如果是 L2，进行归一化
        self.index.add(embeddings)  # 添加向量到索引
        self.docs = docs  # 保存文档内容以便打印
  • 嵌入生成：使用 SentenceTransformer 模型将输入的文档列表 docs 转换成向量嵌入列表，嵌入维度为 embedding_dim。
  • 归一化操作：当使用 l2（欧氏距离）度量时，对向量进行 L2 归一化，以确保计算的距离反映出向量间的相对相似性。
  • 添加到索引：通过 self.index.add 将嵌入向量存储到 FAISS 索引中。self.index 中保存了这些向量数据。
  • 存储文档内容：将 docs 保存在 self.docs 中，以便后续查询中打印原始文档内容。

3. 查询 search

    def search(self, query: str, top_k: int = 5, score_threshold: float = 0.5):
        """
        在索引中搜索与查询最匹配的文档，并打印所有文档的相似度得分
        """
        print(f"\n查询内容: {query}")
        
        query_embedding = self.model.encode([query])  # 将查询转换为嵌入向量
        if self.distance_metric == "l2":
            query_embedding = np.array(query_embedding, dtype=np.float32)
            faiss.normalize_L2(query_embedding)

        # 执行检索，获取所有文档的距离和索引
        distances, indices = self.index.search(query_embedding, len(self.docs))

        results = []
        print("\n所有文档的相似度得分：")
        for dist, idx in zip(distances[0], indices[0]):
            if self.distance_metric == "ip":
                score = dist  # 对于内积，分数直接是距离
                is_valid = score >= score_threshold  # 内积的分数是相似度，分数越高越相似
            else:
                score = 1 / (1 + dist)  # 欧氏距离转换为相似度分数
                is_valid = score >= score_threshold  # 欧氏距离的分数越低越相似
            
            # 打印所有文档的得分
            print(f"文档: {self.docs[idx]}, 相似度得分: {score}")
            
            if is_valid:
                results.append({'doc_index': idx, 'doc': self.docs[idx], 'score': score})

        print("\n匹配成功的文档（符合 score_threshold 的文档）：")
        for res in results:
            print(f"匹配文档索引: {res['doc_index']}, 内容: {res['doc']}, 匹配得分: {res['score']}")
        return results
search 方法步骤说明
  1. 打印查询内容：显示输入的查询 query。
  2. 嵌入生成：使用 SentenceTransformer 模型将查询转换为嵌入向量。
  3. 归一化：如果使用的是欧氏距离度量（l2），则对查询向量进行 L2 归一化，以便相似度匹配更准确。
  4. 执行检索：使用 self.index.search 方法在索引中找到与查询最接近的向量，并返回距离和索引。
    • distances 是查询向量到索引向量的距离。
    • indices 是对应的向量在索引中的位置。
  5. 打印所有文档的相似度得分：
    • 对每个结果，计算相似度得分 score，根据 distance_metric 判断得分含义：
      • 若为内积，score 直接等于距离。
      • 若为欧氏距离，将距离 dist 转换为相似度 score = 1 / (1 + dist)，分数越高表示越相似。
    • 判断 score 是否满足 score_threshold，若满足，则记录在 results 中。
  6. 打印匹配成功的文档：显示符合 score_threshold 的匹配文档及其得分。

4. 索引保存与加载

    def save_index(self, file_path: str):
        """
        保存FAISS索引到文件
        """
        faiss.write_index(self.index, file_path)

    def load_index(self, file_path: str):
        """
        从文件加载FAISS索引
        """
        self.index = faiss.read_index(file_path)
  • save_index：将当前 FAISS 索引保存到指定文件 file_path。存储索引数据到磁盘，便于后续重用。
  • load_index：从磁盘加载已保存的 FAISS 索引。加载后可以直接进行搜索操作，适合跨会话的数据持久化。

5. 使用示例

if __name__ == "__main__":
    docs = ["你好，这是一个样本文档。", "今天的天气很好。", "这是关于AI的介绍。", "机器学习是人工智能的一个分支。"]
    query = "什么是人工智能？"
    
    # 初始化向量库
    vector_store = VectorStore(distance_metric="ip")  # 选择内积或欧氏距离

    # 添加文档
    vector_store.add_documents(docs)

    # 查询检索
    results = vector_store.search(query, top_k=3, score_threshold=0.5)

    # 保存索引
    vector_store.save_index("faiss_index.bin")
    
    # 加载索引
    # vector_store.load_index("faiss_index.bin")
示例流程
  1. 初始化：创建 VectorStore 实例，选择内积（ip）为距离度量方式。
  2. 添加文档：将 docs 列表中的文本转换为向量并存入 FAISS 索引。
  3. 执行查询：在 FAISS 中检索最匹配的文档，显示每个文档的相似度得分并返回符合 score_threshold 的匹配文档。
  4. 保存和加载索引：提供 save_index 和 load_index 方法，将向量索引持久化存储和重新加载。

通过这种设计，代码实现了文档添加、

查询、相似度过滤、查询和匹配结果打印等一系列向量存储操作，并灵活支持不同的距离度量方式。

# faiss
Logo
EazyDevelop社区

一站式 AI 云服务平台

更多推荐

cover

国内低代码平台：2025 年国内主流平台盘点

avatar EazyDevelop社区
cover

从零开始搭建个人RAG知识库：RAGFlow+DeepSeek保姆级教程！

avatar EazyDevelop社区
cover

5分钟搞定！MySQL/PostgreSQL 到 Elasticsearch 的实时同步

avatar EazyDevelop社区