前言在 AI 应用快速发展的今天向量数据库已成为构建智能检索系统的核心基础设施。Milvus 作为一款开源的高性能向量数据库在 RAGRetrieval-Augmented Generation系统中发挥着关键作用。本文将带你从零开始基于 Milvus 构建一个完整的 RAG 系统涵盖数据准备、向量检索、结果重排、位置优化等核心环节并分享生产环境中的最佳实践和性能优化技巧。一、 Milvus 简介与核心特性1.1 什么是 MilvusMilvus 是一个开源的向量数据库专为 AI 应用设计支持大规模向量数据的存储、索引和检索。它具备以下核心特性高性能毫秒级向量检索支持百万到百亿级向量易用性提供 Python SDKAPI 简洁直观可扩展性支持分布式部署水平扩展多索引支持HNSW、IVF、DiskANN 等多种索引算法多模态友好支持文本、图像、音频等多种数据类型1.2 为什么选择 Milvus在构建 RAG 系统时我们需要一个能够快速检索从大规模文档库中快速找到相关文档精确匹配通过向量相似度找到语义相关的文档易于集成与现有 AI 工具链无缝集成生产就绪支持高并发、高可用部署Milvus 完美满足这些需求已成为 RAG 系统的首选向量数据库。1.3 Milvus Lite vs Milvus ServerMilvus 提供了两种使用方式特性Milvus LiteMilvus Server部署方式本地文件无需启动服务独立服务需要启动适用场景开发、测试、小规模应用生产环境、大规模应用数据存储本地文件.db分布式存储性能适合中小规模100 万向量支持大规模百万到百亿级使用方式MilvusClient(uri./db)connections.connect(hostlocalhost)本文使用 Milvus Lite因为它更简单适合快速上手和开发测试。二、环境准备与安装2.1 系统要求Python 3.88GB RAM推荐 16GB10GB 磁盘空间用于存储向量和模型2.2 安装依赖首先创建虚拟环境推荐# 创建虚拟环境python3 -m venv venv# 激活虚拟环境# macOS/Linux:source venv/bin/activate# Windows:# venv\Scripts\activate安装必要的依赖pip install pymilvus[milvus_lite] # 包含Milvus Litepip install sentence-transformers # Embedding模型pip install transformers # 重排模型pip install torch # PyTorch重排模型需要重要提示必须安装pymilvus[milvus_lite]不是pymilvus这样才能使用 Milvus Lite 的本地文件功能。2.3 验证安装from pymilvus import MilvusClient# 测试Milvus Lite连接client MilvusClient(uri./test.db)print(✓ Milvus Lite 安装成功)三、数据准备从文档到向量3.1 文档预处理在实际应用中我们需要将原始文档转换为向量。这个过程包括文档加载从文件、数据库或 API 加载文档文档分块将长文档切分为较小的块chunks向量化使用 Embedding 模型将文本转换为向量3.2 文档分块策略文档分块是 RAG 系统的关键步骤直接影响检索效果def chunk_document(text, chunk_size512, overlap50): 文档分块 Args: text: 原始文档文本 chunk_size: 每块的大小字符数或tokens overlap: 块之间的重叠大小保证上下文连贯性 Returns: 文档块列表 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] chunks.append(chunk) start end - overlap # 重叠保证上下文连贯 return chunks分块策略选择固定大小分块简单快速适合结构化文档语义分块按句子或段落边界分块保持语义完整性滑动窗口使用重叠窗口避免信息丢失3.3 向量化选择 Embedding 模型选择合适的 Embedding 模型至关重要模型维度语言支持适用场景性能all-MiniLM-L6-v2384英文快速原型、演示⭐⭐⭐BGE-large-en-v1.51024英文生产环境、高精度⭐⭐⭐⭐⭐BGE-large-zh-v1.51024中文中文检索⭐⭐⭐⭐⭐E5-mistral-7b4096多语言长文档检索⭐⭐⭐⭐代码示例from sentence_transformers import SentenceTransformer# 选择模型根据需求embedding_model sentence-transformers/all-MiniLM-L6-v2 # 快速80MB# embedding_model BAAI/bge-large-en-v1.5 # 高精度1.3GB# 加载模型encoder SentenceTransformer(embedding_model)# 生成向量texts [Milvus is a vector database..., RAG combines retrieval and generation...]embeddings encoder.encode(texts, normalize_embeddingsTrue)print(f向量维度: {embeddings.shape}) # (2, 384) 或 (2, 1024)3.4 创建 Milvus 集合在插入数据之前需要先创建集合Collectionfrom pymilvus import MilvusClient# 连接Milvus自动使用Milvus Liteclient MilvusClient(uri./milvus_demo.db)# 集合配置collection_name documentsdimension 384# 根据Embedding模型选择all-MiniLM-L6-v2384, BGE-large1024# 检查集合是否存在if client.has_collection(collection_name): print(f集合 {collection_name} 已存在删除旧集合...) client.drop_collection(collection_name)# 创建集合client.create_collection( collection_namecollection_name, dimensiondimension, # 向量维度 metric_typeL2, # 距离度量L2欧氏距离或IP内积 auto_idTrue, # 自动生成ID)print(f✓ 集合 {collection_name} 创建成功)3.5 插入数据将向量和元数据插入 Milvus# 准备数据documents [ { text: Milvus is an open-source vector database..., doc_id: doc_001, title: Introduction to Milvus }, # ... 更多文档]# 生成向量texts [doc[text] for doc in documents]embeddings encoder.encode(texts, normalize_embeddingsTrue)# 准备插入数据data []for i, (emb, doc) in enumerate(zip(embeddings, documents)): data.append({ vector: emb.tolist(), # 向量必须 text: doc[text], # 文本内容 doc_id: doc[doc_id], # 业务ID title: doc[title], # 标题 })# 插入数据client.insert(collection_namecollection_name, datadata)print(f✓ 成功插入 {len(data)} 个文档)完整的数据准备脚本完整的数据准备示例from pymilvus import MilvusClientfrom sentence_transformers import SentenceTransformerdef prepare_data(): # 1. 连接Milvus client MilvusClient(uri./milvus_demo.db) collection_name documents # 2. 准备文档 documents [ Milvus is an open-source vector database..., RAG combines information retrieval with language models..., # ... 更多文档 ] # 3. 初始化Embedding模型 encoder SentenceTransformer(sentence-transformers/all-MiniLM-L6-v2) dimension 384 # 4. 创建集合 if client.has_collection(collection_name): client.drop_collection(collection_name) client.create_collection( collection_namecollection_name, dimensiondimension, metric_typeL2, auto_idTrue, ) # 5. 生成向量并插入 embeddings encoder.encode(documents, normalize_embeddingsTrue) data [ {vector: emb.tolist(), text: doc} for emb, doc in zip(embeddings, documents) ] client.insert(collection_namecollection_name, datadata) print(f✓ 数据准备完成共 {len(data)} 个文档)if __name__ __main__: prepare_data()四、核心实现构建 RAG 系统4.1 RAG 系统架构一个完整的 RAG 系统包含以下组件用户查询 ↓向量检索Milvus← 知识库 ↓结果重排可选 ↓上下文构建 ↓LLM生成答案4.2 基础 RAG 实现让我们从最简单的 RAG 系统开始from pymilvus import MilvusClientfrom sentence_transformers import SentenceTransformerclass SimpleRAGSystem: 基础RAG系统 def __init__(self, milvus_uri./milvus_demo.db, collection_namedocuments): # 连接Milvus self.client MilvusClient(urimilvus_uri) self.collection_name collection_name # 初始化Embedding模型 self.encoder SentenceTransformer(sentence-transformers/all-MiniLM-L6-v2) def retrieve(self, query: str, top_k: int 10): 向量检索 # 1. 编码查询 query_vector self.encoder.encode(query, normalize_embeddingsTrue) query_vector query_vector.tolist() # 2. 在Milvus中搜索 results self.client.search( collection_nameself.collection_name, data[query_vector], limittop_k, search_params{metric_type: L2, params: {}}, output_fields[text, doc_id, title] ) # 3. 格式化结果 retrieved_docs [] for hit in results[0]: retrieved_docs.append({ id: hit.get(id, ), text: hit.get(entity, {}).get(text, ), score: hit.get(distance, 0.0), doc_id: hit.get(entity, {}).get(doc_id, ), title: hit.get(entity, {}).get(title, ), }) return retrieved_docs def query(self, user_query: str, top_k: int 5): 查询流程 # 1. 检索 retrieved_docs self.retrieve(user_query, top_ktop_k) # 2. 构建上下文 context \n\n.join([doc[text] for doc in retrieved_docs]) # 3. 返回结果这里只返回上下文实际应用中会调用LLM return { query: user_query, context: context, documents: retrieved_docs }# 使用示例rag SimpleRAGSystem()result rag.query(What is Milvus?)print(result[context])4.3 增强版 RAG添加重排向量检索虽然快速但可能不够精确。我们可以添加重排Reranking步骤来提升准确性from transformers import AutoModelForSequenceClassification, AutoTokenizerimport torchclass EnhancedRAGSystem(SimpleRAGSystem): 增强版RAG系统带重排 def __init__(self, milvus_uri./milvus_demo.db, collection_namedocuments): super().__init__(milvus_uri, collection_name) # 初始化重排模型可选 self.reranker None self.reranker_tokenizer None try: print(正在加载重排模型...) reranker_model BAAI/bge-reranker-base self.reranker AutoModelForSequenceClassification.from_pretrained(reranker_model) self.reranker_tokenizer AutoTokenizer.from_pretrained(reranker_model) self.reranker.eval() print(✓ 重排模型加载成功) except Exception as e: print(f⚠️ 重排模型加载失败: {e}将跳过重排步骤) def rerank(self, query: str, documents: List[str], top_k: int 10): 重排使用BGE-Reranker对检索结果进行精排 if self.reranker isNoneor len(documents) 0: return documents[:top_k] # 构建查询-文档对 pairs [[query, doc] for doc in documents] # Tokenize with torch.no_grad(): inputs self.reranker_tokenizer( pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512 ) # 计算相关性分数 scores self.reranker(**inputs).logits.squeeze(-1) # 按分数排序 ranked_indices scores.argsort(descendingTrue) # 返回Top-K reranked_docs [documents[idx] for idx in ranked_indices[:top_k]] return reranked_docs def query(self, user_query: str, retrieve_top_k: int 100, rerank_top_k: int 10): 查询流程检索 → 重排 → 构建上下文 # 1. 向量检索粗排 retrieved_docs self.retrieve(user_query, top_kretrieve_top_k) # 2. 重排精排 doc_texts [doc[text] for doc in retrieved_docs] reranked_texts self.rerank(user_query, doc_texts, top_krerank_top_k) # 3. 构建上下文 context \n\n.join(reranked_texts) return { query: user_query, context: context, retrieved_count: len(retrieved_docs), reranked_count: len(reranked_texts) }重排的优势提升准确性重排模型考虑查询和文档的交互比单纯向量相似度更准确灵活调整可以调整重排后的 Top-K平衡准确性和成本效果显著通常能提升 15-22%的准确率这份完整版的大模型 AI 学习和面试资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】五、关键优化突破 U 型陷阱5.1 什么是 U 型陷阱研究发现长上下文语言模型存在位置偏差Position Bias开头位置Primacy Bias准确率 ~75.8% ✅中间位置Lost in the Middle准确率 ~53.8% ❌结尾位置Recency Bias准确率 ~63.2% ✅这意味着如果关键信息放在中间位置模型可能无法有效利用导致性能下降。5.2 位置优化策略通过位置优化我们可以突破 U 型陷阱核心策略最相关文档 → 开头利用 Primacy Bias次相关文档 → 中间低优先级用户问题 → 结尾利用 Recency Bias5.3 实现位置优化class OptimizedRAGSystem(EnhancedRAGSystem): 优化版RAG系统位置优化 def build_context(self, query: str, retrieved_docs: List[Dict[str, Any]]) - str: 构建上下文位置优化突破U型陷阱的关键步骤 策略 - 最相关文档 → 开头利用Primacy Bias - 次相关文档 → 中间低优先级 - 用户问题 → 结尾利用Recency Bias if len(retrieved_docs) 0: returnf系统提示请回答问题。\n\n用户问题{query} # 按相关性排序最相关的在前 sorted_docs sorted( retrieved_docs, keylambda x: x.get(score, 0) if isinstance(x.get(score), (int, float)) else0, reverseTrue# 分数越高越好如果是相似度分数 ) # 构建上下文 context_parts [] # 系统提示 context_parts.append(系统提示请基于以下文档回答问题。\n) # 最相关文档开头位置 - 利用Primacy Bias context_parts.append(# 最相关文档开头位置\n) top_docs sorted_docs[:min(3, len(sorted_docs))] for i, doc in enumerate(top_docs, 1): text doc.get(text, ) context_parts.append(f文档 {i}{text}\n) # 次相关文档中间位置 if len(sorted_docs) 3: context_parts.append(\n# 次相关文档中间位置\n) secondary_docs sorted_docs[3:min(7, len(sorted_docs))] for i, doc in enumerate(secondary_docs, 4): text doc.get(text, ) context_parts.append(f文档 {i}{text}\n) # 用户问题结尾位置 - 利用Recency Bias context_parts.append(\n# 用户问题结尾位置\n) context_parts.append(f用户问题{query}) return\n.join(context_parts) def query(self, user_query: str, retrieve_top_k: int 100, rerank_top_k: int 10): 完整查询流程检索 → 重排 → 位置优化 → 生成 print(f\n查询{user_query}\n) # 1. 向量检索 print(f步骤 1: 向量检索Top-{retrieve_top_k}...) retrieved_docs self.retrieve(user_query, top_kretrieve_top_k) print(f✓ 检索到 {len(retrieved_docs)} 个文档) if len(retrieved_docs) 0: return {query: user_query, context: , answer: 未找到相关文档。} # 2. 重排可选 if self.reranker isnotNone: print(f\n步骤 2: 重排Top-{rerank_top_k}...) doc_texts [doc[text] for doc in retrieved_docs] reranked_texts self.rerank(user_query, doc_texts, top_krerank_top_k) # 更新文档列表只保留重排后的文档 reranked_docs [ doc for doc in retrieved_docs if doc[text] in reranked_texts ] # 按重排顺序重新排序 text_to_doc {doc[text]: doc for doc in reranked_docs} reranked_docs [text_to_doc[text] for text in reranked_texts] print(f✓ 重排完成返回 Top-{len(reranked_docs)} 个文档) else: print(f\n步骤 2: 跳过重排未配置Reranker...) reranked_docs retrieved_docs[:rerank_top_k] print(f✓ 使用检索结果返回 Top-{len(reranked_docs)} 个文档) # 3. 位置优化构建上下文 print(f\n步骤 3: 位置优化构建上下文...) context self.build_context(user_query, reranked_docs) print(f✓ 上下文构建完成长度{len(context)} 字符) # 4. 生成答案这里只返回上下文实际应用中会调用LLM print(f\n步骤 4: 上下文已准备就绪\n) return { query: user_query, retrieved_docs: retrieved_docs[:5], # 只返回前5个用于展示 reranked_docs: reranked_docs, context: context, answer: 【提示】未配置LLM仅返回上下文。\n\n context }5.4 效果对比位置优化带来的效果提升指标优化前中间位置优化后开头位置提升准确率53.8%75.8%22%信息利用率低高显著提升关键结论位置优化是突破 U 型陷阱的核心步骤生产环境必须包含六、性能优化实战6.1 索引选择Milvus 支持多种索引类型选择合适的索引对性能至关重要索引类型特点适用场景推荐参数HNSW高精度、低延迟高精度要求、低延迟需求M16, ef64IVF_FLAT成本友好大规模数据、成本敏感nlist1024, nprobe10IVF_PQ压缩存储超大规模数据nlist1024, m8DiskANN大规模场景百亿级向量适合磁盘存储代码示例使用 HNSW 索引# 注意MilvusClient方式创建集合时索引参数在create_collection中设置# 如果需要更精细的索引控制可以使用pymilvus的Collection方式from pymilvus import Collection, connections, FieldSchema, CollectionSchema, DataType# 连接Milvus Server不是Liteconnections.connect(aliasdefault, hostlocalhost, port19530)# 定义Schemafields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim384), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length1000),]schema CollectionSchema(fieldsfields, descriptionDocuments collection)collection Collection(namedocuments, schemaschema)# 创建HNSW索引index_params { metric_type: L2, index_type: HNSW, params: { M: 16, # 每个节点的连接数 efConstruction: 200# 构建时的搜索范围 }}collection.create_index(field_namevector, index_paramsindex_params)# 搜索时设置ef参数search_params {metric_type: L2, params: {ef: 64}}6.2 批量处理对于大量查询使用批量处理可以显著提升吞吐量def batch_search(self, queries: List[str], batch_size: int 32): 批量检索 all_results [] # 批量编码 query_vectors self.encoder.encode(queries, normalize_embeddingsTrue) # 批量搜索 for i in range(0, len(query_vectors), batch_size): batch_vectors query_vectors[i:ibatch_size] batch_queries queries[i:ibatch_size] results self.client.search( collection_nameself.collection_name, databatch_vectors.tolist(), limit10, search_params{metric_type: L2, params: {}}, output_fields[text] ) all_results.extend(results) return all_results6.3 缓存策略实现查询缓存可以大幅降低延迟和成本from functools import lru_cacheimport hashlibimport jsonclass CachedRAGSystem(OptimizedRAGSystem): 带缓存的RAG系统 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.cache {} # 简单的内存缓存生产环境建议使用Redis def _get_cache_key(self, query: str, top_k: int) - str: 生成缓存键 key_data f{query}_{top_k} return hashlib.md5(key_data.encode()).hexdigest() def retrieve(self, query: str, top_k: int 10, use_cache: bool True): 带缓存的检索 if use_cache: cache_key self._get_cache_key(query, top_k) if cache_key in self.cache: print(f✓ 使用缓存结果) return self.cache[cache_key] # 执行检索 results super().retrieve(query, top_k) # 存入缓存 if use_cache: self.cache[cache_key] results return results6.4 异步处理对于高并发场景使用异步处理可以提升性能import asynciofrom concurrent.futures import ThreadPoolExecutorclass AsyncRAGSystem(OptimizedRAGSystem): 异步RAG系统 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.executor ThreadPoolExecutor(max_workers4) asyncdef async_retrieve(self, query: str, top_k: int 10): 异步检索 loop asyncio.get_event_loop() results await loop.run_in_executor( self.executor, self.retrieve, query, top_k ) return results asyncdef async_query(self, user_query: str, retrieve_top_k: int 100, rerank_top_k: int 10): 异步查询 # 异步检索 retrieved_docs await self.async_retrieve(user_query, top_kretrieve_top_k) # 其他步骤可以继续异步化... # 这里简化处理 return { query: user_query, documents: retrieved_docs }# 使用示例asyncdef main(): rag AsyncRAGSystem() result await rag.async_query(What is Milvus?) print(result)# asyncio.run(main())七、生产环境最佳实践7.1 错误处理与重试生产环境必须包含完善的错误处理import timefrom typing import Optionalclass ProductionRAGSystem(OptimizedRAGSystem): 生产环境RAG系统带错误处理和重试 def retrieve_with_retry( self, query: str, top_k: int 10, max_retries: int 3, retry_delay: float 1.0 ): 带重试的检索 for attempt in range(max_retries): try: return self.retrieve(query, top_k) except Exception as e: if attempt max_retries - 1: raise# 最后一次重试失败抛出异常 print(f检索失败尝试 {attempt 1}/{max_retries}{e}) time.sleep(retry_delay * (attempt 1)) # 指数退避 return [] def query(self, user_query: str, **kwargs): 生产环境查询带错误处理 try: return super().query(user_query, **kwargs) except Exception as e: return { query: user_query, error: str(e), context: , answer: 抱歉查询过程中出现错误请稍后重试。 }7.2 日志记录添加详细的日志记录便于监控和调试import loggingfrom datetime import datetimeclass LoggedRAGSystem(ProductionRAGSystem): 带日志的RAG系统 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(rag_system.log), logging.StreamHandler() ] ) self.logger logging.getLogger(__name__) def query(self, user_query: str, **kwargs): 带日志的查询 start_time datetime.now() self.logger.info(f开始查询: {user_query}) try: result super().query(user_query, **kwargs) elapsed_time (datetime.now() - start_time).total_seconds() self.logger.info( f查询完成: {user_query}, f耗时: {elapsed_time:.2f}秒, f检索到: {len(result.get(retrieved_docs, []))} 个文档 ) return result except Exception as e: self.logger.error(f查询失败: {user_query}, 错误: {e}) raise7.3 监控指标收集关键性能指标from collections import defaultdictimport timeclass MonitoredRAGSystem(LoggedRAGSystem): 带监控的RAG系统 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.metrics { total_queries: 0, total_retrieval_time: 0, total_rerank_time: 0, total_query_time: 0, errors: 0, } def query(self, user_query: str, **kwargs): 带监控的查询 self.metrics[total_queries] 1 start_time time.time() try: # 监控检索时间 retrieval_start time.time() retrieved_docs self.retrieve(user_query, top_kkwargs.get(retrieve_top_k, 100)) self.metrics[total_retrieval_time] time.time() - retrieval_start # 监控重排时间 if self.reranker: rerank_start time.time() # ... 重排逻辑 self.metrics[total_rerank_time] time.time() - rerank_start result super().query(user_query, **kwargs) self.metrics[total_query_time] time.time() - start_time return result except Exception as e: self.metrics[errors] 1 raise def get_metrics(self): 获取监控指标 total self.metrics[total_queries] if total 0: return {} return { total_queries: total, avg_retrieval_time: self.metrics[total_retrieval_time] / total, avg_rerank_time: self.metrics[total_rerank_time] / total, avg_query_time: self.metrics[total_query_time] / total, error_rate: self.metrics[errors] / total, }7.4 配置管理使用配置文件管理参数# config.yamlmilvus:uri:./milvus_demo.dbcollection_name:documentsembedding:model:sentence-transformers/all-MiniLM-L6-v2dimension:384reranker:enabled:truemodel:BAAI/bge-reranker-baseretrieval:top_k:100rerank_top_k:10position_optimization:enabled:truetop_relevant:3secondary_relevant:5 plaintext import yamlclass ConfigurableRAGSystem(OptimizedRAGSystem): 可配置的RAG系统 def __init__(self, config_path: str config.yaml): # 加载配置 with open(config_path, r) as f: config yaml.safe_load(f) # 使用配置初始化 super().__init__( milvus_uriconfig[milvus][uri], collection_nameconfig[milvus][collection_name] ) self.config config八、常见问题与解决方案8.1 集合不存在问题Collection documents does not exist解决方案# 检查集合是否存在if not client.has_collection(collection_name): print(f集合 {collection_name} 不存在请先创建集合并插入数据) # 创建集合 client.create_collection(...) # 插入数据 client.insert(...)8.2 向量维度不匹配问题Dimension mismatch: expected 384, got 1024解决方案# 确保Embedding模型维度与集合维度一致embedding_model sentence-transformers/all-MiniLM-L6-v2 # 384维# 或embedding_model BAAI/bge-large-en-v1.5 # 1024维# 创建集合时使用正确的维度dimension 384 if all-MiniLM-L6-v2 in embedding_model else 1024client.create_collection(..., dimensiondimension)8.3 内存不足问题加载大模型时内存不足解决方案使用更小的模型# 使用小模型80MB vs 1.3GBencoder SentenceTransformer(sentence-transformers/all-MiniLM-L6-v2)延迟加载class LazyRAGSystem: def __init__(self): self._encoder None self._reranker None property def encoder(self): if self._encoder is None: self._encoder SentenceTransformer(...) return self._encoder使用 CPU 模式如果 GPU 内存不足encoder SentenceTransformer(..., devicecpu)8.4 检索结果不准确问题检索到的文档与查询不相关解决方案使用更好的 Embedding 模型# 从all-MiniLM-L6-v2升级到BGE-largeencoder SentenceTransformer(BAAI/bge-large-en-v1.5)添加重排# 使用重排模型提升准确性reranker AutoModelForSequenceClassification.from_pretrained(BAAI/bge-reranker-base)调整检索参数# 增加检索数量然后重排retrieved_docs self.retrieve(query, top_k200) # 增加检索数量reranked_docs self.rerank(query, retrieved_docs, top_k10) # 重排后取Top-108.5 性能优化问题查询速度慢解决方案使用合适的索引# HNSW索引适合低延迟场景index_params { index_type: HNSW, params: {M: 16, efConstruction: 200}}批量处理# 批量查询比单个查询更高效results batch_search(queries, batch_size32)缓存结果# 缓存常见查询的结果cached_results cache.get(query)if cached_results: return cached_results九、总结本文介绍了如何从零开始基于 Milvus 构建高性能 RAG 系统涵盖了数据准备、向量检索、结果重排、位置优化等核心环节。关键要点✅位置优化是关键通过将最相关文档放在开头可以突破 U 型陷阱提升 22%的准确率✅重排提升准确性使用 BGE-Reranker 等重排模型可以显著提升检索效果✅索引选择很重要根据数据规模和性能需求选择合适的索引类型HNSW、IVF 等✅生产环境需完善错误处理、日志记录、性能监控是生产环境必备功能希望本文能帮助你构建高性能的 RAG 系统。更多技术细节和代码示例请参考文中各章节内容。最后唠两句为什么AI大模型成为越来越多程序员转行就业、升职加薪的首选很简单这些岗位缺人且高薪智联招聘的最新数据给出了最直观的印证2025年2月AI领域求职人数同比增幅突破200% 远超其他行业平均水平整个人工智能行业的求职增速达到33.4%位居各行业榜首其中人工智能工程师岗位的求职热度更是飙升69.6%。AI产业的快速扩张也让人才供需矛盾愈发突出。麦肯锡报告明确预测到2030年中国AI专业人才需求将达600万人人才缺口可能高达400万人这一缺口不仅存在于核心技术领域更蔓延至产业应用的各个环节。那0基础普通人如何学习大模型 深耕科技一线十二载亲历技术浪潮变迁。我见证那些率先拥抱AI的同行如何建立起效率与薪资的代际优势。如今我将积累的大模型面试真题、独家资料、技术报告与实战路线系统整理分享于此为你扫清学习困惑共赴AI时代新程。我整理出这套 AI 大模型突围资料包【允许白嫖】✅从入门到精通的全套视频教程✅AI大模型学习路线图0基础到项目实战仅需90天✅大模型书籍与技术文档PDF✅各大厂大模型面试题目详解✅640套AI大模型报告合集✅大模型入门实战训练这份完整版的大模型 AI 学习和面试资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】①从入门到精通的全套视频教程包含提示词工程、RAG、Agent等技术点② AI大模型学习路线图0基础到项目实战仅需90天全过程AI大模型学习路线③学习电子书籍和技术文档市面上的大模型书籍确实太多了这些是我精选出来的④各大厂大模型面试题目详解⑤640套AI大模型报告合集⑥大模型入门实战训练如果说你是以下人群中的其中一类都可以来智泊AI学习人工智能找到高薪工作一次小小的“投资”换来的是终身受益应届毕业生‌无工作经验但想要系统学习AI大模型技术期待通过实战项目掌握核心技术。零基础转型‌非技术背景但关注AI应用场景计划通过低代码工具实现“AI行业”跨界‌。业务赋能 ‌突破瓶颈传统开发者Java/前端等学习Transformer架构与LangChain框架向AI全栈工程师转型‌。获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】