向量数据库选择

import pandas as pd
import numpy as np
from typing import Dict, List, Any, Tuple
import matplotlib.pyplot as plt

class VectorDatabaseComparator:
    """向量数据库对比分析器"""
    
    def __init__(self):
        self.databases = {
            "Milvus": {
                "类型": "开源",
                "部署方式": ["自部署", "云服务"],
                "GPU加速": True,
                "分布式": True,
                "多向量支持": True,
                "混合搜索": True,
                "索引类型": ["IVF", "HNSW", "FLAT", "GPU_BRUTE_FORCE"],
                "最大向量维度": 32768,
                "QPS性能": "10,000+",
                "延迟": "1-10ms",
                "数据规模": "10亿+向量",
                "开发语言": ["Python", "Java", "Go", "Node.js"],
                "企业支持": "Zilliz Cloud",
                "学习成本": "中等",
                "适用场景": ["大规模生产", "高性能要求", "复杂查询"],
                "优势": ["最强性能", "GPU加速", "完整生态"],
                "劣势": ["部署复杂", "资源消耗大"]
            },
            
            "Pinecone": {
                "类型": "托管服务",
                "部署方式": ["SaaS"],
                "GPU加速": True,
                "分布式": True,
                "多向量支持": False,
                "混合搜索": True,
                "索引类型": ["优化索引算法"],
                "最大向量维度": 20000,
                "QPS性能": "10,000+",
                "延迟": "小于100ms (p99)",
                "数据规模": "数十亿向量",
                "开发语言": ["Python", "JavaScript", "Java"],
                "企业支持": "完整托管",
                "学习成本": "低",
                "适用场景": ["快速上线", "中小规模", "SaaS应用"],
                "优势": ["零运维", "稳定可靠", "快速部署"],
                "劣势": ["成本较高", "厂商锁定", "定制受限"]
            },
            
            "Weaviate": {
                "类型": "开源",
                "部署方式": ["自部署", "云服务"],
                "GPU加速": False,
                "分布式": True,
                "多向量支持": True,
                "混合搜索": True,
                "索引类型": ["HNSW"],
                "最大向量维度": 65536,
                "QPS性能": "1,000-5,000",
                "延迟": "10-50ms",
                "数据规模": "千万级向量",
                "开发语言": ["Python", "JavaScript", "Go", "Java"],
                "企业支持": "Weaviate Cloud",
                "学习成本": "中等",
                "适用场景": ["知识图谱", "语义搜索", "复杂查询"],
                "优势": ["GraphQL接口", "模块化设计", "语义理解"],
                "劣势": ["性能中等", "GPU支持有限"]
            },
            
            "Chroma": {
                "类型": "开源",
                "部署方式": ["本地", "自部署"],
                "GPU加速": False,
                "分布式": False,
                "多向量支持": False,
                "混合搜索": False,
                "索引类型": ["HNSW", "FLAT"],
                "最大向量维度": 2048,
                "QPS性能": "100-1,000",
                "延迟": "1-20ms",
                "数据规模": "百万级向量",
                "开发语言": ["Python", "JavaScript"],
                "企业支持": "社区",
                "学习成本": "低",
                "适用场景": ["原型开发", "小规模应用", "本地部署"],
                "优势": ["极简设计", "快速上手", "轻量级"],
                "劣势": ["扩展性有限", "功能简单", "无分布式"]
            },
            
            "Qdrant": {
                "类型": "开源", 
                "部署方式": ["自部署", "云服务"],
                "GPU加速": False,
                "分布式": True,
                "多向量支持": True,
                "混合搜索": True,
                "索引类型": ["HNSW"],
                "最大向量维度": 65536,
                "QPS性能": "5,000-10,000",
                "延迟": "5-20ms",
                "数据规模": "亿级向量",
                "开发语言": ["Python", "Rust", "JavaScript"],
                "企业支持": "Qdrant Cloud",
                "学习成本": "低",
                "适用场景": ["中等规模", "实时应用", "边缘部署"],
                "优势": ["资源占用小", "Rust性能", "API友好"],
                "劣势": ["生态较新", "社区相对小"]
            }
        }
        
        self.performance_benchmarks = self.load_performance_data()
    
    def load_performance_data(self) -> Dict[str, Dict[str, float]]:
        """加载性能基准数据"""
        return {
            "Milvus": {
                "插入性能_万条/秒": 50.0,
                "查询性能_QPS": 12000,
                "召回率_95": 0.95,
                "索引大小_GB": 1.5,
                "内存使用_GB": 8.0,
                "CPU使用率": 0.6
            },
            
            "Pinecone": {
                "插入性能_万条/秒": 30.0,
                "查询性能_QPS": 8000,
                "召回率_95": 0.94,
                "索引大小_GB": 1.2,
                "内存使用_GB": 6.0,
                "CPU使用率": 0.4
            },
            
            "Weaviate": {
                "插入性能_万条/秒": 20.0,
                "查询性能_QPS": 4000,
                "召回率_95": 0.92,
                "索引大小_GB": 0.8,  # 最小索引
                "内存使用_GB": 4.0,
                "CPU使用率": 0.5
            },
            
            "Chroma": {
                "插入性能_万条/秒": 10.0,
                "查询性能_QPS": 1000,
                "召回率_95": 0.90,
                "索引大小_GB": 1.0,
                "内存使用_GB": 2.0,
                "CPU使用率": 0.3
            },
            
            "Qdrant": {
                "插入性能_万条/秒": 25.0,
                "查询性能_QPS": 6000,
                "召回率_95": 0.93,
                "索引大小_GB": 1.1,
                "内存使用_GB": 3.0,
                "CPU使用率": 0.35
            }
        }
    
    def get_comparison_matrix(self) -> pd.DataFrame:
        """获取对比矩阵"""
        comparison_data = []
        
        for db_name, specs in self.databases.items():
            row = {
                "数据库": db_name,
                "类型": specs["类型"],
                "GPU加速": "✅" if specs["GPU加速"] else "❌",
                "分布式": "✅" if specs["分布式"] else "❌",
                "混合搜索": "✅" if specs["混合搜索"] else "❌",
                "QPS性能": specs["QPS性能"],
                "延迟": specs["延迟"],
                "数据规模": specs["数据规模"],
                "学习成本": specs["学习成本"],
                "适用场景": "、".join(specs["适用场景"][:2])  # 显示前2个场景
            }
            comparison_data.append(row)
        
        return pd.DataFrame(comparison_data)
    
    def recommend_database(self, requirements: Dict[str, Any]) -> Dict[str, Any]:
        """推荐数据库"""
        data_scale = requirements.get("data_scale", "medium")  # small, medium, large
        performance_priority = requirements.get("performance_priority", "medium")  # low, medium, high
        deployment_preference = requirements.get("deployment", "self_hosted")  # self_hosted, managed
        budget_level = requirements.get("budget", "medium")  # low, medium, high
        team_expertise = requirements.get("team_expertise", "medium")  # low, medium, high
        
        recommendations = []
        
        # 基于需求评分各个数据库
        for db_name, specs in self.databases.items():
            score = 0
            reasoning = []
            
            # 数据规模评分
            if data_scale == "large":
                if specs["数据规模"] == "10亿+向量":
                    score += 30
                    reasoning.append("支持大规模数据")
                elif specs["数据规模"] == "数十亿向量":
                    score += 25
                    reasoning.append("支持超大规模数据")
            elif data_scale == "medium":
                if "千万级" in specs["数据规模"] or "亿级" in specs["数据规模"]:
                    score += 25
                    reasoning.append("适合中等规模数据")
            else:  # small
                score += 20  # 所有数据库都支持小规模
                reasoning.append("支持小规模数据")
            
            # 性能要求评分
            if performance_priority == "high":
                if "10,000+" in specs["QPS性能"]:
                    score += 25
                    reasoning.append("高性能表现")
            elif performance_priority == "medium":
                if any(perf in specs["QPS性能"] for perf in ["1,000-5,000", "5,000-10,000"]):
                    score += 20
                    reasoning.append("中等性能表现")
            
            # 部署偏好评分
            if deployment_preference == "managed" and "SaaS" in specs["部署方式"]:
                score += 20
                reasoning.append("提供托管服务")
            elif deployment_preference == "self_hosted" and "自部署" in specs["部署方式"]:
                score += 20
                reasoning.append("支持自部署")
            
            # 学习成本评分
            if team_expertise == "low" and specs["学习成本"] == "低":
                score += 15
                reasoning.append("学习成本低")
            elif team_expertise == "high":
                score += 10  # 高技能团队适应任何工具
            
            recommendations.append({
                "database": db_name,
                "score": score,
                "reasoning": reasoning,
                "specs": specs
            })
        
        # 按评分排序
        recommendations.sort(key=lambda x: x["score"], reverse=True)
        
        return {
            "requirements": requirements,
            "top_recommendations": recommendations[:3],
            "detailed_analysis": recommendations
        }
    
    def performance_benchmark_comparison(self) -> Dict[str, Any]:
        """性能基准对比"""
        # 创建性能对比图表
        metrics = list(self.performance_benchmarks[list(self.databases.keys())[0]].keys())
        databases = list(self.databases.keys())
        
        benchmark_df = pd.DataFrame(self.performance_benchmarks).T
        
        # 标准化分数（0-1）
        normalized_scores = {}
        for metric in metrics:
            max_val = benchmark_df[metric].max()
            min_val = benchmark_df[metric].min()
            
            for db in databases:
                if db not in normalized_scores:
                    normalized_scores[db] = {}
                
                if max_val != min_val:
                    normalized_scores[db][metric] = (
                        benchmark_df.loc[db, metric] - min_val
                    ) / (max_val - min_val)
                else:
                    normalized_scores[db][metric] = 1.0
        
        return {
            "raw_performance": self.performance_benchmarks,
            "normalized_scores": normalized_scores,
            "performance_df": benchmark_df,
            "winner_by_metric": self.identify_winners_by_metric(),
            "overall_ranking": self.calculate_overall_ranking(normalized_scores)
        }
    
    def identify_winners_by_metric(self) -> Dict[str, str]:
        """识别各指标获胜者"""
        winners = {}
        
        for metric in self.performance_benchmarks[list(self.databases.keys())[0]].keys():
            best_db = max(
                self.databases.keys(),
                key=lambda db: self.performance_benchmarks[db][metric]
            )
            winners[metric] = best_db
        
        return winners
    
    def calculate_overall_ranking(self, normalized_scores: Dict[str, Dict[str, float]]) -> List[Tuple[str, float]]:
        """计算综合排名"""
        # 权重设置
        weights = {
            "插入性能_万条/秒": 0.2,
            "查询性能_QPS": 0.3,
            "召回率_95": 0.25,
            "索引大小_GB": 0.1,  # 越小越好，需要反转
            "内存使用_GB": 0.1,   # 越小越好，需要反转  
            "CPU使用率": 0.05     # 越小越好，需要反转
        }
        
        overall_scores = {}
        
        for db in self.databases.keys():
            score = 0
            for metric, weight in weights.items():
                metric_score = normalized_scores[db][metric]
                
                # 对于"越小越好"的指标，反转分数
                if metric in ["索引大小_GB", "内存使用_GB", "CPU使用率"]:
                    metric_score = 1 - metric_score
                
                score += metric_score * weight
            
            overall_scores[db] = score
        
        # 排序
        ranking = sorted(overall_scores.items(), key=lambda x: x[1], reverse=True)
        
        return ranking

