Advanced RAG 完全指南：從基礎到進階的 10 倍效能提升

# 需要檢索
"公司的部署流程是什麼？" → ✅ 需要查文件

# 不需要檢索  
"你好！" → ❌ 閒聊，直接回答
"2+2等於多少？" → ❌ 常識，直接回答

def need_retrieval(query):
    """簡單的規則判斷"""
    # 問候語
    greetings = ["你好", "嗨", "hello", "hi"]
    if any(word in query.lower() for word in greetings):
        return False
    
    # 數學計算
    if any(op in query for op in ["+", "-", "*", "/", "等於"]):
        return False
    
    # 包含疑問詞 → 可能需要檢索
    question_words = ["什麼", "如何", "為什麼", "怎麼", "哪裡"]
    if any(word in query for word in question_words):
        return True
    
    return True  # 預設需要檢索

def llm_need_retrieval(query):
    """用 LLM 判斷是否需要檢索"""
    prompt = f"""
判斷以下問題是否需要查詢知識庫。

規則：
- 閒聊、問候 → 不需要
- 常識問題 → 不需要
- 需要特定領域知識 → 需要

問題：{query}

回答（只說「需要」或「不需要」）：
"""
    response = llm.generate(prompt)
    return "需要" in response

對話歷史：
使用者：「Django 是什麼？」
AI：「Django 是 Python 網頁框架...」
使用者：「它怎麼安裝？」← 「它」指什麼？

# 原始查詢（失敗）
query = "它怎麼安裝"
→ 檢索結果可能找到各種「安裝」的文件

# 重寫後（成功）
rewritten = "Django 怎麼安裝"
→ 精準找到 Django 安裝文件

# 原始查詢（口語）
"那個 Python 框架咋搞啊？"

# 重寫後（正式）
"如何使用 Python 網頁框架 Django"

# 原始查詢（複雜）
"Django 和 Flask 哪個好，怎麼選，學習曲線如何？"

# 分解為多個子查詢
sub_queries = [
    "Django 與 Flask 的比較",
    "Django 與 Flask 的選擇建議",
    "Django 的學習曲線",
    "Flask 的學習曲線"
]
→ 每個子查詢分別檢索，結果更全面

def rewrite_query_with_history(query, chat_history):
    """根據對話歷史重寫查詢"""
    prompt = f"""
對話歷史：
{chat_history}

當前問題：
{query}

任務：
1. 如果問題中有代名詞（它、這個、那個），替換為具體名稱
2. 補充缺失的上下文
3. 使用正式用語

重寫後的查詢：
"""
    return llm.generate(prompt)

# 使用範例
chat_history = """
使用者：Django 是什麼？
AI：Django 是 Python 網頁框架...
"""

query = "它怎麼安裝？"
rewritten = rewrite_query_with_history(query, chat_history)
# 輸出："Django 如何安裝"

def hypothetical_document_embedding(query):
    """
    HyDE：讓 LLM 先生成「假設的答案」
    然後用這個答案去檢索（效果更好）
    """
    prompt = f"""
假設你要回答這個問題：{query}

請寫一段簡短的答案（2-3 句話）：
"""
    hypothetical_answer = llm.generate(prompt)
    
    # 用假設的答案檢索（而非原始問題）
    return hypothetical_answer

# 範例
query = "Django 怎麼安裝？"
hypo_doc = hypothetical_document_embedding(query)
# 輸出："使用 pip install django 安裝 Django。
#       首先確保已安裝 Python..."

# 用這段文字去檢索，會找到更相關的文檔！

def generate_multiple_queries(query):
    """生成多個不同角度的查詢"""
    prompt = f"""
原始問題：{query}

請生成 3 個不同角度的相關查詢，用於搜尋更全面的資訊：

1. [同義改寫]
2. [更具體的問法]
3. [相關的延伸問題]
"""
    response = llm.generate(prompt)
    queries = parse_queries(response)
    return queries

# 範例
query = "Django 好用嗎？"
queries = generate_multiple_queries(query)
# 輸出：
# 1. Django 框架的優缺點
# 2. Django 適合什麼類型的專案
# 3. Django 與其他框架的比較

