2.3 向量空间理论

向量空间的基本概念

什么是向量空间？

向量空间（Vector Space）是线性代数中的基本概念，是指具有特定运算规则的空间。在AI搜索中，向量空间是指所有文本向量构成的空间。

向量空间的定义：

• 向量空间是一个集合V，其中每个元素都是向量
• 定义了两种基本运算：向量加法和标量乘法
• 满足8条基本运算规律（结合律、交换律、分配律等）

向量空间的性质

基本运算：

1. 向量加法：v + w = w + v
2. 标量乘法：c·(v + w) = c·v + c·w
3. 零向量：v + 0 = v
4. 负向量：v + (-v) = 0

维数定义：
向量空间的维数是指基向量的数量。在文本向量中，维数通常对应词嵌入的维度。

向量空间的距离度量

常见距离度量

欧氏距离（Euclidean Distance）：

d(v, w) = √(Σ(v_i - w_i)²)
​

曼哈顿距离（Manhattan Distance）：

d(v, w) = Σ|v_i - w_i|
​

余弦距离（Cosine Distance）：

d(v, w) = 1 - (v · w) / (||v|| ||w||)
​

切比雪夫距离（Chebyshev Distance）：

d(v, w) = max(|v_i - w_i|)
​

距度量的选择策略

向量空间的维度约简

主成分分析（PCA）

PCA是最常用的降维方法，通过线性变换将数据投影到低维空间：

from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler

class VectorPCA:
    def __init__(self, n_components=0.95):
        self.pca = PCA(n_components=n_components)
        self.scaler = StandardScaler()
        
    def fit(self, vectors):
        """训练PCA模型"""
        # 标准化
        normalized = self.scaler.fit_transform(vectors)
        # PCA降维
        self.pca.fit(normalized)
        return self.pca.explained_variance_ratio_
    
    def transform(self, vectors):
        """降维转换"""
        normalized = self.scaler.transform(vectors)
        return self.pca.transform(normalized)
    
    def fit_transform(self, vectors):
        """训练并转换"""
        normalized = self.scaler.fit_transform(vectors)
        return self.pca.fit_transform(normalized)
​

t-SNE与UMAP

t-SNE：主要用于数据可视化，保留局部结构：

from sklearn.manifold import TSNE

def tsne_visualization(vectors, perplexity=30):
    tsne = TSNE(n_components=2, perplexity=perplexity)
    return tsne.fit_transform(vectors)
​

UMAP：相比t-SNE，UMAP在保持局部结构的同时，也能更好地保持全局结构：

import umap

def umap_visualization(vectors, n_neighbors=15):
    reducer = umap.UMAP(n_neighbors=n_neighbors)
    return reducer.fit_transform(vectors)
​

向量空间的索引结构

空间分割技术

KD-Tree：

from sklearn.neighbors import KDTree

class VectorIndex:
    def __init__(self, vectors):
        self.vectors = vectors
        self.tree = KDTree(vectors)
        
    def search(self, query_vector, k=10):
        """搜索最近邻"""
        distances, indices = self.tree.query(query_vector.reshape(1, -1), k=k)
        return indices[0], distances[0]
​

Ball Tree：

from sklearn.neighbors import BallTree

class BallTreeIndex:
    def __init__(self, vectors):
        self.vectors = vectors
        self.tree = BallTree(vectors)
        
    def search(self, query_vector, k=10):
        """搜索最近邻"""
        distances, indices = self.tree.query(query_vector.reshape(1, -1), k=k)
        return indices[0], distances[0]
​

近似最近邻（ANN）

HNSW（Hierarchical Navigable Small World）：
HNSW是一种高效的近似最近邻搜索算法，具有良好的时间和空间复杂度。

FAISS中的HNSW实现：

import faiss

