AI问答系统崩溃？这篇RAG优化实战指南，教你解决90%的检索问题

import openaiimport jsonimport redef detect_intent(query):    """使用大模型进行意图识别"""    prompt = f"""用户Query: "{query}"  请从[流程,材料,资格,费用,其他]中选择最相关的意图标签，若无匹配则输出"其他"输出格式: JSON {{"intent": "标签"}}"""        try:        response = openai.chat.completions.create(            model="gpt-3.5-turbo",            messages=[{"role": "user", "content": prompt}],            temperature=0        )                result = response.choices[0].message.content        # 提取JSON格式的结果        json_match = re.search(r'\{.*\}', result)        if json_match:            intent_data = json.loads(json_match.group())            return intent_data.get("intent", "其他")        else:            return "其他"    except Exception as e:        print(f"意图识别出错: {e}")        return "其他"# 测试意图识别query = "如何办信用卡？"intent = detect_intent(query)print(f"识别意图: {intent}")

def expand_query(query, intent):    """根据意图扩展查询"""    expansion_rules = {        "流程": "信用卡申请的具体步骤和办理流程",        "材料": "申请信用卡需要提供的文件和材料清单",         "资格": "信用卡申请者需要满足的条件和要求",        "费用": "信用卡年费、手续费等费用标准"    }        # 根据识别的意图扩展查询    expanded_query = expansion_rules.get(intent, query)    return expanded_query# 完整的查询转换流程def transform_query(original_query):    """完整的查询转换流程"""    # 步骤1: 意图识别    intent = detect_intent(original_query)        # 步骤2: 查询扩展    expanded_query = expand_query(original_query, intent)        print(f"原始查询: {original_query}")    print(f"识别意图: {intent}")    print(f"扩展查询: {expanded_query}")        return expanded_query, intent# 测试完整流程result = transform_query("如何办信用卡？")

import numpy as npfrom sklearn.feature_extraction.text import TfidfVectorizerfrom sklearn.metrics.pairwise import cosine_similarityfrom sentence_transformers import SentenceTransformerimport jiebafrom collections import defaultdictclass HybridRetriever:    def __init__(self, documents):        self.documents = documents        self.setup_vector_search()        self.setup_keyword_search()        def setup_vector_search(self):        """初始化向量检索"""        # 使用sentence-transformers进行向量化        self.vector_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')        self.doc_embeddings = self.vector_model.encode(self.documents)        def setup_keyword_search(self):        """初始化关键词检索（使用TF-IDF模拟BM25）"""        # 中文分词处理        tokenized_docs = [' '.join(jieba.cut(doc)) for doc in self.documents]        self.tfidf_vectorizer = TfidfVectorizer()        self.tfidf_matrix = self.tfidf_vectorizer.fit_transform(tokenized_docs)        def vector_search(self, query, top_k=50):        """向量检索"""        query_embedding = self.vector_model.encode([query])        similarities = cosine_similarity(query_embedding, self.doc_embeddings)[0]                # 获取top_k个最相似的文档        top_indices = np.argsort(similarities)[::-1][:top_k]        results = []        for idx in top_indices:            results.append({                'doc_id': idx,                'content': self.documents[idx],                'score': similarities[idx],                'method': 'vector'            })        return results        def keyword_search(self, query, top_k=50):        """关键词检索（TF-IDF）"""        query_tokens = ' '.join(jieba.cut(query))        query_vector = self.tfidf_vectorizer.transform([query_tokens])        similarities = cosine_similarity(query_vector, self.tfidf_matrix)[0]                # 获取top_k个最相似的文档        top_indices = np.argsort(similarities)[::-1][:top_k]        results = []        for idx in top_indices:            if similarities[idx] > 0:  # 只返回有匹配的文档                results.append({                    'doc_id': idx,                    'content': self.documents[idx],                    'score': similarities[idx],                    'method': 'keyword'                })        return results        def deduplicate_and_merge(self, vector_results, keyword_results):        """去重合并结果"""        doc_scores = defaultdict(list)                # 收集所有文档的分数        for result in vector_results + keyword_results:            doc_scores[result['doc_id']].append(result)                # 合并分数（取平均值）        merged_results = []        for doc_id, results in doc_scores.items():            if len(results) == 1:                merged_results.append(results[0])            else:                # 多种方法都检索到的文档，合并分数                avg_score = sum(r['score'] for r in results) / len(results)                methods = '+'.join(set(r['method'] for r in results))                merged_results.append({                    'doc_id': doc_id,                    'content': results[0]['content'],                    'score': avg_score * 1.2,  # 多方法匹配加权                    'method': methods                })                # 按分数排序        merged_results.sort(key=lambda x: x['score'], reverse=True)        return merged_results        def hybrid_retrieve(self, query, k=50):        """混合检索主函数"""        # 并行执行两种检索        vector_results = self.vector_search(query, top_k=k)        keyword_results = self.keyword_search(query, top_k=k)                # 去重合并结果        all_results = self.deduplicate_and_merge(vector_results, keyword_results)        return all_results[:k]# 使用示例documents = [    "信用卡申请需要提供身份证、收入证明、工作证明等材料",    "申请信用卡的基本流程：填写申请表、提交材料、银行审核、制卡邮寄",    "信用卡申请条件：年满18周岁、有稳定收入、信用记录良好",    "白金信用卡年费标准：普通卡免年费，金卡200元，白金卡600元",    "网上申请信用卡更方便，审批速度快，通常7-15个工作日"]# 初始化检索器retriever = HybridRetriever(documents)# 执行检索query = "信用卡申请的具体步骤和办理流程"results = retriever.hybrid_retrieve(query, k=3)print("检索结果：")for i, result in enumerate(results, 1):    print(f"{i}. [{result['method']}] 分数: {result['score']:.3f}")    print(f"   内容: {result['content']}")    print()