# 多個查詢一起檢索，結果更全面

# 你的文檔
"""
...
Django 的部署流程包括三個步驟。
首先，配置生產環境的設定檔。
其次，執行 python manage.py migrate 建立資料表。
最後，使用 gunicorn 啟動應用。
...
"""

# 基礎 RAG 只檢索到
"執行 python manage.py migrate 建立資料表。"

# 問題
❌ 使用者不知道這是第二步
❌ 使用者不知道之前要做什麼
❌ 使用者不知道之後要做什麼

class SentenceWindowRetriever:
    """檢索時自動包含前後句子"""
    
    def __init__(self, documents, window_size=3):
        """
        window_size: 前後各包含幾句
        """
        self.window_size = window_size
        self.sentences = self._split_into_sentences(documents)
        
    def retrieve(self, query, top_k=5):
        # 1. 檢索最相關的句子
        relevant_sentences = self._vector_search(query, top_k)
        
        # 2. 擴充：加入前後句子
        enriched_results = []
        for sent_idx in relevant_sentences:
            start = max(0, sent_idx - self.window_size)
            end = min(len(self.sentences), sent_idx + self.window_size + 1)
            
            context = self.sentences[start:end]
            enriched_results.append({
                'main_sentence': self.sentences[sent_idx],
                'context': ' '.join(context)
            })
        
        return enriched_results

# 使用
retriever = SentenceWindowRetriever(documents, window_size=2)
results = retriever.retrieve("如何建立資料表")

# 結果（自動包含前後 2 句）：
"""
Django 的部署流程包括三個步驟。
首先，配置生產環境的設定檔。
其次，執行 python manage.py migrate 建立資料表。← 核心句子
最後，使用 gunicorn 啟動應用。
記得檢查 ALLOWED_HOSTS 設定。
"""

class ParentDocumentRetriever:
    """
    索引：小片段（精準檢索）
    返回：大片段（完整上下文）
    """
    
    def __init__(self):
        self.small_chunks = []  # 用於檢索
        self.parent_chunks = [] # 用於返回
        self.chunk_to_parent = {} # 映射關係
    
    def add_document(self, document):
        # 1. 切分成大片段（父文檔）
        parent_chunks = split_text(document, chunk_size=2000)
        
        # 2. 每個大片段再切成小片段
        for parent_idx, parent_chunk in enumerate(parent_chunks):
            small_chunks = split_text(parent_chunk, chunk_size=500)
            
            for small_chunk in small_chunks:
                small_idx = len(self.small_chunks)
                self.small_chunks.append(small_chunk)
                self.chunk_to_parent[small_idx] = parent_idx
        
        self.parent_chunks = parent_chunks
    
    def retrieve(self, query, top_k=3):
        # 1. 用小片段檢索（精準）
        small_indices = self._vector_search(query, top_k)
        
        # 2. 返回對應的大片段（完整）
        parent_indices = [self.chunk_to_parent[i] for i in small_indices]
        return [self.parent_chunks[i] for i in set(parent_indices)]

# 優勢
✅ 檢索精準（小片段匹配度高）
✅ 上下文完整（返回大片段）

# 場景：搜尋「Django ORM 教學」

# BM25 結果（關鍵字匹配）
1. Django ORM 完整指南 ✓（完美匹配）
2. ORM 是什麼？ ~（部分匹配）
3. Django 與 SQLAlchemy 的 ORM 比較 ✓（相關）

# 向量搜尋結果（語義理解）
1. Django 資料庫操作入門 ✓（語義相關）
2. Python ORM 框架介紹 ~（語義相關但不夠具體）
3. Django Model 使用教學 ✓（高度相關）

# 混合結果（最佳）
1. Django ORM 完整指南 ✓✓（兩者都高分）
2. Django 資料庫操作入門 ✓（向量高分）
3. Django Model 使用教學 ✓（向量高分）

from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

