# 🚀 Day 2 實作摘要：從原型到可維護系統

> **給 Claude Code 的教學指引**：
> 這份文件採用「引導式學習」方法。在實作前，請先詢問使用者對概念的理解，讓他們先思考「為什麼要這樣做」，再提供程式碼。當使用者遇到問題時，用問題引導他們找到答案，而非直接給解法。

---

## 🎯 Day 2 核心學習目標

### 技術目標
- 將 CLI 程式改造成 RESTful API
- 加入系統監控與 logging 機制
- 深入理解向量資料庫的運作原理
- 建立評估 RAG 效果的方法

### 思維目標
- 從「能跑」到「可維護」的工程思維
- 學會「如何知道系統好不好」（可觀測性）
- 理解「debugg RAG 系統」的系統化方法

---

## 📋 上午任務（3-4 小時）

### 🤔 開始前的思考題

**在開始寫程式前，請先思考並回答：**

1. **為什麼要把 CLI 改成 API？**
   - 想想看：CLI 和 API 的使用情境有什麼不同？
   - 如果你要把這個 RAG 系統給別人用，哪種方式比較好？

2. **什麼是 Logging？為什麼需要它？**
   - 想像你的 RAG 系統出錯了，你要怎麼知道哪裡出問題？
   - 如果有 1000 個使用者同時問問題，你要怎麼追蹤每個請求？

3. **API 應該回傳什麼資訊？**
   - 只回傳「答案」夠嗎？
   - 使用者可能還想知道什麼？（提示：來源、可信度、回應時間...）

---

### ⏰ Hour 1-2：將 RAG 包裝成 API

#### 💡 概念引導

**在寫程式前，讓我們先設計 API：**

**問題 1：API 應該有幾個 endpoint？**
- 最少需要什麼功能？（提示：問答、健康檢查）
- 需不需要「更新知識庫」的 endpoint？

**問題 2：請求和回應的格式應該是什麼？**
```
使用者發送：{ "query": "..." }
系統回傳：{ ??? }  ← 你覺得應該包含什麼欄位？
```

**問題 3：如果系統出錯，應該如何回應？**
- HTTP 狀態碼應該用哪個？
- 錯誤訊息應該包含什麼資訊？

---

#### 💻 實作：FastAPI 包裝

**檔案：`api_rag.py`**

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_google_genai import GoogleGenerativeAI
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from dotenv import load_dotenv
import logging
from datetime import datetime
import time

load_dotenv()

# === 💡 思考點 1：為什麼要在程式啟動時就載入模型？===
print("🔄 初始化 RAG 系統...")
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

vectordb = Chroma(
    persist_directory="./db",
    embedding_function=embeddings
)

llm = GoogleGenerativeAI(
    model="gemini-1.5-flash",
    temperature=0
)

retriever = vectordb.as_retriever(search_kwargs={"k": 3})
qa = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    return_source_documents=True
)

print("✅ RAG 系統就緒")

# === FastAPI 應用 ===
app = FastAPI(title="RAG API", version="1.0")

# === 💡 思考點 2：為什麼要定義 Pydantic Model？===
class Question(BaseModel):
    query: str
    
    class Config:
        json_schema_extra = {
            "example": {
                "query": "這個人有什麼專長？"
            }
        }

class Answer(BaseModel):
    query: str
    answer: str
    sources: list[str]
    response_time: float
    timestamp: str

# === Endpoints ===

@app.get("/")
def root():
    """健康檢查"""
    return {
        "status": "healthy",
        "message": "RAG API is running",
        "version": "1.0"
    }

@app.post("/ask", response_model=Answer)
def ask(question: Question):
    """
    問答 endpoint
    
    💡 思考點 3：這個函數做了哪些事？能不能拆分成更小的函數？
    """
    try:
        start_time = time.time()
        
        # 執行 RAG
        result = qa({"query": question.query})
        
        response_time = time.time() - start_time
        
        # 整理回應
        return Answer(
            query=question.query,
            answer=result['result'],
            sources=[doc.page_content[:200] for doc in result['source_documents']],
            response_time=round(response_time, 3),
            timestamp=datetime.now().isoformat()
        )
        
    except Exception as e:
        # 💡 思考點 4：為什麼不直接回傳錯誤訊息給使用者？
        raise HTTPException(status_code=500, detail="Internal server error")

