TurboVec 是什麼?
TurboVec 是一個用 Rust 編寫、提供 Python 綁定的高效能向量索引庫,基於 Google Research 的 TurboQuant 量化演算法建構。它的核心目標是解決 RAG(檢索增強生成)系統中的兩大痛點:記憶體佔用和搜尋速度。
為什麼需要 TurboVec?
在傳統的向量搜尋場景中,如果你有一個包含 1000 萬條文件的語料庫,使用 float32 格式儲存向量需要約 31 GB 的 RAM。而 TurboVec 透過資料無關量化(data-oblivious quantization)技術,將同樣的資料集壓縮到僅 4 GB —— 記憶體佔用降低了 87%,同時搜尋速度還超越了 FAISS。
TurboVec vs FAISS
| 特性 | FAISS | TurboVec |
|---|---|---|
| 記憶體佔用 | 高(float32 或 PQ 量化) | 極低(TurboQuant 量化) |
| 是否需要訓練 | ✅ 需要訓練階段 | ❌ 無需訓練,即插即用 |
| 線上添加向量 | ⚠️ 需重建索引 | ✅ 即時添加,無需重建 |
| 過濾搜尋 | 需後處理 | ✅ 核心層級支援,無效能損失 |
| SIMD 最佳化 | 有 | 手寫 NEON (ARM) + AVX-512BW (x86) |
| 純本機執行 | ✅ | ✅ |
| Python 綁定 | ✅ | ✅ |
| Rust 原生支援 | ❌ | ✅ |
核心技術:TurboQuant 演算法
TurboQuant 是 Google Research 在 2025 年提出的新型量化演算法,關鍵特點包括:
- 資料無關(Data-Oblivious):不需要針對特定資料集訓練碼本(codebook),避免了傳統 PQ(Product Quantization)的訓練開銷
- 逼近 Shannon 下界:在失真度上接近理論最佳值
- 位寬靈活:支援 2-bit、4-bit、8-bit 等多種量化精度,平衡召回率和記憶體佔用
在基準測試中,TurboVec 在 ARM 架構上比 FAISS IndexPQFastScan 快 12-20%,在 x86 架構上持平或略優。
安裝 TurboVec
TurboVec 提供了 Python 和 Rust 兩種語言的介面。
Python 安裝
pip install turbovec
如果需要與 LangChain、LlamaIndex 等框架整合,可以安裝對應的額外依賴:
# LangChain 整合
pip install turbovec[langchain]
# LlamaIndex 整合
pip install turbovec[llama-index]
# Haystack 整合
pip install turbovec[haystack]
# Agno 整合
pip install turbovec[agno]
Rust 安裝
在 Cargo.toml 中添加依賴:
[dependencies]
turbovec = "0.1"
快速上手:Python 基礎用法
建立索引並添加向量
import numpy as np
from turbovec import TurboQuantIndex
# 建立索引:維度 1536(OpenAI embedding 預設維度),4-bit 量化
index = TurboQuantIndex(dim=1536, bit_width=4)
# 生成示例向量(實際使用中替換為你的 embedding 向量)
vectors = np.random.rand(10000, 1536).astype(np.float32)
# 添加向量到索引
index.add(vectors)
# 可以繼續添加更多向量,無需重建索引
more_vectors = np.random.rand(5000, 1536).astype(np.float32)
index.add(more_vectors)
print(f"索引中共有 {len(index)} 個向量")
執行搜尋
# 生成查詢向量
query = np.random.rand(1536).astype(np.float32)
# 搜尋最相似的 10 個向量
scores, indices = index.search(query, k=10)
print("相似度分數:", scores)
print("向量索引:", indices)
持久化儲存與載入
# 儲存索引到磁碟
index.write("my_index.tq")
# 從磁碟載入索引
loaded_index = TurboQuantIndex.load("my_index.tq")
# 驗證載入成功
scores, indices = loaded_index.search(query, k=10)
進階功能一:使用外部 ID 對應
在實際應用中,你通常需要將向量索引與你資料庫中的文件 ID 關聯起來。TurboVec 提供了 IdMapIndex 來支援這一需求。
添加帶 ID 的向量
import numpy as np
from turbovec import IdMapIndex
# 建立支援外部 ID 的索引
index = IdMapIndex(dim=1536, bit_width=4)
# 假設你有 3 個向量,對應的外部 ID 為 1001, 1002, 1003
vectors = np.random.rand(3, 1536).astype(np.float32)
external_ids = np.array([1001, 1002, 1003], dtype=np.uint64)
# 添加向量和對應的外部 ID
index.add_with_ids(vectors, external_ids)
# 搜尋返回的是外部 ID,而非內部索引
query = np.random.rand(1536).astype(np.float32)
scores, ids = index.search(query, k=10)
print("返回的外部 ID:", ids) # [1001, 1003, 1002, ...]
刪除向量
IdMapIndex 支援透過外部 ID 直接刪除向量,時間複雜度為 O(1):
# 刪除 ID 為 1002 的向量
index.remove(1002)
# 再次搜尋,1002 不會再出現在結果中
scores, ids = index.search(query, k=10)
print("刪除後的 ID:", ids) # 不再包含 1002
持久化帶 ID 的索引
# 儲存
index.write("my_index.tvim")
# 載入
loaded_index = IdMapIndex.load("my_index.tvim")
進階功能二:過濾搜尋(Filter-at-Search)
這是 TurboVec 的核心亮點之一。在傳統向量資料庫中,如果你想限制搜尋結果只來自某個租戶或某個時間範圍,通常需要先搜尋出大量候選結果,然後在應用層進行過濾 —— 這會導致召回率下降和效能浪費。
TurboVec 在核心層級支援過濾,透過 allowlist 參數傳入允許的 ID 列表,SIMD 核心會直接在計算過程中跳過不允許的槽位。
場景:多租戶 RAG 系統
假設你有一個多租戶的 RAG 系統,每個租戶只能存取自己的文件:
import numpy as np
from turbovec import IdMapIndex
# 建立索引
idx = IdMapIndex(dim=1536, bit_width=4)
# 假設有 10000 個向量,每個向量對應一個文件 ID
vectors = np.random.rand(10000, 1536).astype(np.float32)
doc_ids = np.arange(1, 10001, dtype=np.uint64)
idx.add_with_ids(vectors, doc_ids)
# 模擬資料庫查詢:取得租戶 A 的文件 ID 列表
# 實際場景中,這會是從 PostgreSQL / MySQL 查詢的結果
tenant_a_docs = np.array([1, 5, 10, 15, 20, 25, 30, 35, 40, 45], dtype=np.uint64)
# 在租戶 A 的文件範圍內進行搜尋
query = np.random.rand(1536).astype(np.float32)
scores, ids = idx.search(query, k=5, allowlist=tenant_a_docs)
print("租戶 A 的最相關文件:", ids)
# 輸出只會包含 tenant_a_docs 中的 ID
效能優勢
過濾發生在 SIMD 核心內部,採用 32 向量塊粒度的短路機制:
- 如果某個塊中沒有任何允許的槽位,直接跳過整個塊的 LUT 查找和評分計算
- 如果塊中有部分允許的槽位,只對允許的槽位進行評分
- 對於選擇性很強的過濾條件(允許的 ID 佔總數比例很小),可以避免大部分 SIMD 計算開銷
輸出結果長度為 min(k, len(allowlist)),當允許列表小於 k 時,返回恰好 len(allowlist) 個結果,而不是用不相關的結果填充。
與主流 RAG 框架整合
TurboVec 提供了對 LangChain、LlamaIndex、Haystack 和 Agno 的無縫整合,只需替換匯入語句即可。
LangChain 整合
from langchain_community.vectorstores import TurboVec
from langchain_openai import OpenAIEmbeddings
# 初始化 embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 建立 TurboVec 向量儲存
vector_store = TurboVec.from_documents(
documents=documents, # 你的 Document 列表
embedding=embeddings,
bit_width=4 # 4-bit 量化
)
# 相似性搜尋
results = vector_store.similarity_search("你的問題", k=5)
# 持久化
vector_store.save_local("turbovec_index")
# 載入
loaded_store = TurboVec.load_local("turbovec_index", embeddings)
LlamaIndex 整合
from llama_index.vector_stores.turbovec import TurboVecVectorStore
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
# 載入文件
documents = SimpleDirectoryReader("./data").load_data()
# 建立 TurboVec 向量儲存
vector_store = TurboVecVectorStore(dim=1536, bit_width=4)
# 建立索引
index = VectorStoreIndex.from_documents(
documents,
vector_store=vector_store
)
# 查詢引擎
query_engine = index.as_query_engine()
response = query_engine.query("你的問題")
print(response)
Haystack 整合
from haystack_integrations.document_stores.turbovec import TurboVecDocumentStore
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack import Pipeline
# 建立文件儲存
document_store = TurboVecDocumentStore(dim=768, bit_width=4)
# 嵌入器
embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
# 建構管道
pipeline = Pipeline()
pipeline.add_component("embedder", embedder)
# ... 繼續添加其他元件
Rust 原生用法
如果你正在用 Rust 建構高效能後端服務,可以直接使用 TurboVec 的原生 Rust API。
基本用法
use turbovec::TurboQuantIndex;
use ndarray::Array2;
fn main() {
// 建立索引:1536 維,4-bit 量化
let mut index = TurboQuantIndex::new(1536, 4);
// 準備向量資料(這裡用亂數示例)
let vectors = Array2::<f32>::random((10000, 1536), &mut rand::thread_rng());
// 添加向量
index.add(&vectors);
// 準備查詢向量
let query = Array1::<f32>::random(1536, &mut rand::thread_rng());
// 搜尋最相似的 10 個向量
let results = index.search(&query, 10);
println!("Top 10 結果: {:?}", results);
// 持久化
index.write("index.tv").unwrap();
// 載入
let loaded = TurboQuantIndex::load("index.tv").unwrap();
}
帶外部 ID 的索引
use turbovec::IdMapIndex;
fn main() {
let mut index = IdMapIndex::new(1536, 4);
let vectors = Array2::<f32>::random((100, 1536), &mut rand::thread_rng());
let ids = vec![1001u64, 1002, 1003, /* ... */];
index.add_with_ids(&vectors, &ids);
let query = Array1::<f32>::random(1536, &mut rand::thread_rng());
let (scores, returned_ids) = index.search(&query, 10);
println!("返回的外部 ID: {:?}", returned_ids);
// 刪除
index.remove(1002);
// 持久化
index.write("index.tvim").unwrap();
let loaded = IdMapIndex::load("index.tvim").unwrap();
}
效能基準測試
根據官方提供的基準測試資料,TurboVec 在不同資料集和位寬下的表現如下:
召回率對比(TurboQuant vs FAISS IndexPQ)
測試條件:100K 向量,k=64
| 資料集 | 位寬 | TurboVec R@1 | FAISS R@1 | 優勢 |
|---|---|---|---|---|
| GloVe d=200 | 4-bit | +0.3 pts | 基準 | TurboVec 更高 |
| OpenAI d=1536 | 2-bit | +0.4 pts | 基準 | TurboVec 更高 |
| OpenAI d=1536 | 4-bit | +1.2 pts | 基準 | TurboVec 更高 |
| OpenAI d=3072 | 4-bit | +3.4 pts | 基準 | TurboVec 顯著更高 |
在所有測試中,兩者在 k=4 時都收斂到召回率 1.0。
速度對比
- ARM (NEON):TurboVec 比 FAISS IndexPQFastScan 快 12-20%
- x86 (AVX-512BW):TurboVec 與 FAISS 持平或略優
記憶體佔用
| 向量數量 | 維度 | float32 記憶體 | TurboVec (4-bit) 記憶體 | 節省 |
|---|---|---|---|---|
| 1000 萬 | 1536 | ~31 GB | ~4 GB | 87% |
| 100 萬 | 1536 | ~3.1 GB | ~400 MB | 87% |
| 10 萬 | 1536 | ~310 MB | ~40 MB | 87% |
實戰案例:建構本機 RAG 系統
下面是一個完整的示例,展示如何使用 TurboVec 建構一個完全本機的 RAG 系統,無需任何雲端服務。
環境準備
pip install turbovec sentence-transformers langchain-community langchain-openai
完整程式碼
import numpy as np
from turbovec import IdMapIndex
from sentence_transformers import SentenceTransformer
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 1. 載入嵌入模型(本機執行,無需 API Key)
print("載入嵌入模型...")
model = SentenceTransformer('all-MiniLM-L6-v2') # 384 維
dim = 384
# 2. 載入和分割文件
print("載入文件...")
loader = TextLoader('./data/my_documents.txt', encoding='utf-8')
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = splitter.split_documents(documents)
print(f"分割成 {len(chunks)} 個文字塊")
# 3. 生成向量並建構索引
print("生成向量並建構索引...")
index = IdMapIndex(dim=dim, bit_width=4)
texts = [chunk.page_content for chunk in chunks]
metadata_list = [chunk.metadata for chunk in chunks]
# 批次生成向量
embeddings = model.encode(texts, show_progress_bar=True)
# 添加向量和詮釋資料(用索引作為外部 ID)
external_ids = np.arange(len(texts), dtype=np.uint64)
index.add_with_ids(embeddings.astype(np.float32), external_ids)
# 儲存索引
index.write("rag_index.tvim")
print("索引已儲存")
# 4. 搜尋函式
def search(query: str, k: int = 5):
"""搜尋最相關的文字塊"""
# 載入索引
idx = IdMapIndex.load("rag_index.tvim")
# 生成查詢向量
query_embedding = model.encode([query])[0].astype(np.float32)
# 搜尋
scores, ids = idx.search(query_embedding, k=k)
# 返回結果
results = []
for score, doc_id in zip(scores, ids):
doc_id = int(doc_id)
results.append({
'text': texts[doc_id],
'score': float(score),
'metadata': metadata_list[doc_id]
})
return results
# 5. 測試搜尋
if __name__ == "__main__":
query = "什麼是 TurboVec?"
results = search(query, k=3)
print("\n=== 搜尋結果 ===")
for i, result in enumerate(results, 1):
print(f"\n[{i}] 相似度: {result['score']:.4f}")
print(f"內容: {result['text'][:200]}...")
添加過濾功能
如果你的文件有分類標籤,可以在搜尋時進行過濾:
def search_with_filter(query: str, category: str, k: int = 5):
"""帶分類過濾的搜尋"""
idx = IdMapIndex.load("rag_index.tvim")
# 找出屬於指定分類的文件 ID
allowed_ids = np.array([
i for i, meta in enumerate(metadata_list)
if meta.get('category') == category
], dtype=np.uint64)
if len(allowed_ids) == 0:
return []
# 生成查詢向量
query_embedding = model.encode([query])[0].astype(np.float32)
# 帶過濾的搜尋
scores, ids = idx.search(query_embedding, k=k, allowlist=allowed_ids)
results = []
for score, doc_id in zip(scores, ids):
doc_id = int(doc_id)
results.append({
'text': texts[doc_id],
'score': float(score),
'metadata': metadata_list[doc_id]
})
return results
# 只搜尋「技術」分類的文件
results = search_with_filter("如何安裝?", category="技術", k=3)
常見問題
Q1: TurboVec 適合什麼場景?
- 記憶體受限的環境:需要在有限 RAM 中儲存大規模向量索引
- 隱私敏感場景:資料不能離開本機或 VPC
- 動態增長的語料庫:需要頻繁添加新向量,無法承受重建索引的開銷
- 多租戶 RAG:需要在搜尋時進行細粒度的權限過濾
Q2: TurboVec 不適合什麼場景?
- 超大規模分散式搜尋:如果需要跨多台機器分散索引,可能需要考慮 Milvus、Weaviate 等分散式向量資料庫
- 需要複雜詮釋資料過濾:TurboVec 目前只支援基於 ID 的過濾,複雜的詮釋資料查詢需要在應用層處理
Q3: 如何選擇 bit_width?
- 4-bit:推薦預設值,在召回率和記憶體之間取得良好平衡
- 2-bit:極致壓縮,召回率略有下降,適合記憶體極度受限的場景
- 8-bit:最高精度,記憶體佔用約為 4-bit 的兩倍,召回率接近 float32
Q4: TurboVec 與 FAISS 可以同時使用嗎?
可以。你可以用 FAISS 做粗召回(coarse search),然後用 TurboVec 做精排(rerank),或者反過來。兩者的 API 設計思路不同,可以互補使用。
總結
TurboVec 是一個值得關注的新興向量搜尋庫,它透過 TurboQuant 演算法在記憶體佔用、搜尋速度和易用性三個方面取得了很好的平衡:
- ✅ 記憶體降低 87%:1000 萬向量從 31GB 降到 4GB
- ✅ 速度超越 FAISS:ARM 上快 12-20%,x86 上持平
- ✅ 無需訓練:即插即用,支援線上添加向量
- ✅ 核心層級過濾:多租戶 RAG 的理想選擇
- ✅ 生態整合:LangChain、LlamaIndex、Haystack 無縫對接
如果你的 RAG 專案面臨記憶體瓶頸或需要純本機部署,TurboVec 值得嘗試。