本文 AI 產出，尚未審核

LangChain 實戰專案：RAG 問答系統 ── 部署與測試

簡介

在資訊爆炸的時代，檢索增強生成（Retrieval‑Augmented Generation，簡稱 RAG） 已成為提升大型語言模型（LLM）回答品質的關鍵技術。透過將外部文件、知識庫或向量資料庫作為「記憶」來源，RAG 能讓模型在回答問題時直接引用真實、最新的資訊，避免產生「幻覺」答案。

本單元聚焦於 LangChain 框架下的 RAG 問答系統，從 本機測試、Docker 部署 到 API 介面 的完整流程，讓讀者能在自己的專案中快速上線、驗證與維運。文章以 Python 為主，搭配 FastAPI、Docker、pytest 等實務工具，適合剛接觸 LangChain 的初學者，也能為中級開發者提供可直接套用的範本。

核心概念

1. RAG 流程概覽

  1. 文件切分：將原始文本切成適合向量化的片段（Chunk）。
  2. 向量化：使用 Embedding 模型（如 OpenAI 的 text-embedding-ada-002）把每個 Chunk 轉成向量。
  3. 向量儲存：將向量與原始文字一起寫入向量資料庫（FAISS、Pinecone、Chroma 等）。
  4. 檢索：使用相似度搜尋找出與使用者問題最相關的 Chunk。
  5. 生成：把檢索結果作為「上下文」傳入 LLM，產生最終答案。

LangChain 提供了 Retriever、Chain、PromptTemplate 等高階抽象，讓上述步驟可以以少量程式碼串起來。

2. 為什麼要把 RAG 系統「容器化」？

  • 環境一致性：開發、測試、上線使用相同的 OS、套件版本。
  • 彈性伸縮：Docker + Kubernetes 可快速水平擴展。
  • 安全隔離：API 金鑰、模型權限可在容器層面加以管控。

3. 測試策略

  • 單元測試：驗證 Chunk 切分、Embedding、向量搜尋的正確性。
  • 整合測試：模擬完整的 HTTP 請求，確保 API 端點返回預期格式。
  • 負載測試：使用 locust 或 hey 觀察在高併發下的回應時間與資源使用。

程式碼範例

以下範例均採用 Python 3.10+、LangChain 0.0.XXX（請自行替換為最新版本），並以 FastAPI 作為服務層。每段程式碼均附有說明註解，方便讀者快速理解。

3.1 建立向量資料庫（FAISS）

# file: vector_store.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from pathlib import Path

def build_faiss_index(docs_path: str, index_path: str) -> FAISS:
    """
    讀取指定資料夾內的所有 .txt 檔，切分、向量化，最後寫入本機 FAISS 索引。
    """
    # 1️⃣ 讀檔
    raw_texts = []
    for txt_file in Path(docs_path).rglob("*.txt"):
        raw_texts.append(txt_file.read_text(encoding="utf-8"))

    # 2️⃣ 切分
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,   # 每個 chunk 最多 500 個字元
        chunk_overlap=50 # 前後保留 50 個字元作為重疊
    )
    chunks = splitter.split_documents([{"page_content": t} for t in raw_texts])

    # 3️⃣ 向量化
    embedder = OpenAIEmbeddings(model="text-embedding-ada-002")
    # 4️⃣ 建索引
    faiss_index = FAISS.from_documents(chunks, embedder)

    # 5️⃣ 永久保存
    faiss_index.save_local(index_path)
    return faiss_index

重點：RecursiveCharacterTextSplitter 能自動避免在中文斷句時切到半個詞，提升檢索品質。

3.2 建立 RetrievalQA Chain

# file: rag_chain.py
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from vector_store import build_faiss_index
import os

# 只在第一次執行時建索引；之後直接載入
if not os.path.isdir("faiss_index"):
    build_faiss_index(docs_path="data", index_path="faiss_index")

# 載入已建立好的 FAISS 索引
from langchain.vectorstores import FAISS
vector_store = FAISS.load_local("faiss_index", OpenAIEmbeddings())

# 取得 Retriever（相似度搜尋器）
retriever = vector_store.as_retriever(search_kwargs={"k": 4})  # 取前 4 個相關 Chunk

# 自訂 Prompt，讓 LLM 知道要「根據檔案內容」回答
qa_prompt = PromptTemplate(
    input_variables=["context", "question"],
    template=(
        "以下是與問題相關的文件內容：\n"
        "'''{context}'''\n"
        "請根據上述內容，直接回答以下問題，若內容不足請說明「資料不足」：\n"
        "{question}"
    ),
)

# 建立 RetrievalQA Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(model_name="gpt-3.5-turbo", temperature=0),
    chain_type="stuff",          # 直接把檢索結果拼接到 Prompt
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": qa_prompt}
)

def answer_question(question: str):
    """
    呼叫 chain 並回傳答案與來源文件。
    """
    result = qa_chain({"query": question})
    return {
        "answer": result["result"],
        "sources": [doc.metadata.get("source", "unknown") for doc in result["source_documents"]]
    }

技巧：return_source_documents=True 能把每筆答案的來源 Chunk 回傳，方便前端顯示「引用」或做後續審核。

3.3 用 FastAPI 包裝成 HTTP API

# file: api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from rag_chain import answer_question

app = FastAPI(title="LangChain RAG 問答 API")

class QueryRequest(BaseModel):
    question: str