# 使用示例
comparator = VectorDatabaseComparator()

# 获取对比矩阵
comparison_df = comparator.get_comparison_matrix()
print("向量数据库对比矩阵:")
print(comparison_df)

# 性能基准对比
performance_analysis = comparator.performance_benchmark_comparison()
print(f"\n各指标获胜者: {performance_analysis['winner_by_metric']}")
print(f"综合排名: {performance_analysis['overall_ranking']}")

# 根据需求推荐
requirements = {
    "data_scale": "large",
    "performance_priority": "high", 
    "deployment": "self_hosted",
    "budget": "high",
    "team_expertise": "high"
}

recommendation = comparator.recommend_database(requirements)
print(f"\n基于需求的推荐:")
for i, rec in enumerate(recommendation["top_recommendations"], 1):
    print(f"{i}. {rec['database']} (评分: {rec['score']}) - {', '.join(rec['reasoning'])}")

class RAGVectorDBSelector:
    """RAG应用向量数据库选择器"""
    
    def __init__(self):
        self.rag_scenarios = {
            "企业知识库": {
                "数据特点": "文档多样、更新频繁、权限控制",
                "性能要求": "中高",
                "推荐方案": ["Milvus", "Weaviate", "Pinecone"],
                "关键考虑": ["混合搜索", "元数据过滤", "安全性"]
            },
            
            "客服机器人": {
                "数据特点": "FAQ结构化、实时查询、高并发",
                "性能要求": "高",
                "推荐方案": ["Pinecone", "Milvus", "Qdrant"],
                "关键考虑": ["低延迟", "高可用", "成本控制"]
            },
            
            "代码搜索": {
                "数据特点": "代码片段、语法结构、版本管理",
                "性能要求": "中",
                "推荐方案": ["Weaviate", "Milvus", "Chroma"],
                "关键考虑": ["语义理解", "代码结构", "版本控制"]
            },
            
            "多媒体内容": {
                "数据特点": "图像文本混合、多模态、大文件",
                "性能要求": "高",
                "推荐方案": ["Milvus", "Weaviate"],
                "关键考虑": ["多向量支持", "GPU加速", "存储优化"]
            },
            
            "实时推荐": {
                "数据特点": "用户行为、实时更新、个性化",
                "性能要求": "极高",
                "推荐方案": ["Milvus", "Qdrant", "Pinecone"],
                "关键考虑": ["实时性", "扩展性", "个性化算法"]
            },
            
            "研究原型": {
                "数据特点": "实验数据、快速迭代、本地开发",
                "性能要求": "低",
                "推荐方案": ["Chroma", "Qdrant", "本地Milvus"],
                "关键考虑": ["易用性", "快速部署", "成本低"]
            }
        }
    
    def select_for_rag_scenario(self, scenario: str, additional_requirements: Dict = None) -> Dict[str, Any]:
        """为RAG场景选择向量数据库"""
        
        if scenario not in self.rag_scenarios:
            return {"error": f"不支持的场景: {scenario}"}
        
        scenario_info = self.rag_scenarios[scenario]
        
        # 基础推荐
        base_recommendations = scenario_info["推荐方案"]
        
        # 根据额外需求调整
        if additional_requirements:
            adjusted_recommendations = self.adjust_recommendations(
                base_recommendations, 
                additional_requirements
            )
        else:
            adjusted_recommendations = base_recommendations
        
        # 生成详细分析
        detailed_analysis = {}
        for db_name in adjusted_recommendations:
            if db_name in comparator.databases:
                db_specs = comparator.databases[db_name]
                
                # 计算场景匹配度
                match_score = self.calculate_scenario_match(db_specs, scenario_info)
                
                detailed_analysis[db_name] = {
                    "match_score": match_score,
                    "pros": db_specs["优势"],
                    "cons": db_specs["劣势"],
                    "deployment_complexity": self.assess_deployment_complexity(db_specs),
                    "estimated_cost": self.estimate_monthly_cost(db_specs, scenario)
                }
        
        return {
            "scenario": scenario,
            "scenario_characteristics": scenario_info,
            "recommended_databases": adjusted_recommendations,
            "detailed_analysis": detailed_analysis,
            "implementation_guide": self.generate_implementation_guide(adjusted_recommendations[0], scenario)
        }
    
    def calculate_scenario_match(self, db_specs: Dict, scenario_info: Dict) -> float:
        """计算场景匹配度"""
        score = 0.0
        
        # 性能要求匹配
        perf_requirement = scenario_info["性能要求"]
        if perf_requirement == "高" and "10,000+" in db_specs["QPS性能"]:
            score += 0.3
        elif perf_requirement == "中" and any(p in db_specs["QPS性能"] for p in ["1,000", "5,000"]):
            score += 0.3
        elif perf_requirement == "低":
            score += 0.3  # 所有数据库都满足
        
        # 关键考虑匹配
        key_considerations = scenario_info["关键考虑"]
        for consideration in key_considerations:
            if consideration == "混合搜索" and db_specs["混合搜索"]:
                score += 0.2
            elif consideration == "GPU加速" and db_specs["GPU加速"]:
                score += 0.2
            elif consideration == "低延迟" and "ms" in db_specs["延迟"]:
                latency_val = int(db_specs["延迟"].split("-")[0])
                if latency_val <= 10:
                    score += 0.2
        
        # 学习成本考虑
        if db_specs["学习成本"] == "低":
            score += 0.1
        
        return min(score, 1.0)
    
    def assess_deployment_complexity(self, db_specs: Dict) -> str:
        """评估部署复杂度"""
        if "SaaS" in db_specs["部署方式"]:
            return "简单（托管服务）"
        elif db_specs["学习成本"] == "低":
            return "中等（易于配置）"
        elif db_specs["GPU加速"]:
            return "复杂（需要GPU环境）"
        else:
            return "中等"
    
    def estimate_monthly_cost(self, db_specs: Dict, scenario: str) -> str:
        """估算月度成本"""
        # 简化的成本估算
        if "SaaS" in db_specs["部署方式"]:
            if scenario in ["企业知识库", "实时推荐"]:
                return "$500-2000/月"
            else:
                return "$100-500/月"
        else:
            if db_specs["GPU加速"]:
                return "$300-1000/月（自部署+GPU）"
            else:
                return "$100-400/月（自部署）"
    
    def generate_implementation_guide(self, recommended_db: str, scenario: str) -> Dict[str, Any]:
        """生成实施指南"""
        
        if recommended_db == "Milvus":
            return {
                "步骤1": "安装Milvus集群（推荐Docker Compose或Kubernetes）",
                "步骤2": "配置GPU加速（如需要）和存储后端",
                "步骤3": "创建Collection并定义索引策略（HNSW推荐）",
                "步骤4": "集成Python/Java客户端并实现批量导入",
                "步骤5": "优化查询参数和缓存策略",
                "参考代码": "使用pymilvus库进行连接和操作",
                "注意事项": ["GPU内存管理", "索引构建时间", "查询并发控制"]
            }
        
        elif recommended_db == "Pinecone":
            return {
                "步骤1": "注册Pinecone账户并创建项目",
                "步骤2": "创建Index并选择合适的向量维度",
                "步骤3": "配置API密钥和客户端连接",
                "步骤4": "实现批量向量上传和元数据管理",
                "步骤5": "集成查询接口和结果处理",
                "参考代码": "使用pinecone-client库",
                "注意事项": ["API配额管理", "数据传输优化", "成本监控"]
            }
        
        else:
            return {
                "步骤1": f"部署{recommended_db}服务",
                "步骤2": "配置数据库连接和认证",
                "步骤3": "设计向量存储Schema",
                "步骤4": "实现数据导入和索引构建",
                "步骤5": "优化查询性能和监控",
                "参考代码": f"使用{recommended_db.lower()}官方客户端",
                "注意事项": ["性能调优", "数据备份", "版本升级"]
            }

