Multi-Step RAG 完全指南：解決複雜問題的多跳檢索與生成

# 基礎 RAG 的流程
def basic_rag(query):
    # 1. 檢索一次
    docs = retrieve(query)
    
    # 2. 生成答案
    answer = llm.generate(f"根據：{docs}\n回答：{query}")
    
    return answer

# 對於 2-hop 問題
query = "Django 創始人的出生地？"
docs = retrieve(query)
# 找到：「Django 由 Adrian Holovaty 創建」
# 但沒有出生地資訊！

answer = llm.generate(...)
# LLM：「文檔中沒有提到出生地」
# ❌ 失敗

class FixedStepRAG:
    """固定步驟的 Multi-step RAG"""
    
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
    
    def answer_2hop(self, query):
        """
        處理 2-hop 問題
        例如：「Django 創始人的出生地？」
        """
        # Step 1：分解問題
        step1_query = self._extract_first_hop(query)
        # 「Django 的創始人是誰？」
        
        # Step 2：第一次檢索
        step1_docs = self.retriever.retrieve(step1_query)
        step1_answer = self._extract_answer(step1_docs, step1_query)
        # 「Adrian Holovaty」
        
        # Step 3：構造第二個查詢
        step2_query = self._construct_second_query(query, step1_answer)
        # 「Adrian Holovaty 的出生地在哪裡？」
        
        # Step 4：第二次檢索
        step2_docs = self.retriever.retrieve(step2_query)
        step2_answer = self._extract_answer(step2_docs, step2_query)
        # 「美國伊利諾州」
        
        # Step 5：生成最終答案
        final_answer = self._generate_final_answer(
            query, step1_answer, step2_answer
        )
        
        return final_answer
    
    def _extract_first_hop(self, query):
        """提取第一個子問題"""
        prompt = f"""
問題：{query}

這是一個需要多步推理的問題。請提取第一個子問題。

範例：
問題：「Django 創始人的出生地？」
第一個子問題：「Django 的創始人是誰？」

第一個子問題：
"""
        return self.llm.generate(prompt).strip()
    
    def _extract_answer(self, docs, query):
        """從文檔中提取答案"""
        prompt = f"""
文檔：
{docs}

問題：{query}

請簡潔回答（只說答案，不要解釋）：
"""
        return self.llm.generate(prompt).strip()
    
    def _construct_second_query(self, original_query, first_answer):
        """根據第一個答案構造第二個查詢"""
        prompt = f"""
原始問題：{original_query}
第一步答案：{first_answer}

請構造第二個查詢來回答原始問題。

第二個查詢：
"""
        return self.llm.generate(prompt).strip()
    
    def _generate_final_answer(self, query, answer1, answer2):
        """整合答案"""
        prompt = f"""
問題：{query}

收集到的資訊：
1. {answer1}
2. {answer2}

請根據這些資訊回答原始問題：
"""
        return self.llm.generate(prompt).strip()

# 使用範例
rag = FixedStepRAG(retriever, llm)
answer = rag.answer_2hop("Django 創始人的出生地在哪裡？")
print(answer)
# 輸出：「美國伊利諾州」

