向量数据库与嵌入模型：从原理到实践

网络小白不怕黑

979人浏览 · 2025-12-12 12:30:00

网络小白不怕黑 · 2025-12-12 12:30:00 发布

引言

在人工智能飞速发展的今天，如何让机器理解人类语言并进行智能化检索成为了关键技术挑战。传统的基于关键词的搜索在面对复杂语义时显得力不从心，而向量数据库与嵌入模型的结合为这一问题提供了革命性的解决方案。本文将深入探讨从文本向量化到向量数据库应用的全链路技术，助你掌握这一前沿技术栈。

一、嵌入模型：将世界转化为数学的语言

1. 向量表征：万物皆可数学化

1.1 向量表征的本质

核心思想：将现实世界中的复杂信息（文本、图像、声音等）转化为高维数学向量，使AI系统能够以统一的数学语言理解和处理多元信息。

技术定义：

Embedding（嵌入）：将离散数据映射到连续向量空间的数学表示
向量空间：由无数个高维点构成的数学空间，每个点代表一个实体的语义位置
语义相似性：相似的含义在向量空间中距离更近

1.2 向量数学基础

向量的数学表示：

# 二维向量：从原点(0,0)到点(x,y)的有向线段
vector_2d = np.array([3, 4])

# 三维向量
vector_3d = np.array([1, 2, 3])

# 高维向量（如768维，常见于BERT模型）
vector_high_dim = np.random.randn(768)  # 正态分布的768维向量

# 向量的基本属性
magnitude = np.linalg.norm(vector_2d)  # 模长：√(3² + 4²) = 5
direction = vector_2d / magnitude      # 方向单位向量：[0.6, 0.8]

维度的重要性：

维度级别	典型场景	特点
50-300维	Word2Vec/GloVe	捕获词语的基本语义
768维	BERT-base	标准的预训练模型维度
1536维	OpenAI text-embedding-v1/v2	更丰富的特征表示
3072维	OpenAI text-embedding-3-large	最详细的语义编码

1.3 文本向量化实战

import numpy as np
from typing import List

def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """计算余弦相似度"""
    dot_product = np.dot(vec_a, vec_b)
    norm_a = np.linalg.norm(vec_a)
    norm_b = np.linalg.norm(vec_b)
    return dot_product / (norm_a * norm_b)