class HybridRetriever:
    def __init__(self, documents, alpha=0.5):
        """
        alpha: 權重參數
        - 0.0 = 完全 BM25
        - 1.0 = 完全向量
        - 0.5 = 各半
        """
        self.documents = documents
        self.alpha = alpha
        
        # 初始化 BM25
        tokenized_docs = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        
        # 初始化向量搜尋
        self.encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
        self.doc_vectors = self.encoder.encode(documents)
    
    def retrieve(self, query, top_k=5):
        # 1. BM25 檢索
        bm25_scores = self.bm25.get_scores(query.split())
        bm25_scores = self._normalize(bm25_scores)
        
        # 2. 向量檢索
        query_vector = self.encoder.encode(query)
        vector_scores = cosine_similarity([query_vector], self.doc_vectors)[0]
        vector_scores = self._normalize(vector_scores)
        
        # 3. 融合分數
        final_scores = (
            (1 - self.alpha) * bm25_scores + 
            self.alpha * vector_scores
        )
        
        # 4. 排序並返回
        top_indices = np.argsort(final_scores)[::-1][:top_k]
        return [self.documents[i] for i in top_indices]
    
    def _normalize(self, scores):
        """將分數標準化到 0-1"""
        min_score = np.min(scores)
        max_score = np.max(scores)
        if max_score - min_score == 0:
            return scores
        return (scores - min_score) / (max_score - min_score)

# 使用
retriever = HybridRetriever(documents, alpha=0.7)
# alpha=0.7 表示 30% BM25 + 70% 向量
results = retriever.retrieve("Django ORM", top_k=5)

def reciprocal_rank_fusion(rankings_list, k=60):
    """
    RRF: 更穩健的融合方法
    不依賴原始分數，只看排名
    """
    fused_scores = {}
    
    for rankings in rankings_list:
        for rank, doc_id in enumerate(rankings):
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0
            
            # RRF 公式
            fused_scores[doc_id] += 1 / (k + rank + 1)
    
    # 排序
    sorted_docs = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_id for doc_id, score in sorted_docs]

# 使用
bm25_rankings = [3, 7, 1, 15, 22]  # BM25 的 Top-5 文檔 ID
vector_rankings = [1, 3, 9, 12, 18]  # 向量的 Top-5 文檔 ID

fused_rankings = reciprocal_rank_fusion([bm25_rankings, vector_rankings])
# 輸出：[1, 3, 7, 9, 12, ...] （融合後的排名）

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.vectorstores import Chroma

# BM25 檢索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 向量檢索器
vectordb = Chroma.from_documents(documents, embeddings)
vector_retriever = vectordb.as_retriever(search_kwargs={"k": 5})

# 混合檢索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # 30% BM25, 70% 向量
)

# 使用
results = ensemble_retriever.get_relevant_documents("Django ORM")

# 向量搜尋（快但粗糙）
問題："Django 如何處理文件上傳？"

初步檢索 Top 10：
1. Django 文件上傳教學 ← 相關度 95%
2. Django 表單處理 ← 相關度 70%
3. Python 文件操作 ← 相關度 60%
4. Django 處理 POST 請求 ← 相關度 75%
5. 文件上傳安全注意事項 ← 相關度 80%
...

# 問題：排序不夠精確
# 第 1 名是對的，但 2-10 名順序有誤

from sentence_transformers import CrossEncoder

class RerankerRetriever:
    def __init__(self, base_retriever):
        self.base_retriever = base_retriever
        
        # 初始化 Reranker
        self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
    
    def retrieve(self, query, top_k=5):
        # 階段 1：初步檢索 Top 50
        candidates = self.base_retriever.get_relevant_documents(
            query, 
            k=50
        )
        
        # 階段 2：Reranking
        # 為每個候選文檔計算與查詢的精確相關度
        pairs = [[query, doc.page_content] for doc in candidates]
        scores = self.reranker.predict(pairs)
        
        # 根據新分數排序
        ranked_indices = np.argsort(scores)[::-1][:top_k]
        return [candidates[i] for i in ranked_indices]

# 使用
retriever = RerankerRetriever(base_retriever)
results = retriever.retrieve("Django 文件上傳", top_k=5)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# 基礎檢索器
base_retriever = vectordb.as_retriever(search_kwargs={"k": 50})

