綜合實戰:個人文件問答系統上線

啟動向量資料庫

本課用 Qdrant 當向量庫(Docker 一行啟動,不需要先有 Postgres):

# docker-compose.yml
version: "3.9"
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - ./qdrant_data:/qdrant/storage

docker compose up -d

啟動後 http://localhost:6333/dashboard 可以開 Qdrant 的管理介面確認正常。

.env 裡放好 API 金鑰:

OPENAI_API_KEY=sk-...
COHERE_API_KEY=...
ANTHROPIC_API_KEY=sk-ant-...

安裝依賴:

pip install fastapi uvicorn python-multipart \
    langchain langchain-openai langchain-community \
    qdrant-client cohere anthropic PyPDF2 python-dotenv tiktoken

索引管線:文件上傳到向量庫

# backend/ingest.py
import hashlib
import os
from pathlib import Path
from typing import List

import PyPDF2
from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, PointStruct, VectorParams
import tiktoken

load_dotenv()

COLLECTION_NAME = "my_docs"
EMBED_MODEL = "text-embedding-3-small"
EMBED_DIM = 1536
CHUNK_SIZE = 512
CHUNK_OVERLAP = 64

embeddings = OpenAIEmbeddings(model=EMBED_MODEL)
qdrant = QdrantClient(host="localhost", port=6333)

enc = tiktoken.get_encoding("cl100k_base")

def ensure_collection():
    """確保 collection 存在,避免重複建立"""
    existing = [c.name for c in qdrant.get_collections().collections]
    if COLLECTION_NAME not in existing:
        qdrant.create_collection(
            collection_name=COLLECTION_NAME,
            vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.COSINE),
        )
        print(f"[ingest] 建立 collection: {COLLECTION_NAME}")

def file_hash(content: bytes) -> str:
    """用 SHA-256 當文件指紋,實現增量更新時的去重"""
    return hashlib.sha256(content).hexdigest()

def extract_text(filepath: str) -> str:
    """支援 PDF、Markdown、純文字"""
    if filepath.endswith(".pdf"):
        with open(filepath, "rb") as f:
            reader = PyPDF2.PdfReader(f)
            return "\n\n".join(p.extract_text() or "" for p in reader.pages)
    else:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()

def chunk_and_index(filepath: str, source_name: str) -> int:
    """切塊、embed、存入 Qdrant。回傳寫入的 chunk 數。"""
    ensure_collection()

    raw_text = extract_text(filepath)
    if not raw_text.strip():
        print(f"[ingest] 警告:{filepath} 萃取文字為空,跳過")
        return 0

    with open(filepath, "rb") as f:
        doc_hash = file_hash(f.read())

    # 用 token 計數切塊
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=lambda text: len(enc.encode(text)),
        separators=["\n\n", "\n", "。", ".", " ", ""],
    )
    chunks = splitter.create_documents(
        [raw_text],
        metadatas=[{"source": source_name, "doc_hash": doc_hash, "filepath": filepath}],
    )

    # 過濾太短的 chunk(防止空白或只有換行的垃圾片段)
    chunks = [c for c in chunks if len(c.page_content.strip()) >= 50]
    if not chunks:
        return 0

    # 批次 embed(每批 100 個,避免單次請求超過 API 限制)
    texts = [c.page_content for c in chunks]
    vectors = embeddings.embed_documents(texts)

    points = []
    for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
        points.append(PointStruct(
            id=abs(hash(f"{doc_hash}_{i}")) % (2**63),  # 穩定的數字 ID
            vector=vector,
            payload={
                "text": chunk.page_content,
                "source": source_name,
                "doc_hash": doc_hash,
                "chunk_index": i,
                "total_chunks": len(chunks),
            },
        ))

    qdrant.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"[ingest] {source_name} → {len(points)} 個 chunk 寫入 Qdrant")
    return len(points)

增量更新的關鍵在 doc_hash。每份文件算一個 SHA-256 fingerprint。下次同名文件進來,先比對 hash:一樣就跳過,不一樣就刪掉舊的 hash 對應的 points,再重新索引。

查詢 API:檢索 + Rerank + 生成

