RAGFlow 集成 Milvus向量库操作指南

一、前置条件检查

1. 版本兼容性

   RAGFlow版本	Milvus版本	Python SDK版本
   ≥v1.2.0	2.4.x	pymilvus≥2.4.0
   ≥v1.3.5	2.5.x	pymilvus≥2.5.1

2. 网络连通性测试

   # 在 RAGFlow 容器内执行
   docker exec -it ragflow_container bash
   curl -v telnet://milvus-host:19530
   nc -zv milvus-host 19530
   

二、配置变更全流程

1. 修改 docker-compose.yaml

   services:
     ragflow:
       environment:
   +      VECTOR_DB_TYPE: "milvus"
   +      MILVUS_HOST: "milvus-prod"
   +      MILVUS_PORT: "19530"
   +      MILVUS_USER: "admin"
   +      MILVUS_PASSWORD: "SecureP@ss123!"
       depends_on:
   +      - milvus
   
   +  milvus:
   +    image: milvusdb/milvus:v2.5.3
   +    ports:
   +      - "19530:19530"
   +    volumes:
   +      - milvus_data:/var/lib/milvus
   +      - milvus_conf:/etc/milvus
   

2. 创建 Milvus 专用账户

   docker exec -it milvus-prod bash
   milvus-cli --user root --password milvusroot
   CREATE USER 'admin' IDENTIFIED BY 'SecureP@ss123!';
   GRANT ALL ON *.* TO admin;
   

三、向量库迁移实操

1. 从 Chroma 导出数据

   from chromadb.api import ClientAPI
   
   client = ClientAPI()
   collections = client.list_collections()
   for col in collections:
       data = col.get(include=["embeddings", "metadatas"])
       with open(f"/backup/{col.name}.json", "w") as f:
           json.dump(data, f)
   

2. 向 Milvus 导入数据

   from pymilvus import utility, Collection
   
   collection = Collection("rag_docs")
   for file in os.listdir("/backup"):
       with open(file) as f:
           data = json.load(f)
           entities = [
               {
                   "id": hash(item["metadata"]["source"]),
                   "text": item["document"],
                   "vector": item["embedding"],
                   "source": item["metadata"]["source"]
               } for item in data
           ]
           collection.insert(entities)
   utility.wait_for_loading_complete("rag_docs")
   

四、检索适配层改造

1. 修改检索器实现

   # 原 Chroma 实现
   from chromadb import QueryResult
   
   class ChromaRetriever:
       def search(self, query):
           return self.collection.query(query_texts=[query])
   
   # 改为 Milvus 实现
   from pymilvus import SearchRequest, AnnsField
   
   class MilvusRetriever:
       def __init__(self):
           self.collection = Collection("rag_docs")
           
       def search(self, query, top_k=5):
           req = SearchRequest(
               data=[self._encode(query)],
               anns_field=AnnsField("vector"),
               param={"metric_type": "IP", "params": {"nprobe": 32}},
               limit=top_k
           )
           return self.collection.search(req)
   

2. 混合检索增强

   def hybrid_search(query):
       # 向量检索
       vector_results = milvus_retriever.search(query)
       
       # 关键词检索
       keyword_results = self._keyword_search(query)
       
       # 结果融合 (Weighted Reciprocal Rank Fusion)
       fused = []
       for res in [vector_results, keyword_results]:
           for i, item in enumerate(res):
               score = item.score * (1 / (i + 1))
               fused.append((item.id, score))
               
       return sorted(fused, key=lambda x: -x[1])[:10]
   

五、性能优化配置

1. 索引参数优化

   index_params = {
       "index_type": "GPU_IVF_PQ",
       "params": {
           "nlist": 4096,       # 平衡查询速度和内存
           "m": 32,             # PQ 子量化器数量
           "nbits": 8           # 每个向量的存储位数
       },
       "metric_type": "IP"
   }
   

2. 缓存策略

   # milvus.yaml 配置
   cache:
     cache_size: 16GB        # 分配内存缓存
     insert_buffer_size: 2GB # 写入缓冲区
     preload_collection: rag_docs # 启动时预加载
   

六、验证与监控

1. 集成验证脚本

   # 测试向量插入
   python -c "from pymilvus import Collection; c=Collection('rag_docs'); print(c.num_entities)"
   
   # 测试检索时延
   ab -n 1000 -c 10 -p query.json http://ragflow:9380/api/search
   

2. 监控看板配置

   # Prometheus 配置
   - job_name: 'milvus'
     static_configs:
       - targets: ['milvus-prod:9090']
   
   # Grafana 导入看板
   Dashboard ID: 13613  # 官方监控模板
   

故障排查指南：

1. 连接失败 → 检查防火墙规则和SELinux状态
2. 检索超时 → 增加 search.params.nprobe 值
3. 内存不足 → 启用 GPU_IVF_PQ 索引减少内存占用
4. 数据不一致 → 使用 utility.flush(['rag_docs']) 强制同步