Hugging Face 實戰（五）：端到端實戰 — 打造一個完整的 RAG 客服系統

 1# app/chunking.py
 2import re
 3from dataclasses import dataclass, field, asdict
 4
 5
 6@dataclass
 7class Chunk:
 8    id: str
 9    text: str                       # 用來檢索的小片段
10    parent_text: str                # 送給 LLM 的完整章節
11    doc_id: str
12    title: str                      # 完整標題路徑，例如「退換貨 > 流程 > 期限」
13    source: str                     # 檔案路徑或 URL
14    meta: dict = field(default_factory=dict)
15
16
17HEADING = re.compile(r"^(#{1,4})\s+(.+)$", re.MULTILINE)
18
19
20def split_markdown(text: str, doc_id: str, source: str,
21                   max_chars: int = 400, overlap: int = 80) -> list[Chunk]:
22    """依標題結構切分，保留標題路徑作為上下文"""
23    sections, stack, pos = [], [], 0
24
25    for m in HEADING.finditer(text):
26        if pos < m.start():
27            body = text[pos:m.start()].strip()
28            if body:
29                sections.append((list(stack), body))
30        level, title = len(m.group(1)), m.group(2).strip()
31        stack = stack[:level - 1] + [title]
32        pos = m.end()
33
34    tail = text[pos:].strip()
35    if tail:
36        sections.append((list(stack), tail))
37
38    chunks = []
39    for path, body in sections:
40        title_path = " > ".join(path) if path else "（無標題）"
41
42        # 章節本身不長就整段當一個片段
43        if len(body) <= max_chars:
44            pieces = [body]
45        else:
46            pieces = _sliding_split(body, max_chars, overlap)
47
48        for i, piece in enumerate(pieces):
49            chunks.append(Chunk(
50                id=f"{doc_id}::{len(chunks)}",
51                # 檢索文本前綴標題路徑，大幅提升短片段的可辨識度
52                text=f"{title_path}\n{piece}",
53                parent_text=f"# {title_path}\n\n{body}",   # 父片段 = 完整章節
54                doc_id=doc_id,
55                title=title_path,
56                source=source,
57                meta={"section_index": i, "section_total": len(pieces)},
58            ))
59    return chunks
60
61
62def _sliding_split(text: str, max_chars: int, overlap: int) -> list[str]:
63    """在句子邊界切，不要切在句子中間"""
64    sentences = re.split(r"(?<=[。！？；\n])", text)
65    out, buf = [], ""
66    for s in sentences:
67        if len(buf) + len(s) > max_chars and buf:
68            out.append(buf.strip())
69            buf = buf[-overlap:] + s      # 保留尾端重疊
70        else:
71            buf += s
72    if buf.strip():
73        out.append(buf.strip())
74    return out

 1# app/indexer.py
 2import json
 3import pickle
 4from pathlib import Path
 5
 6import faiss
 7import numpy as np
 8from sentence_transformers import SentenceTransformer
 9from rank_bm25 import BM25Okapi