class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/qa", response_model=AnswerResponse)
async def ask_qa(req: QueryRequest):
    """
    接收使用者問題，回傳 LLM 產生的答案與引用來源。
    """
    try:
        result = answer_question(req.question)
        return AnswerResponse(**result)
    except Exception as e:
        # 在正式環境建議寫入 log，這裡直接回傳 500
        raise HTTPException(status_code=500, detail=str(e))

安全建議：若 API 只供內部使用，可在 FastAPI 加入 OAuth2、API Key 機制，避免金鑰外洩。

3.4 Dockerfile：一鍵容器化

# Dockerfile
FROM python:3.11-slim

# 1️⃣ 建立工作目錄
WORKDIR /app

# 2️⃣ 複製需求檔與程式
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 3️⃣ 設定環境變數（OpenAI 金鑰須在執行時注入）
ENV OPENAI_API_KEY=${OPENAI_API_KEY}

# 4️⃣ 暴露 8000 埠（FastAPI 預設）
EXPOSE 8000

# 5️⃣ 使用 uvicorn 啟動
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt 建議內容：

fastapi
uvicorn[standard]
langchain
openai
faiss-cpu
pydantic

注意：若在 GPU 環境部署，可改用 faiss-gpu 以及 torch，Dockerfile 只需要對 FROM 換成支援 CUDA 的映像。

3.5 使用 pytest 撰寫整合測試

# tests/test_api.py
import pytest
from fastapi.testclient import TestClient
from api import app

client = TestClient(app)

def test_qa_success():
    payload = {"question": "什麼是 LangChain 的 Retriever？"}
    response = client.post("/qa", json=payload)
    assert response.status_code == 200
    data = response.json()
    assert "answer" in data
    assert "sources" in data
    # 簡單檢查答案不會是空字串
    assert len(data["answer"].strip()) > 0

def test_qa_missing_question():
    response = client.post("/qa", json={})
    assert response.status_code == 422  # Pydantic 會自動驗證

執行方式：pytest -s，在 CI 流程中加入此步驟，可確保每次部署前 API 功能正常。

常見陷阱與最佳實踐

陷阱 為什麼會發生 解決方案
向量維度不一致 不同 Embedding 模型產生的向量長度不同，導致 FAISS 建索引失敗。 固定模型（如 text-embedding-ada-002）或在切換模型前重新建索引。
Chunk 太大 LLM 的 Prompt 長度上限（如 4096 token）被單一 Chunk 吃光，無法加入檢索結果。 使用 RecursiveCharacterTextSplitter 調整 chunk_size，確保每個 Chunk ≤ 800 token。
金鑰洩漏 Dockerfile 直接寫入 ENV OPENAI_API_KEY=xxxx，映像檔被推至公共 Registry。 不在 Dockerfile 中硬編碼金鑰，改用 Docker runtime -e 或 Kubernetes secret。
檔案更新未同步 新增或修改文件後，未重新建索引，導致舊資料仍被檢索。 在 CI/CD pipeline 中加入「資料變更 → 重新建索引」步驟，或使用 incremental indexing。
回傳來源太多 前端一次顯示過多 Chunk，使用者體驗差。 在 API 中限制 k（如 3），或在前端自行做摘要。

最佳實踐：

  1. 分層日誌：將檢索過程、LLM 輸出、API 請求分別寫入不同等級的日誌，便於除錯與監控。
  2. 健康檢查端點：在 FastAPI 加入 /healthz，回傳向量庫是否可用、OpenAI 金鑰是否有效。
  3. 資源限制：使用 gunicorn + uvicorn workers，根據 CPU 核心數設定 worker 數量，避免單一容器過載。
  4. 版本化索引：每次建索引時加入時間戳或 Git commit hash，讓部署時能確保使用最新索引。

實際應用場景

行業 典型需求 RAG 解決方案
金融 法規、投資說明書快速查詢 把最新的法規 PDF 轉為 Chunk，結合 LLM 產生合規回覆。
醫療 病歷、藥品說明書檢索 使用 HIPAA‑compliant 向量資料庫，保護患者隱私，同時提供即時藥品資訊。
客服 常見問題與內部手冊 把公司內部 Wiki 建成向量索引，讓客服機器人即時引用官方說明。
教育 課程教材、歷年考題 把教材與解答檔案向量化，學生可用自然語言詢問「第 5 章的重點是什麼？」
法律 合同條款比對 把過往合同條款向量化，律師可快速找出相似條款與先例。

以上案例皆可透過 LangChain + FastAPI + Docker 快速構建原型，後續再根據流量與安全需求做橫向擴展（K8s、Istio、API Gateway）。

總結

本篇從 概念說明、程式碼實作、容器化部署、測試驗證 四個層面，完整呈現了 LangChain RAG 問答系統的部署與測試 流程。關鍵要點包括：

  • 使用 RecursiveCharacterTextSplitter 產生適合 LLM 的 Chunk。
  • 以 FAISS 或其他向量資料庫保存 Embedding，確保檢索效率。
  • 透過 RetrievalQA Chain 把檢索結果注入 Prompt，提升答案真實性。
  • 使用 FastAPI 包裝成 RESTful 介面，配合 Docker、pytest 實現 CI/CD。
  • 注意 金鑰安全、向量維度一致、索引更新 等常見陷阱，並遵循 日誌、健康檢查、資源限制 的最佳實踐。

掌握以上技巧後，你就能把 語言模型 與 企業知識庫 無縫結合，打造出 即時、可信、可擴展 的智慧問答服務。未來可以進一步探索 多模態 RAG（加入圖片、表格）或 自動化重建索引（使用 Airflow、Prefect），讓系統更加彈性與智慧。祝開發順利，期待看到你在實務上運用 LangChain 產生的更多創新應用！