class IterativeRAG:
    """迭代式 Multi-step RAG"""
    
    def __init__(self, retriever, llm, max_iterations=5):
        self.retriever = retriever
        self.llm = llm
        self.max_iterations = max_iterations
    
    def answer(self, query):
        """
        迭代式問答
        自動決定需要幾次檢索
        """
        # 初始化
        context = []
        iteration = 0
        
        while iteration < self.max_iterations:
            iteration += 1
            
            # 決策：是否需要更多資訊？
            need_more = self._need_more_info(query, context)
            
            if not need_more:
                # 資訊足夠，生成答案
                break
            
            # 決定下一個檢索查詢
            next_query = self._generate_next_query(query, context)
            
            # 檢索
            docs = self.retriever.retrieve(next_query)
            
            # 加入 context
            context.append({
                'query': next_query,
                'docs': docs,
                'iteration': iteration
            })
            
            print(f"🔍 迭代 {iteration}: {next_query}")
        
        # 生成最終答案
        final_answer = self._generate_answer(query, context)
        
        return {
            'answer': final_answer,
            'iterations': iteration,
            'retrieval_history': context
        }
    
    def _need_more_info(self, query, context):
        """判斷是否需要更多資訊"""
        if not context:
            # 第一次，肯定需要檢索
            return True
        
        # 讓 LLM 判斷
        context_summary = self._summarize_context(context)
        
        prompt = f"""
原始問題：{query}

已收集的資訊：
{context_summary}

請判斷：這些資訊是否足夠回答問題？

請只回答「足夠」或「不足夠」：
"""
        decision = self.llm.generate(prompt, temperature=0).strip()
        
        return "不足夠" in decision
    
    def _generate_next_query(self, original_query, context):
        """生成下一個檢索查詢"""
        context_summary = self._summarize_context(context)
        
        prompt = f"""
原始問題：{original_query}

已收集的資訊：
{context_summary}

為了回答原始問題，下一步應該檢索什麼？

請只說查詢內容（不要解釋）：
"""
        return self.llm.generate(prompt, temperature=0).strip()
    
    def _summarize_context(self, context):
        """摘要已收集的資訊"""
        if not context:
            return "（尚無資訊）"
        
        summary = []
        for i, item in enumerate(context, 1):
            summary.append(f"{i}. 查詢「{item['query']}」")
            summary.append(f"   找到：{item['docs'][:200]}...")
        
        return '\n'.join(summary)
    
    def _generate_answer(self, query, context):
        """根據所有 context 生成答案"""
        context_text = '\n\n'.join([
            f"檢索 {i+1}：{item['docs']}"
            for i, item in enumerate(context)
        ])
        
        prompt = f"""
問題：{query}

收集到的資訊：
{context_text}

請根據這些資訊回答問題：
"""
        return self.llm.generate(prompt).strip()

# 使用範例
rag = IterativeRAG(retriever, llm, max_iterations=5)
result = rag.answer("Django 創始人出生的州的首府是哪裡？")

print(f"🤖 答案：{result['answer']}")
print(f"📊 迭代次數：{result['iterations']}")
print(f"🔍 檢索歷史：")
for item in result['retrieval_history']:
    print(f"  - {item['query']}")

# 輸出：
# 🔍 迭代 1: Django 的創始人是誰
# 🔍 迭代 2: Adrian Holovaty 的出生地
# 🔍 迭代 3: 伊利諾州的首府
# 🤖 答案：Springfield
# 📊 迭代次數：3

from typing import List, Dict

class ReActRAG:
    """
    ReAct: Reasoning and Acting
    讓 LLM 自己決定何時檢索、檢索什麼
    """
    
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
    
    def answer(self, query, max_steps=10):
        """
        ReAct 循環
        
        每步：思考 → 行動 → 觀察
        """
        # 初始化
        thought_action_history = []
        
        for step in range(1, max_steps + 1):
            # 生成當前狀態的提示詞
            prompt = self._build_prompt(query, thought_action_history)
            
            # LLM 思考並決定行動
            response = self.llm.generate(prompt)
            
            # 解析回應
            thought, action, action_input = self._parse_response(response)
            
            print(f"\n--- Step {step} ---")
            print(f"💭 Thought: {thought}")
            print(f"🎬 Action: {action}")
            
            # 執行行動
            if action == "Search":
                # 檢索
                observation = self.retriever.retrieve(action_input)
                print(f"👀 Observation: {observation[:100]}...")
                
            elif action == "Finish":
                # 完成，返回答案
                print(f"✅ Final Answer: {action_input}")
                return {
                    'answer': action_input,
                    'steps': step,
                    'history': thought_action_history
                }
            
            else:
                observation = f"Unknown action: {action}"
            
            # 記錄
            thought_action_history.append({
                'thought': thought,
                'action': action,
                'action_input': action_input,
                'observation': observation
            })
        
        # 超過最大步驟數
        return {
            'answer': "Unable to answer within max steps",
            'steps': max_steps,
            'history': thought_action_history
        }
    
    def _build_prompt(self, query, history):
        """構建 ReAct 提示詞"""
        base_prompt = f"""
你是一個問答助手。請回答以下問題：

問題：{query}

你可以使用以下工具：
- Search[query]: 搜尋知識庫
- Finish[answer]: 給出最終答案

請按照以下格式回答：
Thought: [你的思考過程]
Action: [Search 或 Finish]
Action Input: [搜尋內容 或 最終答案]

範例：
Thought: 我需要知道 Django 的創始人
Action: Search
Action Input: Django 創始人

---
"""
        
        # 加入歷史
        for i, item in enumerate(history, 1):
            base_prompt += f"""
Step {i}:
Thought: {item['thought']}
Action: {item['action']}
Action Input: {item['action_input']}
Observation: {item['observation'][:200]}

"""
        
        base_prompt += "\nNext step:\n"
        return base_prompt
    
    def _parse_response(self, response):
        """解析 LLM 的回應"""
        lines = response.strip().split('\n')
        
        thought = ""
        action = ""
        action_input = ""
        
        for line in lines:
            if line.startswith("Thought:"):
                thought = line.replace("Thought:", "").strip()
            elif line.startswith("Action:"):
                action = line.replace("Action:", "").strip()
            elif line.startswith("Action Input:"):
                action_input = line.replace("Action Input:", "").strip()
        
        return thought, action, action_input