10import jieba
11
12from .chunking import Chunk
13
14EMBED_MODEL = "BAAI/bge-m3"      # 多語、1024 維、支援 8192 token
15INDEX_DIR = Path("data/index")
16
17
18class Indexer:
19    def __init__(self, model_name: str = EMBED_MODEL, device: str = "cuda"):
20        self.encoder = SentenceTransformer(model_name, device=device)
21        self.dim = self.encoder.get_sentence_embedding_dimension()
22
23    def build(self, chunks: list[Chunk]):
24        INDEX_DIR.mkdir(parents=True, exist_ok=True)
25        texts = [c.text for c in chunks]
26
27        # ── 1. 向量索引 ──
28        vecs = self.encoder.encode(
29            texts,
30            batch_size=64,
31            normalize_embeddings=True,        # 正規化後內積 = 餘弦相似度
32            show_progress_bar=True,
33        ).astype("float32")
34
35        if len(chunks) < 50_000:
36            index = faiss.IndexFlatIP(self.dim)        # 精確搜尋，小資料集夠快
37        else:
38            quantizer = faiss.IndexFlatIP(self.dim)
39            nlist = int(4 * np.sqrt(len(chunks)))
40            index = faiss.IndexIVFFlat(quantizer, self.dim, nlist,
41                                       faiss.METRIC_INNER_PRODUCT)
42            index.train(vecs)
43        index.add(vecs)
44        faiss.write_index(index, str(INDEX_DIR / "faiss.index"))
45
46        # ── 2. BM25 索引（補向量檢索抓不到的精確詞：型號、錯誤碼）──
47        tokenized = [list(jieba.cut(t)) for t in texts]
48        with open(INDEX_DIR / "bm25.pkl", "wb") as f:
49            pickle.dump(BM25Okapi(tokenized), f)
50
51        # ── 3. 中繼資料 ──
52        with open(INDEX_DIR / "chunks.jsonl", "w", encoding="utf-8") as f:
53            for c in chunks:
54                f.write(json.dumps(c.__dict__, ensure_ascii=False) + "\n")
55
56        print(f"索引完成：{len(chunks):,} 個片段，維度 {self.dim}")

 1# scripts/build_index.py
 2from pathlib import Path
 3from app.chunking import split_markdown
 4from app.indexer import Indexer
 5
 6chunks = []
 7for path in Path("data/docs").rglob("*.md"):
 8    text = path.read_text(encoding="utf-8")
 9    chunks += split_markdown(
10        text,
11        doc_id=path.stem,
12        source=str(path.relative_to("data/docs")),
13    )
14
15print(f"共 {len(chunks):,} 個片段，平均 {sum(len(c.text) for c in chunks)//len(chunks)} 字")
16Indexer().build(chunks)

 1# app/retriever.py
 2import json
 3import pickle
 4from dataclasses import dataclass
 5from pathlib import Path
 6
 7import faiss
 8import jieba
 9import numpy as np
10import torch
11from sentence_transformers import SentenceTransformer
12from transformers import AutoTokenizer, AutoModelForSequenceClassification
13
14INDEX_DIR = Path("data/index")
15
16
17@dataclass
18class Hit:
19    chunk: dict
20    score: float
21    source_rank: dict          # 各路檢索的名次，方便除錯
22
23
24class HybridRetriever:
25    def __init__(self,
26                 embed_model="BAAI/bge-m3",
27                 rerank_model="BAAI/bge-reranker-v2-m3",
28                 device="cuda"):
29        self.device = device
30        self.encoder = SentenceTransformer(embed_model, device=device)
31        self.index = faiss.read_index(str(INDEX_DIR / "faiss.index"))
32        with open(INDEX_DIR / "bm25.pkl", "rb") as f:
33            self.bm25 = pickle.load(f)
34        self.chunks = [json.loads(l) for l in
35                       open(INDEX_DIR / "chunks.jsonl", encoding="utf-8")]
36
37        self.rk_tok = AutoTokenizer.from_pretrained(rerank_model)
38        self.rk = AutoModelForSequenceClassification.from_pretrained(
39            rerank_model, torch_dtype=torch.float16
40        ).to(device).eval()
41
42    # ── 第一階段：召回 ──
43    def _dense(self, query: str, k: int) -> list[int]:
44        # bge-m3 建議查詢端不加前綴；若用 bge-large-zh 則要加 "為這個句子生成表示："
45        qv = self.encoder.encode([query], normalize_embeddings=True).astype("float32")
46        _, idx = self.index.search(qv, k)
47        return idx[0].tolist()
48
49    def _sparse(self, query: str, k: int) -> list[int]:
50        scores = self.bm25.get_scores(list(jieba.cut(query)))
51        return np.argsort(scores)[::-1][:k].tolist()
52
53    @staticmethod
54    def _rrf(rank_lists: dict[str, list[int]], k: int = 60) -> dict[int, float]:
55        """Reciprocal Rank Fusion：不需要分數校準，只用名次"""
56        fused = {}
57        for lst in rank_lists.values():
58            for rank, doc_id in enumerate(lst):
59                fused[doc_id] = fused.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
60        return fused
61
62    # ── 第二階段：精排 ──
63    @torch.no_grad()
64    def _rerank(self, query: str, doc_ids: list[int]) -> list[tuple[int, float]]:
65        pairs = [[query, self.chunks[i]["text"]] for i in doc_ids]
66        enc = self.rk_tok(pairs, padding=True, truncation=True,
67                          max_length=512, return_tensors="pt").to(self.device)
68        logits = self.rk(**enc).logits.view(-1).float()
69        scores = torch.sigmoid(logits).cpu().numpy()     # 轉成 0–1 便於設閾值
70        order = np.argsort(scores)[::-1]
71        return [(doc_ids[i], float(scores[i])) for i in order]
72
73    # ── 對外介面 ──
74    def search(self, query: str, top_k: int = 5,
75               recall_k: int = 20, min_score: float = 0.35) -> list[Hit]:
76        dense = self._dense(query, recall_k)
77        sparse = self._sparse(query, recall_k)
78        fused = self._rrf({"dense": dense, "sparse": sparse})
79
80        candidates = sorted(fused, key=fused.get, reverse=True)[:recall_k * 2]
81        reranked = self._rerank(query, candidates)
82
83        hits = []
84        for doc_id, score in reranked[:top_k]:
85            if score < min_score:            # 相關度閘門
86                break
87            hits.append(Hit(
88                chunk=self.chunks[doc_id],
89                score=score,
90                source_rank={
91                    "dense": dense.index(doc_id) if doc_id in dense else None,
92                    "sparse": sparse.index(doc_id) if doc_id in sparse else None,
93                },
94            ))
95        return hits

 1GATE = {
 2    "answer": 0.55,      # 最高分 ≥ 0.55 → 正常回答
 3    "hedge":  0.35,      # 0.35–0.55  → 回答但明確標註「以下資訊可能不完整」
 4    # < 0.35 → 直接拒答，不呼叫 LLM
 5}
 6
 7def decide(hits: list[Hit]) -> str:
 8    if not hits:
 9        return "refuse"