@app.get("/stats")
def stats():
    """系統統計"""
    return {
        "total_documents": vectordb._collection.count(),
        "embedding_model": "paraphrase-multilingual-MiniLM-L12-v2",
        "llm_model": "gemini-1.5-flash"
    }
```

**執行測試：**
```bash
# 安裝 FastAPI
pip install fastapi uvicorn

# 啟動 API
uvicorn api_rag:app --reload --port 8000

# 測試（另開終端）
curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"query":"這個人有什麼專長？"}'
```

---

#### 🧪 實作後的反思問題

**請在測試完成後回答：**

1. **API 比 CLI 好在哪裡？**
   - 你觀察到什麼優勢？
   - 有什麼劣勢嗎？

2. **`response_model=Answer` 做了什麼？**
   - 試試看把它拿掉，有什麼差別？

3. **如果 1000 人同時呼叫 API，會發生什麼？**
   - 向量資料庫會被重複載入嗎？
   - 為什麼要在程式啟動時就載入模型？

---

### ⏰ Hour 3-4：加入 Logging 系統

#### 💡 概念引導

**在實作前，請先思考：**

**情境題：你的 RAG 系統上線了，但使用者抱怨「回答怪怪的」**

你需要什麼資訊來 debug？
- [ ] 使用者問了什麼問題？
- [ ] 系統檢索到哪些文件？
- [ ] LLM 生成的答案是什麼？
- [ ] 花了多少時間？
- [ ] 有沒有發生錯誤？

**這就是 Logging 要記錄的！**

---

#### 💻 實作：完整的 Logging

**檔案：`api_rag_with_logging.py`**

```python
import logging
import json
from datetime import datetime
from pathlib import Path

# === 💡 思考點 5：為什麼要用結構化的 log？===
# 設定 logging
log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)

# 設定兩個 handler：一個給控制台，一個給檔案
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_dir / f"rag_{datetime.now().strftime('%Y%m%d')}.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# === 修改 ask endpoint ===
@app.post("/ask", response_model=Answer)
def ask(question: Question):
    request_id = datetime.now().strftime('%Y%m%d%H%M%S%f')  # 唯一 ID
    
    logger.info(f"[{request_id}] 收到問題: {question.query}")
    
    try:
        start_time = time.time()
        
        result = qa({"query": question.query})
        
        response_time = time.time() - start_time
        
        # === 💡 思考點 6：log 應該記錄什麼層級的資訊？===
        log_data = {
            "request_id": request_id,
            "query": question.query,
            "answer": result['result'],
            "sources_count": len(result['source_documents']),
            "response_time": response_time,
            "timestamp": datetime.now().isoformat()
        }
        
        logger.info(f"[{request_id}] 完成回答，耗時 {response_time:.3f}秒")
        
        # 儲存詳細 log 到 JSON
        with open(log_dir / f"queries_{datetime.now().strftime('%Y%m%d')}.jsonl", "a") as f:
            f.write(json.dumps(log_data, ensure_ascii=False) + "\n")
        
        return Answer(
            query=question.query,
            answer=result['result'],
            sources=[doc.page_content[:200] for doc in result['source_documents']],
            response_time=round(response_time, 3),
            timestamp=datetime.now().isoformat()
        )
        
    except Exception as e:
        logger.error(f"[{request_id}] 錯誤: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")
```

---

#### 📊 Log 分析腳本

**檔案：`analyze_logs.py`**

```python
import json
from pathlib import Path
from datetime import datetime

def analyze_today_logs():
    """
    分析今天的查詢 log
    
    💡 思考點 7：這個分析能告訴我們什麼？
    """
    log_file = Path("logs") / f"queries_{datetime.now().strftime('%Y%m%d')}.jsonl"
    
    if not log_file.exists():
        print("今天還沒有查詢紀錄")
        return
    
    queries = []
    with open(log_file) as f:
        for line in f:
            queries.append(json.loads(line))
    
    print(f"📊 今日統計 ({len(queries)} 筆查詢)")
    print(f"─" * 50)
    
    # 平均回應時間
    avg_time = sum(q['response_time'] for q in queries) / len(queries)
    print(f"平均回應時間: {avg_time:.3f} 秒")
    
    # 最慢的查詢
    slowest = max(queries, key=lambda x: x['response_time'])
    print(f"\n最慢查詢 ({slowest['response_time']:.3f}秒):")
    print(f"  問題: {slowest['query']}")
    
    # 常見問題
    print(f"\n問題類型分布:")
    # 這裡可以加入分類邏輯