def euclidean_distance(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """计算欧氏距离"""
    return np.linalg.norm(vec_a - vec_b)

def dot_product(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """计算点积"""
    return np.sum(vec_a * vec_b)

# 示例：相似度计算对比
text_vectors = {
    "这个多少钱": np.array([0.95, 0.31, -0.22]),
    "这个什么价格": np.array([0.92, 0.26, -0.14]),
    "我想买这个": np.array([0.88, 0.47, 0.09]),
    "天气真好": np.array([-0.52, 0.74, 0.41])
}

query = "报价是多少"
query_vector = np.array([0.93, 0.23, -0.18])

print(" 相似度计算结果:")
print("-" * 50)
for text, vector in text_vectors.items():
    cos_sim = cosine_similarity(query_vector, vector)
    euc_dist = euclidean_distance(query_vector, vector)
    print(f"{text:15} | 余弦相似度: {cos_sim:.3f} | 欧氏距离: {euc_dist:.3f}")

三种相似度度量方法对比：

度量方法	公式	范围	特点	适用场景
余弦相似度	cos(θ) = A·B/(\|A\|\|B\|)	[-1, 1]	只关心方向，忽略大小	文本检索、文档相似度
欧氏距离	d = √Σ(Aᵢ-Bᵢ)²	[0, +∞)	空间中的直线距离	图像检索、聚类分析
点积	A·B = ΣAᵢBᵢ	(-∞, +∞)	包含方向和大小信息	推荐系统、排序任务

2. 嵌入模型技术与实践

2.1 嵌入模型的训练原理

训练代码示例：

import torch
import torch.nn.functional as F
from transformers import AutoModel, AutoTokenizer

class SiameseNetwork(torch.nn.Module):
    """孪生网络用于文本相似度学习"""
    def __init__(self, model_name="bert-base-chinese"):
        super().__init__()
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.pooling = torch.nn.AdaptiveAvgPool1d(1)
        
    def forward(self, texts):
        inputs = self.tokenizer(texts, padding=True, truncation=True, 
                               return_tensors="pt", max_length=512)
        with torch.no_grad():
            outputs = self.model(**inputs)
            embeddings = outputs.last_hidden_state.mean(dim=1)  # 均值池化
        return embeddings
    
    def compute_similarity(self, text1, text2):
        emb1 = self.forward([text1])
        emb2 = self.forward([text2])
        return F.cosine_similarity(emb1, emb2).item()

# 使用示例
model = SiameseNetwork()
similarity = model.compute_similarity(
    "如何使用Python连接数据库",
    "Python数据库连接教程"
)
print(f"语义相似度: {similarity:.4f}")

2.2 主流嵌入模型选型指南

主流模型性能对比：

模型	维度	语言	MTEB评分	上下文长度	推荐场景
BGE-M3	1024	多语言	69.5	8K	企业级知识库
NV-Embed-v2	768	多语言	62.7	8K	高精度检索
BGE-large-zh	1024	中文	71.3	512	中文专业文档
text-embedding-3-large	3072	多语言	64.6	8K	OpenAI生态
nomic-embed-text	768	多语言	61.6	8192	轻量化部署

2.3 阿里云百炼平台实践

import os
import dashscope
from dashscope import TextEmbedding

# 配置环境变量
os.environ['DASHSCOPE_API_KEY'] = 'sk-your-api-key-here'
os.environ['DASHSCOPE_BASE_URL'] = 'https://dashscope.aliyuncs.com/compatible-mode/v1'

def get_embeddings_bailian(texts, model="text-embedding-v3", dimensions=1024):
    """
    使用阿里云百炼平台获取文本向量
    Args:
        texts: 文本列表
        model: 模型版本 v1/v2/v3
        dimensions: 向量维度，仅v3支持
    Returns:
        向量列表
    """
    if model in ["text-embedding-v1", "text-embedding-v2"]:
        dimensions = None
    
    try:
        if dimensions:
            resp = TextEmbedding.call(
                model=model,
                input=texts,
                text_type="document",
                parameters={"dimensions": dimensions}
            )
        else:
            resp = TextEmbedding.call(
                model=model,
                input=texts,
                text_type="document"
            )
        
        if resp.status_code == 200:
            embeddings = [output['embedding'] for output in resp.output['embeddings']]
            return embeddings
        else:
            print(f"API调用失败: {resp.code} - {resp.message}")
            return None
            
    except Exception as e:
        print(f"获取向量时出错: {str(e)}")
        return None

# 使用示例
texts = [
    "聚客AI-用科技力量，构建智能未来！",
    "向量数据库是实现语义检索的核心技术",
    "人工智能正在深刻改变各行各业"
]

embeddings = get_embeddings_bailian(texts, model="text-embedding-v3", dimensions=1024)
if embeddings:
    print(f"向量维度: {len(embeddings[0])}")
    print(f"第一个向量的前10个值: {embeddings[0][:10]}")

二、向量数据库：海量向量的智能管家

1. 向量数据库核心技术

1.1 与传统数据库的本质区别

1.2 主流向量数据库对比

数据库	类型	核心特性	适用场景
Chroma	轻量级	易于集成，Python原生	原型开发，小型应用
Milvus	企业级	分布式架构，功能完整	大规模生产环境
Pinecone	云服务	全托管，无需运维	初创公司快速上线
Qdrant	开源	Rust编写，性能优异	高性能要求场景
Weaviate	开源	支持GraphQL，自带ML	复杂关系查询
PGVector	扩展	Postgres插件，兼容SQL	已有PG生态

2. ChromaDB深度实践

2.1 安装与初始化

# 安装ChromaDB
pip install chromadb
# 可选：安装额外的嵌入模型支持
pip install sentence-transformers

# 验证安装
python -c "import chromadb; print(f'Chroma版本: {chromadb.__version__}')"

2.2 核心概念与操作

import chromadb
from chromadb.config import Settings
import numpy as np

class ChromaVectorStore:
    """ChromaDB向量存储管理类"""
    
    def __init__(self, persist_directory="./chroma_db"):
        """
        初始化Chroma客户端
        Args:
            persist_directory: 持久化存储目录
        """
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(
                anonymized_telemetry=False,  # 禁用匿名遥测
                allow_reset=True
            )
        )
        print(f" ChromaDB客户端已初始化，数据将保存到: {persist_directory}")
    
    def create_collection(self, collection_name, embedding_model="all-MiniLM-L6-v2"):
        """
        创建向量集合
        Args:
            collection_name: 集合名称
            embedding_model: 嵌入模型名称
        """
        try:
            # 配置集合参数
            collection = self.client.create_collection(
                name=collection_name,
                metadata={"description": f"{collection_name} 知识库"},
                embedding_function=self._get_embedding_function(embedding_model)
            )
            print(f" 集合 '{collection_name}' 创建成功")
            return collection
        except Exception as e:
            print(f" 创建集合失败: {str(e)}")
            # 如果集合已存在，则获取它
            return self.client.get_collection(collection_name)
    
    def _get_embedding_function(self, model_name):
        """获取嵌入函数"""
        from chromadb.utils import embedding_functions
        
        if model_name.startswith("text-embedding"):
            # OpenAI格式的嵌入模型
            return embedding_functions.OpenAIEmbeddingFunction(
                api_key=os.getenv("OPENAI_API_KEY"),
                model_name=model_name
            )
        else:
            # Sentence Transformers模型
            return embedding_functions.SentenceTransformerEmbeddingFunction(
                model_name=model_name
            )
    
    def add_documents(self, collection_name, documents, metadatas=None, ids=None):
        """
        向集合中添加文档
        Args:
            collection_name: 集合名称
            documents: 文档列表
            metadatas: 元数据列表
            ids: 文档ID列表
        """
        collection = self.client.get_collection(collection_name)
        
        # 自动生成ID和元数据
        if ids is None:
            ids = [f"doc_{i}" for i in range(len(documents))]
        
        if metadatas is None:
            metadatas = [{"source": "uploaded", "index": i} 
                        for i in range(len(documents))]
        
        # 添加文档
        collection.add(
            documents=documents,
            metadatas=metadatas,
            ids=ids
        )
        print(f" 已添加 {len(documents)} 个文档到集合 '{collection_name}'")
    
    def search(self, collection_name, query_text, n_results=5, where_filter=None):
        """
        语义搜索
        Args:
            collection_name: 集合名称
            query_text: 查询文本
            n_results: 返回结果数量
            where_filter: 过滤条件
        Returns:
            搜索结果
        """
        collection = self.client.get_collection(collection_name)
        
        results = collection.query(
            query_texts=[query_text],
            n_results=n_results,
            where=where_filter,
            include=["documents", "metadatas", "distances"]
        )
        
        return self._format_results(results)
    
    def _format_results(self, results):
        """格式化搜索结果"""
        formatted = []
        if results and results['documents']:
            for i, doc in enumerate(results['documents'][0]):
                formatted.append({
                    "rank": i + 1,
                    "score": 1 - results['distances'][0][i],  # 距离转相似度
                    "content": doc,
                    "metadata": results['metadatas'][0][i] if results['metadatas'] else {},
                    "distance": results['distances'][0][i]
                })
        return formatted
    
    def delete_collection(self, collection_name):
        """删除集合"""
        try:
            self.client.delete_collection(collection_name)
            print(f" 集合 '{collection_name}' 已删除")
        except Exception as e:
            print(f" 删除集合失败: {str(e)}")
    
    def get_collection_info(self, collection_name):
        """获取集合信息"""
        collection = self.client.get_collection(collection_name)
        return {
            "name": collection_name,
            "count": collection.count(),
            "metadata": collection.metadata
        }

# 使用示例
def demo_chroma_operations():
    """ChromaDB操作演示"""
    # 1. 初始化
    store = ChromaVectorStore(persist_directory="./my_knowledge_base")
    
    # 2. 创建集合
    collection = store.create_collection(
        collection_name="technical_docs",
        embedding_model="all-MiniLM-L6-v2"
    )
    
    # 3. 添加文档
    documents = [
        "RAG（Retrieval-Augmented Generation）是一种检索增强生成技术",
        "向量数据库用于存储和检索文档的嵌入表示",
        "嵌入模型将文本转换为高维向量",
        "语义搜索通过向量相似度找到相关内容",
        "Transformer架构是现代NLP模型的基础"
    ]
    
    metadatas = [
        {"category": "rag", "importance": "high"},
        {"category": "vector_db", "importance": "high"},
        {"category": "embedding", "importance": "medium"},
        {"category": "search", "importance": "medium"},
        {"category": "nlp", "importance": "high"}
    ]
    
    store.add_documents(
        collection_name="technical_docs",
        documents=documents,
        metadatas=metadatas
    )
    
    # 4. 搜索演示
    queries = [
        "什么是检索增强生成？",
        "如何存储向量数据？",
        "文本怎么变成向量？"
    ]
    
    for query in queries:
        print(f"\n 查询: '{query}'")
        results = store.search("technical_docs", query, n_results=3)
        
        for result in results:
            print(f"  排名{result['rank']} (相似度: {result['score']:.3f}):")
            print(f"    内容: {result['content'][:60]}...")
            print(f"    元数据: {result['metadata']}")
    
    # 5. 集合信息
    info = store.get_collection_info("technical_docs")
    print(f"\n 集合信息: {info}")
    
    # 6. 条件过滤搜索
    print("\n 条件过滤搜索 (category='rag'):")
    filtered_results = store.search(
        "technical_docs",
        "检索技术",
        where_filter={"category": "rag"}
    )
    for result in filtered_results:
        print(f"  • {result['content'][:50]}...")

if __name__ == "__main__":
    demo_chroma_operations()

3. Milvus企业级部署

3.1 Docker Compose部署方案

# docker-compose.yml - Milvus单机部署
version: '3.5'

services:
  etcd:
    container_name: milvus-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:
      - ./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:
    container_name: milvus-minio
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ./volumes/minio:/minio_data
    command: minio server /minio_data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 20s
      retries: 3

  standalone:
    container_name: milvus-standalone
    image: milvusdb/milvus:v2.3.3
    command: ["milvus", "run", "standalone"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    volumes:
      - ./volumes/milvus:/var/lib/milvus
    ports:
      - "19530:19530"
      - "9091:9091"
    depends_on:
      - "etcd"
      - "minio"

networks:
  default:
    name: milvus

volumes:
  etcd:
  minio:
  milvus:

3.2 Python客户端操作

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

class MilvusManager:
    """Milvus向量数据库管理器"""
    
    def __init__(self, host='localhost', port='19530'):
        self.host = host
        self.port = port
        self.connect()
    
    def connect(self):
        """连接Milvus服务器"""
        try:
            connections.connect(
                alias="default", 
                host=self.host, 
                port=self.port
            )
            print(f" 已连接到Milvus服务器 {self.host}:{self.port}")
        except Exception as e:
            print(f" 连接失败: {str(e)}")
            raise
    
    def create_collection(self, collection_name, dim=768):
        """
        创建集合
        Args:
            collection_name: 集合名称
            dim: 向量维度
        """
        # 定义字段
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
            FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
            FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=256),
            FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=128),
            FieldSchema(name="created_at", dtype=DataType.INT64)
        ]
        
        # 创建schema
        schema = CollectionSchema(fields=fields, description=f"{collection_name} 知识库")
        
        # 创建集合
        collection = Collection(
            name=collection_name, 
            schema=schema, 
            using='default', 
            shards_num=2
        )
        
        # 创建索引
        index_params = {
            "metric_type": "COSINE",
            "index_type": "IVF_FLAT",
            "params": {"nlist": 16384}
        }
        
        collection.create_index(field_name="embedding", index_params=index_params)
        print(f" 集合 '{collection_name}' 创建成功，向量维度: {dim}")
        
        return collection
    
    def insert_data(self, collection_name, embeddings, contents, titles=None, categories=None):
        """
        插入数据
        Args:
            collection_name: 集合名称
            embeddings: 向量列表
            contents: 内容列表
            titles: 标题列表（可选）
            categories: 分类列表（可选）
        """
        collection = Collection(collection_name)
        
        import time
        current_time = int(time.time())
        
        # 准备数据
        entities = []
        
        # 添加向量
        entities.append(embeddings)
        
        # 添加内容
        entities.append(contents)
        
        # 添加标题（如果提供）
        if titles:
            entities.append(titles)
        else:
            entities.append([""] * len(contents))
        
        # 添加分类（如果提供）
        if categories:
            entities.append(categories)
        else:
            entities.append(["unknown"] * len(contents))
        
        # 添加时间戳
        entities.append([current_time] * len(contents))
        
        # 插入数据
        collection.insert(entities)
        
        # 刷新使数据可搜索
        collection.flush()
        
        print(f" 已插入 {len(contents)} 条数据到集合 '{collection_name}'")
        
        return len(contents)
    
    def search(self, collection_name, query_vector, top_k=10, filter_expr=None):
        """
        向量搜索
        Args:
            collection_name: 集合名称
            query_vector: 查询向量
            top_k: 返回结果数量
            filter_expr: 过滤表达式
        """
        collection = Collection(collection_name)
        
        # 加载集合到内存
        collection.load()
        
        # 搜索参数
        search_params = {
            "metric_type": "COSINE",
            "params": {"nprobe": 32}
        }
        
        # 执行搜索
        results = collection.search(
            data=[query_vector],
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            expr=filter_expr,
            output_fields=["content", "title", "category", "created_at"]
        )
        
        # 格式化结果
        formatted_results = []
        for hits in results:
            for hit in hits:
                formatted_results.append({
                    "id": hit.id,
                    "score": hit.score,
                    "content": hit.entity.get("content"),
                    "title": hit.entity.get("title"),
                    "category": hit.entity.get("category"),
                    "created_at": hit.entity.get("created_at")
                })
        
        # 释放内存
        collection.release()
        
        return formatted_results
    
    def hybrid_search(self, collection_name, query_text, query_embedding, 
                     vector_weight=0.7, keyword_weight=0.3, top_k=10):
        """
        混合搜索：向量 + 关键词
        Args:
            collection_name: 集合名称
            query_text: 查询文本（用于关键词匹配）
            query_embedding: 查询向量（用于向量搜索）
            vector_weight: 向量搜索权重
            keyword_weight: 关键词搜索权重
            top_k: 返回结果数量
        """
        # 1. 向量搜索
        vector_results = self.search(collection_name, query_embedding, top_k=top_k*2)
        
        # 2. 关键词搜索（通过Milvus的表达式过滤）
        # 这里简化处理，实际中可以结合倒排索引
        collection = Collection(collection_name)
        collection.load()
        
        # 使用Milvus的表达式进行简单关键词匹配
        keywords = query_text.split()
        expr_parts = []
        for keyword in keywords[:3]:  # 限制关键词数量
            expr_parts.append(f'content like "%{keyword}%"')
        
        if expr_parts:
            filter_expr = " or ".join(expr_parts)
            
            keyword_results = collection.query(
                expr=filter_expr,
                output_fields=["content", "title", "category", "created_at"],
                limit=top_k*2
            )
        else:
            keyword_results = []
        
        collection.release()
        
        # 3. 结果融合（简化版）
        # 实际项目需要实现更复杂的融合算法
        combined_results = []
        
        # 为向量结果添加权重
        for vr in vector_results:
            vr["final_score"] = vr["score"] * vector_weight
            vr["search_type"] = "vector"
            combined_results.append(vr)
        
        # 为关键词结果添加权重
        # 这里需要计算关键词匹配度
        from collections import Counter
        
        query_word_counts = Counter(query_text.lower().split())
        
        for kr in keyword_results:
            content_words = Counter(str(kr.get("content", "")).lower().split())
            
            # 计算Jaccard相似度
            intersection = sum((query_word_counts & content_words).values())
            union = sum((query_word_counts | content_words).values())
            
            if union > 0:
                keyword_score = intersection / union
                final_score = keyword_score * keyword_weight
                
                combined_results.append({
                    "id": kr.get("id"),
                    "final_score": final_score,
                    "content": kr.get("content"),
                    "title": kr.get("title"),
                    "category": kr.get("category"),
                    "created_at": kr.get("created_at"),
                    "search_type": "keyword",
                    "score": keyword_score
                })
        
        # 按最终分数排序
        combined_results.sort(key=lambda x: x["final_score"], reverse=True)
        
        # 去重（基于ID）
        seen_ids = set()
        deduplicated = []
        
        for result in combined_results:
            if result["id"] not in seen_ids:
                seen_ids.add(result["id"])
                deduplicated.append(result)
        
        return deduplicated[:top_k]
    
    def get_collection_stats(self, collection_name):
        """获取集合统计信息"""
        collection = Collection(collection_name)
        
        stats = {
            "name": collection_name,
            "num_entities": collection.num_entities,
            "primary_field": collection.primary_field.name,
            "description": collection.description
        }
        
        return stats
    
    def delete_collection(self, collection_name):
        """删除集合"""
        try:
            utility.drop_collection(collection_name)
            print(f" 集合 '{collection_name}' 已删除")
        except Exception as e:
            print(f" 删除集合失败: {str(e)}")

