2.2 检索算法基础

2.2.1 传统关键词检索方法

传统关键词检索方法是信息检索领域的基础技术，虽然现在RAG系统主要使用语义检索，但理解传统方法对于构建混合检索系统至关重要。

TF-IDF算法原理

TF-IDF（Term Frequency-Inverse Document Frequency）是最经典的文本检索算法，它通过计算词项在文档中的重要性来评估文档的相关性。

TF-IDF数学表达式：

TF-IDF(t,d) = TF(t,d) × IDF(t)
​

其中：

• TF(t,d)：词项t在文档d中的词频
• IDF(t)：词项t的逆文档频率

import math
from collections import Counter
from typing import List, Dict

def calculate_tf(word: str, document: str) -> float:
    """计算词频（Term Frequency）"""
    words = document.split()
    word_count = Counter(words)
    total_words = len(words)
    return word_count[word] / total_words

def calculate_idf(word: str, all_documents: List[str]) -> float:
    """计算逆文档频率（Inverse Document Frequency）"""
    containing_docs = sum(1 for doc in all_documents if word in doc)
    total_docs = len(all_documents)
    return math.log(total_docs / (1 + containing_docs))

def calculate_tfidf(word: str, document: str, all_documents: List[str]) -> float:
    """计算TF-IDF值"""
    tf = calculate_tf(word, document)
    idf = calculate_idf(word, all_documents)
    return tf * idf

# 使用示例
documents = [
    "机器学习是人工智能的重要分支",
    "深度学习基于神经网络",
    "自然语言处理属于AI领域"
]

query = "机器学习"
document_scores = []

for doc in documents:
    tfidf_score = calculate_tfidf(query, doc, documents)
    document_scores.append((doc, tfidf_score))

# 按TF-IDF分数排序
document_scores.sort(key=lambda x: x[1], reverse=True)
print("TF-IDF检索结果:")
for doc, score in document_scores:
    print(f"{score:.4f}: {doc}")
​

TF-IDF的优缺点分析：

优点：

• 计算简单高效
• 对文档长度不敏感
• 容易理解和实现
• 在特定领域效果良好

缺点：

• 无法理解语义相似性
• 忽略词序和语法结构
• 无法处理一词多义
• 对新词敏感度低

BM25算法改进

BM25（Best Match 25）是TF-IDF的改进版本，它通过引入参数化公式来改进检索效果。

BM25数学表达式：

Score(D,Q) = Σ IDF(qi) × (f(qi,D) × (k1 + 1)) / (f(qi,D) + k1 × (1 - b + b × |D|/avgdl))
​

其中：

• f(qi,D)：查询词qi在文档D中的频率
• k1：词频饱和参数（通常1.2-2.0）
• b：文档长度归一化参数（通常0.75）
• |D|：文档D的长度
• avgdl：平均文档长度

def calculate_bm25_score(query: str, document: str, all_documents: List[str], 
                        k1: float = 1.2, b: float = 0.75) -> float:
    """计算BM25分数"""
    words = query.split()
    doc_words = document.split()
    
    # 计算文档长度和平均长度
    avgdl = sum(len(doc.split()) for doc in all_documents) / len(all_documents)
    doc_length = len(doc_words)
    
    score = 0.0
    for word in words:
        if word in doc_words:
            # 词频
            tf = doc_words.count(word)
            
            # 逆文档频率
            containing_docs = sum(1 for doc in all_documents if word in doc)
            idf = math.log((len(all_documents) - containing_docs + 0.5) / 
                          (containing_docs + 0.5) + 1.0)
            
            # BM25公式
            numerator = tf * (k1 + 1)
            denominator = tf + k1 * (1 - b + b * doc_length / avgdl)
            
            score += idf * numerator / denominator
    
    return score

# BM25检索示例
query = "机器学习"
bm25_scores = []

for doc in documents:
    score = calculate_bm25_score(query, doc, documents)
    bm25_scores.append((doc, score))

