第 14 章 RAG 系统设计¶

"""文档解析：多格式统一处理"""
from dataclasses import dataclass
from typing import List

@dataclass  # @dataclass自动生成__init__等方法
class Document:
    content: str
    metadata: dict  # 来源、页码、标题等

class DocumentParser:
    """多格式文档解析器"""

    @staticmethod  # @staticmethod不需要实例即可调用
    def parse_pdf(path: str) -> List[Document]:
        """PDF解析 - 使用PyMuPDF提取文本和表格"""
        import fitz  # PyMuPDF
        docs = []
        pdf = fitz.open(path)
        for page_num, page in enumerate(pdf):  # enumerate同时获取索引和元素
            text = page.get_text("text")
            if text.strip():
                docs.append(Document(
                    content=text.strip(),
                    metadata={"source": path, "page": page_num + 1, "type": "pdf"}
                ))
        pdf.close()
        return docs

    @staticmethod
    def parse_markdown(path: str) -> List[Document]:
        """Markdown解析 - 按标题层级拆分"""
        import re
        with open(path, 'r', encoding='utf-8') as f:  # with自动管理文件关闭
            content = f.read()
        # 按一级/二级标题拆分
        sections = re.split(r'\n(?=#{1,2}\s)', content)
        docs = []
        for section in sections:
            if section.strip():
                title_match = re.match(r'^(#{1,2})\s+(.+)', section)
                title = title_match.group(2) if title_match else "untitled"
                docs.append(Document(
                    content=section.strip(),
                    metadata={"source": path, "title": title, "type": "markdown"}
                ))
        return docs

    @staticmethod
    def parse_html(path: str) -> List[Document]:
        """HTML解析 - 使用BeautifulSoup提取正文"""
        from bs4 import BeautifulSoup
        with open(path, 'r', encoding='utf-8') as f:
            soup = BeautifulSoup(f.read(), 'html.parser')
        # 移除script和style标签
        for tag in soup(['script', 'style', 'nav', 'footer']):
            tag.decompose()
        text = soup.get_text(separator='\n', strip=True)
        return [Document(content=text, metadata={"source": path, "type": "html"})]