def create_hnsw_index(vectors, m=16, ef=200):
    """创建HNSW索引"""
    dimension = vectors.shape[1]
    index = faiss.IndexHNSWFlat(dimension, m)
    index.hnsw.ef = ef
    index.train(vectors)
    index.add(vectors)
    return index
​

向量空间的度量学习

度量学习的概念

度量学习（Metric Learning）是学习一个距离度量函数，使得相似样本在向量空间中更接近。

目标：

• 相似样本之间的距离尽可能小
• 不相似样本之间的距离尽可能大

常见的度量学习方法

Siamese Network：

import torch
import torch.nn as nn

class SiameseNetwork(nn.Module):
    def __init__(self, input_dim, hidden_dim=128):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        
    def forward(self, x1, x2):
        # 编码两个输入
        emb1 = self.encoder(x1)
        emb2 = self.encoder(x2)
        
        # 计算距离
        distance = torch.abs(emb1 - emb2)
        
        return distance
​

Triplet Network：

class TripletNetwork(nn.Module):
    def __init__(self, input_dim, hidden_dim=128):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        
    def forward(self, anchor, positive, negative):
        # 编码三个样本
        anchor_emb = self.encoder(anchor)
        positive_emb = self.encoder(positive)
        negative_emb = self.encoder(negative)
        
        return anchor_emb, positive_emb, negative_emb
​

向量空间的应用场景

文本相似度计算

def text_similarity(text1, text2, vectorizer):
    """计算两个文本的相似度"""
    # 向量化
    vec1 = vectorizer.encode(text1)
    vec2 = vectorizer.encode(text2)
    
    # 计算余弦相似度
    similarity = cosine_similarity(vec1, vec2)
    
    return similarity

# 使用示例
vectorizer = SentenceVectorizer()
similarity = text_similarity("机器学习算法", "深度学习模型", vectorizer)
print(f"相似度: {similarity:.4f}")
​

文档聚类

from sklearn.cluster import KMeans

class DocumentCluster:
    def __init__(self, n_clusters=5):
        self.clusterer = KMeans(n_clusters=n_clusters, random_state=42)
        self.vectorizer = SentenceVectorizer()
        
    def fit(self, documents):
        """训练聚类模型"""
        # 向量化所有文档
        vectors = self.vectorizer.encode_documents(documents)
        
        # 聚类
        self.clusterer.fit(vectors)
        
        return self.clusterer.labels_
    
    def predict(self, document):
        """预测文档所属类别"""
        vector = self.vectorizer.encode_document(document)
        return self.clusterer.predict([vector])[0]
​

异常检测

from sklearn.ensemble import IsolationForest

class AnomalyDetector:
    def __init__(self, contamination=0.1):
        self.detector = IsolationForest(contamination=contamination, random_state=42)
        self.vectorizer = SentenceVectorizer()
        
    def fit(self, normal_documents):
        """训练异常检测模型"""
        # 向量化正常文档
        vectors = self.vectorizer.encode_documents(normal_documents)
        
        # 训练模型
        self.detector.fit(vectors)
        
    def detect(self, document):
        """检测文档是否为异常"""
        vector = self.vectorizer.encode_document(document)
        
        # 预测 (-1为异常，1为正常)
        prediction = self.detector.predict([vector])[0]
        
        return prediction == -1
​

向量空间优化技术

内存优化

import pickle
import gzip

class VectorStorage:
    def __init__(self, compression_level=6):
        self.vectors = {}
        self.compression_level = compression_level
        
    def add_vector(self, key, vector):
        """添加向量"""
        # 压缩存储
        compressed = gzip.compress(pickle.dumps(vector), 
                                 compresslevel=self.compression_level)
        self.vectors[key] = compressed
        
    def get_vector(self, key):
        """获取向量"""
        if key not in self.vectors:
            raise KeyError(f"Key {key} not found")
            
        # 解压
        compressed = self.vectors[key]
        vector = pickle.loads(gzip.decompress(compressed))
        
        return vector
    
    def save_to_file(self, filepath):
        """保存到文件"""
        with open(filepath, 'wb') as f:
            pickle.dump(self.vectors, f)
            
    def load_from_file(self, filepath):
        """从文件加载"""
        with open(filepath, 'rb') as f:
            self.vectors = pickle.load(f)