bm25_scores.sort(key=lambda x: x[1], reverse=True)
print("BM25检索结果:")
for doc, score in bm25_scores:
    print(f"{score:.4f}: {doc}")
​

BM25的改进特点：

• 考虑了文档长度归一化
• 词频有饱和效应，避免高频词过度影响
• 参数可调，适应不同数据集
• 在传统IR任务中表现优异

布尔检索模型

布尔检索基于布尔逻辑运算，使用AND、OR、NOT等操作符组合查询条件。

def boolean_search(query: str, documents: List[Dict]) -> List[Dict]:
    """布尔检索实现"""
    result = []
    query_terms = query.split()
    
    for doc in documents:
        # 构建文档词集
        doc_terms = set(doc['content'].lower().split())
        
        # 检查查询条件
        satisfied = True
        
        # 简化的布尔逻辑处理
        # 这里实现AND逻辑
        for term in query_terms:
            if term.lower() not in doc_terms:
                satisfied = False
                break
        
        if satisfied:
            result.append(doc)
    
    return result

# 文档数据示例
documents = [
    {"id": 1, "content": "机器学习算法包括监督学习和无监督学习"},
    {"id": 2, "content": "深度学习是机器学习的重要分支"},
    {"id": 3, "content": "自然语言处理使用机器学习技术"}
]

# 布尔检索结果
query = "机器学习 学习"
results = boolean_search(query, documents)
print("布尔检索结果:")
for doc in results:
    print(f"文档{doc['id']}: {doc['content']}")
​

布尔检索的特点：

• 精确匹配，结果可解释性强
• 支持复杂的逻辑组合
• 结果可能过于严格或宽松
• 不支持语义相似性

2.2.2 语义检索的原理与优势

语义检索是现代RAG系统的核心技术，它通过向量表示和相似度计算来实现语义层面的匹配。

语义检索的基本原理

语义空间的概念

语义检索将文本映射到高维语义空间，在这个空间中，语义相似的文本在向量空间中的距离较近。

import numpy as np
from sentence_transformers import SentenceTransformer

# 加载预训练模型
model = SentenceTransformer('all-MiniLM-L6-v2')

# 示例文本
sentences = [
    "机器学习是人工智能的重要分支",
    "深度学习基于神经网络架构",
    "自然语言处理处理文本数据",
    "计算机视觉分析图像信息",
    "数据科学结合统计和机器学习"
]

# 生成句子向量
sentence_vectors = model.encode(sentences)

# 计算相似度矩阵
def cosine_similarity_matrix(vectors):
    """计算余弦相似度矩阵"""
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    normalized_vectors = vectors / norms
    return np.dot(normalized_vectors, normalized_vectors.T)

similarity_matrix = cosine_similarity_matrix(sentence_vectors)

print("语义相似度矩阵:")
print(similarity_matrix)

# 找到最相似的句子对
max_sim = 0
best_pair = (0, 1)
for i in range(len(sentences)):
    for j in range(i+1, len(sentences)):
        if similarity_matrix[i][j] > max_sim:
            max_sim = similarity_matrix[i][j]
            best_pair = (i, j)

print(f"\n最相似的句子对: {best_pair}")
print(f"相似度: {max_sim:.3f}")
print(f"句子1: {sentences[best_pair[0]]}")
print(f"句子2: {sentences[best_pair[1]]}")
​

语义检索的核心步骤：

1. 文本预处理

   • 分词和标准化
   • 停用词移除
   • 词形还原和词干提取

2. 向量编码

   • 使用预训练模型生成向量
   • 可选：领域微调优化

3. 相似度计算

   • 余弦相似度
   • 欧氏距离
   • 内积

4. 结果排序

   • 相似度分数排序
   • 可选：重排序优化

语义检索的优势

1. 语义理解能力

# 语义理解示例
semantic_examples = [
    ("人工智能的应用领域", "AI在各个行业的应用"),
    ("机器学习算法类型", "ML方法的分类"),
    ("深度学习原理", "神经网络工作机制"),
    ("自然语言处理技术", "NLP实现方法"),
    ("计算机视觉应用", "CV实际用途")
]