# 使用示例
def demo_milvus_operations():
    """Milvus操作演示"""
    # 1. 连接Milvus
    milvus = MilvusManager(host='localhost', port='19530')
    
    # 2. 创建集合
    collection_name = "demo_documents"
    collection = milvus.create_collection(collection_name, dim=1536)
    
    # 3. 模拟数据
    import numpy as np
    
    # 生成模拟向量
    num_docs = 1000
    embeddings = np.random.randn(num_docs, 1536).tolist()
    
    # 生成模拟内容
    categories = ["technology", "science", "business", "education", "health"]
    contents = []
    titles = []
    doc_categories = []
    
    for i in range(num_docs):
        category = categories[i % len(categories)]
        content = f"这是关于{category}的第{i}篇文档，讨论了相关的重要议题和发展趋势。"
        title = f"{category.title()} 文档 {i+1}"
        
        contents.append(content)
        titles.append(title)
        doc_categories.append(category)
    
    # 4. 插入数据
    inserted_count = milvus.insert_data(
        collection_name=collection_name,
        embeddings=embeddings,
        contents=contents,
        titles=titles,
        categories=doc_categories
    )
    
    # 5. 搜索演示
    print("\n 搜索演示:")
    
    # 生成查询向量
    query_vector = np.random.randn(1536).tolist()
    query_text = "科技发展趋势"
    
    # 向量搜索
    vector_results = milvus.search(
        collection_name=collection_name,
        query_vector=query_vector,
        top_k=5
    )
    
    print("\n向量搜索结果:")
    for i, result in enumerate(vector_results[:3]):
        print(f"  {i+1}. [分数: {result['score']:.3f}] {result['title']}")
        print(f"     内容: {result['content'][:50]}...")
    
    # 混合搜索
    hybrid_results = milvus.hybrid_search(
        collection_name=collection_name,
        query_text=query_text,
        query_embedding=query_vector,
        vector_weight=0.7,
        keyword_weight=0.3,
        top_k=5
    )
    
    print("\n混合搜索结果:")
    for i, result in enumerate(hybrid_results[:3]):
        search_type = result.get("search_type", "unknown")
        print(f"  {i+1}. [{search_type}, 分数: {result['final_score']:.3f}] {result['title']}")
        print(f"     内容: {result['content'][:50]}...")
    
    # 6. 集合统计
    stats = milvus.get_collection_stats(collection_name)
    print(f"\n 集合统计: {stats}")
    
    # 7. 清理（可选）
    # milvus.delete_collection(collection_name)