# 使用範例
react = ReActRAG(retriever, llm)
result = react.answer("Django 創始人出生的州的首府是哪裡？")

# 輸出範例：
# --- Step 1 ---
# 💭 Thought: 我需要先知道 Django 的創始人是誰
# 🎬 Action: Search
# 👀 Observation: Django 由 Adrian Holovaty 創建...
#
# --- Step 2 ---
# 💭 Thought: 現在我需要知道 Adrian Holovaty 的出生地
# 🎬 Action: Search
# 👀 Observation: Adrian Holovaty 出生於美國伊利諾州...
#
# --- Step 3 ---
# 💭 Thought: 我需要知道伊利諾州的首府
# 🎬 Action: Search
# 👀 Observation: 伊利諾州首府是 Springfield...
#
# --- Step 4 ---
# 💭 Thought: 我已經收集到所有資訊，可以回答了
# 🎬 Action: Finish
# ✅ Final Answer: Springfield

class SelfRAG:
    """
    Self-RAG: 自我反思的 RAG 系統
    
    特點：
    1. 檢查檢索結果是否相關
    2. 檢查生成的答案是否有依據
    3. 必要時重新檢索
    """
    
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
    
    def answer(self, query, max_attempts=3):
        """
        Self-RAG 流程
        """
        for attempt in range(1, max_attempts + 1):
            print(f"\n=== Attempt {attempt} ===")
            
            # 1. 檢索
            docs = self.retriever.retrieve(query)
            
            # 2. 評估檢索結果的相關性
            relevance = self._check_relevance(query, docs)
            print(f"📊 Relevance: {relevance}/10")
            
            if relevance < 5:
                # 檢索結果不相關，改寫查詢重試
                query = self._rewrite_query(query, docs)
                print(f"🔄 Rewriting query: {query}")
                continue
            
            # 3. 生成答案
            answer = self._generate_answer(query, docs)
            
            # 4. 檢查答案是否有依據
            is_grounded = self._check_grounded(answer, docs)
            print(f"✓ Grounded: {is_grounded}")
            
            if is_grounded:
                # 答案有依據，檢查是否完整
                is_complete = self._check_complete(query, answer)
                print(f"✓ Complete: {is_complete}")
                
                if is_complete:
                    # 完美！返回答案
                    return {
                        'answer': answer,
                        'attempts': attempt,
                        'confidence': 'high'
                    }
                else:
                    # 不完整，需要更多資訊
                    print("⚠️ Answer incomplete, need more info")
                    # 生成補充查詢
                    follow_up = self._generate_followup(query, answer, docs)
                    additional_docs = self.retriever.retrieve(follow_up)
                    
                    # 重新生成答案
                    all_docs = docs + additional_docs
                    answer = self._generate_answer(query, all_docs)
                    
                    return {
                        'answer': answer,
                        'attempts': attempt,
                        'confidence': 'medium'
                    }
            else:
                # 答案沒有依據（幻覺），重試
                print("❌ Answer not grounded, retrying...")
                continue
        
        # 超過最大嘗試次數
        return {
            'answer': "Unable to provide a reliable answer",
            'attempts': max_attempts,
            'confidence': 'low'
        }
    
    def _check_relevance(self, query, docs):
        """檢查文檔相關性（0-10）"""
        prompt = f"""
問題：{query}

文檔：
{docs[:500]}

請評估文檔與問題的相關程度（0-10 分）：
"""
        score = self.llm.generate(prompt, temperature=0).strip()
        return int(score)
    
    def _check_grounded(self, answer, docs):
        """檢查答案是否有依據"""
        prompt = f"""
答案：{answer}

文檔：
{docs}

問題：答案中的每個事實都能在文檔中找到依據嗎？

請只回答「是」或「否」：
"""
        response = self.llm.generate(prompt, temperature=0).strip()
        return "是" in response
    
    def _check_complete(self, query, answer):
        """檢查答案是否完整"""
        prompt = f"""
問題：{query}

答案：{answer}

問題：這個答案是否完整回答了問題？

請只回答「是」或「否」：
"""
        response = self.llm.generate(prompt, temperature=0).strip()
        return "是" in response
    
    def _rewrite_query(self, query, docs):
        """改寫查詢（當檢索結果不相關時）"""
        prompt = f"""
原始問題：{query}

檢索結果不相關：
{docs[:300]}

請改寫問題，使其更容易找到相關資訊：
"""
        return self.llm.generate(prompt).strip()
    
    def _generate_followup(self, query, partial_answer, docs):
        """生成補充查詢"""
        prompt = f"""
問題：{query}

部分答案：{partial_answer}

已有文檔：
{docs[:300]}

為了完整回答問題，還需要檢索什麼資訊？

補充查詢：
"""
        return self.llm.generate(prompt).strip()
    
    def _generate_answer(self, query, docs):
        """生成答案"""
        prompt = f"""
根據以下文檔回答問題。

文檔：
{docs}

問題：{query}

答案：
"""
        return self.llm.generate(prompt).strip()