# 编码和相似度计算
for query, doc in semantic_examples:
    query_vec = model.encode([query])[0]
    doc_vec = model.encode([doc])[0]
    similarity = np.dot(query_vec, doc_vec)
    
    print(f"查询: '{query}'")
    print(f"文档: '{doc}'")
    print(f"语义相似度: {similarity:.3f}")
    print("-" * 50)
​

2. 处理一词多义

# 一词多义处理示例
polysemy_examples = [
    "苹果公司的产品",  # 科技公司
    "苹果是一种水果",   # 水果
    "苹果手机很受欢迎", # 科技产品
    "苹果树的种植方法"  # 植物
]

# 查询
queries = ["苹果公司", "苹果水果", "苹果手机", "苹果种植"]

# 分析语义差异
print("一词多义分析:")
for query in queries:
    query_vec = model.encode([query])[0]
    similarities = []
    
    for doc in polysemy_examples:
        doc_vec = model.encode([doc])[0]
        sim = np.dot(query_vec, doc_vec)
        similarities.append(sim)
    
    print(f"\n查询: '{query}'")
    for i, (doc, sim) in enumerate(zip(polysemy_examples, similarities)):
        print(f"  文档{i+1}: '{doc}' -> 相似度: {sim:.3f}")
​

3. 跨语言检索能力

# 跨语言检索示例
cross_lingual_examples = [
    {"en": "Machine learning algorithms", "zh": "机器学习算法"},
    {"en": "Deep learning models", "zh": "深度学习模型"},
    {"en": "Natural language processing", "zh": "自然语言处理"},
    {"en": "Computer vision applications", "zh": "计算机视觉应用"}
]

# 使用多语言模型
multilingual_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

print("跨语言检索分析:")
for item in cross_lingual_examples:
    en_vec = multilingual_model.encode([item["en"]])[0]
    zh_vec = multilingual_model.encode([item["zh"]])[0]
    similarity = np.dot(en_vec, zh_vec)
    
    print(f"英文: '{item['en']}'")
    print(f"中文: '{item['zh']}'")
    print(f"跨语言相似度: {similarity:.3f}")
    print("-" * 30)
​

语义检索的实现方法

1. 基于余弦相似度的检索

def semantic_search(query: str, document_vectors: np.ndarray, 
                   document_texts: List[str], top_k: int = 5) -> List[tuple]:
    """基于语义的检索"""
    # 编码查询
    query_vector = model.encode([query])[0]
    
    # 计算相似度
    similarities = np.dot(document_vectors, query_vector)
    
    # 获取top-k结果
    top_indices = np.argsort(similarities)[-top_k:][::-1]
    
    results = []
    for idx in top_indices:
        results.append((document_texts[idx], similarities[idx]))
    
    return results

# 使用示例
query = "深度学习技术"
documents = [
    "机器学习是AI的核心技术",
    "深度学习使用神经网络",
    "自然语言处理处理文本",
    "计算机视觉识别图像",
    "深度学习在图像识别中的应用"
]

document_vectors = model.encode(documents)
results = semantic_search(query, document_vectors, documents)

print("语义检索结果:")
for i, (doc, score) in enumerate(results, 1):
    print(f"{i}. 相似度: {score:.3f} - {doc}")
​

2. 基于欧氏距离的检索

def euclidean_search(query: str, document_vectors: np.ndarray, 
                    document_texts: List[str], top_k: int = 5) -> List[tuple]:
    """基于欧氏距离的检索"""
    query_vector = model.encode([query])[0]
    
    # 计算欧氏距离（越小越好）
    distances = np.linalg.norm(document_vectors - query_vector, axis=1)
    
    # 获取top-k结果（距离最小的）
    top_indices = np.argsort(distances)[:top_k]
    
    results = []
    for idx in top_indices:
        results.append((document_texts[idx], distances[idx]))
    
    return results