# backend/retriever.py
import os
import cohere
from langchain_openai import OpenAIEmbeddings
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

COLLECTION_NAME = "my_docs"
EMBED_MODEL = "text-embedding-3-small"
TOP_K_RETRIEVE = 20   # 先寬鬆撈 20 個
TOP_K_RERANK = 5      # rerank 後只留 5 個送 LLM

embeddings = OpenAIEmbeddings(model=EMBED_MODEL)
qdrant = QdrantClient(host="localhost", port=6333)
co = cohere.Client(os.environ["COHERE_API_KEY"])

def retrieve_and_rerank(query: str, source_filter: str | None = None) -> list[dict]:
    """
    1. 向量搜尋召回 TOP_K_RETRIEVE 個候選
    2. Cohere Rerank v3 二次排序
    3. 回傳 TOP_K_RERANK 個帶 source 的結果
    """
    query_vector = embeddings.embed_query(query)

    search_filter = None
    if source_filter:
        search_filter = Filter(
            must=[FieldCondition(key="source", match=MatchValue(value=source_filter))]
        )

    hits = qdrant.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_vector,
        limit=TOP_K_RETRIEVE,
        query_filter=search_filter,
        with_payload=True,
    )

    if not hits:
        return []

    # Cohere Rerank v3
    docs_for_rerank = [hit.payload["text"] for hit in hits]
    rerank_results = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=docs_for_rerank,
        top_n=TOP_K_RERANK,
    )

    return [
        {
            "text": hits[r.index].payload["text"],
            "source": hits[r.index].payload["source"],
            "relevance_score": r.relevance_score,
        }
        for r in rerank_results.results
    ]

# backend/llm.py
import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """你是一個根據提供文件回答問題的助理。
規則:
1. 只根據「參考文件」裡的內容回答,不要加入你自己的知識
2. 每一個主要論點後面用 [來源:文件名] 標注引用來源
3. 如果文件裡找不到答案,說「在提供的文件中找不到相關資訊」
4. 回答用繁體中文"""

def generate_answer(query: str, contexts: list[dict]) -> str:
    if not contexts:
        return "找不到相關文件內容,無法回答。"

    context_block = "\n\n".join(
        f"[來源:{c['source']}]\n{c['text']}"
        for c in contexts
    )

    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{
            "role": "user",
            "content": f"參考文件:\n{context_block}\n\n問題:{query}"
        }],
    )
    return message.content[0].text

FastAPI 主程式:上傳 + 查詢 endpoint

# backend/main.py
import os
import tempfile
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from dotenv import load_dotenv

load_dotenv()

from ingest import chunk_and_index
from retriever import retrieve_and_rerank
from llm import generate_answer

app = FastAPI(title="My RAG API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],   # 正式環境請限縮來源
    allow_methods=["*"],
    allow_headers=["*"],
)

class QueryRequest(BaseModel):
    question: str
    source_filter: str | None = None   # 可以限定只查某份文件

class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]

@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """接收 PDF/MD/TXT,跑索引管線"""
    allowed_exts = {".pdf", ".md", ".txt"}
    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in allowed_exts:
        raise HTTPException(status_code=400, detail=f"不支援的檔案格式:{ext}")

    with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    try:
        chunk_count = chunk_and_index(tmp_path, file.filename)
    finally:
        os.unlink(tmp_path)

    return {"filename": file.filename, "chunks_indexed": chunk_count}

@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
    """接收問題,回傳答案 + 引用來源"""
    if not req.question.strip():
        raise HTTPException(status_code=400, detail="問題不能是空白")

    contexts = retrieve_and_rerank(req.question, req.source_filter)
    answer = generate_answer(req.question, contexts)

    return QueryResponse(
        answer=answer,
        sources=[{"source": c["source"], "score": c["relevance_score"]} for c in contexts],
    )

@app.get("/health")
async def health():
    return {"status": "ok"}

啟動後端:

cd backend
uvicorn main:app --reload --port 8000

開 http://localhost:8000/docs 可以直接在 Swagger UI 上傳文件、測試查詢。

簡易前端:一個 HTML 檔搞定

