3.3 多向量表示（Multivector）

3.3 多向量表示（Multivector）— 多维度数据建模与增强检索

本节导读：深入理解Qdrant的多向量表示技术，学习如何为同一对象创建多种向量表示，实现更精准的数据建模和增强检索能力，掌握多向量查询的实际应用场景。

学习目标

• 理解多向量表示的概念和优势
• 掌握Qdrant中多向量配置的方法
• 学习多向量查询的策略和权重分配
• 实践多向量在实际业务中的应用
• 了解多向量表示的性能优化技巧

核心概念

多向量表示（Multivector）是Qdrant的一项重要特性，允许为同一个对象创建多种不同的向量表示，每种向量捕获数据的不同特征。这种技术特别适合复杂场景下的多维度数据建模，能够显著提升检索的准确性和覆盖范围。

![多向量表示示意图：展示同一文档的多种向量表示，包括文本语义向量、关键词向量、主题向量等，每种向量代表不同的数据特征]

多向量基础理论

3.3.1 多向量的定义与优势

什么是多向量表示

多向量表示是指为同一个实体（如文档、图片、用户等）生成多个不同的向量，每个向量代表该实体的不同特征维度：

class DocumentMultivector:
    def __init__(self, document_id):
        self.document_id = document_id
        self.vectors = {
            'semantic': None,    # 语义向量：由BERT生成
            'keyword': None,    # 关键词向量：TF-IDF生成
            'topic': None,      # 主题向量：LDA生成
            'style': None,      # 风格向量：文体特征
            'entity': None      # 实体向量：命名实体识别
        }
        
    def add_vector(self, vector_type, vector):
        """添加特定类型的向量"""
        if vector_type in self.vectors:
            self.vectors[vector_type] = vector
        else:
            raise ValueError(f"不支持的向量类型: {vector_type}")
            
    def get_vector(self, vector_type):
        """获取特定类型的向量"""
        return self.vectors.get(vector_type)
        
    def has_vector(self, vector_type):
        """检查是否包含特定类型的向量"""
        return vector_type in self.vectors and self.vectors[vector_type] is not None
​

多向量的核心优势

1. 特征多样性：单一向量难以捕捉数据的全部特征，多向量可以从不同角度表示数据
2. 检索灵活性：可以根据查询需求选择最合适的向量类型
3. 容错能力：某些向量类型效果不佳时，其他向量可以提供补充
4. 场景适配：不同应用场景最适合不同的向量表示

3.3.2 多向量的应用场景

搜索引擎优化

class SearchEngineMultivector:
    def __init__(self, qdrant_client):
        self.client = qdrant_client
        self.collection_name = "search_engine"
        
    def create_multivector_collection(self):
        """创建支持多向量的集合"""
        # 配置多向量
        vectors_config = {
            "semantic": VectorParams(size=768, distance=Distance.COSINE),
            "keyword": VectorParams(size=512, distance=Distance.COSINE),
            "topic": VectorParams(size=128, distance=Distance.COSINE),
            "style": VectorParams(size=256, distance=Distance.COSINE)
        }
        
        self.client.recreate_collection(
            collection_name=self.collection_name,
            vectors_config=vectors_config
        )
        
    def add_document_with_multivectors(self, doc_id, doc_content, metadata):
        """添加带有多向量的文档"""
        # 生成多种向量
        vectors = self.generate_multivectors(doc_content)
        
        # 构建批量插入点
        points = []
        for vector_type, vector in vectors.items():
            points.append({
                "id": f"{doc_id}_{vector_type}",
                "vector": vector.tolist(),
                "payload": {
                    "doc_id": doc_id,
                    "vector_type": vector_type,
                    "content": doc_content,
                    "metadata": metadata,
                    "created_at": datetime.now().isoformat()
                }
            })
            
        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )
​

Qdrant多向量配置

3.3.3 多向量集合创建

基本配置方法

from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams, CreateCollection

class QdrantMultivectorConfig:
    def __init__(self, client):
        self.client = client
        
    def create_multivector_collection(self, collection_name, vector_configs):
        """
        创建支持多向量的集合
        """
        self.client.recreate_collection(
            collection_name=collection_name,
            vectors_config=vector_configs
        )
        
    def example_configs():
        """示例配置"""
        # 电商搜索场景
        ecommerce_config = {
            "product_image": VectorParams(size=512, distance=Distance.COSINE),
            "product_text": VectorParams(size=768, distance=Distance.COSINE),
            "category": VectorParams(size=128, distance=Distance.COSINE),
            "price_range": VectorParams(size=32, distance=Distance.EUCLIDEAN)
        }
        
        # 文档搜索场景
        document_config = {
            "semantic": VectorParams(size=768, distance=Distance.COSINE),
            "keywords": VectorParams(size=256, distance=Distance.COSINE),
            "topics": VectorParams(size=64, distance=Distance.COSINE),
            "entities": VectorParams(size=128, distance=Distance.COSINE)
        }
        
        return {
            "ecommerce": ecommerce_config,
            "document": document_config
        }