def fixed_size_chunking(text: str, chunk_size: int = 512,
                        chunk_overlap: int = 128) -> List[str]:
    """固定大小分块：按字符数切分，含重叠"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk.strip())
        start += chunk_size - chunk_overlap
    return [c for c in chunks if c]

def recursive_chunking(text: str, chunk_size: int = 512,
                       chunk_overlap: int = 128,
                       separators: List[str] = None) -> List[str]:
    """递归分块：按层级分隔符逐步切分"""
    if separators is None:
        separators = ["\n\n", "\n", "。", "；", "，", " ", ""]

    chunks = []
    sep = separators[0]
    remaining_seps = separators[1:]

    # 按当前分隔符拆分
    splits = text.split(sep) if sep else list(text)

    current_chunk = ""
    for split in splits:
        candidate = current_chunk + sep + split if current_chunk else split
        if len(candidate) <= chunk_size:
            current_chunk = candidate
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            # 如果单个split仍然超长，递归使用下级分隔符
            if len(split) > chunk_size and remaining_seps:
                sub_chunks = recursive_chunking(
                    split, chunk_size, chunk_overlap, remaining_seps
                )
                chunks.extend(sub_chunks)
                current_chunk = ""
            else:
                current_chunk = split

    if current_chunk:
        chunks.append(current_chunk.strip())
    return [c for c in chunks if c]

import numpy as np

def semantic_chunking(sentences: List[str], embeddings: np.ndarray,
                      threshold: float = 0.5,
                      max_chunk_size: int = 512) -> List[str]:
    """语义分块：基于句间相似度断点切分

    Args:
        sentences: 句子列表
        embeddings: 每个句子的嵌入向量 (n_sentences, dim)
        threshold: 相似度断点阈值（低于此值则切分）
        max_chunk_size: 最大chunk字符数
    """
    # 计算相邻句子的余弦相似度
    similarities = []
    for i in range(len(embeddings) - 1):
        sim = np.dot(embeddings[i], embeddings[i + 1]) / (  # np.dot矩阵/向量点乘
            np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i + 1])  # np.linalg线性代数运算
        )
        similarities.append(sim)

    # 在相似度低谷处切分
    chunks = []
    current_chunk_sents = [sentences[0]]
    for i, sim in enumerate(similarities):
        current_text = "".join(current_chunk_sents)
        next_sent = sentences[i + 1]
        if sim < threshold or len(current_text) + len(next_sent) > max_chunk_size:
            chunks.append("".join(current_chunk_sents))
            current_chunk_sents = [next_sent]
        else:
            current_chunk_sents.append(next_sent)

    if current_chunk_sents:
        chunks.append("".join(current_chunk_sents))
    return chunks

@dataclass
class HierarchicalChunk:
    content: str
    level: int          # 0=文档, 1=章节, 2=段落, 3=句子
    children: list
    parent: 'HierarchicalChunk' = None

    def get_context_window(self, levels_up: int = 1) -> str:
        """获取包含上级上下文的文本"""
        node = self
        for _ in range(levels_up):
            if node.parent:
                node = node.parent
        return node.content

"""BGE-M3 多向量检索示例"""
# 注意：需要安装支持 `BGEM3FlagModel` 的较新 FlagEmbedding 版本
# 安装命令请以官方文档为准
# API可能随版本变化，请参考官方文档：https://github.com/FlagOpen/FlagEmbedding
from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)

queries = ["什么是检索增强生成？"]
documents = [
    "RAG（Retrieval-Augmented Generation）通过检索外部知识库来增强大模型的回答。",
    "强化学习是一种通过与环境交互来学习高回报策略的方法。",
    "BGE-M3支持超过100种语言的文本嵌入。"
]

# 编码时同时返回三种表示
q_output = model.encode(queries, return_dense=True, return_sparse=True,
                         return_colbert_vecs=True)
d_output = model.encode(documents, return_dense=True, return_sparse=True,
                         return_colbert_vecs=True)

# Dense检索得分
dense_scores = q_output['dense_vecs'] @ d_output['dense_vecs'].T
print("Dense scores:", dense_scores)

# Sparse检索得分（词级别精确匹配）
sparse_scores = model.compute_lexical_matching_score(
    q_output['lexical_weights'][0], d_output['lexical_weights'][0]
)

# ColBERT检索得分（token级别MaxSim）
colbert_scores = model.colbert_score(
    q_output['colbert_vecs'][0], d_output['colbert_vecs'][0]
)

"""Embedding模型性能对比benchmark"""
import time
import numpy as np
from sentence_transformers import SentenceTransformer

# 准备测试数据
test_sentences = [
    "什么是Transformer架构？",
    "BERT模型如何进行预训练？",
    "注意力机制的计算复杂度是多少？",
    "大语言模型的涌现能力有哪些？",
    "如何解决大模型的幻觉问题？",
] * 100  # 500条句子

# 对比不同模型
models_to_test = {
    "BGE-small-zh": "BAAI/bge-small-zh-v1.5",
    "BGE-large-zh": "BAAI/bge-large-zh-v1.5",
    "GTE-large-zh": "thenlper/gte-large-zh",
}

results = {}
for name, model_id in models_to_test.items():
    model = SentenceTransformer(model_id)

    # 计时编码
    start = time.time()
    embeddings = model.encode(test_sentences, batch_size=64,
                               show_progress_bar=False,
                               normalize_embeddings=True)
    elapsed = time.time() - start

    results[name] = {
        "维度": embeddings.shape[1],
        "编码速度(句/秒)": len(test_sentences) / elapsed,
        "耗时(秒)": round(elapsed, 2),
    }

    # 测试语义相似度（query与第一个句子最相关）
    query_emb = model.encode(["Transformer的结构是什么样的？"],
                              normalize_embeddings=True)
    similarities = query_emb @ embeddings[:5].T  # 切片操作，取前n个元素
    results[name]["Top1相似度"] = round(float(similarities[0][0]), 4)

# 展示结果
for name, metrics in results.items():
    print(f"\n{name}:")
    for k, v in metrics.items():
        print(f"  {k}: {v}")

"""Hybrid检索完整实现：BM25 + 向量 + RRF融合"""
import numpy as np
from dataclasses import dataclass, field
from typing import List, Tuple, Dict
from collections import defaultdict
import math
import re

@dataclass
class SearchResult:
    doc_id: int
    content: str
    score: float
    source: str  # "dense" / "sparse" / "hybrid"

class BM25Retriever:
    """BM25稀疏检索器"""

    def __init__(self, k1: float = 1.2, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.documents: List[str] = []
        self.doc_freqs: Dict[str, int] = {}  # 词 -> 包含该词的文档数
        self.doc_term_freqs: List[Dict[str, int]] = []  # 每个文档的词频
        self.avgdl: float = 0.0
        self.n_docs: int = 0

    def _tokenize(self, text: str) -> List[str]:
        """中文简单分词（生产中建议用jieba）"""
        # 按非中文字符分割 + 单字切分
        tokens = re.findall(r'[\u4e00-\u9fff]|[a-zA-Z0-9]+', text.lower())  # re.findall返回所有匹配项列表
        return tokens

    def index(self, documents: List[str]):
        """建立BM25索引"""
        self.documents = documents
        self.n_docs = len(documents)
        doc_lengths = []

        for doc in documents:
            tokens = self._tokenize(doc)
            doc_lengths.append(len(tokens))
            term_freq = defaultdict(int)  # defaultdict访问不存在的键时返回默认值
            for token in tokens:
                term_freq[token] += 1
            self.doc_term_freqs.append(dict(term_freq))
            for token in set(tokens):
                self.doc_freqs[token] = self.doc_freqs.get(token, 0) + 1

        self.avgdl = sum(doc_lengths) / len(doc_lengths) if doc_lengths else 0

    def search(self, query: str, top_k: int = 10) -> List[SearchResult]:
        """BM25检索"""
        query_tokens = self._tokenize(query)
        scores = []

        for doc_id in range(self.n_docs):
            score = 0.0
            doc_len = sum(self.doc_term_freqs[doc_id].values())
            for token in query_tokens:
                if token not in self.doc_freqs:
                    continue
                df = self.doc_freqs[token]
                idf = math.log((self.n_docs - df + 0.5) / (df + 0.5) + 1)
                tf = self.doc_term_freqs[doc_id].get(token, 0)
                tf_norm = (tf * (self.k1 + 1)) / (
                    tf + self.k1 * (1 - self.b + self.b * doc_len / self.avgdl)
                )
                score += idf * tf_norm
            scores.append(score)

        # 排序取Top-K
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [
            SearchResult(
                doc_id=int(idx), content=self.documents[idx],
                score=scores[idx], source="sparse"
            )
            for idx in top_indices if scores[idx] > 0
        ]

class DenseRetriever:
    """Dense向量检索器"""

    def __init__(self, model_name: str = "BAAI/bge-small-zh-v1.5"):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer(model_name)
        self.doc_embeddings: np.ndarray = None
        self.documents: List[str] = []

    def index(self, documents: List[str]):
        """建立向量索引"""
        self.documents = documents
        self.doc_embeddings = self.model.encode(
            documents, normalize_embeddings=True, show_progress_bar=False
        )

    def search(self, query: str, top_k: int = 10) -> List[SearchResult]:
        """向量检索"""
        query_emb = self.model.encode(
            [query], normalize_embeddings=True, show_progress_bar=False
        )
        scores = (query_emb @ self.doc_embeddings.T)[0]
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [
            SearchResult(
                doc_id=int(idx), content=self.documents[idx],
                score=float(scores[idx]), source="dense"
            )
            for idx in top_indices
        ]

class HybridRetriever:
    """混合检索器：Dense + Sparse + RRF融合"""

    def __init__(self, dense_retriever: DenseRetriever,
                 sparse_retriever: BM25Retriever, rrf_k: int = 60):
        self.dense = dense_retriever
        self.sparse = sparse_retriever
        self.rrf_k = rrf_k

    def search(self, query: str, top_k: int = 10,
               dense_weight: float = 0.5,
               sparse_weight: float = 0.5) -> List[SearchResult]:
        """混合检索 + RRF融合"""
        dense_results = self.dense.search(query, top_k=top_k * 2)
        sparse_results = self.sparse.search(query, top_k=top_k * 2)

        # RRF融合
        rrf_scores = defaultdict(float)
        doc_contents = {}

        for rank, result in enumerate(dense_results):
            rrf_scores[result.doc_id] += dense_weight / (self.rrf_k + rank + 1)
            doc_contents[result.doc_id] = result.content

        for rank, result in enumerate(sparse_results):
            rrf_scores[result.doc_id] += sparse_weight / (self.rrf_k + rank + 1)
            doc_contents[result.doc_id] = result.content

        # 按RRF分数排序
        sorted_docs = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)  # lambda匿名函数

        return [
            SearchResult(
                doc_id=doc_id, content=doc_contents[doc_id],
                score=score, source="hybrid"
            )
            for doc_id, score in sorted_docs[:top_k]
        ]

# --- 使用示例 ---
if __name__ == "__main__":
    documents = [
        "RAG通过检索外部知识库增强大语言模型的回答质量，减少幻觉。",
        "BM25是经典的稀疏检索算法，基于词频和逆文档频率计算相关性。",
        "BERT是Google提出的预训练语言模型，采用Transformer编码器架构。",
        "向量数据库如Milvus、Qdrant支持高效的近似最近邻搜索。",
        "Reranker使用Cross-Encoder对初步检索结果进行精细排序。",
        "知识图谱可以为RAG提供结构化的实体关系信息。",
        "Embedding模型将文本映射到稠密向量空间用于语义检索。",
    ]

    # 初始化检索器
    sparse = BM25Retriever()
    sparse.index(documents)

    dense = DenseRetriever(model_name="BAAI/bge-small-zh-v1.5")
    dense.index(documents)

    hybrid = HybridRetriever(dense, sparse)

    # 对比三种检索
    query = "什么是BM25检索算法？"
    print(f"Query: {query}\n")

    for name, results in [
        ("BM25 (Sparse)", sparse.search(query, top_k=3)),
        ("Dense", dense.search(query, top_k=3)),
        ("Hybrid (RRF)", hybrid.search(query, top_k=3)),
    ]:
        print(f"--- {name} ---")
        for r in results:
            print(f"  [{r.score:.4f}] {r.content[:50]}...")
        print()

"""使用BGE-Reranker进行重排序"""
from typing import List, Tuple
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

class BGEReranker:
    """BGE Cross-Encoder重排器"""

    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()  # eval()评估模式

    @torch.no_grad()  # 禁用梯度计算，节省内存
    def rerank(self, query: str, documents: List[str],
               top_k: int = 5) -> List[Tuple[int, str, float]]:
        """对候选文档重排序

        Returns:
            List of (原始索引, 文档内容, 重排分数)
        """
        pairs = [[query, doc] for doc in documents]
        inputs = self.tokenizer(
            pairs, padding=True, truncation=True,
            max_length=512, return_tensors="pt"
        )
        scores = self.model(**inputs).logits.squeeze(-1).tolist()  # squeeze压缩维度

        # 按分数降序排列
        indexed_scores = [(i, documents[i], s) for i, s in enumerate(scores)]
        indexed_scores.sort(key=lambda x: x[2], reverse=True)
        return indexed_scores[:top_k]

# 使用示例
reranker = BGEReranker()

query = "RAG系统如何减少幻觉？"
candidates = [
    "RAG通过引入外部知识源，让模型基于检索到的事实生成回答，显著减少幻觉。",
    "大语言模型的幻觉问题是指模型生成看似合理但实际错误的内容。",
    "向量数据库支持高效的相似度搜索，是RAG系统的核心组件。",
    "通过检索增强，模型回答可以锚定在可验证的文档上，降低捏造风险。",
    "Transformer注意力机制通过Q、K、V矩阵计算注意力权重。",
]

results = reranker.rerank(query, candidates, top_k=3)
print(f"Query: {query}\n")
for idx, doc, score in results:
    print(f"  [{score:.4f}] (原第{idx}条) {doc[:60]}...")

def hyde_retrieval(query: str, llm, retriever) -> List[str]:
    """HyDE：用假设性文档增强检索"""
    prompt = f"请详细回答以下问题；如果证据不足，请明确说明不确定性，不要编造：\n{query}"
    hypothetical_doc = llm.generate(prompt)
    # 用假设文档做检索（语义更接近答案文档的表述）
    results = retriever.search(hypothetical_doc, top_k=5)
    return results

def multi_query_retrieval(query: str, llm, retriever) -> List[str]:
    """Multi-Query：多角度查询扩展"""
    prompt = f"""请将以下问题改写为3个不同角度的查询，每行一个：