# RAG场景选择示例
rag_selector = RAGVectorDBSelector()

# 企业知识库场景
kb_recommendation = rag_selector.select_for_rag_scenario(
    "企业知识库",
    additional_requirements={
        "security_priority": "high",
        "update_frequency": "daily",
        "user_count": 1000
    }
)

print("企业知识库场景推荐:")
print(f"推荐数据库: {kb_recommendation['recommended_databases']}")

for db_name, analysis in kb_recommendation['detailed_analysis'].items():
    print(f"\n{db_name}:")
    print(f"  匹配度: {analysis['match_score']:.2f}")
    print(f"  部署复杂度: {analysis['deployment_complexity']}")
    print(f"  预估成本: {analysis['estimated_cost']}")
    print(f"  优势: {', '.join(analysis['pros'])}")

# 客服机器人场景
chatbot_recommendation = rag_selector.select_for_rag_scenario(
    "客服机器人",
    additional_requirements={
        "latency_requirement": "low",
        "concurrent_users": 5000,
        "availability_requirement": "99.9%"
    }
)

print(f"\n客服机器人场景推荐: {chatbot_recommendation['recommended_databases'][0]}")

# Milvus生产环境配置
import pymilvus
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

class MilvusProductionSetup:
    """Milvus生产环境配置"""
    
    def __init__(self, host: str, port: int = 19530):
        self.host = host
        self.port = port
        self.connection_alias = "default"
        self.setup_connection()
    
    def setup_connection(self):
        """建立连接"""
        connections.connect(
            alias=self.connection_alias,
            host=self.host,
            port=self.port,
            timeout=60
        )
        
        print(f"已连接到Milvus: {self.host}:{self.port}")
    
    def create_optimized_collection(self, 
                                  collection_name: str,
                                  vector_dim: int = 1536,
                                  index_type: str = "HNSW",
                                  metric_type: str = "IP") -> Collection:
        """创建优化的集合"""
        
        # 定义字段schema
        fields = [
            FieldSchema(
                name="id",
                dtype=DataType.INT64,
                is_primary=True,
                auto_id=True,
                description="主键ID"
            ),
            FieldSchema(
                name="vector",
                dtype=DataType.FLOAT_VECTOR,
                dim=vector_dim,
                description="向量数据"
            ),
            FieldSchema(
                name="text",
                dtype=DataType.VARCHAR,
                max_length=8192,
                description="原始文本"
            ),
            FieldSchema(
                name="metadata",
                dtype=DataType.JSON,
                description="元数据信息"
            ),
            FieldSchema(
                name="timestamp",
                dtype=DataType.INT64,
                description="时间戳"
            )
        ]
        
        # 创建集合schema
        schema = CollectionSchema(
            fields=fields,
            description=f"RAG应用集合: {collection_name}",
            enable_dynamic_field=True
        )
        
        # 创建集合
        collection = Collection(
            name=collection_name,
            schema=schema,
            using=self.connection_alias
        )
        
        print(f"集合 {collection_name} 创建成功")
        
        # 创建索引
        self.create_optimized_index(collection, index_type, metric_type)
        
        return collection
    
    def create_optimized_index(self, 
                             collection: Collection,
                             index_type: str = "HNSW",
                             metric_type: str = "IP"):
        """创建优化索引"""
        
        # 不同索引类型的参数
        index_params = {
            "HNSW": {
                "metric_type": metric_type,
                "index_type": "HNSW",
                "params": {
                    "M": 16,          # 连接数
                    "efConstruction": 500  # 构建参数
                }
            },
            "IVF_FLAT": {
                "metric_type": metric_type,
                "index_type": "IVF_FLAT",
                "params": {
                    "nlist": 1024    # 聚类数量
                }
            },
            "GPU_IVF_FLAT": {
                "metric_type": metric_type,
                "index_type": "GPU_IVF_FLAT",
                "params": {
                    "nlist": 1024
                }
            }
        }
        
        if index_type not in index_params:
            index_type = "HNSW"  # 默认使用HNSW
        
        # 创建向量索引
        collection.create_index(
            field_name="vector",
            index_params=index_params[index_type]
        )
        
        # 创建标量索引（用于过滤）
        collection.create_index(
            field_name="timestamp",
            index_params={"index_type": "STL_SORT"}
        )
        
        print(f"索引创建完成: {index_type}")
    
    def batch_insert_optimized(self, 
                             collection: Collection,
                             vectors: List[List[float]],
                             texts: List[str],
                             metadata_list: List[Dict],
                             batch_size: int = 1000) -> Dict[str, Any]:
        """优化的批量插入"""
        
        total_vectors = len(vectors)
        inserted_count = 0
        
        for i in range(0, total_vectors, batch_size):
            batch_end = min(i + batch_size, total_vectors)
            
            batch_data = {
                "vector": vectors[i:batch_end],
                "text": texts[i:batch_end], 
                "metadata": metadata_list[i:batch_end],
                "timestamp": [int(time.time())] * (batch_end - i)
            }
            
            try:
                insert_result = collection.insert(batch_data)
                inserted_count += len(insert_result.primary_keys)
                
                print(f"批次 {i//batch_size + 1}: 插入 {batch_end - i} 条记录")
                
            except Exception as e:
                print(f"批次 {i//batch_size + 1} 插入失败: {e}")
        
        # 刷新集合
        collection.flush()
        
        return {
            "total_attempted": total_vectors,
            "successfully_inserted": inserted_count,
            "batch_size": batch_size,
            "collection_size": collection.num_entities
        }
    
    def optimized_search(self, 
                        collection: Collection,
                        query_vectors: List[List[float]],
                        top_k: int = 10,
                        search_params: Dict = None) -> List[Dict[str, Any]]:
        """优化的搜索"""
        
        # 加载集合到内存
        collection.load()
        
        # 默认搜索参数
        if search_params is None:
            search_params = {
                "metric_type": "IP",
                "params": {
                    "ef": 128,        # HNSW搜索参数
                    "search_k": -1    # 自动选择
                },
                "offset": 0
            }
        
        # 执行搜索
        search_results = collection.search(
            data=query_vectors,
            anns_field="vector",
            param=search_params,
            limit=top_k,
            output_fields=["text", "metadata", "timestamp"]
        )
        
        # 格式化结果
        formatted_results = []
        for i, hits in enumerate(search_results):
            query_result = {
                "query_index": i,
                "results": []
            }
            
            for hit in hits:
                query_result["results"].append({
                    "id": hit.id,
                    "score": hit.score,
                    "text": hit.entity.get("text", ""),
                    "metadata": hit.entity.get("metadata", {}),
                    "timestamp": hit.entity.get("timestamp", 0)
                })
            
            formatted_results.append(query_result)
        
        return formatted_results