​

3.3.4 多向量数据插入

批量插入策略

class MultivectorDataInserter:
    def __init__(self, client, collection_name):
        self.client = client
        self.collection_name = collection_name
        
    def insert_document_multivectors(self, doc_id, document_data):
        """插入单个文档的多向量数据"""
        points = []
        
        # 为每种向量类型创建点
        for vector_type, vector in document_data.vectors.items():
            point_id = f"{doc_id}_{vector_type}"
            
            points.append({
                "id": point_id,
                "vector": vector.tolist(),
                "payload": {
                    "doc_id": doc_id,
                    "vector_type": vector_type,
                    "title": document_data.title,
                    "content": document_data.content,
                    "metadata": document_data.metadata,
                    "created_at": datetime.now().isoformat()
                }
            })
        
        # 批量插入
        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )
​

多向量查询策略

3.3.5 多向量查询原理

查询方式类型

1. 单向量查询：使用单一向量类型进行查询
2. 多向量融合查询：融合多个向量的查询结果
3. 条件查询：根据条件选择特定的向量类型

多向量融合算法

class MultivectorQueryFusion:
    def __init__(self):
        self.fusion_strategies = {
            'weighted_average': self.weighted_average_fusion,
            'rank_fusion': self.rank_fusion,
            'score_combination': self.score_combination,
            'voting': self.voting_fusion
        }
        
    def weighted_average_fusion(self, query_results, weights):
        """
        加权平均融合策略
        """
        fused_results = {}
        
        # 对每个文档，计算加权平均分数
        for doc_id in set().union(*[set(results.keys()) for results in query_results.values()]):
            weighted_score = 0
            total_weight = 0
            
            for vector_type, results in query_results.items():
                if doc_id in results:
                    score = results[doc_id]['score']
                    weight = weights.get(vector_type, 1.0)
                    weighted_score += score * weight
                    total_weight += weight
            
            if total_weight > 0:
                fused_results[doc_id] = {
                    'score': weighted_score / total_weight,
                    'details': {vt: results.get(doc_id, {}) for vt, results in query_results.items()}
                }
        
        # 排序
        sorted_results = dict(sorted(fused_results.items(), 
                                   key=lambda x: x[1]['score'], 
                                   reverse=True))
        return sorted_results
        
    def rank_fusion(self, query_results, k=10):
        """
        排名融合策略（Rank Fusion）
        """
        fused_scores = {}
        
        # 为每个文档计算排名分数
        for doc_id in set().union(*[set(results.keys()) for results in query_results.values()]):
            rank_score = 0
            
            for vector_type, results in query_results.items():
                if doc_id in results:
                    # 基于排名的分数计算
                    rank = list(results.keys()).index(doc_id) + 1
                    rank_score += 1 / rank
            
            fused_scores[doc_id] = rank_score
        
        # 排序
        sorted_results = dict(sorted(fused_scores.items(), 
                                   key=lambda x: x[1], 
                                   reverse=True))
        
        # 取前k个
        return {doc_id: {'score': score, 'details': {}} 
                for doc_id, score in list(sorted_results.items())[:k]}
​

3.3.6 实际查询实现

多向量查询器

class MultivectorSearchEngine:
    def __init__(self, client, collection_name):
        self.client = client
        self.collection_name = collection_name
        self.fusion_engine = MultivectorQueryFusion()
        
    def search_with_multivectors(self, query_vectors, strategy='weighted_average', weights=None, limit=10):
        """
        使用多向量进行查询
        
        Args:
            query_vectors: 各向量类型的查询向量
            strategy: 融合策略 ('weighted_average', 'rank_fusion', 'score_combination')
            weights: 各向量类型的权重
            limit: 返回结果数量
        """
        # 1. 分别对各向量类型进行查询
        query_results = {}
        
        for vector_type, query_vector in query_vectors.items():
            result = self.search_with_single_vector(query_vector, vector_type, limit * 2)
            query_results[vector_type] = result
            
        # 2. 根据策略融合结果
        if strategy == 'weighted_average':
            if weights is None:
                weights = {vt: 1.0 for vt in query_vectors.keys()}
            fused_results = self.fusion_engine.weighted_average_fusion(query_results, weights)
        elif strategy == 'rank_fusion':
            fused_results = self.fusion_engine.rank_fusion(query_results, limit)
        elif strategy == 'score_combination':
            fused_results = self.fusion_engine.score_combination(query_results)
        else:
            raise ValueError(f"不支持的融合策略: {strategy}")
            
        # 3. 过滤和排序最终结果
        final_results = []
        for doc_id, result_data in list(fused_results.items())[:limit]:
            final_results.append({
                'doc_id': doc_id,
                'score': result_data['score'],
                'details': result_data['details'],
                'payload': self.get_doc_payload(doc_id)
            })
            
        return final_results