问题：{query}

改写："""
    sub_queries = llm.generate(prompt).strip().split('\n')

    all_results = {}
    for sub_q in sub_queries:
        results = retriever.search(sub_q.strip(), top_k=5)
        for r in results:
            if r.doc_id not in all_results:
                all_results[r.doc_id] = r
    return list(all_results.values())

def corrective_rag(query: str, retriever, llm, web_search_fn):
    """Corrective RAG：检索质量自动纠正"""
    results = retriever.search(query, top_k=5)

    # 评估检索质量
    eval_prompt = f"""判断以下检索结果与查询的相关性。
查询：{query}
检索结果：{results[0].content[:200]}
回答 CORRECT / INCORRECT / AMBIGUOUS："""

    assessment = llm.generate(eval_prompt).strip()

    if assessment == "CORRECT":
        context = "\n".join([r.content for r in results])
    elif assessment == "INCORRECT":
        # 降级到Web搜索
        web_results = web_search_fn(query)
        context = "\n".join(web_results)
    else:  # AMBIGUOUS
        context = "\n".join([r.content for r in results])
        web_results = web_search_fn(query)
        context += "\n" + "\n".join(web_results)

    return llm.generate(f"基于以下信息回答问题：\n{context}\n\n问题：{query}")

"""Agentic RAG概念示例"""

class AgenticRAG:
    """Agent驱动的RAG系统"""

    def __init__(self, retrievers: dict, llm):
        self.retrievers = retrievers  # {"vector": ..., "bm25": ..., "graph": ...}
        self.llm = llm

    def route(self, query: str) -> str:
        """Agent决策：选择检索策略"""
        prompt = f"""分析查询类型并选择更合适的检索策略：