# 使用示例
results = euclidean_search(query, document_vectors, documents)
print("欧氏距离检索结果:")
for i, (doc, distance) in enumerate(results, 1):
    print(f"{i}. 距离: {distance:.3f} - {doc}")
​

2.2.3 混合检索策略的设计

混合检索策略结合了传统关键词检索和语义检索的优势，通过多阶段检索和结果融合来提高检索效果。

多阶段检索架构

第一阶段：快速预检索

class HybridRetriever:
    def __init__(self, keyword_retriever, semantic_retriever):
        self.keyword_retriever = keyword_retriever
        self.semantic_retriever = semantic_retriever
    
    def two_stage_retrieval(self, query: str, top_k: int = 10):
        """两阶段检索"""
        # 第一阶段：关键词快速检索
        keyword_results = self.keyword_retriever.retrieve(query, top_k * 2)
        
        # 第二阶段：在候选集上语义重排序
        candidate_docs = [doc['content'] for doc in keyword_results]
        doc_ids = [doc['id'] for doc in keyword_results]
        
        semantic_scores = self.semantic_retriever.score(query, candidate_docs)
        
        # 融合分数
        fused_results = []
        for i, (doc_id, content) in enumerate(zip(doc_ids, candidate_docs)):
            keyword_score = keyword_results[i]['score']
            semantic_score = semantic_scores[i]
            
            # 加权融合
            fused_score = 0.6 * keyword_score + 0.4 * semantic_score
            fused_results.append({
                'id': doc_id,
                'content': content,
                'score': fused_score,
                'keyword_score': keyword_score,
                'semantic_score': semantic_score
            })
        
        # 排序返回
        fused_results.sort(key=lambda x: x['score'], reverse=True)
        return fused_results[:top_k]
​

分数融合策略

class ScoreFusion:
    @staticmethod
    def weighted_fusion(keyword_scores: List[float], semantic_scores: List[float], 
                       weights: tuple = (0.5, 0.5)) -> List[float]:
        """加权分数融合"""
        return [w1 * k + w2 * s for k, s, (w1, w2) in 
                zip(keyword_scores, semantic_scores, [weights] * len(keyword_scores))]
    
    @staticmethod
    def rank_fusion(rank_lists: List[List[int]]) -> List[int]:
        """排序融合（Reciprocal Rank Fusion）"""
        from collections import defaultdict
        
        score_dict = defaultdict(float)
        
        for rank_list in rank_lists:
            for rank, doc_id in enumerate(rank_list):
                score_dict[doc_id] += 1.0 / (rank + 1)
        
        # 按融合分数排序
        fused_rank = sorted(score_dict.items(), key=lambda x: x[1], reverse=True)
        return [doc_id for doc_id, score in fused_rank]
    
    @staticmethod
    def probability_fusion(probabilities: List[List[float]]) -> List[float]:
        """概率融合"""
        import math
        
        fused_probs = []
        for probs in zip(*probabilities):
            # 对数概率融合
            log_prob = sum(math.log(p + 1e-10) for p in probs)
            fused_prob = math.exp(log_prob / len(probs))
            fused_probs.append(fused_prob)
        
        return fused_probs
​

查询理解与重写

查询扩展技术

class QueryExpansion:
    def __init__(self, embedding_model):
        self.model = embedding_model
        self.similarity_threshold = 0.7
    
    def expand_with_synonyms(self, query: str, synonym_dict: Dict[str, List[str]]) -> List[str]:
        """基于同义词扩展"""
        expanded_queries = [query]
        
        for word in query.split():
            if word in synonym_dict:
                synonyms = synonym_dict[word]
                for synonym in synonyms:
                    expanded_query = query.replace(word, synonym)
                    if expanded_query != query:
                        expanded_queries.append(expanded_query)
        
        return expanded_queries
    
    def expand_with_embeddings(self, query: str, candidate_terms: List[str], top_k: int = 3) -> List[str]:
        """基于向相似度的查询扩展"""
        query_vector = self.model.encode([query])[0]
        term_vectors = self.model.encode(candidate_terms)
        
        # 计算相似度
        similarities = np.dot(term_vectors, query_vector)
        
        # 选择最相似的术语
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        similar_terms = [candidate_terms[i] for i in top_indices if similarities[i] > self.similarity_threshold]
        
        # 构建扩展查询
        expanded_queries = [query]
        for term in similar_terms:
            expanded_query = f"{query} {term}"
            expanded_queries.append(expanded_query)
        
        return expanded_queries