# 生产环境配置示例
milvus_setup = MilvusProductionSetup("milvus-cluster.company.com", 19530)

# 创建生产集合
collection = milvus_setup.create_optimized_collection(
    collection_name="enterprise_knowledge_base",
    vector_dim=1536,
    index_type="HNSW",
    metric_type="IP"
)

# 批量插入示例数据
sample_vectors = [[0.1] * 1536 for _ in range(10000)]  # 1万个示例向量
sample_texts = [f"文档内容 {i}" for i in range(10000)]
sample_metadata = [{"doc_id": i, "category": "tech"} for i in range(10000)]

insert_result = milvus_setup.batch_insert_optimized(
    collection, sample_vectors, sample_texts, sample_metadata
)

print(f"插入结果: {insert_result}")

# 优化搜索
query_vector = [[0.1] * 1536]  # 查询向量
search_results = milvus_setup.optimized_search(
    collection, query_vector, top_k=5
)

print(f"搜索结果: {search_results[0]['results'][:2]}")  # 显示前2个结果

class VectorDBCostAnalyzer:
    """向量数据库成本分析器"""
    
    def __init__(self):
        self.cost_models = {
            "Milvus": {
                "部署成本": {"自部署": 0, "Zilliz Cloud": 200},
                "运营成本_月": {"小规模": 300, "中规模": 1000, "大规模": 3000},
                "人力成本_月": {"运维": 8000, "开发": 5000},
                "硬件需求": {"CPU": "8核", "内存": "32GB", "存储": "1TB SSD", "GPU": "可选"}
            },
            
            "Pinecone": {
                "部署成本": {"托管": 0},
                "运营成本_月": {"小规模": 200, "中规模": 800, "大规模": 2500},
                "人力成本_月": {"运维": 2000, "开发": 3000},
                "硬件需求": {"无": "托管服务"}
            },
            
            "Weaviate": {
                "部署成本": {"自部署": 0, "Weaviate Cloud": 150},
                "运营成本_月": {"小规模": 250, "中规模": 800, "大规模": 2200},
                "人力成本_月": {"运维": 6000, "开发": 4000},
                "硬件需求": {"CPU": "6核", "内存": "16GB", "存储": "500GB SSD"}
            },
            
            "Chroma": {
                "部署成本": {"自部署": 0},
                "运营成本_月": {"小规模": 100, "中规模": 300, "大规模": "不适用"},
                "人力成本_月": {"运维": 3000, "开发": 2000},
                "硬件需求": {"CPU": "4核", "内存": "8GB", "存储": "200GB SSD"}
            },
            
            "Qdrant": {
                "部署成本": {"自部署": 0, "Qdrant Cloud": 100},
                "运营成本_月": {"小规模": 150, "中规模": 500, "大规模": 1500},
                "人力成本_月": {"运维": 4000, "开发": 3000},
                "硬件需求": {"CPU": "4核", "内存": "16GB", "存储": "500GB SSD"}
            }
        }
    
    def calculate_3_year_tco(self, 
                           database: str, 
                           scale: str = "中规模",
                           deployment_type: str = "自部署") -> Dict[str, Any]:
        """计算3年总拥有成本"""
        
        if database not in self.cost_models:
            return {"error": f"数据库 {database} 不在成本模型中"}
        
        cost_model = self.cost_models[database]
        
        # 初始部署成本
        deploy_cost = cost_model["部署成本"].get(deployment_type, 0)
        
        # 月度运营成本
        monthly_ops = cost_model["运营成本_月"].get(scale, 0)
        
        # 人力成本（假设需要0.5个运维FTE + 0.3个开发FTE）
        monthly_human_cost = (
            cost_model["人力成本_月"]["运维"] * 0.5 + 
            cost_model["人力成本_月"]["开发"] * 0.3
        )
        
        # 3年总成本
        total_operational_cost = (monthly_ops + monthly_human_cost) * 36
        total_cost = deploy_cost + total_operational_cost
        
        return {
            "database": database,
            "scale": scale,
            "deployment_type": deployment_type,
            "initial_deployment_cost": deploy_cost,
            "monthly_operational_cost": monthly_ops,
            "monthly_human_cost": monthly_human_cost,
            "total_monthly_cost": monthly_ops + monthly_human_cost,
            "three_year_total_cost": total_cost,
            "cost_breakdown": {
                "部署": f"${deploy_cost:,}",
                "运营": f"${monthly_ops * 36:,}",
                "人力": f"${monthly_human_cost * 36:,}",
                "总计": f"${total_cost:,}"
            }
        }
    
    def compare_tco_all_databases(self, scale: str = "中规模") -> pd.DataFrame:
        """比较所有数据库的TCO"""
        
        tco_data = []
        
        for db_name in self.cost_models.keys():
            # 选择最经济的部署方式
            deployment_options = list(self.cost_models[db_name]["部署成本"].keys())
            
            best_tco = None
            best_deployment = None
            
            for deployment in deployment_options:
                tco = self.calculate_3_year_tco(db_name, scale, deployment)
                
                if "error" not in tco:
                    if best_tco is None or tco["three_year_total_cost"] < best_tco["three_year_total_cost"]:
                        best_tco = tco
                        best_deployment = deployment
            
            if best_tco:
                tco_data.append({
                    "数据库": db_name,
                    "部署方式": best_deployment,
                    "月度成本": f"${best_tco['total_monthly_cost']:,.0f}",
                    "3年总成本": f"${best_tco['three_year_total_cost']:,.0f}",
                    "运营占比": f"{best_tco['monthly_operational_cost']/best_tco['total_monthly_cost']:.1%}",
                    "人力占比": f"{best_tco['monthly_human_cost']/best_tco['total_monthly_cost']:.1%}"
                })
        
        df = pd.DataFrame(tco_data)
        return df.sort_values("3年总成本")

# 成本分析示例
cost_analyzer = VectorDBCostAnalyzer()

# 计算中规模部署的TCO对比
tco_comparison = cost_analyzer.compare_tco_all_databases("中规模")
print("3年TCO对比（中规模）:")
print(tco_comparison)

# 详细分析特定数据库
milvus_tco = cost_analyzer.calculate_3_year_tco("Milvus", "大规模", "自部署")
print(f"\nMilvus大规模自部署TCO分析:")
for key, value in milvus_tco["cost_breakdown"].items():
    print(f"  {key}: {value}")