# 使用範例
self_rag = SelfRAG(retriever, llm)
result = self_rag.answer("Django 創始人的主要成就有哪些？")

print(f"\n✅ 最終答案：{result['answer']}")
print(f"📊 嘗試次數：{result['attempts']}")
print(f"🎯 信心度：{result['confidence']}")

# 避免無限循環
max_iterations = 5  # 通常 3-5 步足夠

# 如果超過，強制結束
if iteration > max_iterations:
    return "問題過於複雜，無法在限定步驟內回答"

class CachedMultiStepRAG:
    def __init__(self):
        self.cache = {}  # 快取檢索結果
    
    def retrieve(self, query):
        if query in self.cache:
            print(f"✅ Cache hit: {query}")
            return self.cache[query]
        
        docs = self.retriever.retrieve(query)
        self.cache[query] = docs
        return docs

def detect_loop(history):
    """檢測是否在重複相同的檢索"""
    if len(history) < 2:
        return False
    
    recent_queries = [item['query'] for item in history[-3:]]
    
    # 如果最近 3 次有重複查詢
    if len(recent_queries) != len(set(recent_queries)):
        return True
    
    return False

# 使用
if detect_loop(retrieval_history):
    print("⚠️ 檢測到循環，嘗試換個角度")
    query = rewrite_query_differently(query)

import asyncio

async def parallel_retrieve(queries):
    """並行執行多個檢索（節省時間）"""
    tasks = [
        retriever.retrieve_async(query)
        for query in queries
    ]
    results = await asyncio.gather(*tasks)
    return results

# 範例
queries = [
    "Django 創始人",
    "Adrian Holovaty 作品",
    "Python 網頁框架歷史"
]

# 並行檢索（vs 依序檢索）
results = await parallel_retrieve(queries)
# 時間：600ms（vs 1800ms 依序）

from langchain.agents import Tool, AgentExecutor, LLMSingleActionAgent

# 定義工具
tools = [
    Tool(
        name="Search",
        func=retriever.retrieve,
        description="搜尋知識庫。輸入：查詢字串"
    ),
    Tool(
        name="Calculator",
        func=calculator,
        description="計算數學問題。輸入：數學表達式"
    ),
    Tool(
        name="WebSearch",
        func=web_search,
        description="搜尋網路。輸入：查詢字串"
    )
]

# 建立 Agent
agent = AgentExecutor.from_agent_and_tools(
    agent=LLMSingleActionAgent(...),
    tools=tools,
    verbose=True
)

# 使用
response = agent.run(
    "Django 創始人出生在哪一年？那一年發生了什麼重大事件？"
)

# Agent 會自動：
# 1. Search: Django 創始人
# 2. Search: Adrian Holovaty 出生年份
# 3. WebSearch: 1981 年重大事件
# 4. 整合答案

def plan_retrieval_path(query):
    """
    規劃檢索路徑
    類似 A* 搜尋演算法
    """
    # 1. 分析問題
    sub_questions = decompose_question(query)
    
    # 2. 評估每個子問題的難度
    difficulties = [estimate_difficulty(q) for q in sub_questions]
    
    # 3. 規劃順序（先易後難）
    sorted_questions = sort_by_difficulty(sub_questions, difficulties)
    
    # 4. 執行
    answers = {}
    for q in sorted_questions:
        answer = retrieve_and_answer(q, answers)  # 利用已有答案
        answers[q] = answer
    
    # 5. 整合
    final_answer = integrate_answers(query, answers)
    return final_answer