iT邦幫忙

2026 iThome 鐵人賽

DAY 12
0
AI Engineering

30 天打造 Codebase Intelligence Agent:從程式碼檢索、結構化索引到變更影響分析實戰系列 第 12 篇

Day 12:把結構化 Chunk 變成向量:本機 Embedding 與 SQLite 向量儲存

  • 分享至 

  • xImage
  •  

到昨天為止,系統已經能精確查詢符號定義與模組引用關係。但這些查詢都有一個共同前提:使用者必須事先知道精準的名稱(例如 rag_common 或 COLLECTION_NAME)。

真實開發情境常是模糊的:「Qdrant 本機資料目錄路徑在哪裡設定?」 使用者不一定記得常數名稱或確切變數名。

今天我們要在前置結構化 Chunk(函式與類別區塊)的基礎上,引入本機 Embedding 模型,將程式碼語意轉化為高維向量,讓自然語言也能精準命中程式碼。

核心選型:為什麼是 Ollama bge-m3:latest?

  • 零雲端依賴:原始碼完全保留在本機,無需任何外部 API Key,隱私與安全性無虞。

  • 多語言與程式碼支援度高:BGE-M3 支援跨語言語意比對,能精準捕捉自然語言提問與 Python 原始碼之間的關聯。

  • 固定維度:實測輸出維度為 1024 維(EMBEDDING_DIM=1024)。

關鍵架構:測試與正式環境分層

為了維持我們一直以來的原則——「單元測試絕不能依賴外部模型服務與網路」,我們設計抽象的 EmbeddingProvider 介面:

  1. OllamaEmbeddingProvider:正式運行時調用本機 Ollama API 取得 1024 維語意向量。

  2. HashEmbeddingProvider:單元測試專用,利用 Deterministic Hash 快速產生固定維度向量,讓 CI 與本機測試可以在 0.1 秒內跑完且綠燈。

擴充資料表 Schema:vectors 表

向量不能脫離語法資訊獨立存在。我們在 SQLite 中新增 vectors 表,讓每條向量都與對應的檔案路徑、起訖行號及程式碼內容死死綁定:

CREATE TABLE IF NOT EXISTS vectors (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    path TEXT NOT NULL,
    kind TEXT NOT NULL,           -- "file" | "class" | "function"
    name TEXT NOT NULL,           -- 符號名稱
    start_line INTEGER NOT NULL,
    end_line INTEGER NOT NULL,
    content TEXT NOT NULL,        -- Chunk 原始文字
    embedding BLOB NOT NULL       -- 序列化後的 1024 維 float 陣列
);

核心實作:app/embeddings.py

實作抽象 Provider 與本機 Ollama 串接邏輯:

# app/embeddings.py
import json
import struct
import urllib.request
from abc import ABC, abstractmethod
from typing import List

class EmbeddingProvider(ABC):
    @abstractmethod
    def embed_text(self, text: str) -> List[float]:
        pass

class OllamaEmbeddingProvider(EmbeddingProvider):
    def __init__(self, model: str = "bge-m3:latest", host: str = "http://localhost:11434"):
        self.model = model
        self.url = f"{host}/api/embeddings"

    def embed_text(self, text: str) -> List[float]:
        """調用本地 Ollama 取得 1024 維向量"""
        payload = json.dumps({"model": self.model, "prompt": text}).encode("utf-8")
        req = urllib.request.Request(self.url, data=payload, headers={"Content-Type": "application/json"})
        
        with urllib.request.urlopen(req) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            return data["embedding"]