import torchfrom transformers import AutoTokenizer, AutoModelForSequenceClassificationimport numpy as npclass CrossEncoderReranker:    def __init__(self, model_name='BAAI/bge-reranker-large'):        """初始化重排模型"""        try:            self.tokenizer = AutoTokenizer.from_pretrained(model_name)            self.model = AutoModelForSequenceClassification.from_pretrained(model_name)            self.model.eval()            print(f"成功加载模型: {model_name}")        except:            # 如果无法加载BGE模型，使用备用方案            print("无法加载BGE模型，使用备用重排方案")            self.use_fallback = True        def compute_score(self, query_doc_pairs):        """计算查询-文档对的相关性分数"""        if hasattr(self, 'use_fallback'):            return self._fallback_rerank(query_doc_pairs)                scores = []        with torch.no_grad():            for query, doc in query_doc_pairs:                # 构造输入                inputs = self.tokenizer(                    query, doc,                     return_tensors='pt',                    truncation=True,                    max_length=512,                    padding=True                )                                # 前向传播                outputs = self.model(**inputs)                score = torch.sigmoid(outputs.logits).item()                scores.append(score)                return scores        def _fallback_rerank(self, query_doc_pairs):        """备用重排方案：基于关键词匹配"""        scores = []        for query, doc in query_doc_pairs:            # 简单的关键词匹配评分            query_words = set(jieba.cut(query.lower()))            doc_words = set(jieba.cut(doc.lower()))                        # 计算交集比例            intersection = len(query_words & doc_words)            union = len(query_words | doc_words)                        # 避免除零错误            score = intersection / union if union > 0 else 0            scores.append(score)                return scores        def rerank(self, query, candidate_docs, top_k=5):        """重排文档并返回topk结果"""        if not candidate_docs:            return []                # 构造查询-文档对        query_doc_pairs = [[query, doc['content']] for doc in candidate_docs]                # 计算相关性分数        scores = self.compute_score(query_doc_pairs)                # 更新文档分数        for i, doc in enumerate(candidate_docs):            doc['rerank_score'] = scores[i]                # 按重排分数排序        reranked_docs = sorted(candidate_docs, key=lambda x: x['rerank_score'], reverse=True)                return reranked_docs[:top_k]# 完整的RAG检索pipelineclass RAGPipeline:    def __init__(self, documents):        self.retriever = HybridRetriever(documents)        self.reranker = CrossEncoderReranker()        def search(self, query, top_k=5):        """完整的RAG检索流程"""        print(f"开始处理查询: {query}")                # 步骤1: 查询转换        expanded_query, intent = transform_query(query)                # 步骤2: 混合检索        print("执行混合检索...")        candidate_docs = self.retriever.hybrid_retrieve(expanded_query, k=20)        print(f"召回文档数量: {len(candidate_docs)}")                # 步骤3: 智能重排        print("执行智能重排...")        final_results = self.reranker.rerank(expanded_query, candidate_docs, top_k=top_k)                return final_results, intent# 使用示例# 初始化完整pipelinerag_pipeline = RAGPipeline(documents)# 执行完整检索query = "如何办信用卡？"results, intent = rag_pipeline.search(query, top_k=3)print("\n=== 最终检索结果 ===")for i, result in enumerate(results, 1):    print(f"{i}. 重排分数: {result['rerank_score']:.3f}")    print(f"   检索方法: {result['method']}")    print(f"   内容: {result['content']}")    print()