​

查询重写技术

class QueryRewriting:
    def __init__(self, model):
        self.model = model
    
    def paraphrase_rewriting(self, query: str) -> List[str]:
        """同义改写"""
        # 使用T5或BERT进行查询改写
        paraphrased = []
        # 这里使用简单的同义词替换作为示例
        paraphrase_map = {
            "机器学习": "ML",
            "深度学习": "DL",
            "自然语言处理": "NLP",
            "计算机视觉": "CV"
        }
        
        for word, replacement in paraphrase_map.items():
            if word in query:
                rewritten = query.replace(word, replacement)
                if rewritten != query:
                    paraphrased.append(rewritten)
        
        return paraphrased + [query]  # 保留原查询
    
    def semantic_rewriting(self, query: str) -> str:
        """语义重写 - 识别查询意图"""
        # 这里简化处理，实际可以使用更复杂的NLP技术
        intent_keywords = {
            "什么是": "解释概念",
            "如何": "操作方法",
            "为什么": "原因分析",
            "区别": "对比分析"
        }
        
        for keyword, intent in intent_keywords.items():
            if keyword in query:
                return f"{intent}: {query.replace(keyword, '')}"
        
        return query
​

混合检索的评估与调优

混合检索性能评估

class HybridRetrievalEvaluator:
    def __init__(self):
        self.evaluation_metrics = []
    
    def evaluate_hybrid_retrieval(self, query, relevant_docs, 
                                 keyword_results, semantic_results, 
                                 hybrid_results):
        """评估混合检索效果"""
        metrics = {}
        
        # 关键词检索评估
        keyword_metrics = {
            'precision': precision_at_k([doc['id'] for doc in keyword_results], 
                                      relevant_docs, 5),
            'recall': recall_at_k([doc['id'] for doc in keyword_results], 
                                 relevant_docs, 5)
        }
        
        # 语义检索评估
        semantic_metrics = {
            'precision': precision_at_k([doc['id'] for doc in semantic_results], 
                                      relevant_docs, 5),
            'recall': recall_at_k([doc['id'] for doc in semantic_results], 
                                 relevant_docs, 5)
        }
        
        # 混合检索评估
        hybrid_metrics = {
            'precision': precision_at_k([doc['id'] for doc in hybrid_results], 
                                      relevant_docs, 5),
            'recall': recall_at_k([doc['id'] for doc in hybrid_results], 
                                 relevant_docs, 5)
        }
        
        metrics = {
            'keyword': keyword_metrics,
            'semantic': semantic_metrics,
            'hybrid': hybrid_metrics
        }
        
        return metrics
    
    def optimize_hybrid_weights(self, queries, relevant_docs_list,
                              keyword_results_list, semantic_results_list):
        """优化混合权重"""
        from itertools import product
        
        best_weights = (0.5, 0.5)
        best_score = 0
        
        # 网格搜索优化权重
        for w1, w2 in product([i*0.1 for i in range(10)], [i*0.1 for i in range(10)]):
            if w1 + w2 == 1.0:  # 权重和为1
                total_score = 0
                
                for query, rel_docs, kw_results, sem_results in zip(
                    queries, relevant_docs_list, keyword_results_list, semantic_results_list):
                    
                    # 融合分数
                    fused_scores = ScoreFusion.weighted_fusion(
                        [r['score'] for r in kw_results],
                        [r['score'] for r in sem_results],
                        (w1, w2)
                    )
                    
                    # 计算MAP（平均精确度）
                    score = self.calculate_map(fused_scores, rel_docs)
                    total_score += score
                
                avg_score = total_score / len(queries)
                if avg_score > best_score:
                    best_score = avg_score
                    best_weights = (w1, w2)
        
        return best_weights, best_score
    
    def calculate_map(self, scores, relevant_docs):
        """计算平均精确度"""
        # 简化实现，实际需要更复杂的MAP计算
        precision_sum = 0
        relevant_count = 0
        
        for i, score in enumerate(scores):
            if score in relevant_docs:
                relevant_count += 1
                precision_sum += relevant_count / (i + 1)
        
        return precision_sum / len(relevant_docs) if relevant_count > 0 else 0