​

查询优化

class OptimizedVectorSearch:
    def __init__(self, vectors, index_type='hnsw'):
        self.vectors = vectors
        self.index_type = index_type
        
        # 根据类型创建索引
        if index_type == 'hnsw':
            self.index = self._create_hnsw_index()
        elif index_type == 'ivf':
            self.index = self._create_ivf_index()
        else:
            self.index = self._create_linear_index()
    
    def _create_hnsw_index(self):
        """创建HNSW索引"""
        import faiss
        dimension = self.vectors.shape[1]
        index = faiss.IndexHNSWFlat(dimension, 32)
        index.hnsw.ef = 200
        index.train(self.vectors)
        index.add(self.vectors)
        return index
    
    def _create_ivf_index(self):
        """创建IVF索引"""
        import faiss
        dimension = self.vectors.shape[1]
        nlist = 100  # 聚类数量
        quantizer = faiss.IndexFlatL2(dimension)
        index = faiss.IndexIVFFlat(quantizer, dimension, nlist)
        index.train(self.vectors)
        index.add(self.vectors)
        return index
    
    def search(self, query_vector, k=10):
        """搜索最近邻"""
        if self.index_type == 'hnsw':
            distances, indices = self.index.search(query_vector.reshape(1, -1), k)
        elif self.index_type == 'ivf':
            distances, indices = self.index.search(query_vector.reshape(1, -1), k)
        else:
            distances, indices = self.index.search(query_vector.reshape(1, -1), k)
        
        return indices[0], distances[0]
​

向量空间的未来趋势

多模态向量空间

随着AI技术的发展，多模态向量空间成为一个重要方向：

class MultimodalVectorSpace:
    def __init__(self):
        self.text_vectorizer = SentenceVectorizer()
        self.image_vectorizer = ImageVectorizer()
        self.audio_vectorizer = AudioVectorizer()
        
    def encode_multimodal(self, text=None, image=None, audio=None):
        """编码多模态数据"""
        features = []
        
        if text:
            text_vector = self.text_vectorizer.encode_sentence(text)
            features.append(text_vector)
            
        if image:
            image_vector = self.image_vectorizer.encode_image(image)
            features.append(image_vector)
            
        if audio:
            audio_vector = self.audio_vectorizer.encode_audio(audio)
            features.append(audio_vector)
        
        # 融合多模态特征
        if len(features) > 1:
            multimodal_vector = np.mean(features, axis=0)
        else:
            multimodal_vector = features[0]
            
        return multimodal_vector
​

时序向量空间

对于时序数据，需要特殊的向量空间表示：

class TemporalVectorSpace:
    def __init__(self, window_size=10):
        self.window_size = window_size
        self.sequence_encoder = nn.LSTM(768, 256, batch_first=True)
        
    def encode_sequence(self, text_sequence):
        """编码文本序列"""
        # 向量化每个时间步
        vectors = []
        for text in text_sequence:
            vector = self.text_vectorizer.encode_sentence(text)
            vectors.append(vector)
        
        # 序列编码
        vectors = torch.FloatTensor(np.array(vectors))
        packed = pack_padded_sequence(vectors, [len(text_sequence)], 
                                    batch_first=True, enforce_sorted=False)
        output, (hidden, cell) = self.sequence_encoder(packed)
        
        return hidden[-1]  # 使用最后一个隐藏状态
​

通过本章的学习，我们深入理解了向量表示技术的核心原理，包括词嵌入模型、文本向量化方法、向量空间理论以及各种优化技术。这些技术为AI搜索系统的向量检索提供了坚实的理论基础和实践指导。