def dynamic_k_adjustment(query, base_k=20):    """根据查询复杂度动态调整K值"""    # 根据查询长度调整    query_length_factor = len(query.split())        # 根据查询中的关键词数量调整    keywords = list(jieba.cut(query))    keyword_factor = len([w for w in keywords if len(w) > 1])        # 计算动态K值    dynamic_k = min(100, max(10, base_k + query_length_factor * 3 + keyword_factor * 2))        print(f"查询: {query}")    print(f"动态K值: {dynamic_k} (基础K: {base_k}, 长度因子: {query_length_factor}, 关键词因子: {keyword_factor})")        return dynamic_k# 测试动态K值调整test_queries = [    "信用卡",  # 简单查询    "如何申请信用卡",  # 中等复杂度    "银行信用卡申请需要什么材料和条件"  # 复杂查询]for query in test_queries:    k = dynamic_k_adjustment(query)    print("-" * 50)

def multi_granularity_chunking(text, large_chunk_size=1024, small_chunk_size=256, overlap_ratio=0.1):    """多粒度文档分块"""    chunks = []        # 大块分割（保留逻辑连贯性）    large_overlap = int(large_chunk_size * overlap_ratio)    large_chunks = []    for i in range(0, len(text), large_chunk_size - large_overlap):        chunk = text[i:i + large_chunk_size]        if len(chunk.strip()) > 100:  # 过滤太短的块            large_chunks.append({                'content': chunk,                'type': 'large',                'size': len(chunk),                'start_pos': i            })        # 小块分割（提升定位精度）    small_overlap = int(small_chunk_size * overlap_ratio)    small_chunks = []    for i in range(0, len(text), small_chunk_size - small_overlap):        chunk = text[i:i + small_chunk_size]        if len(chunk.strip()) > 50:  # 过滤太短的块            small_chunks.append({                'content': chunk,                'type': 'small',                'size': len(chunk),                'start_pos': i            })        return large_chunks + small_chunks# 业务规则注入示例def apply_business_rules(query, docs):    """在重排阶段加入业务规则"""    enhanced_docs = []        for doc in docs:        enhanced_doc = doc.copy()        original_score = doc.get('rerank_score', doc.get('score', 0))                # 规则1: 关键业务词组合加权        if "年费" in query and "白金卡" in doc['content']:            enhanced_doc['rerank_score'] = original_score * 1.5            enhanced_doc['boost_reason'] = "业务关键词匹配"                # 规则2: 流程类问题优先级        elif "流程" in query or "步骤" in query:            if any(word in doc['content'] for word in ["第一步", "首先", "然后", "最后"]):                enhanced_doc['rerank_score'] = original_score * 1.3                enhanced_doc['boost_reason'] = "流程步骤匹配"                # 规则3: 材料清单优先级        elif "材料" in query or "文件" in query:            if any(word in doc['content'] for word in ["需要", "提供", "准备"]):                enhanced_doc['rerank_score'] = original_score * 1.2                enhanced_doc['boost_reason'] = "材料要求匹配"                else:            enhanced_doc['rerank_score'] = original_score            enhanced_doc['boost_reason'] = "无规则加权"                enhanced_docs.append(enhanced_doc)        # 重新排序    enhanced_docs.sort(key=lambda x: x['rerank_score'], reverse=True)    return enhanced_docs# 测试业务规则test_query = "白金卡年费多少"test_docs = [    {'content': '普通信用卡免年费，金卡年费200元', 'rerank_score': 0.7},    {'content': '白金卡年费600元，享受更多权益', 'rerank_score': 0.6},    {'content': '申请信用卡需要身份证等材料', 'rerank_score': 0.8}]enhanced_results = apply_business_rules(test_query, test_docs)print("业务规则加权后的结果：")for i, doc in enumerate(enhanced_results, 1):    print(f"{i}. 分数: {doc['rerank_score']:.3f} ({doc['boost_reason']})")    print(f"   内容: {doc['content']}")    print()

