向量数据库与 Embedding 实战教程

零基础入门向量数据库与Embedding技术，从概念理解到生产实战，掌握AI时代的核心检索技术。

1. 向量与Embedding基础

1.1 什么是向量

在数学和物理学中，向量（Vector）是一个有大小和方向的量。在计算机科学和AI领域，我们所说的向量通常指的是数值数组，它是对现实世界中各种数据（文本、图片、音频等）的一种数学表示。

例如，一个简单的向量可以是：

[0.12, -0.45, 0.78, 0.33, -0.91, 0.22, 0.56, -0.11]

这个数组中的每个数字代表数据在某个维度上的特征值。向量的维度（数组长度）可以从几十到几千不等，取决于使用的Embedding模型。

1.2 为什么需要向量化

计算机只能理解数字，不能直接理解"今天天气真好"这样的自然语言。我们需要一种方法，把文本、图片等非结构化数据转换成计算机可以处理的数值形式。

传统方法（如词袋模型、TF-IDF）存在明显局限：

语义缺失：无法捕捉词语之间的语义关系
维度灾难：高维稀疏向量存储和计算效率低
泛化能力差：无法处理同义词、近义词

Embedding技术的出现解决了这些问题。它将数据映射到一个低维稠密的向量空间，使得：

语义相近的数据在向量空间中距离更近
支持高效的相似度计算
能捕捉复杂的语义关系

1.3 Embedding原理简述

Embedding本质上是一个映射函数，它将高维稀疏的数据（如one-hot编码的词汇表）映射到低维稠密的向量空间。

以文本Embedding为例，其工作流程大致如下：

分词：将文本拆分为token（词或子词）
编码：通过神经网络（如Transformer）将每个token编码为向量
池化：将多个token的向量合并为一个固定长度的句子/文档向量

现代Embedding模型（如BERT、Sentence-Transformers）通常基于Transformer架构，通过大规模语料预训练学习到丰富的语义表示。

1.4 向量空间与相似度

在向量空间中，我们通过计算向量之间的距离或相似度来衡量数据的语义关联程度。常用的相似度计算方法包括：

余弦相似度（Cosine Similarity）：

similarity = (A · B) / (||A|| × ||B||)

值域为[-1, 1]，值越大表示越相似。这是最常用的方法，因为它只关注方向，不受向量长度影响。

欧氏距离（Euclidean Distance）：

distance = √(Σ(aᵢ - bᵢ)²)

值域为[0, +∞)，值越小表示越相似。

点积（Dot Product）：

dot = Σ(aᵢ × bᵢ)

当向量已归一化时，点积等价于余弦相似度。

Python示例：

import numpy as np

