rag-architect

RAG Architect

核心工作流程

1. 需求分析 — 確認檢索需求、延遲限制、準確度要求與規模
2. 向量資料庫設計 — 選擇資料庫、綱要設計、索引策略、分片方式
3. 分塊策略 — 文件分割、重疊、語意邊界、中繼資料豐富化
4. 檢索管道 — 嵌入選擇、查詢轉換、混合搜尋、重新排序
5. 評估與迭代 — 指標追蹤、檢索除錯、持續最佳化

每個步驟完成後需進行驗證（請見下方檢查點）。

參考指南

根據情境載入詳細指引：

主題	參考文件	載入時機
向量資料庫	references/vector-databases.md	比較 Pinecone、Weaviate、Chroma、pgvector、Qdrant
嵌入模型	references/embedding-models.md	選擇嵌入、微調、維度取捨
分塊策略	references/chunking-strategies.md	文件分割、重疊、語意分塊
檢索最佳化	references/retrieval-optimization.md	混合搜尋、重新排序、查詢擴展、過濾
RAG 評估	references/rag-evaluation.md	指標、評估框架、檢索除錯

實作範例

1. 文件分塊

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 請根據領域資料評估 chunk_size — 切勿盲目使用 512
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.create_documents(
    texts=[doc.page_content for doc in raw_docs],
    metadatas=[{"source": doc.metadata["source"], "timestamp": doc.metadata.get("timestamp")} for doc in raw_docs],
)

檢查點： assert all(c.metadata.get("source") for c in chunks), "Missing source metadata"

2. 生成嵌入向量與索引

from openai import OpenAI
import qdrant_client
from qdrant_client.models import VectorParams, Distance, PointStruct

client = OpenAI()
qdrant = qdrant_client.QdrantClient("localhost", port=6333)

# 建立集合
qdrant.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(input=chunks, model=model)
    return [r.embedding for r in response.data]

# 透過確定性 ID 實現冪等寫入與去重
import hashlib, uuid

points = []
for i, chunk in enumerate(chunks):
    doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))
    embedding = embed_chunks([chunk.page_content])[0]
    points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))

qdrant.upsert(collection_name="knowledge_base", points=points)

檢查點： assert qdrant.count("knowledge_base").count == len(set(p.id for p in points)), "Deduplication failed"

3. 混合搜尋（向量 + BM25）

from qdrant_client.models import Filter, FieldCondition, MatchValue, SparseVector
from rank_bm25 import BM25Okapi

def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
    # 稠密檢索
    query_embedding = embed_chunks([query])[0]
    tenant_filter = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))])
    dense_results = qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_embedding,
        query_filter=tenant_filter,
        limit=top_k,
    )

    # 稀疏檢索（BM25）
    corpus = [r.payload.get("text", "") for r in dense_results]
    bm25 = BM25Okapi([doc.split() for doc in corpus])
    bm25_scores = bm25.get_scores(query.split())

    # 互惠排名融合
    ranked = sorted(
        zip(dense_results, bm25_scores),
        key=lambda x: 0.6 * x[0].score + 0.4 * x[1],
        reverse=True,
    )
    return [r for r, _ in ranked[:top_k]]

檢查點： assert len(hybrid_search("test query", tenant_id="demo")) > 0, "Hybrid search returned no results"

4. 對 Top-K 結果重新排序

import cohere

co = cohere.Client("YOUR_API_KEY")

def rerank(query: str, results: list, top_n: int = 5) -> list:
    docs = [r.payload.get("text", "") for r in results]
    reranked = co.rerank(query=query, documents=docs, top_n=top_n, model="rerank-english-v3.0")
    return [results[r.index] for r in reranked.results]

5. 檢索評估

# 針對標記好的評估集執行 precision@k 與 recall@k
# python evaluate.py --metrics precision@10 recall@10 mrr --collection knowledge_base

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from datasets import Dataset

eval_dataset = Dataset.from_dict({
    "question": questions,
    "contexts": retrieved_contexts,
    "answer": generated_answers,
    "ground_truth": ground_truth_answers,
})

results = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(results)

檢查點： 在進入 LLM 整合前，目標為 context_precision >= 0.7 且 context_recall >= 0.6。

限制

必須做

• 在您的領域資料上評估多個嵌入模型後再決定
• 為生產系統實作混合搜尋（向量 + 關鍵字）
• 加入中繼資料過濾以支援多租戶或特定領域檢索
• 衡量檢索指標（precision@k、recall@k、MRR、NDCG）
• 在將上下文傳遞給 LLM 前，對 top-k 結果進行重新排序
• 實作具去重功能的冪等資料寫入（確定性 ID）
• 持續監控檢索延遲與品質
• 為嵌入模型進行版本管理，並規劃模型遷移

禁止做

• 未經領域資料評估就使用預設分塊大小（512）
• 跳過中繼資料豐富化（來源、時間戳、章節）
• 只關注 LLM 輸出品質而忽略檢索品質指標
• 未經前處理/清理就直接儲存原始文件
• 在複雜多領域檢索中僅使用餘弦相似度
• 未在接近生產的資料量下測試就部署
• 忘記處理邊界情況（空結果、格式錯誤的文件）
• 將嵌入模型與應用程式程式碼緊密耦合

輸出範本

設計 RAG 架構時，應提供：

1. 系統架構圖（資料寫入 + 檢索管道）
2. 向量資料庫選擇與取捨分析
3. 分塊策略（含範例與理由）
4. 檢索管道設計（查詢 → 結果流程）
5. 評估計畫（含指標、基準、通過/失敗門檻）

文件