if __name__ == "__main__":
    demo_milvus_operations()

三、生产环境部署与管理

1. ChromaDB服务器部署

#!/bin/bash
# deploy-chromadb-server.sh

# 创建数据目录
DATA_DIR="/data/chromadb"
LOG_DIR="/var/log/chromadb"
mkdir -p $DATA_DIR $LOG_DIR
chmod 755 $DATA_DIR $LOG_DIR

# 创建systemd服务文件
cat > /etc/systemd/system/chromadb.service << EOF
[Unit]
Description=ChromaDB Vector Database Server
After=network.target

[Service]
Type=simple
User=chroma
Group=chroma
WorkingDirectory=$DATA_DIR
Environment="PATH=/usr/local/bin:/usr/bin:/bin"
ExecStart=/usr/local/bin/chroma run --path $DATA_DIR --host 0.0.0.0 --port 8000
Restart=always
RestartSec=10
StandardOutput=append:$LOG_DIR/chromadb.log
StandardError=append:$LOG_DIR/chromadb.error.log

[Install]
WantedBy=multi-user.target
EOF

# 创建专用用户
useradd -r -s /bin/false -d $DATA_DIR chroma
chown -R chroma:chroma $DATA_DIR $LOG_DIR

# 重新加载systemd配置
systemctl daemon-reload