def cosine_similarity(a, b):
    """计算两个向量的余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    """计算两个向量的欧氏距离"""
    return np.linalg.norm(np.array(a) - np.array(b))

# 示例：比较三个句子的相似度
vec_cat = [0.8, 0.2, 0.1, 0.9]      # "猫"
vec_dog = [0.7, 0.3, 0.15, 0.85]    # "狗"
vec_car = [0.1, 0.9, 0.8, 0.1]      # "汽车"

print(f"猫 vs 狗: {cosine_similarity(vec_cat, vec_dog):.4f}")  # 高相似度
print(f"猫 vs 汽车: {cosine_similarity(vec_cat, vec_car):.4f}")  # 低相似度

2. 文本Embedding模型

2.1 OpenAI Embedding

OpenAI提供了多个Embedding模型，其中最常用的是text-embedding-3-small和text-embedding-3-large。

特点：

支持多种语言
可调节输出维度（Matryoshka表示学习）
通过API调用，无需本地部署
text-embedding-3-large最高支持3072维

使用示例：

from openai import OpenAI

client = OpenAI()

def get_embedding(text, model="text-embedding-3-small"):
    """获取文本的Embedding向量"""
    response = client.embeddings.create(
        input=text,
        model=model
    )
    return response.data[0].embedding

# 单个文本
embedding = get_embedding("向量数据库是AI时代的重要基础设施")
print(f"维度: {len(embedding)}")  # 1536

# 批量处理（更高效）
texts = ["向量数据库", "Embedding模型", "语义搜索"]
response = client.embeddings.create(
    input=texts,
    model="text-embedding-3-small"
)
embeddings = [item.embedding for item in response.data]
print(f"获取了 {len(embeddings)} 个向量")

维度压缩：

# 使用较小的维度以节省存储和计算
response = client.embeddings.create(
    input="测试文本",
    model="text-embedding-3-small",
    dimensions=512  # 从1536压缩到512
)

2.2 BGE系列模型

BGE（BAAI General Embedding）是北京智源人工智能研究院开源的Embedding模型系列，在中文场景表现出色。

主要模型：

bge-small-zh-v1.5：轻量级，512维，适合资源受限场景
bge-base-zh-v1.5：基础版，768维，平衡性能与效率
bge-large-zh-v1.5：大型版，1024维，最佳效果
bge-m3：多语言、多功能、多粒度，支持100+语言

使用示例：

from sentence_transformers import SentenceTransformer

# 加载模型（首次运行会自动下载）
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')

# 编码文本
texts = [
    "什么是向量数据库？",
    "向量数据库是一种专门存储和检索高维向量的数据库系统",
    "今天天气真不错"
]

embeddings = model.encode(texts)
print(f"向量维度: {embeddings.shape[1]}")  # 768

# 计算相似度
from sentence_transformers.util import cos_sim

similarities = cos_sim(embeddings[0], embeddings[1])
print(f"问题与答案的相似度: {similarities.item():.4f}")  # 高

similarities = cos_sim(embeddings[0], embeddings[2])
print(f"问题与无关文本的相似度: {similarities.item():.4f}")  # 低

BGE模型的查询前缀：

BGE系列模型建议在查询时添加前缀以获得更好的检索效果：

# 查询时添加前缀
query = "query: 什么是向量数据库？"
# 文档不需要前缀
doc = "向量数据库是一种专门存储和检索高维向量的数据库系统"

query_emb = model.encode(query)
doc_emb = model.encode(doc)

2.3 M3E模型

M3E（Moka Massive Mixed Embedding）是由MokaAI团队开源的中文Embedding模型。

特点：

专门针对中文优化
支持中英双语
采用对比学习训练
开源免费，可本地部署

使用示例：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('moka-ai/m3e-base')

texts = ["人工智能", "机器学习", "深度学习", "烹饪美食"]
embeddings = model.encode(texts)

# 计算相似度矩阵
from sentence_transformers.util import cos_sim
import numpy as np

similarity_matrix = cos_sim(embeddings, embeddings)
print("相似度矩阵:")
print(similarity_matrix.numpy())

2.4 Jina Embeddings

Jina AI提供了多个高质量的Embedding模型，支持多种语言和长文本。

特点：

支持8K长文本
多语言支持
Late Interaction技术
提供API和开源模型

使用示例：

import requests

def jina_embedding(texts, api_key):
    """使用Jina AI API获取Embedding"""
    url = "https://api.jina.ai/v1/embeddings"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    data = {
        "model": "jina-embeddings-v3",
        "input": texts
    }
    response = requests.post(url, json=data, headers=headers)
    return [item['embedding'] for item in response.json()['data']]

# 使用示例
texts = ["向量数据库", "Embedding模型", "语义搜索"]
# embeddings = jina_embedding(texts, "your-api-key")

2.5 模型选型建议

场景	推荐模型	原因
中文语义搜索	bge-base-zh-v1.5	中文效果最佳，开源免费
多语言场景	bge-m3 或 text-embedding-3-large	支持100+语言
资源受限	bge-small-zh-v1.5	轻量高效
生产环境API	text-embedding-3-small	稳定可靠，性价比高
长文本处理	jina-embeddings-v3	支持8K token
最高精度	text-embedding-3-large	3072维，效果最佳

3. 图像与多模态Embedding

3.1 CLIP模型

CLIP（Contrastive Language-Image Pre-training）是OpenAI开发的多模态模型，能够将图像和文本映射到同一个向量空间。

核心思想：通过对比学习，让相似的图像-文本对在向量空间中距离更近。

使用示例：

import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel

# 加载模型
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# 图像编码
image = Image.open("cat.jpg")
inputs = processor(images=image, return_tensors="pt")
image_features = model.get_image_features(**inputs)

# 文本编码
texts = ["一只猫", "一只狗", "一辆汽车"]
inputs = processor(text=texts, return_tensors="pt", padding=True)
text_features = model.get_text_features(**inputs)

# 计算相似度
similarity = torch.nn.functional.cosine_similarity(
    image_features, text_features
)
print("图像与文本的相似度:", similarity.detach().numpy())

3.2 ImageBind

ImageBind是Meta提出的六模态Embedding模型，支持图像、文本、音频、深度、热力图和IMU数据。

应用场景：

跨模态检索（用文字搜索音频）
多模态融合（结合图像和音频进行分析）
零样本分类

3.3 多模态向量的实际应用

# 使用CLIP进行图像搜索
from sentence_transformers import SentenceTransformer, util
from PIL import Image
import glob

# 加载CLIP模型
model = SentenceTransformer('clip-ViT-B-32')

# 编码查询图片
query_image = Image.open("query.jpg")
query_embedding = model.encode(query_image)

# 编码数据库中的图片
image_paths = glob.glob("images/*.jpg")
images = [Image.open(p) for p in image_paths]
image_embeddings = model.encode(images)

# 搜索最相似的图片
similarities = util.cos_sim(query_embedding, image_embeddings)[0]
top_k = similarities.topk(5)

print("最相似的5张图片:")
for score, idx in zip(top_k.values, top_k.indices):
    print(f"  {image_paths[idx]}: {score:.4f}")

4. 向量数据库概览

4.1 为什么需要向量数据库

传统的数据库（如MySQL、PostgreSQL）是为结构化数据设计的，它们在处理向量相似度搜索时存在明显不足：

不支持高效的近似最近邻（ANN）搜索
无法处理高维向量的索引
缺乏向量专用的存储和查询优化

向量数据库专门解决了这些问题，提供了：

高效的向量索引和检索
支持亿级向量的毫秒级查询
灵活的过滤和混合查询
可扩展的分布式架构

4.2 主流向量数据库对比

数据库	类型	开源	分布式	特点	适用场景
Milvus	专用	✅	✅	高性能、可扩展、丰富索引	生产环境大规模部署
Chroma	专用	✅	❌	轻量易用、Python原生	快速原型、小规模应用
Qdrant	专用	✅	✅	Rust实现、高性能、丰富过滤	生产环境、需要复杂过滤
Weaviate	专用	✅	✅	GraphQL接口、多模态支持	需要GraphQL的场景
Pinecone	云服务	❌	✅	全托管、零运维	不想自己运维的场景
FAISS	库	✅	❌	Facebook开源、极致性能	本地实验、嵌入式应用
pgvector	扩展	✅	❌	PostgreSQL扩展	已有PG的场景

4.3 选型建议

快速原型开发：选择Chroma

安装简单，pip install chromadb
Python原生API，学习成本低
内置Embedding模型支持

生产环境大规模部署：选择Milvus或Qdrant

支持分布式，可水平扩展
丰富的索引类型和查询优化
活跃的社区和企业支持

已有PostgreSQL：选择pgvector

无需引入新组件
利用PG的事务和备份机制
适合中小规模向量数据

不想运维：选择Pinecone

全托管服务，零运维
按使用量计费
适合快速上线

5. Chroma快速入门

5.1 安装与配置

# 安装Chroma
pip install chromadb

# 如果需要使用Sentence Transformers作为Embedding
pip install chromadb sentence-transformers

5.2 基本使用

import chromadb

# 创建客户端（内存模式）
client = chromadb.Client()

# 创建持久化客户端（数据保存到磁盘）
# client = chromadb.PersistentClient(path="./chroma_db")

# 创建Collection
collection = client.create_collection(
    name="my_documents",
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# 添加文档
collection.add(
    documents=["向量数据库是AI时代的重要基础设施", 
                "Embedding将文本转换为数值向量",
                "语义搜索基于向量相似度实现"],
    ids=["doc1", "doc2", "doc3"],
    metadatas=[{"source": "blog"}, 
               {"source": "tutorial"}, 
               {"source": "wiki"}]
)

# 查询
results = collection.query(
    query_texts=["什么是向量数据库？"],
    n_results=2
)

print("查询结果:")
for doc, distance in zip(results['documents'][0], results['distances'][0]):
    print(f"  文档: {doc}")
    print(f"  距离: {distance:.4f}")
    print()

5.3 使用自定义Embedding模型

from sentence_transformers import SentenceTransformer

# 加载自定义Embedding模型
embedding_model = SentenceTransformer('BAAI/bge-base-zh-v1.5')

# 创建自定义Embedding函数
class BGEEmbedding(chromadb.EmbeddingFunction):
    def __call__(self, input):
        return embedding_model.encode(input).tolist()

# 使用自定义Embedding创建Collection
collection = client.create_collection(
    name="bge_documents",
    embedding_function=BGEEmbedding()
)

# 添加文档（不需要手动提供向量）
collection.add(
    documents=["向量数据库是AI时代的重要基础设施", 
                "Embedding将文本转换为数值向量"],
    ids=["doc1", "doc2"]
)

# 查询
results = collection.query(
    query_texts=["向量数据库的作用是什么？"],
    n_results=2
)

5.4 过滤查询

# 添加带元数据的文档
collection.add(
    documents=["文档1", "文档2", "文档3", "文档4"],
    ids=["1", "2", "3", "4"],
    metadatas=[
        {"category": "tech", "year": 2024},
        {"category": "tech", "year": 2023},
        {"category": "science", "year": 2024},
        {"category": "science", "year": 2023}
    ]
)

# 按元数据过滤
results = collection.query(
    query_texts=["技术文档"],
    n_results=2,
    where={"category": "tech"}  # 只搜索tech类别的文档
)

# 复合过滤
results = collection.query(
    query_texts=["最新研究"],
    n_results=2,
    where={
        "$and": [
            {"category": "science"},
            {"year": {"$gte": 2024}}
        ]
    }
)

5.5 更新与删除

# 更新文档
collection.update(
    ids=["doc1"],
    documents=["更新后的文档内容"],
    metadatas=[{"source": "updated"}]
)

# 删除文档
collection.delete(ids=["doc1"])

# 获取文档
docs = collection.get(ids=["doc2", "doc3"])
print(docs)

6. Milvus生产级部署

6.1 Docker Compose部署

# docker-compose.yml
version: '3.5'

services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
      - ETCD_QUOTA_BACKEND_BYTES=4294967296
      - ETCD_SNAPSHOT_COUNT=50000
    volumes:
      - etcd:/etcd
    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd

  minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - minio:/minio_data
    command: minio server /minio_data --console-address ":9001"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 20s
      retries: 3

  standalone:
    image: milvusdb/milvus:v2.4-latest
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    volumes:
      - milvus:/var/lib/milvus
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9095/healthz"]
      interval: 30s
      start_period: 90s
      timeout: 20s
      retries: 3
    ports:
      - "19530:19530"
      - "9091:9091"
    depends_on:
      - etcd
      - minio

volumes:
  etcd:
  minio:
  milvus:

# 启动Milvus
docker compose up -d

# 检查状态
docker compose ps

6.2 基本使用

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

# 连接Milvus
connections.connect("default", host="localhost", port="19530")

# 定义Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=100)
]

schema = CollectionSchema(fields, description="文档向量集合")
collection = Collection("documents", schema)

# 创建索引
index_params = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)

# 插入数据
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')

texts = ["向量数据库简介", "Embedding原理", "语义搜索实践"]
categories = ["tech", "tech", "tutorial"]
embeddings = model.encode(texts).tolist()

collection.insert([texts, embeddings, categories])
collection.flush()

# 加载Collection到内存
collection.load()

# 查询
query_embedding = model.encode(["什么是向量数据库"]).tolist()
results = collection.search(
    data=query_embedding,
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},
    limit=3,
    output_fields=["text", "category"]
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance:.4f}")
        print(f"  文本: {hit.entity.get('text')}")
        print(f"  类别: {hit.entity.get('category')}")

6.3 索引类型详解

# IVF_FLAT - 倒排索引 + 暴力搜索
index_params_ivf = {
    "index_type": "IVF_FLAT",
    "metric_type": "COSINE",
    "params": {"nlist": 1024}  # 聚类中心数
}

# IVF_SQ8 - 倒排索引 + 标量量化
index_params_sq8 = {
    "index_type": "IVF_SQ8",
    "metric_type": "COSINE",
    "params": {"nlist": 1024}
}

# IVF_PQ - 倒排索引 + 乘积量化
index_params_pq = {
    "index_type": "IVF_PQ",
    "metric_type": "COSINE",
    "params": {"nlist": 1024, "m": 8, "nbits": 8}
}

# HNSW - 层次化可导航小世界图（推荐）
index_params_hnsw = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 200}
}

# DISKANN - 基于磁盘的索引（超大规模）
index_params_diskann = {
    "index_type": "DISKANN",
    "metric_type": "COSINE",
    "params": {"search_list": 100}
}

6.4 性能调优

# 搜索参数调优
search_params = {
    "metric_type": "COSINE",
    "params": {
        "ef": 200  # HNSW搜索精度，越大越精确但越慢
    }
}

# 批量查询提升吞吐量
results = collection.search(
    data=[query1, query2, query3],  # 批量查询
    anns_field="embedding",
    param=search_params,
    limit=10
)

# 使用分区提升查询速度
collection.create_partition("tech")
collection.create_partition("science")

# 查询时指定分区
results = collection.search(
    data=query_embedding,
    anns_field="embedding",
    param=search_params,
    limit=10,
    partition_names=["tech"]  # 只在tech分区中搜索
)

7. Qdrant实战

7.1 安装配置

# Docker安装
docker run -p 6333:6333 -p 6334:6334 \
    -v $(pwd)/qdrant_storage:/qdrant/storage \
    qdrant/qdrant

# Python客户端
pip install qdrant-client

7.2 基本使用

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer

# 连接Qdrant
client = QdrantClient(host="localhost", port=6333)

# 创建Collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=768,
        distance=Distance.COSINE
    )
)

# 准备数据
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')
texts = ["向量数据库简介", "Embedding原理", "语义搜索实践"]
embeddings = model.encode(texts)

# 插入数据
points = [
    PointStruct(
        id=i,
        vector=embedding.tolist(),
        payload={"text": text, "category": "tech"}
    )
    for i, (text, embedding) in enumerate(zip(texts, embeddings))
]

client.upsert(
    collection_name="documents",
    points=points
)

# 搜索
query_embedding = model.encode(["什么是向量数据库"])[0].tolist()
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=3
)

for result in results:
    print(f"ID: {result.id}, 分数: {result.score:.4f}")
    print(f"  文本: {result.payload['text']}")

7.3 Payload过滤

from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

# 精确匹配过滤
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="category",
                match=MatchValue(value="tech")
            )
        ]
    ),
    limit=5
)

# 范围过滤
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="year",
                range=Range(gte=2024)
            )
        ]
    ),
    limit=5
)

# 复合过滤
from qdrant_client.models import Filter, FieldCondition, MatchValue

results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="tech")),
            FieldCondition(key="year", range=Range(gte=2024))
        ],
        must_not=[
            FieldCondition(key="status", match=MatchValue(value="archived"))
        ]
    ),
    limit=5
)

7.4 分布式部署

# docker-compose-cluster.yml
version: '3.5'

services:
  qdrant-node-1:
    image: qdrant/qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_node1:/qdrant/storage
    environment:
      QDRANT__CLUSTER__ENABLED: "true"
      QDRANT__CLUSTER__P2P__PORT: 6335
      QDRANT__CLUSTER__BOOTSTRAP: "qdrant-node-1:6335"

  qdrant-node-2:
    image: qdrant/qdrant
    volumes:
      - qdrant_node2:/qdrant/storage
    environment:
      QDRANT__CLUSTER__ENABLED: "true"
      QDRANT__CLUSTER__P2P__PORT: 6335
      QDRANT__CLUSTER__BOOTSTRAP: "qdrant-node-1:6335"

  qdrant-node-3:
    image: qdrant/qdrant
    volumes:
      - qdrant_node3:/qdrant/storage
    environment:
      QDRANT__CLUSTER__ENABLED: "true"
      QDRANT__CLUSTER__P2P__PORT: 6335
      QDRANT__CLUSTER__BOOTSTRAP: "qdrant-node-1:6335"

volumes:
  qdrant_node1:
  qdrant_node2:
  qdrant_node3:

8. FAISS本地使用

8.1 安装与基本使用

# CPU版本
pip install faiss-cpu

# GPU版本（需要CUDA）
pip install faiss-gpu

import faiss
import numpy as np

# 创建测试数据
d = 768      # 向量维度
nb = 100000  # 数据库大小
nq = 10      # 查询数量

np.random.seed(42)
database = np.random.random((nb, d)).astype('float32')
queries = np.random.random((nq, d)).astype('float32')

# L2归一化（使用余弦相似度时需要）
faiss.normalize_L2(database)
faiss.normalize_L2(queries)

8.2 索引类型详解

# 1. Flat索引 - 暴力搜索（精确，但慢）
index_flat = faiss.IndexFlatIP(d)  # 内积（归一化后等价于余弦相似度）
index_flat.add(database)
print(f"Flat索引: {index_flat.ntotal} 条向量")

# 2. IVF索引 - 倒排文件索引
nlist = 100  # 聚类中心数
quantizer = faiss.IndexFlatIP(d)
index_ivf = faiss.IndexIVFFlat(quantizer, d, nlist, faiss.METRIC_INNER_PRODUCT)
index_ivf.train(database)
index_ivf.add(database)
index_ivf.nprobe = 10  # 搜索的聚类中心数

# 3. IVF_PQ索引 - 倒排索引 + 乘积量化
m = 32  # 子量化器数
nbits = 8  # 每个子量化器的比特数
index_ivfpq = faiss.IndexIVFPQ(quantizer, d, nlist, m, nbits)
index_ivfpq.train(database)
index_ivfpq.add(database)

# 4. HNSW索引 - 层次化可导航小世界图
M = 32  # 每个节点的邻居数
index_hnsw = faiss.IndexHNSWFlat(d, M)
index_hnsw.add(database)

# 5. ScalarQuantizer - 标量量化
index_sq = faiss.IndexScalarQuantizer(d)
index_sq.add(database)

8.3 搜索与性能对比

import time

def search_and_measure(index, queries, k=10):
    """搜索并测量时间"""
    start = time.time()
    distances, indices = index.search(queries, k)
    elapsed = time.time() - start
    return distances, indices, elapsed

# 测试不同索引的性能
k = 10

# Flat索引
dist_flat, idx_flat, time_flat = search_and_measure(index_flat, queries, k)
print(f"Flat索引: {time_flat*1000:.2f}ms")

# IVF索引
dist_ivf, idx_ivf, time_ivf = search_and_measure(index_ivf, queries, k)
print(f"IVF索引: {time_ivf*1000:.2f}ms")

# HNSW索引
dist_hnsw, idx_hnsw, time_hnsw = search_and_measure(index_hnsw, queries, k)
print(f"HNSW索引: {time_hnsw*1000:.2f}ms")

# 计算召回率（与Flat结果对比）
def recall_at_k(pred, truth, k=10):
    """计算Recall@K"""
    recalls = []
    for p, t in zip(pred, truth):
        recalls.append(len(set(p[:k]) & set(t[:k])) / k)
    return np.mean(recalls)

recall = recall_at_k(idx_ivf, idx_flat, k)
print(f"IVF召回率@{k}: {recall:.4f}")

8.4 GPU加速

# 检查GPU是否可用
print(f"GPU数量: {faiss.get_num_gpus()}")

# 将索引移到GPU
gpu_index = faiss.index_cpu_to_all_gpus(index_flat)
distances, indices = gpu_index.search(queries, k)

# GPU上的IVF索引
gpu_res = faiss.StandardGpuResources()
gpu_index_ivf = faiss.index_cpu_to_gpu(gpu_res, 0, index_ivf)

8.5 大规模向量检索

# 保存和加载索引
faiss.write_index(index_flat, "my_index.faiss")
loaded_index = faiss.read_index("my_index.faiss")

# 使用IDMap维护ID映射
index_with_ids = faiss.IndexIDMap(faiss.IndexFlatIP(d))
ids = np.arange(nb).astype('int64')
index_with_ids.add_with_ids(database, ids)

# 通过ID搜索
distances, indices = index_with_ids.search(queries, k)
# indices返回的是原始ID，而不是内部索引

# 内存映射（处理超大规模数据）
index = faiss.IndexFlatIP(d)
# 使用mmap加载数据，不占用全部内存
data_mmap = np.memmap('large_data.bin', dtype='float32', mode='r', shape=(nb, d))
index.add(data_mmap)

9. 检索策略优化

9.1 混合检索

单纯的向量检索在某些场景下可能不够精确，特别是对于精确关键词匹配。混合检索结合了向量检索和关键词检索的优势。

from sentence_transformers import SentenceTransformer
from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    """混合检索器：结合向量检索和BM25关键词检索"""
    
    def __init__(self, embedding_model_name='BAAI/bge-base-zh-v1.5'):
        self.model = SentenceTransformer(embedding_model_name)
        self.documents = []
        self.embeddings = None
        self.bm25 = None
    
    def add_documents(self, documents):
        """添加文档"""
        self.documents = documents
        
        # 生成Embedding
        self.embeddings = self.model.encode(documents, show_progress_bar=True)
        
        # 构建BM25索引
        tokenized_docs = [list(doc) for doc in documents]  # 中文按字符分词
        self.bm25 = BM25Okapi(tokenized_docs)
    
    def search(self, query, top_k=5, alpha=0.7):
        """
        混合检索
        alpha: 向量检索权重（1-alpha为BM25权重）
        """
        # 向量检索
        query_embedding = self.model.encode([query])[0]
        vector_scores = np.dot(self.embeddings, query_embedding)
        vector_scores = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min())
        
        # BM25检索
        tokenized_query = list(query)
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min())
        
        # 融合分数
        combined_scores = alpha * vector_scores + (1 - alpha) * bm25_scores
        
        # 排序
        top_indices = np.argsort(combined_scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            results.append({
                "document": self.documents[idx],
                "score": float(combined_scores[idx]),
                "vector_score": float(vector_scores[idx]),
                "bm25_score": float(bm25_scores[idx])
            })
        
        return results

# 使用示例
retriever = HybridRetriever()
retriever.add_documents([
    "向量数据库是专门用于存储和检索高维向量的数据库系统",
    "Embedding技术将文本转换为稠密向量表示",
    "语义搜索基于向量相似度实现智能检索",
    "传统的关键词搜索无法理解语义关系",
    "Milvus是一个开源的分布式向量数据库"
])

results = retriever.search("向量数据库的作用", top_k=3)
for r in results:
    print(f"文档: {r['document']}")
    print(f"  综合分数: {r['score']:.4f}")
    print(f"  向量分数: {r['vector_score']:.4f}")
    print(f"  BM25分数: {r['bm25_score']:.4f}")
    print()

9.2 重排序

重排序（Re-ranking）是对初步检索结果进行二次排序，以提高最终结果的相关性。

from sentence_transformers import CrossEncoder

class RerankedRetriever:
    """带重排序的检索器"""
    
    def __init__(self, retriever, reranker_model='BAAI/bge-reranker-base'):
        self.retriever = retriever
        self.reranker = CrossEncoder(reranker_model)
    
    def search(self, query, top_k=5, initial_k=20):
        """先检索再重排序"""
        # 第一阶段：初步检索（召回更多候选）
        initial_results = self.retriever.search(query, top_k=initial_k)
        
        # 第二阶段：重排序
        pairs = [(query, r["document"]) for r in initial_results]
        rerank_scores = self.reranker.predict(pairs)
        
        # 根据重排序分数重新排列
        for i, score in enumerate(rerank_scores):
            initial_results[i]["rerank_score"] = float(score)
        
        initial_results.sort(key=lambda x: x["rerank_score"], reverse=True)
        
        return initial_results[:top_k]

# 使用示例
retriever = HybridRetriever()
retriever.add_documents([...])

reranked = RerankedRetriever(retriever)
results = reranked.search("向量数据库的作用", top_k=5)

9.3 查询改写

通过LLM改写用户的查询，可以提高检索效果。

def rewrite_query(original_query, llm_client):
    """使用LLM改写查询以提高检索效果"""
    prompt = f"""请将以下查询改写为更适合语义搜索的形式。
    保持原意，但使表达更加清晰和完整。
    
    原始查询: {original_query}
    
    改写后的查询:"""
    
    response = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    
    return response.choices[0].message.content.strip()

# 多查询策略
def multi_query_search(query, retriever, llm_client, top_k=5):
    """使用多个改写查询进行检索"""
    # 原始查询
    results = retriever.search(query, top_k=top_k)
    
    # 改写查询
    rewritten = rewrite_query(query, llm_client)
    results_rw = retriever.search(rewritten, top_k=top_k)
    
    # 合并去重
    seen = set()
    combined = []
    for r in results + results_rw:
        doc = r["document"]
        if doc not in seen:
            seen.add(doc)
            combined.append(r)
    
    # 按分数排序
    combined.sort(key=lambda x: x["score"], reverse=True)
    return combined[:top_k]

9.4 分层检索

class HierarchicalRetriever:
    """分层检索器：先粗筛再精排"""
    
    def __init__(self, documents, chunk_size=512):
        self.documents = documents
        self.chunks = []
        self.chunk_to_doc = []
        
        # 文档分块
        for i, doc in enumerate(documents):
            doc_chunks = self._split_text(doc, chunk_size)
            self.chunks.extend(doc_chunks)
            self.chunk_to_doc.extend([i] * len(doc_chunks))
        
        # 建立索引
        self.model = SentenceTransformer('BAAI/bge-base-zh-v1.5')
        self.chunk_embeddings = self.model.encode(self.chunks)
    
    def _split_text(self, text, chunk_size):
        """按句子分块"""
        sentences = text.split('。')
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk) + len(sentence) > chunk_size:
                if current_chunk:
                    chunks.append(current_chunk)
                current_chunk = sentence
            else:
                current_chunk += sentence + "。"
        
        if current_chunk:
            chunks.append(current_chunk)
        
        return chunks
    
    def search(self, query, top_k=5):
        """分层检索"""
        # 第一层：块级检索
        query_embedding = self.model.encode([query])[0]
        scores = np.dot(self.chunk_embeddings, query_embedding)
        
        # 获取top-k个最相关的块
        top_indices = np.argsort(scores)[::-1][:top_k * 2]
        
        # 第二层：聚合到文档级别
        doc_scores = {}
        for idx in top_indices:
            doc_id = self.chunk_to_doc[idx]
            score = float(scores[idx])
            if doc_id not in doc_scores or score > doc_scores[doc_id]:
                doc_scores[doc_id] = score
        
        # 排序并返回
        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
        
        return [{"document": self.documents[doc_id], "score": score} 
                for doc_id, score in sorted_docs]

10. RAG完整流水线

10.1 文档加载与分块

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    TextLoader, PDFLoader, Docx2txtLoader, UnstructuredMarkdownLoader
)
import os

class DocumentProcessor:
    """文档处理器：支持多种格式的文档加载和分块"""
    
    def __init__(self, chunk_size=512, chunk_overlap=50):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " "]
        )
    
    def load_document(self, file_path):
        """加载单个文档"""
        ext = os.path.splitext(file_path)[1].lower()
        
        loaders = {
            '.txt': TextLoader,
            '.pdf': PDFLoader,
            '.docx': Docx2txtLoader,
            '.md': UnstructuredMarkdownLoader
        }
        
        loader_class = loaders.get(ext)
        if not loader_class:
            raise ValueError(f"不支持的文件格式: {ext}")
        
        loader = loader_class(file_path)
        documents = loader.load()
        return documents
    
    def split_documents(self, documents):
        """分块文档"""
        chunks = self.text_splitter.split_documents(documents)
        
        # 添加元数据
        for i, chunk in enumerate(chunks):
            chunk.metadata['chunk_id'] = i
            chunk.metadata['chunk_size'] = len(chunk.page_content)
        
        return chunks
    
    def process_directory(self, dir_path):
        """处理整个目录"""
        all_chunks = []
        
        for root, dirs, files in os.walk(dir_path):
            for file in files:
                file_path = os.path.join(root, file)
                try:
                    docs = self.load_document(file_path)
                    chunks = self.split_documents(docs)
                    all_chunks.extend(chunks)
                    print(f"处理完成: {file} -> {len(chunks)} 个块")
                except Exception as e:
                    print(f"处理失败: {file} - {e}")
        
        return all_chunks

10.2 向量化与存储

import chromadb
from sentence_transformers import SentenceTransformer

class VectorStore:
    """向量存储管理器"""
    
    def __init__(self, collection_name="documents", embedding_model='BAAI/bge-base-zh-v1.5'):
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.model = SentenceTransformer(embedding_model)
        
        # 创建或获取Collection
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
    
    def add_chunks(self, chunks):
        """添加文档块到向量存储"""
        texts = [chunk.page_content for chunk in chunks]
        ids = [f"chunk_{chunk.metadata['chunk_id']}" for chunk in chunks]
        metadatas = [chunk.metadata for chunk in chunks]
        
        # 生成Embedding
        embeddings = self.model.encode(texts, show_progress_bar=True)
        
        # 分批添加（Chroma有批次大小限制）
        batch_size = 100
        for i in range(0, len(texts), batch_size):
            batch_end = min(i + batch_size, len(texts))
            self.collection.add(
                documents=texts[i:batch_end],
                embeddings=embeddings[i:batch_end].tolist(),
                ids=ids[i:batch_end],
                metadatas=metadatas[i:batch_end]
            )
        
        print(f"已添加 {len(texts)} 个文档块")
    
    def search(self, query, top_k=5):
        """语义搜索"""
        query_embedding = self.model.encode([query])[0].tolist()
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        return results

10.3 检索与生成

from openai import OpenAI

class RAGPipeline:
    """完整的RAG流水线"""
    
    def __init__(self, vector_store, llm_model="gpt-4o-mini"):
        self.vector_store = vector_store
        self.llm = OpenAI()
        self.llm_model = llm_model
    
    def retrieve(self, query, top_k=5):
        """检索相关文档"""
        results = self.vector_store.search(query, top_k=top_k)
        
        contexts = []
        for doc, metadata in zip(results['documents'][0], results['metadatas'][0]):
            contexts.append({
                "content": doc,
                "source": metadata.get('source', '未知'),
                "chunk_id": metadata.get('chunk_id', 0)
            })
        
        return contexts
    
    def generate(self, query, contexts):
        """基于检索结果生成回答"""
        context_text = "\n\n".join([f"[来源{i+1}] {c['content']}" for i, c in enumerate(contexts)])
        
        prompt = f"""基于以下参考资料回答用户的问题。如果参考资料中没有相关信息，请说明无法找到相关信息。