10    top = hits[0].score
11    if top >= GATE["answer"]:
12        return "answer"
13    if top >= GATE["hedge"]:
14        return "hedge"
15    return "refuse"

 1# app/generator.py
 2import re
 3import time
 4from dataclasses import dataclass
 5
 6from openai import AsyncOpenAI
 7from .retriever import Hit
 8
 9SYSTEM = """你是產品客服助理。請嚴格遵守以下規則：
10
111. 只根據【參考資料】回答。資料中沒有的內容，回答「文件中未提供此資訊」。
122. 每個事實陳述後面必須標註來源編號，格式為 [1]、[2]。
133. 不要編造訂單編號、日期、金額、型號。
144. 回答控制在 200 字以內，用條列式呈現步驟性內容。
155. 用繁體中文，語氣專業友善。"""
16
17REFUSE_MSG = "抱歉，我在產品文件中找不到相關資訊。已為您記錄此問題，可協助轉接真人客服。"
18
19HEDGE_NOTE = "（以下資訊與您的問題相關度較低，僅供參考）\n\n"
20
21
22def build_context(hits: list[Hit], max_chars: int = 6000) -> tuple[str, list[dict]]:
23    """組裝參考資料區塊，並回傳引用清單。用 parent_text 提供完整上下文。"""
24    blocks, citations, used, seen = [], [], 0, set()
25
26    for hit in hits:
27        c = hit.chunk
28        key = (c["doc_id"], c["title"])
29        if key in seen:                       # 同章節不重複放
30            continue
31        seen.add(key)
32
33        body = c["parent_text"]
34        if used + len(body) > max_chars:
35            body = body[: max_chars - used]
36        if len(body) < 50:
37            break
38
39        n = len(citations) + 1
40        blocks.append(f"[{n}] 出處：{c['source']} — {c['title']}\n{body}")
41        citations.append({
42            "n": n, "source": c["source"], "title": c["title"],
43            "score": round(hit.score, 3),
44        })
45        used += len(body)
46
47    return "\n\n---\n\n".join(blocks), citations
48
49
50def build_messages(question: str, context: str, history: list[dict]) -> list[dict]:
51    msgs = [{"role": "system", "content": SYSTEM}]
52    msgs += history[-4:]                      # 只保留最近兩輪，避免 context 爆掉
53    msgs.append({
54        "role": "user",
55        "content": f"【參考資料】\n{context}\n\n【問題】\n{question}",
56    })
57    return msgs
58
59
60# ── 多輪對話的問題改寫 ──
61REWRITE_PROMPT = """把使用者的最新問題改寫成一個獨立、完整的問題，
62使其在沒有對話歷史的情況下也能被理解。只輸出改寫後的問題，不要任何解釋。
63
64對話歷史：
65{history}
66
67最新問題：{question}"""
68
69
70async def rewrite_query(llm, question: str, history: list[dict]) -> str:
71    if not history:
72        return question
73    hist = "\n".join(f"{m['role']}: {m['content'][:200]}" for m in history[-4:])
74    r = await llm.chat.completions.create(
75        model="support-chat",
76        messages=[{"role": "user",
77                   "content": REWRITE_PROMPT.format(history=hist, question=question)}],
78        temperature=0.0, max_tokens=100,
79    )
80    return r.choices[0].message.content.strip()
81
82
83# ── 引用驗證：擋掉不存在的編號 ──
84def verify_citations(answer: str, citations: list[dict]) -> tuple[str, list[int]]:
85    valid = {c["n"] for c in citations}
86    cited = {int(m) for m in re.findall(r"\[(\d+)\]", answer)}
87    bogus = cited - valid
88    for n in bogus:
89        answer = answer.replace(f"[{n}]", "")      # 移除幻覺引用
90    return answer, sorted(cited & valid)

  1# app/server.py
  2import json
  3import time
  4import hashlib
  5import logging
  6
  7import redis.asyncio as redis
  8from fastapi import FastAPI, HTTPException, Depends, Header
  9from fastapi.responses import StreamingResponse
 10from openai import AsyncOpenAI
 11from pydantic import BaseModel, Field
 12
 13from .retriever import HybridRetriever
 14from .generator import (
 15    build_context, build_messages, rewrite_query,
 16    verify_citations, decide, REFUSE_MSG, HEDGE_NOTE,
 17)
 18
 19log = logging.getLogger("rag")
 20app = FastAPI(title="Support RAG")
 21
 22retriever: HybridRetriever | None = None
 23llm = AsyncOpenAI(base_url="http://vllm:8000/v1", api_key="x")
 24cache = redis.from_url("redis://redis:6379", decode_responses=True)
 25
 26
 27@app.on_event("startup")
 28async def startup():
 29    global retriever
 30    retriever = HybridRetriever()          # 載入索引與 reranker（約 15 秒）
 31    log.info("retriever ready")
 32
 33
 34class AskRequest(BaseModel):
 35    question: str = Field(..., min_length=2, max_length=1000)
 36    history: list[dict] = Field(default_factory=list)
 37    stream: bool = False
 38
 39
 40def auth(x_api_key: str = Header(...)):
 41    # 實務上換成真正的金鑰檢查
 42    if not x_api_key:
 43        raise HTTPException(401, "missing api key")
 44
 45
 46@app.post("/ask", dependencies=[Depends(auth)])
 47async def ask(req: AskRequest):
 48    t0 = time.perf_counter()
 49
 50    # 1) 快取（單輪問題才快取；多輪的 context 不同）
 51    ckey = None
 52    if not req.history:
 53        ckey = "ask:" + hashlib.sha256(req.question.encode()).hexdigest()[:32]
 54        if (cached := await cache.get(ckey)) and not req.stream:
 55            out = json.loads(cached)
 56            out["cached"] = True
 57            return out
 58
 59    # 2) 問題改寫（多輪）
 60    query = await rewrite_query(llm, req.question, req.history)
 61
 62    # 3) 檢索
 63    t_retr = time.perf_counter()
 64    hits = retriever.search(query, top_k=5, recall_k=20)
 65    retrieval_ms = round((time.perf_counter() - t_retr) * 1000)
 66
 67    # 4) 閘門
 68    mode = decide(hits)
 69    if mode == "refuse":
 70        log.info(json.dumps({"event": "refuse", "q": req.question,
 71                             "top_score": hits[0].score if hits else None}))
 72        return {"answer": REFUSE_MSG, "citations": [], "refused": True,
 73                "retrieval_ms": retrieval_ms}
 74
 75    context, citations = build_context(hits)
 76    messages = build_messages(query, context, req.history)
 77
 78    # 5) 生成
 79    if req.stream:
 80        async def gen():
 81            yield f"event: citations\ndata: {json.dumps(citations, ensure_ascii=False)}\n\n"
 82            if mode == "hedge":
 83                yield f"data: {json.dumps({'delta': HEDGE_NOTE})}\n\n"
 84            s = await llm.chat.completions.create(
 85                model="support-chat", messages=messages,
 86                temperature=0.2, max_tokens=512, stream=True,
 87            )
 88            async for chunk in s:
 89                if d := chunk.choices[0].delta.content:
 90                    yield f"data: {json.dumps({'delta': d}, ensure_ascii=False)}\n\n"
 91            yield "data: [DONE]\n\n"
 92        return StreamingResponse(gen(), media_type="text/event-stream")
 93
 94    resp = await llm.chat.completions.create(
 95        model="support-chat", messages=messages,
 96        temperature=0.2, max_tokens=512,
 97    )
 98    answer, used = verify_citations(resp.choices[0].message.content, citations)
 99    if mode == "hedge":
