高级 RAG 技术

检索层问题（最常见）：
  - 块切得太碎：语义被截断，检索时找不到完整答案
  - 向量匹配失败：用户问"涨价了多少"，文档写"调整了定价策略"
  - Top-K 质量差：相关文档排在第 5 条，但只取了 Top-3

生成层问题：
  - 上下文太长：把 10 个 chunk 全塞进去，模型注意力被稀释
  - 无关噪声：检索到的内容和问题只是词汇相关，不是语义相关

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector

# 混合检索：将向量分数和 BM25 分数用 RRF 融合
def hybrid_search(query: str, top_k: int = 10):
    # 向量检索
    dense_results = qdrant.search(
        collection_name="docs",
        query_vector=embed(query),
        limit=top_k * 2,
    )
    # 稀疏向量（BM25）检索
    sparse_results = qdrant.search(
        collection_name="docs",
        query_vector=NamedSparseVector(
            name="text",
            vector=SparseVector(**bm25_encode(query))
        ),
        limit=top_k * 2,
    )
    # Reciprocal Rank Fusion（RRF）融合排序
    return reciprocal_rank_fusion(dense_results, sparse_results, k=60)[:top_k]


def reciprocal_rank_fusion(list_a, list_b, k=60):
    scores = {}
    for rank, doc in enumerate(list_a):
        scores[doc.id] = scores.get(doc.id, 0) + 1 / (rank + k)
    for rank, doc in enumerate(list_b):
        scores[doc.id] = scores.get(doc.id, 0) + 1 / (rank + k)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

def rerank(query: str, candidates: list[str], top_n: int = 3) -> list[str]:
    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_n]]

# 流程：向量检索 Top-20 → Reranker 精选 Top-3 → 传给 LLM
chunks = vector_search(query, top_k=20)
best_chunks = rerank(query, chunks, top_n=3)

import anthropic

client = anthropic.Anthropic()

def hyde_search(question: str, top_k: int = 5) -> list:
    # Step 1：生成假设性文档
    resp = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": (
                f"请为下面的问题写一段可能出现在官方文档中的回答（2-3句话），"
                f"不需要准确，只需要风格和术语接近真实文档。\n问题：{question}"
            )
        }]
    )
    hypothetical_doc = resp.content[0].text

    # Step 2：用假设文档向量化后检索
    results = vector_search(hypothetical_doc, top_k=top_k)
    return results

def multi_query_search(question: str, top_k: int = 5) -> list:
    resp = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": (
                f"为下面的问题生成 3 种不同的改写方式，用于检索文档。"
                f"每行一个，只输出改写后的问题，不要编号和解释。\n原问题：{question}"
            )
        }]
    )
    rewrites = resp.content[0].text.strip().split("\n")
    queries = [question] + rewrites[:3]

    # 对所有查询检索，去重合并
    seen = set()
    results = []
    for q in queries:
        for chunk in vector_search(q, top_k=top_k):
            if chunk.id not in seen:
                seen.add(chunk.id)
                results.append(chunk)
    return results

原始文档（叶节点）
    ↓ 聚类 + 摘要
中层摘要（若干节的概述）
    ↓ 再聚类 + 摘要
顶层摘要（全文概述）

from sklearn.cluster import KMeans
import numpy as np

def build_raptor_tree(chunks: list[str], levels: int = 3) -> dict:
    """构建 RAPTOR 树状摘要索引"""
    current_level = chunks
    tree = {0: chunks}

    for level in range(1, levels + 1):
        embeddings = np.array([embed(c) for c in current_level])
        n_clusters = max(1, len(current_level) // 5)
        kmeans = KMeans(n_clusters=n_clusters).fit(embeddings)

        summaries = []
        for cluster_id in range(n_clusters):
            cluster_chunks = [current_level[i] for i, label in
                              enumerate(kmeans.labels_) if label == cluster_id]
            cluster_text = "\n\n".join(cluster_chunks)

            resp = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=300,
                messages=[{"role": "user",
                           "content": f"请将以下内容概括为 200 字以内的摘要：\n{cluster_text}"}]
            )
            summaries.append(resp.content[0].text)

        tree[level] = summaries
        current_level = summaries

    return tree

def self_rag(question: str) -> str:
    # Step 1：判断是否需要检索
    decision_resp = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=10,
        messages=[{
            "role": "user",
            "content": (
                f"以下问题是否需要查阅外部文档才能回答？"
                f"只回答 yes 或 no。\n问题：{question}"
            )
        }]
    )
    need_retrieval = decision_resp.content[0].text.strip().lower() == "yes"

    if not need_retrieval:
        # 直接回答，无需检索
        return direct_answer(question)

    # Step 2：检索 + 评估相关性
    chunks = hybrid_search(question, top_k=5)
    relevant_chunks = []
    for chunk in chunks:
        relevance_resp = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=10,
            messages=[{
                "role": "user",
                "content": (
                    f"以下段落是否与问题相关？只回答 yes 或 no。\n"
                    f"问题：{question}\n段落：{chunk.text}"
                )
            }]
        )
        if relevance_resp.content[0].text.strip().lower() == "yes":
            relevant_chunks.append(chunk)

    # Step 3：基于相关内容生成回答
    return generate_with_context(question, relevant_chunks)

传统 RAG：问题 → 向量检索 → 相关 chunks → 生成
GraphRAG：问题 → 实体识别 → 图遍历（多跳） → 相关子图 + chunks → 生成

"X 公司的 CEO 和 Y 公司有什么合作关系？"
→ 识别实体：X公司.CEO、Y公司
→ 图遍历：X公司 --CEO--> 张三 --合作--> Y公司
→ 回答基于图路径，而非单纯文本匹配

# 用 RAGAS 快速评估
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall

result = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_recall]
)
print(result)