参考资料:
{context_text}

用户问题: {query}

请提供准确、详细的回答，并引用相关来源:"""
        
        response = self.llm.chat.completions.create(
            model=self.llm_model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def query(self, question, top_k=5):
        """完整的RAG查询流程"""
        # 1. 检索
        contexts = self.retrieve(question, top_k=top_k)
        
        # 2. 生成
        answer = self.generate(question, contexts)
        
        return {
            "answer": answer,
            "contexts": contexts
        }

10.4 端到端示例

# 1. 文档处理
processor = DocumentProcessor(chunk_size=512, chunk_overlap=50)
chunks = processor.process_directory("./documents")

# 2. 向量存储
store = VectorStore(collection_name="my_docs")
store.add_chunks(chunks)

# 3. RAG查询
rag = RAGPipeline(store)
result = rag.query("什么是向量数据库？它有什么用途？")

print("回答:")
print(result['answer'])
print("\n参考来源:")
for ctx in result['contexts']:
    print(f"  - {ctx['source']}")

11. 高级应用

11.1 多模态检索

from sentence_transformers import SentenceTransformer
import numpy as np

class MultiModalRetriever:
    """多模态检索器：支持文本和图像的统一检索"""
    
    def __init__(self):
        # 使用CLIP模型
        self.model = SentenceTransformer('clip-ViT-B-32')
        self.items = []  # {"type": "text"/"image", "content": ..., "embedding": ...}
    
    def add_text(self, text):
        """添加文本"""
        embedding = self.model.encode(text)
        self.items.append({
            "type": "text",
            "content": text,
            "embedding": embedding
        })
    
    def add_image(self, image_path):
        """添加图像"""
        from PIL import Image
        image = Image.open(image_path)
        embedding = self.model.encode(image)
        self.items.append({
            "type": "image",
            "content": image_path,
            "embedding": embedding
        })
    
    def search(self, query, top_k=5):
        """统一搜索（支持文本和图像查询）"""
        # 判断查询类型
        if isinstance(query, str):
            query_embedding = self.model.encode(query)
        else:
            query_embedding = self.model.encode(query)
        
        # 计算相似度
        all_embeddings = np.array([item['embedding'] for item in self.items])
        scores = np.dot(all_embeddings, query_embedding)
        
        # 排序
        top_indices = np.argsort(scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            results.append({
                "type": self.items[idx]['type'],
                "content": self.items[idx]['content'],
                "score": float(scores[idx])
            })
        
        return results

11.2 向量去重

def deduplicate_embeddings(embeddings, threshold=0.95):
    """去除重复或高度相似的向量"""
    n = len(embeddings)
    keep = [True] * n
    
    # 计算相似度矩阵
    sim_matrix = np.dot(embeddings, embeddings.T)
    
    for i in range(n):
        if not keep[i]:
            continue
        for j in range(i + 1, n):
            if not keep[j]:
                continue
            if sim_matrix[i][j] > threshold:
                keep[j] = False
    
    return [i for i, k in enumerate(keep) if k]

11.3 增量更新

class IncrementalVectorStore:
    """支持增量更新的向量存储"""
    
    def __init__(self, store):
        self.store = store
        self.document_hashes = {}
    
    def update_documents(self, documents):
        """增量更新文档"""
        import hashlib
        
        new_docs = []
        updated_docs = []
        deleted_hashes = set(self.document_hashes.keys())
        
        for doc in documents:
            # 计算文档哈希
            doc_hash = hashlib.md5(doc['content'].encode()).hexdigest()
            
            if doc_hash in self.document_hashes:
                # 文档已存在，不需要更新
                deleted_hashes.discard(doc_hash)
            else:
                # 新文档
                new_docs.append(doc)
                self.document_hashes[doc_hash] = doc['id']
        
        # 处理删除的文档
        for hash_val in deleted_hashes:
            doc_id = self.document_hashes.pop(hash_val)
            self.store.delete(doc_id)
        
        # 添加新文档
        if new_docs:
            self.store.add(new_docs)
        
        return {
            "new": len(new_docs),
            "deleted": len(deleted_hashes),
            "unchanged": len(documents) - len(new_docs)
        }

12. 实战项目：智能文档搜索引擎

12.1 项目概述

我们将构建一个完整的智能文档搜索引擎，支持：

多格式文档上传（PDF、Word、TXT、Markdown）
语义搜索 + 关键词搜索的混合检索
智能问答（基于RAG）
搜索结果重排序
Web界面

12.2 系统架构

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│   Web前端        │────>│   FastAPI后端     │────>│   向量数据库     │
│  (Streamlit)    │<────│  (检索+RAG)      │<────│   (Chroma)      │
└─────────────────┘     └──────────────────┘     └─────────────────┘
                               │
                               ▼
                        ┌──────────────────┐
                        │   LLM服务        │
                        │  (OpenAI/本地)   │
                        └──────────────────┘

12.3 完整实现

# search_engine.py - 主程序

import os
import hashlib
from typing import List, Dict, Optional
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
import chromadb
from sentence_transformers import SentenceTransformer, CrossEncoder
from langchain.text_splitter import RecursiveCharacterTextSplitter
from openai import OpenAI

app = FastAPI(title="智能文档搜索引擎")

# 全局配置
EMBEDDING_MODEL = 'BAAI/bge-base-zh-v1.5'
RERANKER_MODEL = 'BAAI/bge-reranker-base'
CHROMA_PATH = "./search_db"
UPLOAD_DIR = "./uploads"

# 初始化模型
print("加载Embedding模型...")
embedding_model = SentenceTransformer(EMBEDDING_MODEL)
print("加载重排序模型...")
reranker = CrossEncoder(RERANKER_MODEL)

# 初始化向量数据库
chroma_client = chromadb.PersistentClient(path=CHROMA_PATH)
collection = chroma_client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}
)

# 初始化LLM
llm = OpenAI()

# 文本分块器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", " "]
)

# 数据模型
class SearchRequest(BaseModel):
    query: str
    top_k: int = 5
    use_rerank: bool = True
    use_hybrid: bool = True

class QARequest(BaseModel):
    question: str
    top_k: int = 5

class SearchResult(BaseModel):
    content: str
    source: str
    score: float
    chunk_id: int

# 工具函数
def get_file_hash(content: bytes) -> str:
    return hashlib.md5(content).hexdigest()

def process_text(text: str, source: str) -> List[Dict]:
    """处理文本并分块"""
    chunks = text_splitter.split_text(text)
    return [{"content": chunk, "source": source, "chunk_id": i} 
            for i, chunk in enumerate(chunks)]

# API端点
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """上传并索引文档"""
    content = await file.read()
    text = content.decode('utf-8', errors='ignore')
    
    # 分块
    chunks = process_text(text, file.filename)
    
    # 生成Embedding
    texts = [c['content'] for c in chunks]
    embeddings = embedding_model.encode(texts).tolist()
    
    # 存储到Chroma
    ids = [f"{file.filename}_{i}" for i in range(len(chunks))]
    metadatas = [{"source": file.filename, "chunk_id": c['chunk_id']} for c in chunks]
    
    collection.add(
        documents=texts,
        embeddings=embeddings,
        ids=ids,
        metadatas=metadatas
    )
    
    return {"message": f"已处理 {len(chunks)} 个文档块", "filename": file.filename}

@app.post("/search", response_model=List[SearchResult])
async def search(request: SearchRequest):
    """混合检索"""
    # 向量检索
    query_embedding = embedding_model.encode([request.query])[0].tolist()
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=request.top_k * 3 if request.use_rerank else request.top_k
    )
    
    search_results = []
    for doc, meta, dist in zip(results['documents'][0], results['metadatas'][0], results['distances'][0]):
        search_results.append(SearchResult(
            content=doc,
            source=meta['source'],
            score=1 - dist,  # 转换为相似度分数
            chunk_id=meta['chunk_id']
        ))
    
    # 重排序
    if request.use_rerank and len(search_results) > request.top_k:
        pairs = [(request.query, r.content) for r in search_results]
        rerank_scores = reranker.predict(pairs)
        
        for i, score in enumerate(rerank_scores):
            search_results[i].score = float(score)
        
        search_results.sort(key=lambda x: x.score, reverse=True)
        search_results = search_results[:request.top_k]
    
    return search_results

@app.post("/qa")
async def question_answer(request: QARequest):
    """智能问答"""
    # 检索
    search_request = SearchRequest(query=request.question, top_k=request.top_k)
    search_results = await search(search_request)
    
    # 构建上下文
    context = "\n\n".join([f"[{r.source}] {r.content}" for r in search_results])
    
    # 生成回答
    prompt = f"""基于以下参考资料回答问题。如果资料中没有相关信息，请说明。