100        answer = HEDGE_NOTE + answer
101
102    out = {
103        "answer": answer,
104        "citations": [c for c in citations if c["n"] in used],
105        "refused": False,
106        "retrieval_ms": retrieval_ms,
107        "total_ms": round((time.perf_counter() - t0) * 1000),
108        "usage": resp.usage.model_dump(),
109    }
110
111    # 6) 記錄（這是之後做 DPO 的資料來源）
112    log.info(json.dumps({
113        "event": "answer", "q": req.question, "rewritten": query,
114        "top_score": hits[0].score, "n_citations": len(used),
115        **{k: out[k] for k in ("retrieval_ms", "total_ms")},
116    }, ensure_ascii=False))
117
118    if ckey:
119        await cache.setex(ckey, 3600, json.dumps(out, ensure_ascii=False))
120    return out
121
122
123@app.post("/feedback")
124async def feedback(payload: dict):
125    """使用者按讚/倒讚 → 直接寫成 KTO 格式，之後可餵給第四篇的訓練流程"""
126    log.info(json.dumps({"event": "feedback", **payload}, ensure_ascii=False))
127    return {"ok": True}
128
129
130@app.get("/healthz")
131async def healthz():
132    if retriever is None:
133        raise HTTPException(503, "index not loaded")
134    await llm.models.list()
135    return {"status": "ok"}

 1# app/ui.py
 2import json
 3import requests
 4import gradio as gr
 5
 6API = "http://api:8080"
 7HEADERS = {"x-api-key": "dev-key"}
 8
 9
