「大語言模型給的是一般性常識，而 RAG 檢索增強技術給的，才是讓 AI Agent 具備特定金融領域深度的靈魂。」

在 Day 04 中，我們介紹了 API 閘道器（gemini_gateway.py）與 FastAPI 的主流程。今天我們將深入探討 Angelina AI Agent 的知識庫心臟——app/services/rag_engine.py。

為了保持全系統 $0 元運營 的承諾，我們沒有選擇 Pinecone 等雲端付費向量資料庫，而是採用本地運行的 ChromaDB，搭配完全免費的 sentence-transformers/all-MiniLM-L6-v2 嵌入模型（Embedding Model）。

本篇重點摘要

1. 為什麼選用 sentence-transformers/all-MiniLM-L6-v2 與 ChromaDB呢？
2. 使用 asyncio.to_thread 將 CPU 密集的 Embedding 運算移出非同步 Event Loop。
3. 實作相似度過濾 (SIMILARITY_THRESHOLD = 0.5) 與餘弦距離 (Cosine Space) 轉換。
4. 自動化文本切分 (tiktoken) 與索引重建機制（rebuild_index）。

一、RAG 引擎架構與選型考量

在 Self-hosted AI Agent 架構中，Embedding 計算（文字向量化）屬於 CPU 密集型任務。如果直接在 FastAPI 的非同步主執行緒（Event Loop）中執行，會導致整體 API 卡頓。

因此，RAGEngine 在設計上採取了以下關鍵措施：
1. 模組級單例 (Singletons)：_sentence_transformer_model 與 _chroma_client 採用 Lazy-initialisation 延遲初始化，並在全系統共享。
2. Event Loop 解耦：所有涉及 Embedding 計算與 ChromaDB 磁碟 I/O 的同步方法，均透過 asyncio.to_thread 包裹，推送到 Python Thread Pool 執行。

二、app/services/rag_engine.py 原始碼解析

1. 向量檢索與相似度計算 (Cosine Similarity)
   ChromaDB 預設使用 HNSW 餘弦空間 (Cosine Space)，其回傳的 distance 與語意相似度 similarity 的轉換公式為：
   $$\text{similarity} = 1.0 - \text{distance}$$

我們設定 SIMILARITY_THRESHOLD = 0.5，低於此相似度的知識片段將會被濾除，避免引導 AI 產生幻覺：

Python
def _sync_search(query: str, top_k: int, persist_path: str) -> list[dict]:
 """同步檢索：將查詢向量化、查詢 ChromaDB 並轉換相似度"""
 _, collection = _get_or_create_collection(persist_path)

 if collection.count() == 0:
     return []

 query_embedding = _embed_texts([query])[0]

 results = collection.query(
     query_embeddings=[query_embedding],
     n_results=min(top_k, collection.count()),
     include=["documents", "metadatas", "distances"],
 )

 output: list[dict] = []
 if not results or not results.get("ids"):
     return output

 ids = results["ids"][0]
 documents = results["documents"][0]
 metadatas = results["metadatas"][0]
 distances = results["distances"][0]

 for chunk_id, text, meta, distance in zip(ids, documents, metadatas, distances):
     # ChromaDB 余弦空間：distance = 1 - similarity
     similarity = 1.0 - float(distance)
     if similarity >= SIMILARITY_THRESHOLD:
         output.append({
             "id": chunk_id,
             "text": text,
             "source_type": meta.get("source_type", "notebooklm"),
             "created_at": meta.get("created_at", ""),
             "similarity": similarity,
         })

 return output

2. 非同步 API 封裝 (asyncio.to_thread)
   在對外公開的 RAGEngine 類別中，所有公開方法皆為 async，透過 asyncio.to_thread 將阻塞運算派發給線程池，確保主流程不塞車：

Python
class RAGEngine:
 """Async RAG Engine backed by ChromaDB and sentence-transformers."""

 def __init__(
     self,
     persist_path: str = VECTOR_STORE_PATH,
     notebooklm_path: str = NOTEBOOKLM_DATA_PATH,
 ) -> None:
     self._persist_path = persist_path
     self._notebooklm_path = notebooklm_path

 async def search(self, query: str, top_k: int = DEFAULT_TOP_K) -> list[Chunk]:
     """語意檢索 top-k 相關知識片段 (非同步無阻塞)"""
     from app.models import Chunk

     raw: list[dict] = await asyncio.to_thread(
         _sync_search, query, top_k, self._persist_path
     )

     chunks = [
         Chunk(
             id=r["id"],
             text=r["text"],
             source_type=r["source_type"],
             created_at=r["created_at"],
             similarity=r["similarity"],
         )
         for r in raw
     ]
     chunks.sort(key=lambda c: c.similarity, reverse=True)
     return chunks

 async def add_chunks(self, chunks: list[Chunk]) -> None:
     """非同步批次向量化並寫入 ChromaDB"""
     if not chunks:
         return

     chunks_data = [
         {
             "id": c.id,
             "text": c.text,
             "source_type": c.source_type,
             "created_at": c.created_at,
         }
         for c in chunks
     ]
     await asyncio.to_thread(_sync_add_chunks, chunks_data, self._persist_path)
     ```
     
3. 自動化文本切分與索引重建 (rebuild_index)
當我們從 NotebookLM 匯出新的筆記（放置於 data/notebooklm/）或觸發重構時，可透過 rebuild_index 方法，利用 RecursiveCharacterTextSplitter 搭配 tiktoken (cl100k_base 模組) 進行精準切分：
   (1) Chunk Size：500 Token
   (2) Overlap Size：50 Token
   (3) 分隔符號：優先以 \n\n、\n、句號 。 與句點 . 進行語法斷句。
   

  Python
  def _sync_rebuild_index(export_path: str, persist_path: str) -> None:
"""重建向量索引：清空舊 Collection、使用 tiktoken 分塊並重建立向量"""
global _chroma_client, _chroma_collection

import chromadb
from langchain_text_splitters import RecursiveCharacterTextSplitter
import tiktoken

# 讀取 NotebookLM 匯出文本
export_text = Path(export_path).read_text(encoding="utf-8")
enc = tiktoken.get_encoding("cl100k_base")

def _token_length(text: str) -> int:
    return len(enc.encode(text))

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    length_function=_token_length,
    separators=["\n\n", "\n", "。", ".", " ", ""],
)

raw_chunks = splitter.split_text(export_text)
# 進行 Embedding 向量化並寫入 ChromaDB...
```

三、今日總結與架構優勢

透過 rag_engine.py 的實作，我們建構了一套輕量且高效的本地 RAG 系統：

1. 零成本：模型與向量資料庫完全在本地端運作，不需支付外包向量雲端服務費用。
2. 高效能：將 CPU 密集運算解耦至線程池（Thread Pool），維護了 FastAPI 事件循環的高響應性。
3. 數據合規：所有私有金融知識均留存在本地 RHEL 宿主機中，不外洩至第三方平台。

明天（Day 06）我們將進入 app/services/conversation_memory.py，探討如何透過 aiosqlite 建立長期對話記憶，以及超過 100 輪對話時的 Auto-Summarization 實務！

明日預告：【Day 06】長效對話記憶：aiosqlite 持久化儲存與自動對話摘要壓縮