# 启动服务
systemctl start chromadb
systemctl enable chromadb

# 检查服务状态
systemctl status chromadb

echo " ChromaDB服务器已部署完成"
echo " 访问地址: http://your-server-ip:8000"
echo " 数据目录: $DATA_DIR"
echo " 日志文件: $LOG_DIR/chromadb.log"

2. 负载均衡配置（Nginx）

# nginx-chromadb.conf
upstream chromadb_cluster {
    # 多实例负载均衡
    server 192.168.1.101:8000 weight=3;
    server 192.168.1.102:8000 weight=2;
    server 192.168.1.103:8000 weight=2;
    
    # 健康检查
    check interval=3000 rise=2 fall=3 timeout=1000 type=http;
    check_http_send "GET /api/v1/heartbeat HTTP/1.0\r\n\r\n";
    check_http_expect_alive http_2xx http_3xx;
}

server {
    listen 80;
    server_name chromadb.yourdomain.com;
    
    # SSL配置（如需HTTPS）
    # listen 443 ssl;
    # ssl_certificate /path/to/cert.pem;
    # ssl_certificate_key /path/to/key.pem;
    
    location / {
        proxy_pass http://chromadb_cluster;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 连接超时设置
        proxy_connect_timeout 30s;
        proxy_send_timeout 120s;
        proxy_read_timeout 120s;
        
        # 缓冲区设置
        proxy_buffering on;
        proxy_buffer_size 4k;
        proxy_buffers 8 4k;
        proxy_busy_buffers_size 8k;
        
        # 启用WebSocket支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
    
    # 监控端点
    location /status {
        stub_status on;
        access_log off;
        allow 127.0.0.1;
        deny all;
    }
    
    # 静态文件缓存
    location ~* \.(js|css|png|jpg|jpeg|gif|ico)$ {
        expires 1y;
        add_header Cache-Control "public, immutable";
    }
}

3. 监控与告警配置

# prometheus-chromadb.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'chromadb'
    static_configs:
      - targets:
        - '192.168.1.101:8000'
        - '192.168.1.102:8000'
        - '192.168.1.103:8000'
    metrics_path: '/api/v1/metrics'
    
  - job_name: 'chromadb_node'
    static_configs:
      - targets:
        - '192.168.1.101:9100'  # Node Exporter
        - '192.168.1.102:9100'
        - '192.168.1.103:9100'

# Grafana仪表板配置
dashboard_json: |
  {
    "panels": [
      {
        "title": "ChromaDB QPS",
        "targets": [
          {
            "expr": "rate(chromadb_api_requests_total[5m])",
            "legendFormat": "{{instance}}"
          }
        ],
        "type": "graph"
      },
      {
        "title": "内存使用率",
        "targets": [
          {
            "expr": "node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes * 100",
            "legendFormat": "{{instance}}"
          }
        ],
        "type": "singlestat"
      },
      {
        "title": "向量搜索延迟",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(chromadb_query_duration_seconds_bucket[5m]))",
            "legendFormat": "P95 Latency"
          }
        ],
        "type": "graph"
      }
    ]
  }