​

2.2.4 检索算法的比较与选择

主流检索算法对比

算法特性对比表

性能对比分析

# 算法性能对比示例
def compare_retrieval_algorithms(queries, documents, relevant_docs):
    """比较不同检索算法的性能"""
    
    results = {}
    
    # TF-IDF检索
    tfidf_scores = []
    for query in queries:
        scores = []
        for doc in documents:
            tfidf_score = calculate_tfidf(query, doc, documents)
            scores.append((doc, tfidf_score))
        scores.sort(key=lambda x: x[1], reverse=True)
        tfidf_scores.append([doc[0] for doc in scores[:5]])
    
    # 语义检索
    semantic_scores = []
    doc_vectors = model.encode(documents)
    
    for query in queries:
        results = semantic_search(query, doc_vectors, documents, top_k=5)
        semantic_scores.append([doc[0] for doc in results])
    
    # BM25检索
    bm25_scores = []
    for query in queries:
        scores = []
        for doc in documents:
            bm25_score = calculate_bm25_score(query, doc, documents)
            scores.append((doc, bm25_score))
        scores.sort(key=lambda x: x[1], reverse=True)
        bm25_scores.append([doc[0] for doc in scores[:5]])
    
    # 计算性能指标
    metrics = {
        'tfidf': {
            'map': calculate_average_precision(tfidf_scores, relevant_docs),
            'ndcg': calculate_ndcg(tfidf_scores, relevant_docs)
        },
        'bm25': {
            'map': calculate_average_precision(bm25_scores, relevant_docs),
            'ndcg': calculate_ndcg(bm25_scores, relevant_docs)
        },
        'semantic': {
            'map': calculate_average_precision(semantic_scores, relevant_docs),
            'ndcg': calculate_ndcg(semantic_scores, relevant_docs)
        }
    }
    
    return metrics

def calculate_average_precision(retrieved_lists, relevant_docs):
    """计算平均精确度"""
    # 简化实现
    ap_sum = 0
    for retrieved, relevant in zip(retrieved_lists, relevant_docs):
        precision = 0
        relevant_count = 0
        
        for i, doc in enumerate(retrieved):
            if doc in relevant:
                relevant_count += 1
                precision += relevant_count / (i + 1)
        
        ap_sum += precision / len(relevant) if len(relevant) > 0 else 0
    
    return ap_sum / len(retrieved_lists)
​

算法选择策略

1. 基于查询类型的选择

def select_algorithm_by_query_type(query):
    """根据查询类型选择合适的检索算法"""
    
    # 精确匹配查询
    if any(word in query for word in ['"', '=', 'exact']):
        return 'tfidf'
    
    # 概念性查询
    if any(word in query for word in ['什么是', '为什么', '如何']):
        return 'semantic'
    
    # 技术性查询
    if any(word in query for word in ['代码', '实现', '方法']):
        return 'bm25'
    
    # 一般性查询
    return 'hybrid'
​

2. 基于数据特性的选择

def select_algorithm_by_data_characteristics(documents):
    """根据数据特性选择检索算法"""
    
    # 文档长度分析
    avg_length = sum(len(doc.split()) for doc in documents) / len(documents)
    
    # 术语多样性分析
    all_terms = set()
    for doc in documents:
        all_terms.update(doc.split())
    vocabulary_size = len(all_terms)
    
    # 选择策略
    if avg_length < 100 and vocabulary_size < 1000:
        return 'tfidf'  # 短文本、小词汇量
    elif avg_length > 1000 and vocabulary_size > 5000:
        return 'semantic'  # 长文本、大词汇量
    else:
        return 'hybrid'  # 一般情况
