向量存储 - 全面指南

pip install semantica
# 或安装所有可选依赖：
pip install semantica[all]

# 安装 Semantica
!pip install -q semantica

from semantica.vector_store import VectorStore
from semantica.embeddings import TextEmbedder
import numpy as np

# 1. 初始化 Embedder（选择提供方和模型）
# 你可以选择 'sentence_transformers' 或 'fastembed'
embedder = TextEmbedder(method="sentence_transformers", model_name="all-MiniLM-L6-v2")
dimension = embedder.get_embedding_dimension()

# 2. 创建向量存储
store = VectorStore(backend="faiss", dimension=dimension)

# 3. 生成真实嵌入
texts = [f"Document {i}" for i in range(100)]
vectors = embedder.embed_batch(texts)

metadata = [
    {"text": txt, "category": "science" if i % 2 == 0 else "technology", "year": 2020 + (i % 4)}
    for i, txt in enumerate(texts)
]
# 4. 存储向量
vector_ids = store.store_vectors(vectors, metadata=metadata)

print(f"Stored {len(vector_ids)} vectors")
print(f"First 3 IDs: {vector_ids[:3]}")

from semantica.vector_store import VectorIndexer, FAISSStore 
import numpy as np 
# 我们使用数据集中的第一个向量作为示例查询
if 'query_vector' not in locals():
    if 'vectors' in locals() and len(vectors) > 0:
        query_vector = vectors[0]
    else:
        # 如果向量也缺失，则回退（安全检查）
        query_vector = np.random.rand(dimension).astype('float32')

# 创建索引器
indexer = VectorIndexer(backend="faiss", dimension=dimension) 

# 创建 HNSW 索引以进行快速近似检索
adapter = FAISSStore(dimension=dimension) 
index = adapter.create_index(index_type="hnsw", metric="L2", m=16) 

# 将向量添加到索引
vectors_array = np.array(vectors).astype('float32') 
# 修复：移除了 'index' 参数。适配器使用其内部的 self.index
adapter.add_vectors(vectors_array, ids=vector_ids) 

# 使用索引进行检索
query_array = np.array(query_vector).astype('float32') 
# 修复：直接在 'index' 对象上调用 search，而不是在适配器上
distances, indices = index.search(query_array.reshape(1, -1), k=10) 

print(f"Index search found {len(indices[0])} results") 
print(f"Distances: {distances[0][:5]}")

from semantica.vector_store import VectorIndexer, FAISSStore

# 从向量获取维度以确保一致性
dimension = len(vectors[0]) if len(vectors) > 0 else 384

# 创建索引器
indexer = VectorIndexer(backend="faiss", dimension=dimension)

# 创建 HNSW 索引以进行快速近似检索
adapter = FAISSStore(dimension=dimension)
index = adapter.create_index(index_type="hnsw", metric="L2", m=16)

# 将向量添加到索引
vectors_array = np.array(vectors).astype('float32')
adapter.add_vectors(vectors_array, ids=vector_ids)

# 使用索引进行检索
query_array = query_vector.astype('float32')
distances, indices = index.search(query_array.reshape(1, -1), k=10)

print(f"Index search found {len(indices[0])} results")
print(f"Distances: {distances[0][:5]}")

from semantica.vector_store import HybridSearch, MetadataFilter 
import numpy as np

# 创建混合检索
hybrid_search = HybridSearch() 

# 创建元数据过滤器
filter = MetadataFilter() \
    .eq("category", "science") \
    .gt("year", 2021) 

# 执行混合检索
# 确保 vectors 和 metadata 可用
if 'vectors' not in locals() or 'metadata' not in locals() or 'vector_ids' not in locals():
    print("Warning: vectors, metadata, or vector_ids are missing. Please run previous cells.")
else:
    hybrid_results = hybrid_search.search( 
        query_vector, 
        vectors, 
        metadata, 
        vector_ids, 
        filter=filter, 
        k=10 
    ) 

    print(f"Hybrid search found {len(hybrid_results)} results") 
    print("\nFiltered results (science, year > 2021):") 
    for i, result in enumerate(hybrid_results[:5], 1): 
        meta = result.get('metadata', {}) 
        print(f"{i}. Category: {meta.get('category')}, Year: {meta.get('year')}, Score: {result['score']:.3f}")

from semantica.vector_store import MetadataStore, MetadataSchema

# 创建元数据存储
meta_store = MetadataStore()

# 存储元数据
for i, vec_id in enumerate(vector_ids[:10]):
    meta_store.store_metadata(vec_id, metadata[i])

# 查询元数据
matching_ids = meta_store.query_metadata(
    {"category": "science"},
    operator="AND"
)

print(f"Found {len(matching_ids)} vectors with category='science'")

# 定义用于验证的模式
schema = MetadataSchema({
    "text": {"type": str, "required": True},
    "category": {"type": str, "required": True},
    "year": {"type": int, "required": True}
})

# 验证元数据
is_valid = schema.validate(metadata[0])
print(f"\nMetadata validation: {is_valid}")

from semantica.vector_store import SearchRanker

# 使用 RRF 策略创建排序器
ranker = SearchRanker(strategy="reciprocal_rank_fusion")

# 模拟多次检索结果
results1 = [
    {"id": "vec_1", "score": 0.9},
    {"id": "vec_2", "score": 0.8},
    {"id": "vec_3", "score": 0.7}
]

results2 = [
    {"id": "vec_2", "score": 0.85},
    {"id": "vec_4", "score": 0.75},
    {"id": "vec_1", "score": 0.7}
]

# 使用 RRF 融合结果
fused_results = ranker.rank([results1, results2], k=60)

print("Fused results using RRF:")
for i, result in enumerate(fused_results, 1):
    print(f"{i}. ID: {result['id']}, Fused Score: {result['score']:.3f}")

from semantica.vector_store import NamespaceManager

# 创建命名空间管理器
ns_manager = NamespaceManager()

# 为不同租户创建命名空间
ns1 = ns_manager.create_namespace("tenant1", "Tenant 1 vectors")
ns2 = ns_manager.create_namespace("tenant2", "Tenant 2 vectors")

# 将向量添加到命名空间
for i in range(5):
    ns_manager.add_vector_to_namespace(f"t1_vec_{i}", "tenant1")
    ns_manager.add_vector_to_namespace(f"t2_vec_{i}", "tenant2")

# 获取命名空间向量
tenant1_vectors = ns_manager.get_namespace_vectors("tenant1")
tenant2_vectors = ns_manager.get_namespace_vectors("tenant2")

print(f"Tenant 1: {len(tenant1_vectors)} vectors")
print(f"Tenant 2: {len(tenant2_vectors)} vectors")

# 设置访问控制
ns1.set_access_control("user1", ["read", "write"])
ns1.set_access_control("user2", ["read"])

print(f"\nUser1 can write: {ns1.has_permission('user1', 'write')}")
print(f"User2 can write: {ns1.has_permission('user2', 'write')}")