# Cohere Reranker（高品質）
compressor = CohereRerank(
    model="rerank-multilingual-v2.0",
    top_n=5
)

# 組合
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

# 使用
results = reranking_retriever.get_relevant_documents("Django 文件上傳")
# 自動從 50 個候選中選出最相關的 5 個

def llm_rerank(query, candidates, top_k=5):
    """用 LLM 直接判斷相關性"""
    prompt = f"""
請為以下文檔評分（0-10），根據它們與查詢的相關程度。

查詢：{query}

文檔：
{format_candidates(candidates)}

請輸出 JSON 格式：
{{"scores": [8, 6, 9, 4, ...]}}
"""
    response = llm.generate(prompt)
    scores = json.loads(response)['scores']
    
    # 排序
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    return [candidates[i] for i in ranked_indices]

# 檢索到 5 個文檔，每個 1000 tokens
總計：5000 tokens

問題：
❌ 超過 LLM 的上下文限制（如 GPT-3.5: 4K）
❌ API 費用高（按 token 計費）
❌ 包含很多無關資訊，稀釋重點
❌ 生成速度慢

理想：
✅ 只保留 500-1000 tokens 的關鍵資訊
✅ 節省 80% 成本
✅ 提升回答品質

from sentence_transformers import SentenceTransformer, util

class SentenceCompressor:
    def __init__(self):
        self.encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
    
    def compress(self, query, documents, threshold=0.5):
        """只保留與查詢相關的句子"""
        query_vector = self.encoder.encode(query)
        
        compressed_docs = []
        for doc in documents:
            # 分句
            sentences = self._split_sentences(doc)
            
            # 計算每句與查詢的相似度
            sent_vectors = self.encoder.encode(sentences)
            similarities = util.cos_sim(query_vector, sent_vectors)[0]
            
            # 只保留相似度 > threshold 的句子
            relevant_sentences = [
                sent for sent, sim in zip(sentences, similarities)
                if sim > threshold
            ]
            
            compressed_docs.append(' '.join(relevant_sentences))
        
        return compressed_docs

# 使用
compressor = SentenceCompressor()
compressed = compressor.compress(
    query="Django ORM 是什麼",
    documents=retrieved_docs,
    threshold=0.5
)

# 結果：從 5000 tokens 壓縮到 800 tokens

def llm_compress(query, documents):
    """用 LLM 提取關鍵資訊"""
    combined_docs = '\n\n'.join(documents)
    
    prompt = f"""
以下是一些文檔片段：

{combined_docs}

任務：
1. 只提取與問題「{query}」相關的資訊
2. 移除無關內容
3. 保持原意，簡潔表達

精簡後的內容：
"""
    return llm.generate(prompt)

# 使用
compressed = llm_compress("Django ORM 是什麼", retrieved_docs)

from llmlingua import PromptCompressor

compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2",
    use_llmlingua2=True
)

# 壓縮
compressed_prompt = compressor.compress_prompt(
    context=retrieved_docs,
    instruction="",
    question=query,
    target_token=1000,  # 目標 token 數
    condition_compare=True,
    condition_in_question='after',
    rank_method='longllmlingua',
    dynamic_context_compression_ratio=0.4  # 壓縮率
)

# 結果
print(f"原始: {compressed_prompt['origin_tokens']} tokens")
print(f"壓縮後: {compressed_prompt['compressed_tokens']} tokens")
print(f"壓縮率: {compressed_prompt['ratio']:.1%}")

# 輸出範例：
# 原始: 5234 tokens
# 壓縮後: 1047 tokens
# 壓縮率: 80.0%

from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever

# 基礎檢索器
base_retriever = vectordb.as_retriever()

# LLM 壓縮器
compressor = LLMChainExtractor.from_llm(llm)

# 組合
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