​

3. 基于性能需求的选择

def select_algorithm_by_performance_requirements():
    """根据性能需求选择算法"""
    
    requirements = {
        'realtime': 'tfidf',      # 实时响应
        'accuracy': 'semantic',   # 高精度
        'balanced': 'hybrid',    # 平衡
        'cost_sensitive': 'bm25'  # 成本敏感
    }
    
    return requirements
​

2.2.5 检索算法的优化策略

性能优化技术

1. 索引优化

class SearchIndexOptimizer:
    def __init__(self):
        self.index_strategies = {
            'inverted': self.create_inverted_index,
            'vector': self.create_vector_index,
            'hybrid': self.create_hybrid_index
        }
    
    def create_inverted_index(self, documents):
        """创建倒排索引"""
        inverted_index = {}
        
        for doc_id, doc in enumerate(documents):
            for term in doc.split():
                if term not in inverted_index:
                    inverted_index[term] = []
                inverted_index[term].append(doc_id)
        
        return inverted_index
    
    def create_vector_index(self, documents, model):
        """创建向量索引"""
        vectors = model.encode(documents)
        return vectors
    
    def optimize_index_structure(self, index, strategy='compressed'):
        """优化索引结构"""
        if strategy == 'compressed':
            # 压缩倒排索引
            for term in index:
                index[term] = list(set(index[term]))
        
        return index
​

2. 查询优化

class QueryOptimizer:
    def __init__(self):
        self.query_cache = {}
    
    def cache_queries(self, query, results):
        """缓存查询结果"""
        self.query_cache[query] = results
    
    def get_cached_results(self, query):
        """获取缓存的查询结果"""
        return self.query_cache.get(query, None)
    
    def optimize_query_structure(self, query):
        """优化查询结构"""
        # 查询分解
        terms = query.split()
        
        # 过滤停用词
        stop_words = {'的', '了', '是', '在', '和', '与', '或'}
        filtered_terms = [term for term in terms if term not in stop_words]
        
        # 查询重构
        optimized_query = ' '.join(filtered_terms)
        
        return optimized_query
​

准确性优化技术

1. 重排序模型

class Reranker:
    def __init__(self, reranker_model):
        self.model = reranker_model
    
    def cross_encoder_rerank(self, query, candidates, top_k=5):
        """使用交叉编码器进行重排序"""
        pairs = [(query, candidate) for candidate in candidates]
        
        # 计算相关性分数
        scores = self.model.predict(pairs)
        
        # 排序并返回top-k结果
        reranked_indices = np.argsort(scores)[-top_k:][::-1]
        reranked_results = [candidates[i] for i in reranked_indices]
        
        return reranked_results
    
    def mono_t5_rerank(self, query, candidates, top_k=5):
        """使用T5进行单塔重排序"""
        # 构建重排序输入
        rerank_input = [f"Query: {query} Document: {doc}" for doc in candidates]
        
        # 预测相关性分数
        scores = self.model.predict(rerank_input)
        
        # 排序
        reranked_indices = np.argsort(scores)[-top_k:][::-1]
        return [candidates[i] for i in reranked_indices]
​

2. 上下文感知检索

class ContextualRetriever:
    def __init__(self, context_window=3):
        self.context_window = context_window
    
    def add_context_awareness(self, query, conversation_history):
        """添加上下文感知能力"""
        if not conversation_history:
            return query
        
        # 获取最近的对话上下文
        recent_context = conversation_history[-self.context_window:]
        
        # 构建增强查询
        contextual_query = f"Context: {' '.join(recent_context)} Query: {query}"
        
        return contextual_query
    
    def resolve_coreference(self, query, document):
        """解决共指消解"""
        # 这里简化处理，实际可以使用更复杂的NLP技术
        coreference_resolution_map = {
            "它": document.split()[-1],  # 简单处理，取最后一个词
            "这个": document.split()[0],  # 取第一个词
            "该": document.split()[0]
        }
        
        resolved_query = query
        for pronoun, replacement in coreference_resolution_map.items():
            if pronoun in query:
                resolved_query = query.replace(pronoun, replacement)
        
        return resolved_query