if __name__ == "__main__":
    analyze_today_logs()
```

---

#### 🧪 實作後的反思問題

**測試幾次 API 後，執行 `python analyze_logs.py`，然後回答：**

1. **log 檔案和 console 輸出的差異是什麼？**
   - 為什麼要分成兩個 handler？
   - 什麼情況下你會只看 console？什麼情況下會看檔案？

2. **為什麼用 JSONL 格式儲存詳細 log？**
   - 比起純文字 log，JSON 有什麼好處？
   - 試試看用 Python 讀取並分析 JSONL 檔案

3. **你還想記錄什麼資訊？**
   - 檢索到的文件內容要全部記錄嗎？
   - 使用者的 IP 要記錄嗎？為什麼？

---

## 📋 下午任務（3-4 小時）

### 🎯 學習目標
- 理解向量資料庫的底層運作
- 學會評估 RAG 系統的效能
- 找出並解決「檢索失敗」的問題

---

### ⏰ Hour 5-6：向量資料庫深入探索

#### 💡 概念引導

**在開始前，讓我們先建立概念地圖：**

**問題 1：向量資料庫跟 MySQL 的差異是什麼？**

| 特性 | MySQL | 向量資料庫 |
|------|-------|------------|
| 搜尋方式 | ??? | ??? |
| 索引類型 | B-Tree | ??? |
| 搜尋結果 | 精確匹配 | ??? |
| 適用情境 | ??? | ??? |

**問題 2：「相似度搜尋」是怎麼做到的？**
- Chroma 怎麼從 10000 個向量中快速找到最相似的 3 個？
- 為什麼不用「暴力比對」每一個向量？

---

#### 💻 探索向量資料庫內部

**檔案：`vectordb_explorer.py`**

```python
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
import numpy as np

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

vectordb = Chroma(
    persist_directory="./db",
    embedding_function=embeddings
)

# === 實驗 1：資料庫基本資訊 ===
print("=== 向量資料庫資訊 ===")
collection = vectordb._collection
print(f"總文件數: {collection.count()}")

# 💡 思考點 8：這個數字跟你 Day 1 切出來的 chunks 數量一樣嗎？

# === 實驗 2：手動搜尋 ===
query = "Python 開發經驗"
print(f"\n=== 搜尋：{query} ===")

# 取得相似度分數
docs_with_scores = vectordb.similarity_search_with_score(query, k=5)

for i, (doc, score) in enumerate(docs_with_scores):
    print(f"\n第 {i+1} 名 (距離: {score:.4f})")
    print(f"內容: {doc.page_content[:150]}...")

# 💡 思考點 9：分數越低代表越相似還是越不相似？

# === 實驗 3：比較不同搜尋方式 ===
queries = [
    "軟體開發",
    "software development",  # 英文
    "寫程式",                # 同義詞
]

print(f"\n=== 不同查詢詞的效果 ===")
for q in queries:
    docs = vectordb.similarity_search(q, k=1)
    print(f"\n查詢: {q}")
    print(f"最相關: {docs[0].page_content[:100]}...")

# 💡 思考點 10：為什麼英文和中文能搜到同樣的文件？
```

---

#### 🔬 向量相似度視覺化（選做）

```python
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA

def visualize_vectors():
    """
    將高維向量降維到 2D 視覺化
    
    💡 思考點 11：384 維的向量要怎麼「畫出來」？
    """
    # 取得一些文件的向量
    all_docs = vectordb.get()
    
    # 將向量降維到 2D
    vectors = np.array(all_docs['embeddings'])
    pca = PCA(n_components=2)
    vectors_2d = pca.fit_transform(vectors)
    
    # 繪圖
    plt.figure(figsize=(10, 8))
    plt.scatter(vectors_2d[:, 0], vectors_2d[:, 1])
    
    # 標註幾個點
    for i in range(min(5, len(vectors_2d))):
        plt.annotate(
            all_docs['documents'][i][:30],
            (vectors_2d[i, 0], vectors_2d[i, 1])
        )
    
    plt.title("文件向量分布（PCA 降維）")
    plt.savefig("vector_visualization.png")
    print("✅ 圖片已儲存為 vector_visualization.png")

