TurboVec 完全指南：比 FAISS 更快的开源向量搜索库

pip install turbovec

# LangChain 集成
pip install turbovec[langchain]

# LlamaIndex 集成
pip install turbovec[llama-index]

# Haystack 集成
pip install turbovec[haystack]

# Agno 集成
pip install turbovec[agno]

[dependencies]
turbovec = "0.1"

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)

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, ...]

# 删除 ID 为 1002 的向量
index.remove(1002)

# 再次搜索，1002 不会再出现在结果中
scores, ids = index.search(query, k=10)
print("删除后的 ID:", ids)  # 不再包含 1002

# 保存
index.write("my_index.tvim")

# 加载
loaded_index = IdMapIndex.load("my_index.tvim")

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

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)

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)

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)
# ... 继续添加其他组件

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();
}

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();
}

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)