# Alertmanager告警规则
groups:
  - name: chromadb_alerts
    rules:
      - alert: HighMemoryUsage
        expr: node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes < 0.2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "内存使用率过高"
          description: "{{ $labels.instance }} 内存使用率超过80%"
      
      - alert: HighQueryLatency
        expr: histogram_quantile(0.95, rate(chromadb_query_duration_seconds_bucket[5m])) > 1
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: "查询延迟过高"
          description: "{{ $labels.instance }} 查询P95延迟超过1秒"
      
      - alert: ServiceDown
        expr: up{job="chromadb"} == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "服务下线"
          description: "{{ $labels.instance }} ChromaDB服务已下线"

四、RAG技术进阶应用

1. 完整的RAG系统架构

from typing import List, Dict, Any
import hashlib

class RAGSystem:
    """完整的RAG系统实现"""
    
    def __init__(self, vector_store, embedding_model, llm_model):
        """
        初始化RAG系统
        Args:
            vector_store: 向量数据库实例
            embedding_model: 嵌入模型
            llm_model: 大语言模型
        """
        self.vector_store = vector_store
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        
    def ingest_documents(self, documents: List[str], chunk_size: int = 500, 
                        overlap: int = 50) -> Dict[str, Any]:
        """
        文档摄取：分割、向量化、存储
        Args:
            documents: 原始文档列表
            chunk_size: 分块大小
            overlap: 块重叠大小
        """
        # 1. 文档分割
        chunks = self._chunk_documents(documents, chunk_size, overlap)
        
        # 2. 生成向量
        embeddings = self.embedding_model.encode(chunks)
        
        # 3. 生成唯一ID
        chunk_ids = []
        for i, chunk in enumerate(chunks):
            chunk_hash = hashlib.md5(chunk.encode()).hexdigest()[:8]
            chunk_ids.append(f"chunk_{i}_{chunk_hash}")
        
        # 4. 存储到向量数据库
        self.vector_store.add_documents(
            documents=chunks,
            embeddings=embeddings,
            ids=chunk_ids
        )
        
        return {
            "total_chunks": len(chunks),
            "average_chunk_size": sum(len(c) for c in chunks) // len(chunks),
            "chunk_ids": chunk_ids
        }
    
    def _chunk_documents(self, documents: List[str], chunk_size: int, 
                        overlap: int) -> List[str]:
        """文档分割"""
        chunks = []
        
        for doc in documents:
            words = doc.split()
            
            for i in range(0, len(words), chunk_size - overlap):
                chunk_words = words[i:i + chunk_size]
                chunk = " ".join(chunk_words)
                
                if len(chunk.strip()) > 20:  # 忽略太短的块
                    chunks.append(chunk)
        
        return chunks
    
    def retrieve_context(self, query: str, top_k: int = 5, 
                        rerank: bool = True) -> List[Dict]:
        """
        检索上下文
        Args:
            query: 查询文本
            top_k: 检索数量
            rerank: 是否重排序
        """
        # 1. 首轮检索（获取更多候选）
        candidates = self.vector_store.search(query, n_results=top_k * 3)
        
        # 2. 重排序（如果需要）
        if rerank and len(candidates) > top_k:
            ranked_results = self._rerank_results(query, candidates)
            final_results = ranked_results[:top_k]
        else:
            final_results = candidates[:top_k]
        
        return final_results
    
    def _rerank_results(self, query: str, candidates: List[Dict]) -> List[Dict]:
        """结果重排序"""
        # 使用交叉编码器进行更准确的相关性评分
        try:
            from sentence_transformers import CrossEncoder
            cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
            
            pairs = [[query, cand['document']] for cand in candidates]
            scores = cross_encoder.predict(pairs)
            
            # 更新分数
            for i, cand in enumerate(candidates):
                cand['rerank_score'] = float(scores[i])
            
            # 按重排序分数排序
            candidates.sort(key=lambda x: x.get('rerank_score', 0), reverse=True)
            
        except ImportError:
            # 回退到简单的基于关键词的重排序
            query_words = set(query.lower().split())
            
            for cand in candidates:
                cand_words = set(cand['document'].lower().split())
                common = len(query_words.intersection(cand_words))
                cand['rerank_score'] = common / max(len(query_words), 1)
            
            candidates.sort(key=lambda x: x['rerank_score'], reverse=True)
        
        return candidates
    
    def generate_answer(self, query: str, contexts: List[Dict], 
                       temperature: float = 0.7) -> Dict[str, Any]:
        """
        生成答案
        Args:
            query: 查询问题
            contexts: 检索到的上下文
            temperature: 生成温度
        """
        # 1. 构建提示词
        prompt = self._build_prompt(query, contexts)
        
        # 2. 调用LLM生成
        response = self.llm_model.generate(
            prompt=prompt,
            temperature=temperature,
            max_tokens=1000
        )
        
        # 3. 解析响应
        answer = self._parse_response(response)
        
        # 4. 构建返回结果
        return {
            "answer": answer,
            "query": query,
            "contexts": contexts,
            "used_context_count": len(contexts),
            "prompt_template": self._get_prompt_template()
        }
    
    def _build_prompt(self, query: str, contexts: List[Dict]) -> str:
        """构建提示词"""
        context_text = "\n\n".join([
            f"[来源 {i+1}] {ctx['document'][:500]}..." 
            for i, ctx in enumerate(contexts)
        ])
        
        prompt_template = """
        基于以下提供的上下文信息，回答问题。如果上下文信息不足以回答问题，
        请诚实地回答"根据提供的信息，我无法回答这个问题"。

        上下文信息:
        {contexts}

        问题: {query}

        请提供详细、准确的回答，并注明答案来源于哪些上下文：
        """
        
        return prompt_template.format(
            contexts=context_text,
            query=query
        )
    
    def rag_pipeline(self, query: str, **kwargs) -> Dict[str, Any]:
        """
        完整的RAG流程
        Args:
            query: 用户查询
            kwargs: 其他参数
        """
        # 1. 检索上下文
        contexts = self.retrieve_context(query, **kwargs)
        
        # 2. 生成答案
        result = self.generate_answer(query, contexts)
        
        # 3. 添加检索指标
        result.update({
            "retrieval_metrics": {
                "query_time_ms": kwargs.get("query_time_ms", 0),
                "context_precision": self._calculate_precision(query, contexts),
                "context_recall": self._calculate_recall(query, contexts),
                "context_relevance": self._calculate_relevance(query, contexts)
            }
        })
        
        return result
    
    def _calculate_precision(self, query: str, contexts: List[Dict]) -> float:
        """计算检索精度（简化版）"""
        # 在实际项目中，这里需要真实的相关性标注
        # 这里使用简单的关键词匹配作为示意
        query_words = set(query.lower().split())
        relevant_count = 0
        
        for ctx in contexts:
            ctx_words = set(ctx['document'].lower().split())
            if len(query_words.intersection(ctx_words)) > 0:
                relevant_count += 1
        
        return relevant_count / max(len(contexts), 1)
    
    def _calculate_recall(self, query: str, contexts: List[Dict]) -> float:
        """计算检索召回率（简化版）"""
        # 在实际项目中，需要知道总的相关文档数
        # 这里返回一个估计值作为示意
        return 0.8
    
    def _calculate_relevance(self, query: str, contexts: List[Dict]) -> float:
        """计算整体相关性"""
        precision = self._calculate_precision(query, contexts)
        recall = self._calculate_recall(query, contexts)
        
        if precision + recall > 0:
            f1 = 2 * precision * recall / (precision + recall)
        else:
            f1 = 0
        
        return f1