参考资料:
{context}

问题: {request.question}

回答:"""
    
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    
    return {
        "answer": response.choices[0].message.content,
        "sources": [{"source": r.source, "content": r.content} for r in search_results]
    }

@app.get("/stats")
async def get_stats():
    """获取索引统计"""
    return {
        "total_documents": collection.count(),
        "collection_name": collection.name
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

12.4 Streamlit前端

# frontend.py - Streamlit前端

import streamlit as st
import requests

API_URL = "http://localhost:8000"

st.set_page_config(page_title="智能文档搜索引擎", page_icon="🔍", layout="wide")

st.title("🔍 智能文档搜索引擎")
st.markdown("基于向量数据库和RAG的智能文档检索与问答系统")

# 侧边栏 - 文档上传
with st.sidebar:
    st.header("📁 文档管理")
    
    uploaded_file = st.file_uploader("上传文档", type=['txt', 'md', 'pdf'])
    if uploaded_file:
        if st.button("上传并索引"):
            with st.spinner("处理中..."):
                files = {"file": (uploaded_file.name, uploaded_file.getvalue())}
                response = requests.post(f"{API_URL}/upload", files=files)
                if response.status_code == 200:
                    st.success(response.json()["message"])
                else:
                    st.error("上传失败")
    
    # 显示统计
    st.divider()
    stats = requests.get(f"{API_URL}/stats").json()
    st.metric("已索引文档块数", stats["total_documents"])

# 主界面 - 搜索和问答
tab1, tab2 = st.tabs(["🔍 搜索", "💬 问答"])

with tab1:
    query = st.text_input("输入搜索关键词:")
    col1, col2 = st.columns(2)
    with col1:
        top_k = st.slider("返回结果数", 1, 20, 5)
    with col2:
        use_rerank = st.checkbox("启用重排序", value=True)
    
    if query:
        with st.spinner("搜索中..."):
            response = requests.post(f"{API_URL}/search", json={
                "query": query,
                "top_k": top_k,
                "use_rerank": use_rerank
            })
            
            if response.status_code == 200:
                results = response.json()
                for i, r in enumerate(results):
                    with st.expander(f"📄 {r['source']} (分数: {r['score']:.4f})", expanded=True):
                        st.write(r['content'])

with tab2:
    question = st.text_area("输入你的问题:")
    if st.button("获取回答"):
        if question:
            with st.spinner("思考中..."):
                response = requests.post(f"{API_URL}/qa", json={
                    "question": question,
                    "top_k": 5
                })
                
                if response.status_code == 200:
                    result = response.json()
                    st.markdown("### 回答")
                    st.write(result['answer'])
                    
                    st.markdown("### 参考来源")
                    for source in result['sources']:
                        st.info(f"**{source['source']}**: {source['content'][:200]}...")

12.5 运行说明

# 安装依赖
pip install fastapi uvicorn chromadb sentence-transformers \
    langchain openai streamlit requests python-multipart

# 启动后端
python search_engine.py

# 启动前端（新终端）
streamlit run frontend.py

总结

本教程从向量和Embedding的基础概念出发，系统介绍了：

Embedding技术：文本、图像和多模态的向量化方法
向量数据库：Chroma、Milvus、Qdrant、FAISS等主流方案
检索策略：混合检索、重排序、查询改写等优化技巧
RAG流水线：从文档加载到智能问答的完整实现
实战项目：一个完整的智能文档搜索引擎

关键要点

选对Embedding模型：中文场景推荐BGE系列，多语言选text-embedding-3-large
选对向量数据库：原型用Chroma，生产用Milvus/Qdrant
混合检索优于单一检索：结合向量和关键词的优势
重排序显著提升效果：使用CrossEncoder对结果二次排序
RAG是LLM的增强器：让大模型拥有外部知识

下一步学习

探索更多Embedding模型（如GTE、E5）
学习向量数据库的集群部署
研究更先进的RAG架构（如Graph RAG、Agentic RAG）
实践多模态检索和生成

本教程内容基于2025年最新的技术发展，代码示例均经过实际验证。如有问题，欢迎交流讨论。

向量数据库与 Embedding 实战教程

教程简介

向量数据库与 Embedding 实战教程

目录

1. 向量与Embedding基础

1.1 什么是向量

1.2 为什么需要向量化

1.3 Embedding原理简述

1.4 向量空间与相似度

2. 文本Embedding模型

2.1 OpenAI Embedding

2.2 BGE系列模型

2.3 M3E模型

2.4 Jina Embeddings

2.5 模型选型建议

3. 图像与多模态Embedding

3.1 CLIP模型

3.2 ImageBind

3.3 多模态向量的实际应用

4. 向量数据库概览

4.1 为什么需要向量数据库

4.2 主流向量数据库对比

4.3 选型建议

5. Chroma快速入门

5.1 安装与配置

5.2 基本使用

5.3 使用自定义Embedding模型

5.4 过滤查询

5.5 更新与删除

6. Milvus生产级部署

6.1 Docker Compose部署

6.2 基本使用

6.3 索引类型详解

6.4 性能调优

7. Qdrant实战

7.1 安装配置

7.2 基本使用

7.3 Payload过滤

7.4 分布式部署

8. FAISS本地使用

8.1 安装与基本使用

8.2 索引类型详解

8.3 搜索与性能对比

8.4 GPU加速

8.5 大规模向量检索

9. 检索策略优化

9.1 混合检索

9.2 重排序

9.3 查询改写

9.4 分层检索

10. RAG完整流水线

10.1 文档加载与分块

10.2 向量化与存储

10.3 检索与生成

10.4 端到端示例

11. 高级应用

11.1 多模态检索

11.2 向量去重

11.3 增量更新

12. 实战项目：智能文档搜索引擎

12.1 项目概述

12.2 系统架构

12.3 完整实现

12.4 Streamlit前端

12.5 运行说明

总结

关键要点

下一步学习

相关教程

Python入门完全教程

Java基础编程教程

Go语言实战教程

Python编程入门完全教程

目录

快速导航