10def respond(message, history):
11    hist = [{"role": m["role"], "content": m["content"]} for m in history]
12    r = requests.post(f"{API}/ask", headers=HEADERS, stream=True,
13                      json={"question": message, "history": hist, "stream": True})
14
15    citations, partial = [], ""
16    for raw in r.iter_lines(decode_unicode=True):
17        if not raw:
18            continue
19        if raw.startswith("event: citations"):
20            continue
21        if raw.startswith("data: "):
22            payload = raw[6:]
23            if payload == "[DONE]":
24                break
25            try:
26                obj = json.loads(payload)
27            except json.JSONDecodeError:
28                continue
29            if isinstance(obj, list):
30                citations = obj
31                continue
32            partial += obj.get("delta", "")
33            yield partial
34
35    if citations:
36        refs = "\n".join(f"- [{c['n']}] {c['title']}（{c['source']}，"
37                         f"相關度 {c['score']}）" for c in citations)
38        yield partial + f"\n\n---\n**參考來源**\n{refs}"
39
40
41demo = gr.ChatInterface(
42    fn=respond,
43    type="messages",
44    title="📚 產品客服助理",
45    description="所有回答都附上文件來源；文件中沒有的資訊會明確告知。",
46    examples=[
47        "退貨的期限是幾天？",
48        "錯誤碼 E-4021 是什麼意思？",
49        "企業版和專業版的差別在哪？",
50    ],
51)
52
53if __name__ == "__main__":
54    demo.launch(server_name="0.0.0.0", server_port=7860)

 1# docker-compose.yml
 2services:
 3  vllm:
 4    image: vllm/vllm-openai:latest
 5    command: >
 6      --model /models/support-awq
 7      --served-model-name support-chat
 8      --max-model-len 8192
 9      --gpu-memory-utilization 0.75      