# visualize_vectors()  # 需要安裝 matplotlib scikit-learn
```

---

#### 🧪 實作後的反思問題

1. **Chroma 使用什麼演算法？**
   - Google 搜尋「HNSW algorithm」
   - 用自己的話解釋：為什麼它比暴力搜尋快？

2. **相似度分數的意義？**
   - 你的實驗中，分數範圍是多少？
   - 分數 0.5 和 0.8 的差異大嗎？

3. **多語言搜尋為什麼有效？**
   - Embedding 模型是怎麼做到的？
   - 如果搜尋日文，會有效嗎？

---

### ⏰ Hour 7-8：建立評估機制

#### 💡 概念引導

**核心問題：怎麼知道你的 RAG 系統「好不好」？**

**情境題：你做了一個改動（例如調整 chunk_size），怎麼知道有沒有變好？**

如果只靠「感覺」，你無法證明系統進步了。需要「量化指標」！

---

#### 📊 RAG 系統的評估維度

```
1. 檢索品質
   - 相關文件有沒有被找到？（Recall）
   - 找到的文件是不是真的相關？（Precision）

2. 回答品質
   - 答案正確嗎？
   - 答案是否基於檢索到的文件？（沒有幻覺）
   - 答案是否完整？

3. 系統效能
   - 回應速度
   - 成本（API 呼叫次數）
```

**💡 思考點 12：這三個維度有衝突嗎？**
- 例如：提高檢索品質（增加 top_k）會影響速度嗎？

---

#### 💻 建立測試集與評估腳本

**檔案：`evaluation.py`**

```python
from langchain_google_genai import GoogleGenerativeAI
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from dotenv import load_dotenv
import time

load_dotenv()

# === 第一步：建立測試集 ===
# 💡 思考點 13：什麼樣的測試案例是好的？
test_cases = [
    {
        "question": "這個人有什麼專長？",
        "expected_keywords": ["Python", "API", "開發"],  # 期望答案包含的關鍵字
        "expected_source": True  # 應該要找到相關文件
    },
    {
        "question": "這個人會做菜嗎？",
        "expected_keywords": [],  # 如果 knowledge.txt 沒寫
        "expected_source": False  # 不應該找到相關文件
    },
    # 💡 請加入 3-5 個你自己的測試案例
]

# === 載入 RAG 系統 ===
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
vectordb = Chroma(persist_directory="./db", embedding_function=embeddings)
llm = GoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
retriever = vectordb.as_retriever(search_kwargs={"k": 3})
qa = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    return_source_documents=True
)

# === 執行評估 ===
results = []

for i, case in enumerate(test_cases):
    print(f"\n{'='*50}")
    print(f"測試案例 {i+1}: {case['question']}")
    
    start_time = time.time()
    result = qa({"query": case['question']})
    response_time = time.time() - start_time
    
    answer = result['result']
    sources = result['source_documents']
    
    # === 評估 1：關鍵字檢查 ===
    keyword_score = sum(kw in answer for kw in case['expected_keywords'])
    keyword_total = len(case['expected_keywords'])
    
    # === 評估 2：來源檢查 ===
    has_source = len(sources) > 0
    source_correct = has_source == case['expected_source']
    
    # === 評估 3：幻覺檢查 ===
    # 簡單版本：如果沒有相關來源，答案應該說「找不到」或類似的話
    if not case['expected_source']:
        hallucination_phrases = ["找不到", "沒有", "不清楚", "無法回答"]
        no_hallucination = any(phrase in answer for phrase in hallucination_phrases)
    else:
        no_hallucination = True  # 有來源時假設沒幻覺
    
    # 記錄結果
    result_data = {
        "question": case['question'],
        "answer": answer,
        "keyword_score": f"{keyword_score}/{keyword_total}",
        "source_correct": source_correct,
        "no_hallucination": no_hallucination,
        "response_time": response_time,
        "pass": keyword_score == keyword_total and source_correct and no_hallucination
    }
    
    results.append(result_data)
    
    # 輸出結果
    print(f"答案: {answer[:150]}...")
    print(f"關鍵字分數: {keyword_score}/{keyword_total}")
    print(f"來源正確: {'✅' if source_correct else '❌'}")
    print(f"無幻覺: {'✅' if no_hallucination else '❌'}")
    print(f"回應時間: {response_time:.3f}秒")
    print(f"通過: {'✅' if result_data['pass'] else '❌'}")