​

多向量应用实战

3.3.7 电商商品搜索系统

完整电商搜索实现

class EcommerceMultivectorSearch:
    def __init__(self, client):
        self.client = client
        self.collection_name = "ecommerce_products"
        self.setup_collection()
        
    def setup_collection(self):
        """设置电商商品多向量集合"""
        vector_configs = {
            "image_features": VectorParams(size=512, distance=Distance.COSINE),
            "title_semantic": VectorParams(size=384, distance=Distance.COSINE),
            "description_semantic": VectorParams(size=768, distance=Distance.COSINE),
            "category_vector": VectorParams(size=128, distance=Distance.COSINE),
            "price_features": VectorParams(size=8, distance=Distance.EUCLIDEAN),
            "brand_features": VectorParams(size=64, distance=Distance.COSINE)
        }
        
        self.client.recreate_collection(
            collection_name=self.collection_name,
            vectors_config=vector_configs
        )
        
    def search_products(self, query, search_strategy="balanced"):
        """多维度商品搜索"""
        
        # 根据查询类型生成不同的查询向量
        query_vectors = self.generate_query_vectors(query)
        
        # 设置不同搜索策略的权重
        if search_strategy == "image_first":
            weights = {
                "image_features": 0.4,
                "title_semantic": 0.2,
                "description_semantic": 0.2,
                "category_vector": 0.1,
                "price_features": 0.05,
                "brand_features": 0.05
            }
        elif search_strategy == "text_first":
            weights = {
                "image_features": 0.1,
                "title_semantic": 0.3,
                "description_semantic": 0.3,
                "category_vector": 0.15,
                "price_features": 0.05,
                "brand_features": 0.1
            }
        elif search_strategy == "balanced":
            weights = {
                "image_features": 0.2,
                "title_semantic": 0.2,
                "description_semantic": 0.2,
                "category_vector": 0.15,
                "price_features": 0.1,
                "brand_features": 0.15
            }
        else:
            weights = None
            
        # 执行多向量搜索
        search_engine = MultivectorSearchEngine(
            self.client, 
            self.collection_name
        )
        
        results = search_engine.search_with_multivectors(
            query_vectors,
            strategy="weighted_average",
            weights=weights,
            limit=20
        )
        
        return results
​

最佳实践与优化

3.3.8 多向量配置优化

内存和性能优化

class MultivectorOptimizer:
    def __init__(self, client):
        self.client = client
        self.performance_metrics = {}
        
    def optimize_vector_dimensions(self, dataset_stats):
        """
        根据数据集统计信息优化向量维度
        """
        recommendations = {}
        
        # 基于数据规模推荐向量维度
        for vector_type, stats in dataset_stats.items():
            data_size = stats['size']
            unique_values = stats['unique_values']
            sparsity = stats['sparsity']
            
            if vector_type in ['semantic', 'text']:
                if data_size < 1000:
                    recommendations[vector_type] = 384
                elif data_size < 10000:
                    recommendations[vector_type] = 768
                else:
                    recommendations[vector_type] = 1024
                    
            elif vector_type in ['keywords', 'topics']:
                if sparsity > 0.9:  # 高稀疏度
                    recommendations[vector_type] = min(256, unique_values // 4)
                else:
                    recommendations[vector_type] = min(512, unique_values // 2)
                    
            elif vector_type in ['entities', 'style']:
                recommendations[vector_type] = min(512, unique_values)
                
        return recommendations
​

本节小结

本节详细讲解了Qdrant多向量表示技术，包括：

1. 基础理论：多向量的定义、优势和多种应用场景，理解多向量表示的价值
2. 配置实现：多向量集合的创建方法和数据插入策略，掌握多向量的配置技术
3. 查询策略：多种融合算法的实际实现，包括加权平均、排名融合、分数组合等
4. 实战应用：电商搜索系统的完整实现，学习多向量在实际业务中的应用
5. 优化技巧：性能优化和配置优化，掌握多向量系统的调优方法

通过多向量表示技术，可以显著提升检索系统的准确性和覆盖范围，满足复杂场景下的多样化需求。

延伸阅读

• 多向量检索综述论文
• Qdrant多向量文档
• 多模态检索技术

关键词：Qdrant, 多向量表示, Multivector, 多维度检索, 向量融合, 多模态搜索, 性能优化
难度：高级
预计阅读：50 分钟