前言
前兩篇解決了「搜尋」的問題。這篇要解決兩個更深層的問題：
- 問題本身就難以直接搜尋：複雜問題需要先轉換或拆解
- Context 品質不夠純粹：塞給 LLM 的資訊裡有太多噪音
這裡介紹三個技術：Step-Back Prompting（退一步提問）、Self-RAG（自我反思 RAG）、Context Compression（上下文壓縮）。
技術 1：查詢轉換（Query Transformation）
問題：複雜問題無法直接搜尋
有些問題不適合直接拿來做向量搜尋：
❌ 直接搜尋困難的問題類型：
「為什麼我們公司的 API 最近變慢了？」
→ 問題太具體，知識庫不可能有這個答案
「除了 Redis，還有哪些快取方案？」
→ 包含否定條件，搜尋容易找到 Redis 的文章
「比較 PostgreSQL 和 MySQL 的優缺點，然後說明哪個適合我們的場景」
→ 多個子問題，一次搜尋無法全部覆蓋
Step-Back Prompting（退一步提問）
核心思想：先把具體問題「退一步」抽象化，搜尋更通用的背景知識，再回答具體問題。
具體問題：「Estrogen 會影響 BRCA1 的 transcription 嗎？」
退一步：「BRCA1 基因的調控機制是什麼？」
→ 先搜尋通用知識，再回答具體問題
1import openai
2
3client = openai.OpenAI(api_key="your-api-key")
4
5
6def step_back_query(original_query: str) -> str:
7 """生成一個比原始問題更抽象的退一步問題"""
8 prompt = f"""你是一個搜尋查詢優化助手。
9將以下具體問題「退一步」，改寫成一個更通用、更基本的問題，
10這個更通用的問題可以幫助找到回答原始問題所需的背景知識。
11
12具體問題：{original_query}
13
14更通用的退一步問題（只輸出問題，不要解釋）："""
15
16 response = client.chat.completions.create(
17 model="gpt-4o-mini",
18 messages=[{"role": "user", "content": prompt}],
19 temperature=0,
20 )
21 return response.choices[0].message.content.strip()
22
23
24def rag_with_step_back(query: str, retriever) -> str:
25 """Step-Back RAG pipeline"""
26 # 退一步問題
27 abstract_query = step_back_query(query)
28 print(f"🔍 原始問題：{query}")
29 print(f"🔙 退一步問題：{abstract_query}")
30
31 # 用兩個問題分別搜尋
32 specific_chunks = retriever.retrieve(query, top_k=3)
33 abstract_chunks = retriever.retrieve(abstract_query, top_k=2)
34
35 # 合併 context
36 context = "\n\n---\n\n".join(specific_chunks + abstract_chunks)
37
38 prompt = f"""請根據以下背景知識回答問題。
39
40【背景知識】
41{context}
42
43【問題】{query}
44
45請先概述相關背景，再回答具體問題："""
46
47 response = client.chat.completions.create(
48 model="gpt-4o",
49 messages=[{"role": "user", "content": prompt}],
50 temperature=0,
51 )
52 return response.choices[0].message.content
Query Decomposition（問題分解）
對複雜的多步驟問題，先分解成子問題再各自搜尋：
1import json
2
3def decompose_query(complex_query: str) -> list[str]:
4 """將複雜問題分解成可獨立搜尋的子問題"""
5 prompt = f"""將以下複雜問題分解成 2-4 個簡單的子問題，
6每個子問題都可以獨立在知識庫中搜尋。
7以 JSON 陣列格式輸出，例如：["子問題1", "子問題2"]
8
9複雜問題：{complex_query}
10
11子問題 JSON："""
12
13 response = client.chat.completions.create(
14 model="gpt-4o-mini",
15 messages=[{"role": "user", "content": prompt}],
16 temperature=0,
17 )
18 try:
19 return json.loads(response.choices[0].message.content)
20 except json.JSONDecodeError:
21 return [complex_query] # fallback：退回原始問題
22
23
24def rag_with_decomposition(complex_query: str, retriever) -> str:
25 """問題分解 RAG pipeline"""
26 sub_questions = decompose_query(complex_query)
27 print(f"📋 子問題：")
28 for q in sub_questions:
29 print(f" - {q}")
30
31 # 每個子問題分別搜尋，收集所有 context
32 all_chunks = []
33 for sub_q in sub_questions:
34 chunks = retriever.retrieve(sub_q, top_k=2)
35 all_chunks.extend(chunks)
36
37 # 去重
38 unique_chunks = list(dict.fromkeys(all_chunks))
39 context = "\n\n---\n\n".join(unique_chunks)
40
41 prompt = f"""請根據以下參考資料，完整回答這個複雜問題。
42
43【參考資料】
44{context}
45
46【問題】{complex_query}"""
47
48 response = client.chat.completions.create(
49 model="gpt-4o",
50 messages=[{"role": "user", "content": prompt}],
51 temperature=0,
52 )
53 return response.choices[0].message.content
54
55
56# 示範
57# complex_q = "比較 Redis 和 Memcached 的架構差異，並說明各自適合哪些使用場景？"
58# answer = rag_with_decomposition(complex_q, retriever)
優點：對複雜、多跳（multi-hop）問題效果顯著
缺點：多次 LLM 呼叫，延遲和成本增加
最佳使用場景：分析類問題、比較類問題、需要推理的多步驟問題
技術 2：Self-RAG（自我反思 RAG）
核心思想
Naive RAG 盲目地把所有檢索到的 chunk 塞給 LLM，不管它們是否真的相關。
Self-RAG 讓 LLM 在整個流程中「自我反思」：
- 是否需要檢索？（有些問題 LLM 直接回答就好）
- 檢索到的文件是否相關？（過濾不相關的 chunk）
- 生成的回答是否有充分根據？（檢查答案是否基於 context）
- 回答是否有用？（自我評分）
問題
↓
[ISREL?] 這個問題需要外部知識嗎？
├── No → 直接生成答案
└── Yes → 檢索文件
↓
[ISREL] 每個文件相關嗎？過濾不相關的
↓
生成多個草稿答案
↓
[ISSUP] 答案有文件支持嗎？
↓
[ISUSE] 哪個答案最有用？
↓
輸出最佳答案
簡化版 Self-RAG 實作
1import openai
2
3client = openai.OpenAI(api_key="your-api-key")
4
5
6def needs_retrieval(question: str) -> bool:
7 """判斷問題是否需要外部知識"""
8 prompt = f"""判斷以下問題是否需要查詢外部知識庫才能回答（Yes/No）。
9
10需要外部知識的例子：詢問具體事實、最新資訊、特定領域知識
11不需要外部知識的例子：數學計算、常識性問題、語言翻譯
12
13問題：{question}
14回答（只輸出 Yes 或 No）："""
15
16 response = client.chat.completions.create(
17 model="gpt-4o-mini",
18 messages=[{"role": "user", "content": prompt}],
19 temperature=0,
20 max_tokens=5,
21 )
22 return response.choices[0].message.content.strip().lower() == "yes"
23
24
25def is_relevant(question: str, chunk: str) -> bool:
26 """判斷文件片段是否與問題相關"""
27 prompt = f"""判斷以下【文件片段】是否包含回答【問題】的有用資訊（Yes/No）。
28
29【問題】：{question}
30【文件片段】：{chunk[:300]}
31
32是否相關（只輸出 Yes 或 No）："""
33
34 response = client.chat.completions.create(
35 model="gpt-4o-mini",
36 messages=[{"role": "user", "content": prompt}],
37 temperature=0,
38 max_tokens=5,
39 )
40 return response.choices[0].message.content.strip().lower() == "yes"
41
42
43def is_grounded(answer: str, context: str) -> bool:
44 """判斷答案是否有 context 的支持，而非憑空捏造"""
45 prompt = f"""判斷以下【回答】是否完全基於【參考資料】中的內容，
46沒有加入參考資料以外的資訊（Yes/No）。
47
48【參考資料】：{context[:500]}
49【回答】：{answer[:300]}
50
51是否有根據（只輸出 Yes 或 No）："""
52
53 response = client.chat.completions.create(
54 model="gpt-4o-mini",
55 messages=[{"role": "user", "content": prompt}],
56 temperature=0,
57 max_tokens=5,
58 )
59 return response.choices[0].message.content.strip().lower() == "yes"
60
61
62def self_rag(question: str, retriever, max_retries: int = 2) -> str:
63 """Self-RAG pipeline"""
64
65 # Step 1: 是否需要檢索？
66 if not needs_retrieval(question):
67 print("💭 直接回答（不需要外部知識）")
68 response = client.chat.completions.create(
69 model="gpt-4o",
70 messages=[{"role": "user", "content": question}],
71 temperature=0,
72 )
73 return response.choices[0].message.content
74
75 print("🔍 需要外部知識，開始檢索...")
76
77 for attempt in range(max_retries):
78 # Step 2: 檢索文件
79 raw_chunks = retriever.retrieve(question, top_k=5)
80
81 # Step 3: 過濾不相關的文件
82 relevant_chunks = [c for c in raw_chunks if is_relevant(question, c)]
83 print(f" 📄 過濾後相關 chunk 數量：{len(relevant_chunks)}/{len(raw_chunks)}")
84
85 if not relevant_chunks:
86 if attempt < max_retries - 1:
87 print(" ⚠️ 找不到相關文件，嘗試重新查詢...")
88 continue
89 return "根據現有資料，無法回答此問題。"
90
91 context = "\n\n---\n\n".join(relevant_chunks)
92
93 # Step 4: 生成回答
94 prompt = f"""請根據以下參考資料回答問題。
95只使用參考資料中的資訊，不要加入其他知識。
96
97【參考資料】
98{context}
99
100【問題】{question}"""
101
102 answer = client.chat.completions.create(
103 model="gpt-4o",
104 messages=[{"role": "user", "content": prompt}],
105 temperature=0,
106 ).choices[0].message.content
107
108 # Step 5: 檢查答案是否有根據
109 if is_grounded(answer, context):
110 print(" ✅ 答案有文件支持")
111 return answer
112 else:
113 print(f" ⚠️ 答案可能包含幻覺，嘗試 {attempt + 1}/{max_retries}")
114
115 return answer # 即使可能有問題，還是返回最後一次的答案
116
117
118# 使用範例
119# answer = self_rag("Python 的 GIL 是什麼？", retriever)
120# answer = self_rag("2 + 2 等於多少？", retriever) # 不需要外部知識的問題
優點：大幅降低幻覺；對不需要 RAG 的問題不浪費搜尋資源；有自我品質把關
缺點：多次 LLM 呼叫，延遲顯著增加；每步判斷的準確度依賴模型能力
最佳使用場景：高可信度要求的問答（醫療、法律、財務）；需要精確溯源的系統
技術 3：Context Compression（上下文壓縮）
核心問題
向量搜尋取回的 chunk 通常包含大量與當前問題無關的「噪音」：
問題：「asyncio 的事件迴圈如何工作？」
取回的 chunk：
「Python 在 3.4 版本引入了 asyncio 模組，
提供非同步 I/O 的支援。asyncio 的核心是事件迴圈（Event Loop），
它透過單執行緒的方式調度協程（coroutine）。
Python 的 asyncio 可以搭配 aiohttp 做非同步 HTTP 請求，
也可以搭配 asyncpg 做非同步資料庫操作。此外，
Python 3.11 對 asyncio 的效能做了大幅改進...」
↑ 很多內容（aiohttp、asyncpg、效能改進）跟問題無關，但都會佔用 context window
Context Compression 的目標：只保留 chunk 中與問題直接相關的部分。
方法一：LLM 提取壓縮
1def compress_with_llm(question: str, chunk: str) -> str | None:
2 """用 LLM 從 chunk 中提取與問題相關的句子"""
3 prompt = f"""從以下【文件片段】中，只提取出與【問題】直接相關的句子。
4如果整個片段都不相關，輸出「不相關」。
5只輸出相關句子，不要加任何解釋或格式。
6
7【問題】：{question}
8【文件片段】：{chunk}
9
10相關句子："""
11
12 response = client.chat.completions.create(
13 model="gpt-4o-mini",
14 messages=[{"role": "user", "content": prompt}],
15 temperature=0,
16 max_tokens=300,
17 )
18 result = response.choices[0].message.content.strip()
19 return None if "不相關" in result else result
20
21
22def rag_with_compression(question: str, retriever, top_k: int = 6) -> str:
23 """帶 Context Compression 的 RAG pipeline"""
24 # 故意取多一點 chunk（因為壓縮後會過濾掉部分）
25 raw_chunks = retriever.retrieve(question, top_k=top_k)
26
27 # 壓縮每個 chunk
28 compressed = []
29 for chunk in raw_chunks:
30 result = compress_with_llm(question, chunk)
31 if result:
32 compressed.append(result)
33
34 print(f"📉 壓縮：{len(raw_chunks)} chunks → {len(compressed)} 個相關片段")
35
36 if not compressed:
37 return "找不到相關資料。"
38
39 context = "\n\n".join(compressed)
40
41 prompt = f"""根據以下資料回答問題：
42
43{context}
44
45問題：{question}"""
46
47 return client.chat.completions.create(
48 model="gpt-4o",
49 messages=[{"role": "user", "content": prompt}],
50 temperature=0,
51 ).choices[0].message.content
方法二：嵌入相似度過濾（無需 LLM 呼叫）
1import numpy as np
2
3def compress_by_sentence_similarity(
4 question: str, chunk: str, threshold: float = 0.75
5) -> str:
6 """把 chunk 切成句子，過濾掉跟問題語意距離遠的句子"""
7 import re
8
9 # 切成句子（簡單版：按句號切）
10 sentences = re.split(r'[。！？\.\!\?]', chunk)
11 sentences = [s.strip() for s in sentences if len(s.strip()) > 10]
12
13 if not sentences:
14 return chunk
15
16 # 取得問題和所有句子的 embedding
17 all_texts = [question] + sentences
18 embeddings = np.array([get_embedding(t) for t in all_texts])
19
20 q_emb = embeddings[0]
21 s_embs = embeddings[1:]
22
23 # 計算每個句子與問題的相似度
24 sims = s_embs @ q_emb / (
25 np.linalg.norm(s_embs, axis=1) * np.linalg.norm(q_emb) + 1e-9
26 )
27
28 # 只保留相似度超過閾值的句子
29 relevant_sentences = [s for s, sim in zip(sentences, sims) if sim >= threshold]
30
31 return "。".join(relevant_sentences) if relevant_sentences else ""
方法三：LangChain ContextualCompressionRetriever
1from langchain.retrievers import ContextualCompressionRetriever
2from langchain.retrievers.document_compressors import LLMChainExtractor
3from langchain_openai import ChatOpenAI, OpenAIEmbeddings
4from langchain_community.vectorstores import Chroma
5
6llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
7base_retriever = Chroma(
8 embedding_function=OpenAIEmbeddings()
9).as_retriever(search_kwargs={"k": 6})
10
11# 建立壓縮器
12compressor = LLMChainExtractor.from_llm(llm)
13
14# 包裝成壓縮式 Retriever
15compression_retriever = ContextualCompressionRetriever(
16 base_compressor=compressor,
17 base_retriever=base_retriever,
18)
19
20# 使用方式跟一般 retriever 相同
21docs = compression_retriever.invoke("asyncio 的事件迴圈如何工作？")
22for doc in docs:
23 print(doc.page_content)
優點：減少 LLM 的 context 長度 → 降低 token 成本、提升回答焦點；有效過濾噪音
缺點：LLM 壓縮方法增加延遲；Embedding 方法可能過度過濾
最佳使用場景：文件較長且包含混雜主題；context window 有限；需要降低 LLM token 成本
組合使用：完整的進階 RAG Pipeline
把這篇和上一篇的技術組合，可以建立一個健壯的生產級 RAG：
1class AdvancedRAGPipeline:
2 """結合 Query Decomposition + Hybrid Search + Reranker + Context Compression"""
3
4 def __init__(self, chunks: list[str]):
5 # 初始化各元件（略，見前幾篇）
6 self.hybrid_retriever = HybridRetriever(chunks)
7 self.reranker = CrossEncoder("BAAI/bge-reranker-base")
8
9 def run(self, question: str) -> dict:
10 # 1. 判斷是否需要問題分解
11 sub_questions = decompose_query(question)
12 is_complex = len(sub_questions) > 1
13
14 if is_complex:
15 print(f"🧩 複雜問題，分解為 {len(sub_questions)} 個子問題")
16 all_candidates = []
17 for sub_q in sub_questions:
18 chunks = self.hybrid_retriever.hybrid_search(sub_q, top_k=10)
19 all_candidates.extend(chunks)
20 candidates = list(dict.fromkeys(all_candidates))[:20]
21 else:
22 candidates = self.hybrid_retriever.hybrid_search(question, top_k=20)
23
24 # 2. Rerank
25 pairs = [[question, c] for c in candidates]
26 scores = self.reranker.predict(pairs)
27 ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
28 top_chunks = [c for c, _ in ranked[:6]]
29
30 # 3. Context Compression
31 compressed = []
32 for chunk in top_chunks:
33 result = compress_with_llm(question, chunk)
34 if result:
35 compressed.append(result)
36
37 if not compressed:
38 return {"answer": "找不到足夠的相關資料。", "sources": []}
39
40 context = "\n\n---\n\n".join(compressed)
41
42 # 4. 生成最終答案
43 prompt = f"""根據以下參考資料詳細回答問題。
44
45【參考資料】
46{context}
47
48【問題】{question}"""
49
50 answer = client.chat.completions.create(
51 model="gpt-4o",
52 messages=[{"role": "user", "content": prompt}],
53 temperature=0,
54 ).choices[0].message.content
55
56 return {"answer": answer, "context_used": compressed}
小結
| 技術 | 核心思想 | 解決的問題 |
|---|---|---|
| Step-Back Prompting | 先抽象再具體 | 具體問題難以直接搜尋 |
| Query Decomposition | 分而治之 | 複雜多跳問題 |
| Self-RAG | 讓 LLM 自我把關 | 幻覺、不相關資料混入 |
| Context Compression | 只保留相關句子 | 噪音過多、token 浪費 |
下一篇（最終篇），我們進入生產級 RAG 評估：如何量化你的 RAG 品質？RAGAS 指標是什麼？以及 Agentic RAG——當 RAG 和 AI Agent 結合，能做到什麼？
系列導覽