10    volumes:
11      - ./models:/models
12    shm_size: "2gb"
13    deploy:
14      resources:
15        reservations:
16          devices: [{driver: nvidia, count: 1, capabilities: [gpu]}]
17    healthcheck:
18      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
19      interval: 15s
20      retries: 20
21
22  api:
23    build: .
24    command: uvicorn app.server:app --host 0.0.0.0 --port 8080
25    environment:
26      HF_HOME: /hf
27    volumes:
28      - ./data:/app/data
29      - hf-cache:/hf
30    ports: ["8080:8080"]
31    depends_on:
32      vllm: {condition: service_healthy}
33      redis: {condition: service_started}
34    deploy:
35      resources:
36        reservations:
37          devices: [{driver: nvidia, count: 1, capabilities: [gpu]}]
38
39  ui:
40    build: .
41    command: python -m app.ui
42    ports: ["7860:7860"]
43    depends_on: [api]
44
45  redis:
46    image: redis:7-alpine
47    command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru
48
49volumes:
50  hf-cache:

 1# scripts/evaluate.py
 2import json
 3import time
 4import statistics
 5import requests
 6
 7API = "http://localhost:8080"
 8HEADERS = {"x-api-key": "dev-key"}
 9
10# 測試集格式：
11# {"question": "...", "expected_sources": ["returns.md"], "must_include": ["七日"],
12#  "should_refuse": false}
13cases = [json.loads(l) for l in open("data/eval/qa.jsonl", encoding="utf-8")]
14
15results, latencies = [], []
16for case in cases:
17    t0 = time.perf_counter()
18    r = requests.post(f"{API}/ask", headers=HEADERS,
19                      json={"question": case["question"]}).json()
20    latencies.append((time.perf_counter() - t0) * 1000)
21
22    answer = r["answer"]
23    got_sources = {c["source"] for c in r.get("citations", [])}
24
25    results.append({
26        "q": case["question"],
27        # 檢索：期望的來源有沒有被引用到
28        "retrieval_ok": bool(set(case.get("expected_sources", [])) & got_sources),
29        # 生成：關鍵字有沒有出現
30        "content_ok": all(k in answer for k in case.get("must_include", [])),
31        # 拒答：該拒的有沒有拒
32        "refusal_ok": r["refused"] == case.get("should_refuse", False),
33        # 引用：有沒有標註來源
34        "cited": len(r.get("citations", [])) > 0,
35        "refused": r["refused"],
36    })
37
38n = len(results)
39print(f"樣本數        : {n}")
40print(f"檢索命中率     : {sum(r['retrieval_ok'] for r in results)/n:.1%}")
41print(f"內容正確率     : {sum(r['content_ok'] for r in results)/n:.1%}")
42print(f"拒答判斷正確率 : {sum(r['refusal_ok'] for r in results)/n:.1%}")
43print(f"引用覆蓋率     : {sum(r['cited'] for r in results if not r['refused'])"
44      f"/max(sum(not r['refused'] for r in results),1):.1%}")
45print(f"P50 延遲      : {statistics.median(latencies):.0f} ms")
46print(f"P95 延遲      : {sorted(latencies)[int(n*0.95)]:.0f} ms")
47
48# 失敗案例輸出，供人工檢視
49with open("data/eval/failures.jsonl", "w", encoding="utf-8") as f:
50    for r in results:
51        if not (r["retrieval_ok"] and r["content_ok"] and r["refusal_ok"]):
52            f.write(json.dumps(r, ensure_ascii=False) + "\n")

 1# scripts/collect_preference.py
 2# 把生產日誌轉成第四篇要用的 KTO 格式
 3import json
 4
 5out = []
 6for line in open("logs/app.jsonl", encoding="utf-8"):
 7    ev = json.loads(line)
 8    if ev.get("event") != "feedback":
 9        continue
10    out.append({
11        "prompt": [{"role": "user", "content": ev["question"]}],
12        "completion": [{"role": "assistant", "content": ev["answer"]}],
13        "label": ev["rating"] == "up",
14    })
15
16with open("data/kto_from_prod.jsonl", "w", encoding="utf-8") as f:
17    for r in out:
18        f.write(json.dumps(r, ensure_ascii=False) + "\n")
19print(f"產出 {len(out):,} 筆（正例 {sum(r['label'] for r in out):,}）")