<!-- frontend/index.html -->
<!DOCTYPE html>
<html lang="zh-Hant">
<head>
  <meta charset="UTF-8">
  <title>個人文件問答</title>
  <style>
    body { font-family: sans-serif; max-width: 720px; margin: 40px auto; padding: 0 20px; }
    textarea { width: 100%; height: 60px; font-size: 16px; }
    button { padding: 8px 20px; font-size: 16px; cursor: pointer; }
    #answer { margin-top: 20px; padding: 16px; background: #f5f5f5; border-radius: 8px; white-space: pre-wrap; }
    #sources { margin-top: 12px; font-size: 13px; color: #666; }
    .upload-area { margin-bottom: 24px; padding: 16px; border: 2px dashed #ccc; border-radius: 8px; }
  </style>
</head>
<body>
  <h2>個人文件問答系統</h2>

  <div class="upload-area">
    <p>上傳文件(PDF / MD / TXT)</p>
    <input type="file" id="fileInput" accept=".pdf,.md,.txt">
    <button onclick="uploadFile()">上傳並索引</button>
    <span id="uploadStatus"></span>
  </div>

  <textarea id="question" placeholder="輸入你的問題..."></textarea>
  <br><br>
  <button onclick="ask()">送出問題</button>

  <div id="answer" style="display:none"></div>
  <div id="sources"></div>

  <script>
    const API = "http://localhost:8000";

    async function uploadFile() {
      const file = document.getElementById("fileInput").files[0];
      if (!file) return alert("請先選擇檔案");
      const form = new FormData();
      form.append("file", file);
      document.getElementById("uploadStatus").textContent = "上傳中...";
      const res = await fetch(`${API}/upload`, { method: "POST", body: form });
      const data = await res.json();
      document.getElementById("uploadStatus").textContent =
        res.ok ? `✓ 索引完成,共 ${data.chunks_indexed} 個片段` : `錯誤:${data.detail}`;
    }

    async function ask() {
      const question = document.getElementById("question").value.trim();
      if (!question) return;
      document.getElementById("answer").style.display = "block";
      document.getElementById("answer").textContent = "查詢中...";
      document.getElementById("sources").textContent = "";

      const res = await fetch(`${API}/query`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ question }),
      });
      const data = await res.json();
      document.getElementById("answer").textContent = data.answer;
      document.getElementById("sources").textContent =
        "引用來源:" + data.sources.map(s => `${s.source}(${s.score.toFixed(2)})`).join("、");
    }
  </script>
</body>
</html>

用 python -m http.server 3000 --directory frontend 啟動,在瀏覽器開 http://localhost:3000 就能用。

增量更新:新文件進來不重建索引

完整的增量更新邏輯建在 ingest.py 的 chunk_and_index 裡,關鍵是 doc_hash。每次上傳時先查 Qdrant 裡有沒有同 hash 的 point:

def is_already_indexed(doc_hash: str) -> bool:
    """查 Qdrant 是否已有這份文件的 hash"""
    results = qdrant.scroll(
        collection_name=COLLECTION_NAME,
        scroll_filter=Filter(
            must=[FieldCondition(key="doc_hash", match=MatchValue(value=doc_hash))]
        ),
        limit=1,
    )
    return len(results[0]) > 0

def delete_by_hash(doc_hash: str):
    """刪除舊版本的所有 chunks"""
    qdrant.delete(
        collection_name=COLLECTION_NAME,
        points_selector=Filter(
            must=[FieldCondition(key="doc_hash", match=MatchValue(value=doc_hash))]
        ),
    )

在 chunk_and_index 最前面加這個判斷:

# 取得新版 hash
with open(filepath, "rb") as f:
    new_hash = file_hash(f.read())

# 查同檔名的舊 hash
old_hash = get_hash_by_source(source_name)   # 自行實作:查 Qdrant scroll

if old_hash == new_hash:
    print(f"[ingest] {source_name} 內容未改變,跳過索引")
    return 0

if old_hash:
    delete_by_hash(old_hash)
    print(f"[ingest] 刪除舊版本 {source_name}({old_hash[:8]}...)")

這樣每次上傳文件:沒變就跳過,有改就刪舊存新——索引永遠保持最新版,成本也不會因為重複上傳暴增。