# === 總結報告 ===
print(f"\n{'='*50}")
print("📊 評估總結")
print(f"{'='*50}")

pass_count = sum(r['pass'] for r in results)
print(f"通過率: {pass_count}/{len(results)} ({pass_count/len(results)*100:.1f}%)")

avg_time = sum(r['response_time'] for r in results) / len(results)
print(f"平均回應時間: {avg_time:.3f}秒")

# 找出失敗的案例
failed = [r for r in results if not r['pass']]
if failed:
    print(f"\n❌ 失敗的案例:")
    for r in failed:
        print(f"  - {r['question']}")
```

---

#### 🧪 實作後的反思問題

**執行完評估後，請思考：**

1. **你的測試集夠好嗎？**
   - 涵蓋了哪些類型的問題？
   - 有沒有「邊界案例」（edge cases）？
   - 還需要加入什麼測試？

2. **如果通過率只有 60%，你會怎麼改進？**
   - 先看失敗的案例
   - 是檢索問題？還是生成問題？
   - 如何系統化地找出問題？

3. **這個評估方法的限制是什麼？**
   - 關鍵字檢查夠準確嗎？
   - 有沒有更好的評估方法？（提示：用另一個 LLM 評分）

---

#### 🚀 進階：自動化評估（選做）

```python
# 用 LLM 評估 LLM 的答案
def llm_judge(question, answer, expected):
    """
    用 Gemini 評估答案品質
    
    💡 思考點 14：讓 AI 評估 AI，可靠嗎？
    """
    judge_llm = GoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
    
    prompt = f"""
請評估以下答案的品質（1-5分）：

問題：{question}
答案：{answer}
期望包含的資訊：{expected}

評分標準：
- 5分：完整、準確、相關
- 3分：部分正確但不完整
- 1分：不相關或錯誤

只回傳數字分數。
"""
    
    score = judge_llm.predict(prompt)
    return int(score.strip())
```

---

## ✅ Day 2 完成檢查清單

### 技術成果
- [ ] RAG 系統已包裝成 API
- [ ] 加入完整的 logging 機制
- [ ] 能夠查詢向量資料庫內部資訊
- [ ] 建立了測試集與評估腳本
- [ ] 執行評估並記錄結果

### 概念理解
- [ ] 能說明 API 相對於 CLI 的優勢
- [ ] 能解釋為什麼需要 logging
- [ ] 能說明向量資料庫的搜尋原理
- [ ] 能列出至少 3 個評估 RAG 的指標
- [ ] 能解釋「檢索失敗」vs「生成失敗」的差異

### 系統思維
- [ ] 知道如何 debug RAG 系統
- [ ] 能根據 log 分析系統問題
- [ ] 能設計測試案例
- [ ] 能解釋「為什麼這個改動讓系統變好/變壞」

---

## 🎯 Day 2 核心學習重點

### 從「能跑」到「可維護」

**今天學到的不只是技術，更是工程思維：**

1. **可觀測性**：系統要能「看見」自己在做什麼
   - Logging 讓你知道系統發生什麼事
   - 評估讓你知道系統做得好不好

2. **系統化 Debug**：
   ```
   問題出現 → 查 log → 找到失敗案例 → 分析原因 → 改進 → 重新評估
   ```

3. **模組化設計**：
   - API endpoint 分離關注點
   - Logging 獨立於業務邏輯
   - 評估系統可重複使用

---

## 💭 睡前思考題（不用馬上回答）

1. **如果有 10 萬份文件，你的系統會遇到什麼瓶頸？**
   - 建索引的時間？
   - 搜尋速度？
   - 記憶體使用？

2. **如何讓「答錯」的機率降到最低？**
   - 改 prompt？
   - 改檢索策略？
   - 加入驗證機制？

3. **你的 RAG 系統現在能上線了嗎？**
   - 還缺什麼？
   - 有哪些風險？

---

## 📝 Day 2 作業（明天開始前完成）

1. **完善你的測試集**
   - 至少 10 個測試案例
   - 涵蓋不同類型的問題

2. **執行評估並記錄**
   - 通過率多少？
   - 哪些案例失敗？為什麼？

3. **分析今天的 log**
   - 平均回應時間多少？
   - 最慢的查詢是什麼？

4. **思考一個改進方案**
   - 如果通過率不是 100%，要怎麼改進？
   - 寫下你的計畫