2. PDF文档处理流水线

import pdfplumber
from PIL import Image
import pytesseract
import io

class PDFProcessor:
    """PDF文档处理流水线"""
    
    def __init__(self, ocr_language='chi_sim+eng'):
        self.ocr_language = ocr_language
    
    def process_pdf(self, pdf_path: str) -> Dict[str, Any]:
        """
        处理PDF文件
        Args:
            pdf_path: PDF文件路径
        Returns:
            处理结果
        """
        results = {
            "text_content": [],
            "images": [],
            "tables": [],
            "metadata": {},
            "processing_summary": {}
        }
        
        try:
            with pdfplumber.open(pdf_path) as pdf:
                # 提取元数据
                results["metadata"] = {
                    "page_count": len(pdf.pages),
                    "author": pdf.metadata.get("Author", ""),
                    "title": pdf.metadata.get("Title", ""),
                    "creation_date": pdf.metadata.get("CreationDate", "")
                }
                
                # 逐页处理
                for page_num, page in enumerate(pdf.pages):
                    page_result = self._process_page(page, page_num)
                    
                    results["text_content"].extend(page_result["text"])
                    results["tables"].extend(page_result["tables"])
                    results["images"].extend(page_result["images"])
                
                # 生成摘要
                results["processing_summary"] = {
                    "total_text_chars": sum(len(t) for t in results["text_content"]),
                    "total_images": len(results["images"]),
                    "total_tables": len(results["tables"]),
                    "avg_text_per_page": sum(len(t) for t in results["text_content"]) / len(pdf.pages)
                }
                
        except Exception as e:
            print(f"处理PDF时出错: {str(e)}")
            results["error"] = str(e)
        
        return results
    
    def _process_page(self, page, page_num: int) -> Dict[str, Any]:
        """处理单个页面"""
        page_result = {
            "text": [],
            "images": [],
            "tables": []
        }
        
        # 1. 提取文本
        page_text = page.extract_text()
        if page_text and page_text.strip():
            page_result["text"].append({
                "page": page_num + 1,
                "content": page_text,
                "type": "direct_text"
            })
        
        # 2. 提取表格
        tables = page.extract_tables()
        for table_num, table in enumerate(tables):
            table_data = {
                "page": page_num + 1,
                "table_number": table_num + 1,
                "rows": len(table),
                "columns": len(table[0]) if table else 0,
                "data": table
            }
            page_result["tables"].append(table_data)
        
        # 3. 提取和OCR图像
        images = page.images
        for img_num, img in enumerate(images):
            image_data = self._process_image(img, page_num, img_num)
            if image_data:
                page_result["images"].append(image_data)
        
        return page_result
    
    def _process_image(self, img, page_num: int, img_num: int) -> Optional[Dict]:
        """处理图像并进行OCR"""
        try:
            # 从PDF中提取图像
            image_obj = img.to_image(resolution=150)
            image_bytes = io.BytesIO()
            image_obj.save(image_bytes, format='PNG')
            image_bytes.seek(0)
            
            # OCR识别
            text = pytesseract.image_to_string(
                Image.open(image_bytes), 
                lang=self.ocr_language
            )
            
            return {
                "page": page_num + 1,
                "image_number": img_num + 1,
                "ocr_text": text.strip(),
                "image_size": image_obj.size,
                "bbox": img["bbox"]
            }
            
        except Exception as e:
            print(f"处理图像时出错: {str(e)}")
            return None
    
    def chunk_pdf_content(self, pdf_result: Dict, chunk_size: int = 500, 
                         overlap: int = 50) -> List[Dict]:
        """
        分割PDF内容为适合向量化的块
        Args:
            pdf_result: PDF处理结果
            chunk_size: 块大小（字数）
            overlap: 重叠大小
        """
        chunks = []
        
        # 1. 处理文本内容
        for text_item in pdf_result["text_content"]:
            text_chunks = self._chunk_text(
                text_item["content"], 
                chunk_size, 
                overlap,
                metadata={
                    "source": "pdf_text",
                    "page": text_item["page"],
                    "type": text_item["type"]
                }
            )
            chunks.extend(text_chunks)
        
        # 2. 处理OCR文本
        for image_item in pdf_result["images"]:
            if image_item["ocr_text"]:
                ocr_chunks = self._chunk_text(
                    image_item["ocr_text"],
                    chunk_size,
                    overlap,
                    metadata={
                        "source": "pdf_ocr",
                        "page": image_item["page"],
                        "image_number": image_item["image_number"]
                    }
                )
                chunks.extend(ocr_chunks)
        
        # 3. 处理表格（转换为文本格式）
        for table_item in pdf_result["tables"]:
            table_text = self._convert_table_to_text(table_item["data"])
            if table_text:
                table_chunks = self._chunk_text(
                    table_text,
                    chunk_size,
                    overlap,
                    metadata={
                        "source": "pdf_table",
                        "page": table_item["page"],
                        "table_number": table_item["table_number"]
                    }
                )
                chunks.extend(table_chunks)
        
        return chunks
    
    def _chunk_text(self, text: str, chunk_size: int, overlap: int, 
                   metadata: Dict) -> List[Dict]:
        """文本分块"""
        chunks = []
        sentences = text.split('。')
        
        current_chunk = ""
        for i in range(len(sentences)):
            sentence = sentences[i] + '。'
            
            if len(current_chunk) + len(sentence) <= chunk_size:
                current_chunk += sentence
            else:
                if current_chunk:
                    chunk_metadata = metadata.copy()
                    chunk_metadata.update({
                        "chunk_position": len(chunks),
                        "char_count": len(current_chunk)
                    })
                    
                    chunks.append({
                        "content": current_chunk,
                        "metadata": chunk_metadata
                    })
                
                # 重叠处理
                if overlap > 0 and i > 0:
                    # 从前一句开始，包含重叠部分
                    prev_sentence = sentences[i-1] + '。'
                    current_chunk = prev_sentence + sentence
                else:
                    current_chunk = sentence
        
        # 添加最后一个块
        if current_chunk:
            chunk_metadata = metadata.copy()
            chunk_metadata.update({
                "chunk_position": len(chunks),
                "char_count": len(current_chunk)
            })
            
            chunks.append({
                "content": current_chunk,
                "metadata": chunk_metadata
            })
        
        return chunks
    
    def _convert_table_to_text(self, table_data: List[List]) -> str:
        """将表格数据转换为文本描述"""
        if not table_data:
            return ""
        
        text_lines = []
        
        # 添加表头
        if table_data and len(table_data) > 0:
            header = " | ".join(str(cell) for cell in table_data[0])
            text_lines.append(f"表头: {header}")
            
            # 添加数据行
            for row_idx, row in enumerate(table_data[1:], 1):
                if row:
                    row_text = " | ".join(str(cell) for cell in row)
                    text_lines.append(f"第{row_idx}行: {row_text}")
        
        return "\n".join(text_lines)