# 使用（自動壓縮）
results = compression_retriever.get_relevant_documents(
    "Django ORM 是什麼"
)

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.retrievers import (
    EnsembleRetriever,
    ContextualCompressionRetriever,
    BM25Retriever
)
from langchain.retrievers.document_compressors import CohereRerank
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class AdvancedRAGSystem:
    """完整的 Advanced RAG 系統"""
    
    def __init__(self, documents):
        self.documents = documents
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        
        # 建立向量資料庫
        embeddings = OpenAIEmbeddings()
        self.vectordb = Chroma.from_documents(documents, embeddings)
        
        # 準備檢索器
        self._setup_retrievers()
    
    def _setup_retrievers(self):
        """設定混合檢索 + Reranking"""
        # 1. BM25 檢索器（關鍵字）
        bm25_retriever = BM25Retriever.from_documents(self.documents)
        bm25_retriever.k = 50
        
        # 2. 向量檢索器（語義）
        vector_retriever = self.vectordb.as_retriever(
            search_kwargs={"k": 50}
        )
        
        # 3. 混合檢索
        ensemble_retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, vector_retriever],
            weights=[0.3, 0.7]
        )
        
        # 4. Reranking
        compressor = CohereRerank(model="rerank-multilingual-v2.0", top_n=5)
        self.retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=ensemble_retriever
        )
    
    def _need_retrieval(self, query):
        """判斷是否需要檢索"""
        greetings = ["你好", "嗨", "hello"]
        if any(g in query.lower() for g in greetings):
            return False
        return True
    
    def _rewrite_query(self, query, chat_history=""):
        """查詢重寫"""
        if not chat_history:
            return query
        
        prompt = f"""
對話歷史：{chat_history}
當前問題：{query}

請重寫問題，補充代名詞和缺失的上下文：
"""
        rewritten = self.llm.predict(prompt)
        return rewritten.strip()
    
    def ask(self, query, chat_history=""):
        """
        完整的 Advanced RAG 流程
        
        Args:
            query: 使用者問題
            chat_history: 對話歷史（可選）
        """
        # ===== Pre-retrieval =====
        
        # 1. 檢索決策
        if not self._need_retrieval(query):
            return self.llm.predict(query)
        
        # 2. 查詢重寫
        rewritten_query = self._rewrite_query(query, chat_history)
        print(f"🔄 重寫查詢: {rewritten_query}")
        
        # 3. 混合檢索 + Reranking（自動完成）
        docs = self.retriever.get_relevant_documents(rewritten_query)
        
        # ===== Post-retrieval =====
        # Reranking 和壓縮已在 retriever 中自動完成
        
        # ===== 生成答案 =====
        prompt_template = """
你是專業的技術助手。請根據以下參考資料回答問題。

參考資料：
{context}

問題：{question}

回答：
"""
        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.retriever,
            chain_type_kwargs={"prompt": prompt}
        )
        
        result = qa_chain({"query": query})
        
        return {
            "answer": result["result"],
            "rewritten_query": rewritten_query,
            "sources": docs
        }


# ==================== 使用範例 ====================

# 初始化系統
rag = AdvancedRAGSystem(documents)

# 對話 1
response1 = rag.ask("Django 是什麼？")
print(response1["answer"])

# 對話 2（利用歷史）
chat_history = """
使用者：Django 是什麼？
AI：Django 是 Python 的網頁框架...
"""

response2 = rag.ask("它怎麼安裝？", chat_history=chat_history)
# 自動重寫為："Django 怎麼安裝？"
print(response2["answer"])

✅ 向量搜尋
✅ 基本 Prompt
→ 快速上線，驗證概念

✅ BM25 + 向量搜尋
✅ 提升關鍵字匹配
→ 準確率 +10%

✅ Cohere Rerank 或 Cross-Encoder
→ 準確率 +15%

✅ Query Rewriting
✅ Context Enrichment
→ 使用者體驗大幅提升

✅ 句子過濾或 LLMLingua
→ 成本降低 50-70%

1. Hybrid Search（免費）
2. Cross-Encoder Reranking（免費）
3. 句子過濾壓縮（免費）

成本：$0（只需 GPU 或 CPU）
效果：準確率 +20%

1. Hybrid Search
2. Cohere Rerank（付費但最好）
3. LLM 查詢重寫
4. LLMLingua 壓縮

成本：~$50/天（10K 查詢）
效果：準確率 +30%，使用者體驗最佳