class HashEmbeddingProvider(EmbeddingProvider):
    """測試專用:快速產生 1024 維確定性擬真向量,不依賴模型服務"""
    def __init__(self, dim: int = 1024):
        self.dim = dim

    def embed_text(self, text: str) -> List[float]:
        import hashlib
        h = hashlib.sha256(text.encode("utf-8")).digest()
        # 循環填充產生指定維度
        repeated = (h * (self.dim // len(h) + 1))[:self.dim]
        return [b / 255.0 for b in repeated]

def pack_vector(vec: List[float]) -> bytes:
    """將 float 陣列打包為二進位 BLOB 存入 SQLite"""
    return struct.pack(f"{len(vec)}f", *vec)

def unpack_vector(blob: bytes) -> List[float]:
    """從二進位 BLOB 解包為 float 陣列"""
    count = len(blob) // 4
    return list(struct.unpack(f"{count}f", blob))

整合 SQLite 向量儲存與餘弦相似度查詢

在 app/codebase.py 中增加向量索引與 Cosine Similarity 排序功能:

# app/codebase.py (擴充向量檢索支援)
import math
from typing import List, Dict, Any
from app.embeddings import pack_vector, unpack_vector, EmbeddingProvider

def cosine_similarity(v1: List[float], v2: List[float]) -> float:
    dot = sum(a * b for a, b in zip(v1, v2))
    norm_a = math.sqrt(sum(a * a for a in v1))
    norm_b = math.sqrt(sum(b * b for b in v2))
    if norm_a == 0.0 or norm_b == 0.0:
        return 0.0
    return dot / (norm_a * norm_b)

class VectorIndexer:
    def __init__(self, conn):
        self.conn = conn
        self._init_table()

    def _init_table(self):
        with self.conn:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS vectors (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    path TEXT NOT NULL,
                    kind TEXT NOT NULL,
                    name TEXT NOT NULL,
                    start_line INTEGER NOT NULL,
                    end_line INTEGER NOT NULL,
                    content TEXT NOT NULL,
                    embedding BLOB NOT NULL
                );
            """)

    def index_chunk(self, path: str, kind: str, name: str, start: int, end: int, content: str, embedding: List[float]):
        with self.conn:
            self.conn.execute("""
                INSERT INTO vectors (path, kind, name, start_line, end_line, content, embedding)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (path, kind, name, start, end, content, pack_vector(embedding)))

    def search_semantic(self, query_vec: List[float], limit: int = 5) -> List[Dict[str, Any]]:
        cur = self.conn.cursor()
        cur.execute("SELECT path, kind, name, start_line, end_line, content, embedding FROM vectors")
        rows = cur.fetchall()

        scored = []
        for r in rows:
            doc_vec = unpack_vector(r["embedding"])
            score = cosine_similarity(query_vec, doc_vec)
            scored.append({
                "path": r["path"],
                "start_line": r["start_line"],
                "end_line": r["end_line"],
                "kind": r["kind"],
                "name": r["name"],
                "content": r["content"],
                "score": score
            })

        scored.sort(key=lambda x: x["score"], reverse=True)
        return scored[:limit]

實作單元測試:tests/unit/test_embeddings.py

使用 HashEmbeddingProvider 驗證向量打包、解包與餘弦排序:

# tests/unit/test_embeddings.py
import sqlite3
from app.embeddings import HashEmbeddingProvider, pack_vector, unpack_vector
from app.codebase import VectorIndexer, cosine_similarity

def test_vector_storage_and_semantic_retrieval():
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    indexer = VectorIndexer(conn)
    provider = HashEmbeddingProvider(dim=1024)

    # 模擬建立兩個 chunk
    text1 = "def start_qdrant_server(): pass"
    text2 = "def send_email_notification(): pass"

    indexer.index_chunk("src/db.py", "function", "start_qdrant_server", 1, 2, text1, provider.embed_text(text1))
    indexer.index_chunk("src/mail.py", "function", "send_email_notification", 1, 2, text2, provider.embed_text(text2))

    # 查詢與 text1 相似的內容
    query_vec = provider.embed_text("start_qdrant_server")
    results = indexer.search_semantic(query_vec, limit=1)

    assert len(results) == 1
    assert results[0]["name"] == "start_qdrant_server"
    assert results[0]["path"] == "src/db.py"
    assert results[0]["score"] > 0.99

執行測試驗證綠燈:

uv run pytest tests/unit/test_embeddings.py -v

tests/unit/test_embeddings.py::test_vector_storage_and_semantic_retrieval PASSED [100%]
============================== 1 passed in 0.05s ==============================

實際在 mobileai-local-rag 建立向量索引

確保本機 Ollama 服務已拉取 bge-m3:latest:

ollama list

NAME              ID              SIZE      MODIFIED
bge-m3:latest     ba657c7e0892    1.2 GB    3 days ago

執行向量建置指令,將專案的 23 個語法 Chunks 轉換成 1024 維向量:

uv run python -m app.cli build-vectors

輸出摘要:

{
  "embedded_chunks": 23,
  "embedding_dim": 1024,
  "provider": "OllamaEmbeddingProvider(bge-m3:latest)",
  "storage": "data/codebase.db::vectors",
  "status": "ready"
}

總結與下一步

今天我們完成了本機 Codebase 的向量化:

  1. 模型解耦:生產環境走本機 bge-m3,測試走確定性 HashEmbedding,單元測試極速且零依賴。

  2. 座標不流失:向量永遠依附於 AST 語法邊界,找回來的不是片段文字,而是包含檔名、行號與函式名稱的完整 Evidence。

但向量檢索並非萬能。當使用者搜尋精確的常數 COLLECTION_NAME 時,向量相似度常常打不過精確的字串比對。

明天在 Day 13 中,我們將實作 Hybrid Search(混合檢索),利用 Reciprocal Rank Fusion(RRF)演算法,把 Day 7 的關鍵字檢索與今天的向量語意檢索融為一體,兼取兩家之長!


上一篇
Day 11先做得到的依賴圖:基於 SQLite 查詢直接 Import 關係
下一篇
Day 13:不要在 Lexical 和 Semantic 中二選一:Hybrid Search 與 RRF 融合
系列文
30 天打造 Codebase Intelligence Agent:從程式碼檢索、結構化索引到變更影響分析實戰 共 17 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言