​

2.2.6 检索算法的评估与监控

评估指标体系

class RetrievalEvaluator:
    def __init__(self):
        self.metrics = {}
    
    def calculate_ndcg(self, retrieved_docs, relevant_docs, k=5):
        """计算归一化折损累计增益"""
        def dcg(relevance_scores):
            return sum(score / np.log2(i + 2) for i, score in enumerate(relevance_scores))
        
        # 理想DCG
        ideal_relevance = sorted(relevant_docs, reverse=True)
        ideal_dcg = dcg(ideal_relevance)
        
        # 实际DCG
        actual_relevance = [1 if doc in relevant_docs else 0 for doc in retrieved_docs[:k]]
        actual_dcg = dcg(actual_relevance)
        
        return actual_dcg / ideal_dcg if ideal_dcg > 0 else 0
    
    def calculate_mrr(self, retrieved_docs, relevant_docs):
        """计算倒数排名"""
        for i, doc in enumerate(retrieved_docs):
            if doc in relevant_docs:
                return 1.0 / (i + 1)
        return 0.0
    
    def calculate_f1(self, retrieved_docs, relevant_docs):
        """计算F1分数"""
        retrieved_set = set(retrieved_docs)
        relevant_set = set(relevant_docs)
        
        tp = len(retrieved_set & relevant_set)
        fp = len(retrieved_set - relevant_set)
        fn = len(relevant_set - retrieved_set)
        
        precision = tp / (tp + fp) if (tp + fp) > 0 else 0
        recall = tp / (tp + fn) if (tp + fn) > 0 else 0
        
        f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
        
        return f1
​

实时监控与反馈

class RetrievalMonitor:
    def __init__(self):
        self.performance_history = []
        self.alert_thresholds = {
            'latency': 1000,  # 毫秒
            'recall': 0.7,    # 召回率
            'precision': 0.6  # 精确度
        }
    
    def log_performance(self, query, results, relevant_docs, execution_time):
        """记录性能数据"""
        performance_data = {
            'query': query,
            'timestamp': time.time(),
            'execution_time': execution_time,
            'retrieved_count': len(results),
            'relevant_retrieved': len(set(results) & set(relevant_docs)),
            'total_relevant': len(relevant_docs)
        }
        
        self.performance_history.append(performance_data)
        
        # 检查告警条件
        self.check_alerts(performance_data)
    
    def check_alerts(self, performance_data):
        """检查性能告警"""
        alerts = []
        
        # 延迟告警
        if performance_data['execution_time'] > self.alert_thresholds['latency']:
            alerts.append(f"高延迟告警: {performance_data['execution_time']}ms")
        
        # 召回率告警
        recall = performance_data['relevant_retrieved'] / performance_data['total_relevant']
        if recall < self.alert_thresholds['recall']:
            alerts.append(f"低召回率告警: {recall:.2f}")
        
        # 精确度告警
        precision = performance_data['relevant_retrieved'] / performance_data['retrieved_count']
        if precision < self.alert_thresholds['precision']:
            alerts.append(f"低精确度告警: {precision:.2f}")
        
        return alerts
    
    def generate_performance_report(self):
        """生成性能报告"""
        if not self.performance_history:
            return "暂无性能数据"
        
        avg_latency = np.mean([p['execution_time'] for p in self.performance_history])
        avg_recall = np.mean([p['relevant_retrieved'] / p['total_relevant'] 
                            for p in self.performance_history if p['total_relevant'] > 0])
        
        report = f"""
性能报告摘要:
- 平均查询延迟: {avg_latency:.2f}ms
- 平均召回率
​