第3章：向量数据库构建

第3章：向量数据库构建

3.1 向量嵌入原理

向量嵌入是将文本转换为数值向量的技术，是RAG系统的核心组件。理解嵌入原理有助于我们选择合适的模型和优化检索效果。

向量嵌入的基本概念

1. 什么是向量嵌入
向量嵌入是将离散的文本、词语或段落转换为连续的数值向量的过程。这些向量在多维空间中能够捕获文本的语义信息。

2. 向量的数学表示

• 向量维度：每个向量由数值组成，维度决定了语义空间的大小
• 向量距离：不同文本之间的语义相似性通过向量距离衡量
• 向量空间：所有文本向量构成的高维语义空间

3. 向量嵌入的特性

• 语义相似性：语义相近的文本在向量空间中距离较近
• 维度统一：所有文本都映射到相同维度的向量空间
• 数值稳定：相同的文本映射为相同的向量值

主流嵌入模型对比

1. OpenAI Embedding系列

text-embedding-ada-002

from openai import OpenAI
import numpy as np

client = OpenAI(api_key="your-api-key")

def get_ada_embedding(text):
    response = client.embeddings.create(
        model="text-embedding-ada-002",
        input=text
    )
    return response.data[0].embedding

# 使用示例
text = "机器学习是人工智能的重要分支"
embedding = get_ada_embedding(text)
print(f"向量维度: {len(embedding)}")  # 1536
​

text-embedding-3-small

def get_embedding_3_small(text):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
        dimensions=1536
    )
    return response.data[0].embedding
​

text-embedding-3-large

def get_embedding_3_large(text):
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=3072
    )
    return response.data[0].embedding
​

2. 开源嵌入模型

Sentence-BERT

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def get_sentence_bert_embeddings(texts):
    return model.encode(texts)

# 使用示例
texts = [
    "机器学习是人工智能的重要分支",
    "深度学习在图像识别中表现优异"
]
embeddings = get_sentence_bert_embeddings(texts)
print(f"向量维度: {embeddings.shape[1]}")  # 384
​

BGE系列

from FlagEmbedding import BGEModel

model = BGEModel('BAAI/bge-large-zh-v1.5', device='cpu')

def get_bge_embeddings(texts):
    return model.encode(texts)

# 使用示例
texts = ["自然语言处理是AI的重要领域"]
embeddings = get_bge_embeddings(texts)
print(f"向量维度: {embeddings.shape[1]}")  # 1024
​

嵌入模型选择指南

1. 选择标准

2. 场景化选择

通用场景

推荐配置:
  模型: text-embedding-3-small
  维度: 1536
  特点: 性价比高、通用性强
  适用: 通用问答、文档检索
  优势: 成本适中、效果稳定
​

中文优化场景

推荐配置:
  模型: bge-large-zh-v1.5
  维度: 1024
  特点: 中文优化、本地部署
  适用: 中文文档处理、中文问答
  优势: 中文语义理解更准确
​

高性能场景

推荐配置:
  模型: text-embedding-3-large
  维度: 3072
  特点: 高精度、语义理解强
  适用: 复杂语义、高精度要求
  优势: 语义理解能力最强
​

向量嵌入优化技巧

1. 批量处理优化

import asyncio
from openai import AsyncOpenAI

async def batch_embeddings_async(texts, model="text-embedding-3-small", batch_size=100):
    """异步批量获取嵌入向量"""
    client = AsyncOpenAI()
    all_embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        
        try:
            response = await client.embeddings.create(
                model=model,
                input=batch
            )
            batch_embeddings = [data.embedding for data in response.data]
            all_embeddings.extend(batch_embeddings)
            
        except Exception as e:
            print(f"批次 {i//batch_size} 处理失败: {str(e)}")
            # 填充零向量
            all_embeddings.extend([[] * 1536 for _ in range(len(batch))])
    
    return all_embeddings
​

2. 本地缓存优化

import json
import os
from hashlib import md5

class EmbeddingCache:
    """嵌入向量缓存"""
    
    def __init__(self, cache_dir="embedding_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def get_cache_key(self, text):
        """生成缓存键"""
        return md5(text.encode('utf-8')).hexdigest()
    
    def get_cached_embedding(self, text):
        """获取缓存的嵌入向量"""
        cache_key = self.get_cache_key(text)
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.json")
        
        if os.path.exists(cache_file):
            with open(cache_file, 'r') as f:
                return json.load(f)
        
        return None
    
    def save_embedding(self, text, embedding):
        """保存嵌入向量到缓存"""
        cache_key = self.get_cache_key(text)
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.json")
        
        with open(cache_file, 'w') as f:
            json.dump(embedding, f)
    
    def get_or_compute_embedding(self, text, compute_func):
        """获取或计算嵌入向量"""
        cached_embedding = self.get_cached_embedding(text)
        
        if cached_embedding is not None:
            return cached_embedding
        
        embedding = compute_func(text)
        self.save_embedding(text, embedding)
        return embedding
​

向量嵌入的最佳实践

1. 文本预处理优化

def preprocess_for_embedding(text):
    """为嵌入优化的文本预处理"""
    import re
    
    # 1. 去除特殊字符（保留中文、英文、数字、基本标点）
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s，。！？；：""''（）【】《》、…—]', '', text)
    
    # 2. 标准化空格
    text = re.sub(r'\s+', ' ', text).strip()
    
    # 3. 文本长度优化
    if len(text) > 8192:  # 大多数模型的最大长度限制
        text = text[:8192]
    
    return text
​

2. 查询优化

def optimize_query_for_embedding(query, max_length=500):
    """优化查询文本以获得更好的嵌入效果"""
    # 1. 去除疑问词（如果不需要）
    query = re.sub(r'^[谁什么哪里为什么怎么何时] ?', '', query)
    
    # 2. 提取关键词（如果适用）
    # 这里可以添加更复杂的提取逻辑
    
    # 3. 长度控制
    if len(query) > max_length:
        query = query[:max_length]
    
    return query
​

总结

向量嵌入是RAG系统的核心组件，选择合适的嵌入模型和优化策略对系统性能至关重要。主要要点包括：

1. 模型选择：根据场景需求选择合适的嵌入模型
2. 批量处理：采用异步批量处理提升性能
3. 缓存机制：使用缓存减少重复计算
4. 质量监控：定期监控嵌入向量质量
5. 多粒度策略：根据需求采用不同粒度的嵌入

通过合理的嵌入策略和优化，可以显著提升RAG系统的检索准确性和性能表现。