前言

前面 26 天我們把 RAG 系統的每個環節都拆解、實作、優化了一遍，從今天開始要進入最後階段：把這些程式碼整合成一個真正可以被使用者操作的系統。

把 RAG 邏輯封裝成 API 服務

目前為止的程式碼大多是腳本形式，第一步是把整套邏輯包裝成一個 Web API，方便前端或其他服務呼叫。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    question: str
    chat_history: list = []

class QueryResponse(BaseModel):
    answer: str
    sources: list

@app.post("/query", response_model=QueryResponse)
def query_rag(request: QueryRequest):
    # 1. 問題改寫（若有對話歷史，套用 Day 19 的技巧）
    rewritten_question = rewrite_with_history(request.chat_history, request.question) \
        if request.chat_history else request.question

    # 2. 檢索（套用 Day 15-20 的優化技巧）
    retrieved = hybrid_search(rewritten_question, top_k=20)
    reranked = rerank_results(rewritten_question, retrieved, top_k=5)

    # 3. 生成（套用 Day 21 的 Prompt 設計）
    system_prompt, user_prompt = build_rag_prompt(request.question, reranked)
    answer = call_llm(system_prompt, user_prompt)

    return QueryResponse(
        answer=answer,
        sources=[{"content": c["content"][:100], "metadata": c["metadata"]} for c in reranked]
    )

uvicorn main:app --reload --port 8000

API 設計時的考量

1. 輸入驗證：使用者輸入的問題長度、格式需要做基本檢查，避免過長或異常輸入拖垮系統
2. 回傳來源資訊：如 Day 21 提到的，回傳答案時附上引用來源，方便前端呈現「參考資料」給使用者
3. 串流輸出（Streaming）：LLM 生成通常需要幾秒鐘，若能用串流方式逐字回傳，使用者體驗會好很多

from fastapi.responses import StreamingResponse

@app.post("/query/stream")
def query_rag_stream(request: QueryRequest):
    def generate():
        for chunk in call_llm_stream(system_prompt, user_prompt):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

簡易前端串接範例

async function askQuestion(question) {
  const response = await fetch("http://localhost:8000/query", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ question, chat_history: [] }),
  });
  const data = await response.json();
  displayAnswer(data.answer, data.sources);
}

前端介面設計上，建議至少包含：

• 輸入框與送出按鈕
• 回答顯示區塊
• 引用來源的展開/收合區塊（讓使用者可以查證答案依據）
• 載入中狀態提示（因為生成通常需要幾秒鐘）

錯誤處理不能忽略

實際上線的系統，需要處理各種異常情況：

@app.post("/query", response_model=QueryResponse)
def query_rag(request: QueryRequest):
    try:
        retrieved = hybrid_search(request.question, top_k=20)
        if not retrieved:
            return QueryResponse(answer="很抱歉，目前查無相關資訊。", sources=[])
        # ... 正常流程
    except Exception as e:
        logger.error(f"查詢發生錯誤:{e}")
        return QueryResponse(answer="系統暫時發生問題，請稍後再試。", sources=[])

小結

把 RAG 邏輯封裝成標準的 API 服務，是讓系統從「一支可以跑的腳本」變成「真正能被使用者操作的產品」的關鍵一步。明天我們要討論上線後不可避免要面對的問題：效能、延遲與成本。