class CompleteRAGSystem:    def __init__(self, documents):        self.documents = documents        self.retriever = HybridRetriever(documents)        self.reranker = CrossEncoderReranker()        def process_query(self, original_query, top_k=5):        """完整的RAG处理流水线"""        print("=" * 60)        print(f"处理查询: {original_query}")        print("=" * 60)                # 步骤1: 查询预处理和转换        print("步骤1: 查询转换")        expanded_query, intent = transform_query(original_query)                # 步骤2: 动态K值调整        print("\n步骤2: 动态K值调整")        dynamic_k = dynamic_k_adjustment(expanded_query)                # 步骤3: 混合检索        print("\n步骤3: 混合检索")        candidate_docs = self.retriever.hybrid_retrieve(expanded_query, k=dynamic_k)        print(f"召回候选文档: {len(candidate_docs)} 个")                # 步骤4: 智能重排        print("\n步骤4: 智能重排")        reranked_docs = self.reranker.rerank(expanded_query, candidate_docs, top_k=top_k*2)                # 步骤5: 业务规则加权        print("\n步骤5: 业务规则加权")        final_docs = apply_business_rules(original_query, reranked_docs)                # 步骤6: 返回最终结果        final_results = final_docs[:top_k]                print(f"\n最终返回文档: {len(final_results)} 个")        print("\n" + "=" * 60)        print("最终检索结果:")        print("=" * 60)                for i, doc in enumerate(final_results, 1):            print(f"{i}. 最终分数: {doc['rerank_score']:.3f}")            print(f"   检索方法: {doc.get('method', 'unknown')}")            print(f"   加权原因: {doc.get('boost_reason', '无')}")            print(f"   内容: {doc['content']}")            print(f"   内容长度: {len(doc['content'])} 字符")            print("-" * 40)                return final_results, intent# 完整系统测试def run_complete_test():    """运行完整的系统测试"""    # 初始化系统    rag_system = CompleteRAGSystem(documents)        # 测试不同类型的查询    test_queries = [        "如何办信用卡？",        "白金卡年费多少？",         "申请需要什么材料？",        "信用卡申请条件"    ]        for query in test_queries:        results, intent = rag_system.process_query(query, top_k=3)        print(f"\n查询意图: {intent}")        print("推荐答案生成基础:")        for i, doc in enumerate(results[:2], 1):  # 只显示前2个最相关的            print(f"  {i}. {doc['content']}")        print("\n" + "="*80 + "\n")# 执行完整测试if __name__ == "__main__":    # 确保安装必要的依赖    print("开始RAG系统完整测试...")    print("依赖检查: jieba, sentence-transformers, transformers, sklearn")        try:        run_complete_test()        print("✅ RAG系统测试完成!")    except Exception as e:        print(f"❌ 测试过程中出现错误: {e}")        print("请确保已安装所需依赖: pip install jieba sentence-transformers transformers scikit-learn")