查询：{query}

可选策略：
- vector: 语义相似的段落检索
- bm25: 关键词精确匹配
- hybrid: 混合检索
- graph: 知识图谱推理
- multi_hop: 多步推理检索
- none: 无需检索（常识问题）

输出策略名称："""
        return self.llm.generate(prompt).strip()

    def execute(self, query: str) -> str:
        """Agent执行RAG Pipeline"""
        strategy = self.route(query)

        if strategy == "none":
            return self.llm.generate(query)

        if strategy == "multi_hop":
            return self._multi_hop(query)

        retriever = self.retrievers.get(strategy,
                                         self.retrievers["hybrid"])
        results = retriever.search(query, top_k=5)
        context = "\n".join([r.content for r in results])

        return self.llm.generate(
            f"基于以下信息回答问题。如果信息不足请说明。\n\n"
            f"参考信息：\n{context}\n\n问题：{query}"
        )

    def _multi_hop(self, query: str) -> str:
        """多跳推理检索"""
        sub_questions = self._decompose(query)
        accumulated_context = ""
        for sub_q in sub_questions:
            enriched_q = f"{sub_q}\n已知信息：{accumulated_context}" if accumulated_context else sub_q
            results = self.retrievers["hybrid"].search(enriched_q, top_k=3)
            answer = self.llm.generate(
                f"简洁回答：{sub_q}\n参考：{results[0].content}"
            )
            accumulated_context += f"\n{sub_q} → {answer}"
        return self.llm.generate(
            f"综合以下信息回答原问题。\n{accumulated_context}\n\n原问题：{query}"
        )

    def _decompose(self, query: str) -> List[str]:
        """将复杂query分解为子问题"""
        prompt = f"将以下复杂问题分解为2-3个简单子问题，每行一个：\n{query}"
        return self.llm.generate(prompt).strip().split('\n')

"""使用RAGAS v0.2+评估RAG系统"""
from ragas import evaluate, EvaluationDataset, SingleTurnSample
from ragas.metrics import (
    Faithfulness,
    ResponseRelevancy,
    LLMContextPrecisionWithoutReference,
    LLMContextRecall,
)

# 准备评估数据集（RAGAS v0.2使用SingleTurnSample）
samples = [
    SingleTurnSample(
        user_input="什么是RAG？",
        response="RAG是检索增强生成，通过检索外部知识库来增强大语言模型的回答，减少幻觉问题。",
        retrieved_contexts=[
            "RAG（Retrieval-Augmented Generation）通过检索外部知识增强LLM回答，有效减少幻觉。"
        ],
        reference="RAG是一种将信息检索与文本生成结合的技术，通过检索外部知识库中的相关文档来增强大语言模型的回答质量。",
    ),
    SingleTurnSample(
        user_input="BM25算法的核心思想是什么？",
        response="BM25基于词频（TF）和逆文档频率（IDF）计算查询与文档的相关性得分，并考虑文档长度归一化。",
        retrieved_contexts=[
            "BM25是基于概率的检索模型，使用TF-IDF变体计算相关性，考虑了词频饱和度和文档长度归一化。"
        ],
        reference="BM25基于词频和逆文档频率计算文档与查询的相关性，核心是TF饱和度和文档长度归一化。",
    ),
    SingleTurnSample(
        user_input="为什么需要Reranker？",
        response="Reranker使用Cross-Encoder对检索结果精排，因为初始检索使用的Bi-Encoder精度有限。",
        retrieved_contexts=[
            "Cross-Encoder将query和文档拼接后联合编码，比Bi-Encoder捕捉更精细的交互信息。"
        ],
        reference="Reranker使用Cross-Encoder对初检结果进行精细排序，解决Bi-Encoder在语义交互上的精度不足。",
    ),
]

eval_dataset = EvaluationDataset(samples=samples)

# 运行评估（需要配置LLM作为评估器，如OpenAI API）
# import os
# os.environ["OPENAI_API_KEY"] = "your-api-key"

# RAGAS v0.2使用类实例而非模块级对象
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        Faithfulness(),              # 生成内容是否忠实于上下文
        ResponseRelevancy(),         # 答案是否相关
        LLMContextPrecisionWithoutReference(),  # 上下文精度
        LLMContextRecall(),          # 上下文召回
    ],
)

print(results)
# 输出示例：
# {'faithfulness': 0.92, 'response_relevancy': 0.88,
#  'llm_context_precision_without_reference': 0.85, 'llm_context_recall': 0.90}

# 查看每条数据的详细得分
df = results.to_pandas()
print(df[['user_input', 'faithfulness', 'response_relevancy']].to_string())

"""轻量级RAG评估（不依赖LLM API）"""
import numpy as np
from typing import List, Set
from rouge_score import rouge_scorer

def retrieval_recall_at_k(retrieved_ids: List[int],
                          relevant_ids: Set[int], k: int) -> float:
    """Recall@K: Top-K中召回的相关文档比例"""
    retrieved_at_k = set(retrieved_ids[:k])
    return len(retrieved_at_k & relevant_ids) / len(relevant_ids)

def mrr(ranked_results: List[List[int]],
        ground_truths: List[Set[int]]) -> float:
    """MRR: 首个相关结果的平均排名倒数"""
    rr_sum = 0.0
    for ranked, truth in zip(ranked_results, ground_truths):  # zip按位置配对
        for rank, doc_id in enumerate(ranked, 1):
            if doc_id in truth:
                rr_sum += 1.0 / rank
                break
    return rr_sum / len(ranked_results)

def answer_similarity(prediction: str, reference: str) -> dict:
    """基于 ROUGE 的答案相似度评估；更适合“有参考答案”的摘要/FAQ 类场景"""
    scorer = rouge_scorer.RougeScorer(
        ['rouge1', 'rouge2', 'rougeL'], use_stemmer=False
    )
    scores = scorer.score(reference, prediction)
    return {
        'rouge1': scores['rouge1'].fmeasure,
        'rouge2': scores['rouge2'].fmeasure,
        'rougeL': scores['rougeL'].fmeasure,
    }

# 使用示例
retrieved = [3, 1, 5, 2, 4]
relevant = {1, 3}
print(f"Recall@3: {retrieval_recall_at_k(retrieved, relevant, k=3)}")  # 1.0
print(f"Recall@1: {retrieval_recall_at_k(retrieved, relevant, k=1)}")  # 0.5

similarity = answer_similarity(
    prediction="RAG通过检索外部知识增强LLM回答",
    reference="RAG是检索增强生成技术，利用外部知识库提升LLM回答质量"
)
print(f"ROUGE-L: {similarity['rougeL']:.4f}")