RAG 技术选型决策树：根据场景选 Embedding、向量库和检索策略

RAG 技术选型决策树根据场景选 Embedding、向量库和检索策略一、从能检索到检索准RAG 技术选型的困境2026 年上半年某法律 AI 平台的 RAG 系统上线后律师们抱怨检索出来的案例跟问题不相关。技术团队尝试了各种办法换了 3 个 Embedding 模型、调整了 5 种分块策略、试了 2 个向量数据库但效果依然不理想。最终发现问题根源技术选型跟场景不匹配。法律文书的检索需要高精度应该用专门的 legal Embedding 模型 混合检索而不是通用的模型 纯向量检索。本文将给出 RAG 技术选型的完整决策树帮助你在不同场景下做出正确选择。二、选型维度一Embedding 模型选择决策因素核心考量语言中文、英文、多语言领域通用、法律、医疗、代码向量维度维度越高精度越高但存储和计算成本也越高推理速度实时检索需要快主流模型对比from typing import Dict, List import numpy as np class EmbeddingModelComparator: Embedding 模型对比 MODELS { text-embedding-3-small: { provider: OpenAI, dimensions: 1536, languages: [多语言], cost_per_1m_tokens: 0.02, # USD speed: 快, accuracy: 高, best_for: 通用场景、快速原型 }, text-embedding-3-large: { provider: OpenAI, dimensions: 3072, languages: [多语言], cost_per_1m_tokens: 0.13, speed: 中, accuracy: 很高, best_for: 高精度要求 }, bge-large-zh-v1.5: { provider: BAAI, dimensions: 1024, languages: [中文], cost_per_1m_tokens: 0, # 开源 speed: 中, accuracy: 高中文, best_for: 中文场景、成本敏感 }, BCEmbedding: { provider: 网易有道, dimensions: 768, languages: [中英双语], cost_per_1m_tokens: 0, speed: 快, accuracy: 高双语, best_for: 中英混合场景 }, voyage-2: { provider: Voyage AI, dimensions: 1024, languages: [多语言], cost_per_1m_tokens: 0.10, speed: 快, accuracy: 高, best_for: RAG 专门优化 } } def recommend(self, scenario: Dict) - List[str]: 根据场景推荐模型 recommendations [] # 规则 1: 语言 if scenario.get(language) 中文: recommendations.append(bge-large-zh-v1.5) recommendations.append(BCEmbedding) elif scenario.get(language) 多语言: recommendations.append(text-embedding-3-small) recommendations.append(voyage-2) # 规则 2: 成本 if scenario.get(cost_sensitive): recommendations [m for m in recommendations if self.MODELS[m][cost_per_1m_tokens] 0] # 规则 3: 精度 if scenario.get(high_accuracy): recommendations sorted( recommendations, keylambda m: self.MODELS[m][dimensions], reverseTrue ) return recommendations # 实测对比 def benchmark_embedding_models(): 对比不同 Embedding 模型的效果 test_cases [ { query: 如何申请退款, relevant: 退款政策用户可以在订单完成后 7 天内申请退款..., irrelevant: 我们的产品支持多种支付方式包括支付宝、微信支付... }, # 更多测试用例... ] models_to_test [text-embedding-3-small, bge-large-zh-v1.5] for model_name in models_to_test: print(f\n测试模型: {model_name}) correct 0 for test in test_cases: # 计算相似度 query_emb get_embedding(test[query], model_name) relevant_emb get_embedding(test[relevant], model_name) irrelevant_emb get_embedding(test[irrelevant], model_name) # 计算余弦相似度 sim_relevant cosine_similarity(query_emb, relevant_emb) sim_irrelevant cosine_similarity(query_emb, irrelevant_emb) if sim_relevant sim_irrelevant: correct 1 accuracy correct / len(test_cases) print(f准确率: {accuracy:.2%}) def get_embedding(text: str, model: str) - np.ndarray: 获取文本 Embedding简化 # 实际应调用对应 API if model text-embedding-3-small: import openai response openai.embeddings.create(inputtext, modelmodel) return np.array(response.data[0].embedding) elif model bge-large-zh-v1.5: from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-large-zh-v1.5) return model.encode(text) # ... def cosine_similarity(v1: np.ndarray, v2: np.ndarray) - float: 余弦相似度 return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))生产级实现动态选择 Embedding 模型from abc import ABC, abstractmethod from typing import List, Optional class EmbeddingModel(ABC): Embedding 模型抽象 abstractmethod def encode(self, texts: List[str], **kwargs) - np.ndarray: 编码文本为向量 pass abstractmethod def get_dimension(self) - int: 返回向量维度 pass class OpenAIEmbedding(EmbeddingModel): OpenAI Embedding 模型 def __init__(self, model: str text-embedding-3-small): self.model model import openai self.client openai.Client() def encode(self, texts: List[str], **kwargs) - np.ndarray: response self.client.embeddings.create( inputtexts, modelself.model ) return np.array([d.embedding for d in response.data]) def get_dimension(self) - int: return 1536 if self.model text-embedding-3-small else 3072 class BGEEmbedding(EmbeddingModel): BGE Embedding 模型开源 def __init__(self, model_name: str BAAI/bge-large-zh-v1.5): from sentence_transformers import SentenceTransformer self.model SentenceTransformer(model_name) def encode(self, texts: List[str], **kwargs) - np.ndarray: return self.model.encode(texts, **kwargs) def get_dimension(self) - int: return 1024 class EmbeddingFactory: Embedding 模型工厂 staticmethod def create(scenario: Dict) - EmbeddingModel: 根据场景创建模型 language scenario.get(language, 英文) cost_sensitive scenario.get(cost_sensitive, False) high_accuracy scenario.get(high_accuracy, False) if language 中文 and cost_sensitive: return BGEEmbedding(BAAI/bge-large-zh-v1.5) elif language 中文 and high_accuracy: return OpenAIEmbedding(text-embedding-3-large) elif language 多语言: return OpenAIEmbedding(text-embedding-3-small) else: # 默认 return BGEEmbedding(BAAI/bge-large-zh-v1.5)三、选型维度二向量数据库选择决策因素核心考量数据规模100万、100万-1亿、1亿查询 QPS100、100-1000、1000成本开源 vs 商业运维复杂度自部署 vs 云服务主流向量库对比生产级实现统一接口from abc import ABC, abstractmethod from typing import List, Tuple class VectorDB(ABC): 向量数据库抽象接口 abstractmethod def create_index(self, dimension: int): 创建索引 pass abstractmethod def add_vectors(self, ids: List[str], vectors: np.ndarray, metadata: List[Dict]): 添加向量 pass abstractmethod def search(self, query_vector: np.ndarray, top_k: int 10) - List[Tuple[str, float]]: 检索 pass class FAISSVectorDB(VectorDB): FAISS 向量库适合小规模 def __init__(self, index_path: Optional[str] None): import faiss self.index None self.id_map {} # ID 映射 self.metadata [] # 元数据 def create_index(self, dimension: int): import faiss # 使用 IndexFlatL2精确搜索或 IndexIVFFlat近似搜索 self.index faiss.IndexFlatL2(dimension) def add_vectors(self, ids: List[str], vectors: np.ndarray, metadata: List[Dict]): if self.index is None: self.create_index(vectors.shape[1]) # 添加向量 self.index.add(vectors.astype(np.float32)) # 保存 ID 映射和元数据 start_id len(self.id_map) for i, (id_str, meta) in enumerate(zip(ids, metadata)): self.id_map[start_id i] id_str self.metadata.append(meta) def search(self, query_vector: np.ndarray, top_k: int 10) - List[Tuple[str, float]]: distances, indices self.index.search( query_vector.astype(np.float32).reshape(1, -1), top_k ) results [] for i, (dist, idx) in enumerate(zip(distances[0], indices[0])): if idx ! -1: id_str self.id_map.get(int(idx), str(idx)) results.append((id_str, float(dist))) return results class MilvusVectorDB(VectorDB): Milvus 向量库适合中大规模 def __init__(self, host: str localhost, port: int 19530): from pymilvus import connections, Collection connections.connect(hosthost, portport) self.collection None def create_index(self, dimension: int): from pymilvus import Collection, FieldSchema, CollectionSchema, DataType fields [ FieldSchema(nameid, dtypeDataType.VARCHAR, is_primaryTrue, max_length100), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimdimension), FieldSchema(namemetadata, dtypeDataType.JSON) ] schema CollectionSchema(fields, descriptionRAG vectors) self.collection Collection(namerag_vectors, schemaschema) # 创建向量索引 self.collection.create_index( field_namevector, index_params{ metric_type: L2, index_type: IVF_FLAT, params: {nlist: 1024} } ) def add_vectors(self, ids: List[str], vectors: np.ndarray, metadata: List[Dict]): if self.collection is None: self.create_index(vectors.shape[1]) # 插入数据 self.collection.insert([ ids, vectors.tolist(), metadata ]) def search(self, query_vector: np.ndarray, top_k: int 10) - List[Tuple[str, float]]: self.collection.load() results self.collection.search( data[query_vector.tolist()], anns_fieldvector, param{metric_type: L2, params: {nprobe: 10}}, limittop_k ) return [(hit.id, hit.distance) for hit in results[0]] class VectorDBFactory: 向量库工厂 staticmethod def create(config: Dict) - VectorDB: 根据配置创建向量库 data_size config.get(data_size, 0) qps config.get(qps, 10) if data_size 1_000_000: return FAISSVectorDB() elif data_size 100_000_000: return MilvusVectorDB() else: # 大规模推荐云服务 raise ValueError(Data size too large, consider using managed service)四、选型维度三检索策略选择策略对比生产级实现混合检索from typing import List, Dict import re class HybridRetriever: 混合检索器 def __init__(self, vector_db: VectorDB, keyword_index: Dict): self.vector_db vector_db self.keyword_index keyword_index # 简化实际应该用 Elasticsearch def retrieve(self, query: str, top_k: int 10) - List[Dict]: 混合检索 # 1. 向量检索 query_vector self._embed(query) vector_results self.vector_db.search(query_vector, top_ktop_k) # 2. 关键词检索BM25 keyword_results self._bm25_search(query, top_ktop_k) # 3. 结果融合Reciprocal Rank Fusion merged self._rrf_merge(vector_results, keyword_results) # 4. 重排序 reranked self._rerank(query, merged[:top_k*2]) return reranked[:top_k] def _embed(self, text: str) - np.ndarray: 文本向量化 # 使用前面定义的 Embedding 模型 pass def _bm25_search(self, query: str, top_k: int) - List[Dict]: BM25 关键词检索 # 简化实现 query_tokens query.lower().split() scores {} for doc_id, doc_tokens in self.keyword_index.items(): score self._calculate_bm25(query_tokens, doc_tokens) scores[doc_id] score # 排序 sorted_results sorted(scores.items(), keylambda x: x[1], reverseTrue) return [{id: doc_id, score: score} for doc_id, score in sorted_results[:top_k]] def _calculate_bm25(self, query_tokens: List[str], doc_tokens: List[str]) - float: 计算 BM25 分数简化 score 0.0 doc_token_set set(doc_tokens) for token in query_tokens: if token in doc_token_set: score 1.0 # 简化实际应该用 BM25 公式 return score def _rrf_merge(self, list1: List[Dict], list2: List[Dict], k: int 60) - List[Dict]: Reciprocal Rank Fusion 融合 scores {} for rank, result in enumerate(list1, 1): doc_id result[id] if doc_id not in scores: scores[doc_id] {id: doc_id, score: 0} scores[doc_id][score] 1 / (k rank) for rank, result in enumerate(list2, 1): doc_id result[id] if doc_id not in scores: scores[doc_id] {id: doc_id, score: 0} scores[doc_id][score] 1 / (k rank) # 排序 merged sorted(scores.values(), keylambda x: x[score], reverseTrue) return merged def _rerank(self, query: str, results: List[Dict]) - List[Dict]: 重排序使用交叉编码器 try: from sentence_transformers import CrossEncoder model CrossEncoder(BAAI/bge-reranker-v1.5) pairs [[query, self._get_doc_text(r[id])] for r in results] rerank_scores model.predict(pairs) # 更新分数 for result, score in zip(results, rerank_scores): result[score] float(score) # 重新排序 results.sort(keylambda x: x[score], reverseTrue) except Exception as e: print(fRerank failed: {e}, skipping) return results def _get_doc_text(self, doc_id: str) - str: 获取文档文本简化 # 实际应该从数据库获取 return fDocument {doc_id} content...五、总结RAG 技术选型决策树Embedding 模型选择中文场景 成本敏感 → BGE / BCE多语言 快速原型 → OpenAI text-embedding-3-small高精度要求 → OpenAI text-embedding-3-large专业领域 → 领域微调模型向量数据库选择 100万 → FAISS100万 - 1亿 → Milvus / Qdrant1亿 → Elasticsearch / 商业云服务检索策略选择高精度优先 → 混合检索 重排序速度优先 → 向量检索IVFFlat 索引简单场景 → 向量检索 Top-5完整选型检查表确定语言和领域选 Embedding 模型评估数据规模和 QPS选向量库确定精度和速度要求选检索策略做 A/B 测试验证效果性能优化建议Embedding 模型批量推理提升 5-10x 速度向量库使用 IVF 索引牺牲 5% 精度换取 10x 速度检索结果缓存Semantic Cache记住没有最好的技术只有最适合场景的技术组合。下一篇我们将深入探讨监控工具选型的终极对比。