高级向量存储——轻松上手

1你将学到什么

本 notebook 向你展示在真实应用中使用向量存储的实用方法。每个示例都很简单且开箱即用。

主题

选择合适的索引——用哪个以及何时用
智能过滤——精确找到你需要的内容
合并结果——合并来自不同来源的搜索
组织数据——将不同用户的数据分开

# 安装 Semantica 包
!pip install semantica

2第 0 部分：设置嵌入

首先，让我们选择嵌入提供方和模型。Semantica 支持多个提供方，例如 Sentence Transformers 和 FastEmbed。

from semantica.embeddings import TextEmbedder

# 选择提供方和模型
embedder = TextEmbedder(method="fastembed", model_name="BAAI/bge-small-en-v1.5")
dimension = embedder.get_embedding_dimension()

print(f"Selected model: {embedder.get_model_info()['model_name']}")
print(f"Embedding dimension: {dimension}")

3第 1 部分：选择合适的索引

可以把索引想象成选择一种归档系统： - Flat：像一个小笔记本——慢但完美 - HNSW：像一个组织良好的图书馆——快速且准确 - IVF：像一个有分区的仓库——对海量集合非常快

简单规则

少于 10,000 项？使用 Flat
在 10,000 到 100 万之间？使用 HNSW ✅（推荐）
超过 100 万？使用 IVF

from semantica.vector_store import FAISSStore
import numpy as np

# 创建一些示例向量（类似文档嵌入）
vectors = np.random.rand(5000, 768).astype('float32')
query = np.random.rand(768).astype('float32')

adapter = FAISSStore(dimension=768)

# HNSW 索引——最适合大多数情况
index = adapter.create_index(index_type="hnsw", metric="L2", m=16)
adapter.add_vectors(vectors, ids=[f"doc_{i}" for i in range(len(vectors))])

# 搜索相似向量
results = adapter.search_similar(query, k=5)

print("Found 5 most similar documents:")
for i, result in enumerate(results, 1):
    print(f"  {i}. Document {result['id']} (distance: {result['distance']:.3f})")

4第 2 部分：使用元数据进行智能过滤

想象一下搜索"相似文章"，但只限于 2024 年且只限于"Technology"类别。这就是元数据过滤的作用！

真实世界示例

你正在构建一个文档搜索，用户想要： - 相似文档（向量搜索） - 来自特定类别（元数据过滤） - 来自最近几年（元数据过滤）

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

# 创建带元数据的示例文档
documents = [
    {"id": 0, "text": "AI in Healthcare", "category": "Technology", "year": 2024},
    {"id": 1, "text": "Machine Learning Basics", "category": "Technology", "year": 2023},
    {"id": 2, "text": "Business Strategy", "category": "Business", "year": 2024},
    {"id": 3, "text": "Data Science Guide", "category": "Technology", "year": 2024},
    {"id": 4, "text": "Marketing Tips", "category": "Business", "year": 2023},
]

# 为每个文档创建向量
vectors = [np.random.rand(768) for _ in documents]
metadata = [{"category": d["category"], "year": d["year"]} for d in documents]
vector_ids = [f"doc_{d['id']}" for d in documents]

# 创建搜索
search = HybridSearch()
query = np.random.rand(768)

# 示例 1：查找 2024 年的 Technology 文章
filter1 = MetadataFilter().eq("category", "Technology").eq("year", 2024)
results = search.search(query, vectors, metadata, vector_ids, filter=filter1, k=10)

print("Technology articles from 2024:")
for r in results:
    doc_id = int(r['id'].split('_')[1])
    print(f"  - {documents[doc_id]['text']}")

# 示例 2：查找 2024 年的任意文章
filter2 = MetadataFilter().eq("year", 2024)
results2 = search.search(query, vectors, metadata, vector_ids, filter=filter2, k=10)

print("\nAll articles from 2024:")
for r in results2:
    doc_id = int(r['id'].split('_')[1])
    print(f"  - {documents[doc_id]['text']} ({documents[doc_id]['category']})")

5第 3 部分：合并搜索结果

有时你想在多个地方搜索并合并结果。就像同时搜索你的电子邮件和文档，然后展示两者的最佳匹配。

何时使用

搜索多个数据库
合并不同的搜索策略
给某些来源更高的权重

from semantica.vector_store import SearchRanker

# 模拟两个不同的搜索
# 搜索 1：最近文档
recent_results = [
    {"id": "doc_3", "score": 0.95, "source": "recent"},
    {"id": "doc_0", "score": 0.90, "source": "recent"},
    {"id": "doc_2", "score": 0.85, "source": "recent"},
]

# 搜索 2：热门文档
popular_results = [
    {"id": "doc_1", "score": 0.92, "source": "popular"},
    {"id": "doc_3", "score": 0.88, "source": "popular"},
    {"id": "doc_4", "score": 0.80, "source": "popular"},
]

# 方法 1：公平组合（RRF）
ranker = SearchRanker(strategy="reciprocal_rank_fusion")
combined = ranker.rank([recent_results, popular_results])

print("Combined results (fair ranking):")
for i, result in enumerate(combined[:3], 1):
    doc_id = int(result['id'].split('_')[1])
    print(f"  {i}. {documents[doc_id]['text']} (score: {result['score']:.3f})")

# 方法 2：偏好最近文档（70% 最近，30% 热门）
weighted_ranker = SearchRanker(strategy="weighted_average")
weighted_combined = weighted_ranker.rank(
    [recent_results, popular_results],
    weights=[0.7, 0.3]
)

print("\nCombined results (prefer recent):")
for i, result in enumerate(weighted_combined[:3], 1):
    doc_id = int(result['id'].split('_')[1])
    print(f"  {i}. {documents[doc_id]['text']} (score: {result['score']:.3f})")

6第 4 部分：将用户数据分开

如果你正在构建一个拥有多个用户或公司的应用，你需要将他们的数据分开。命名空间会自动做到这一点。

真实示例

你正在构建一个 SaaS 应用，其中： - 公司 A 有自己的文档 - 公司 B 有自己的文档 - 它们永远不应看到彼此的数据

from semantica.vector_store import NamespaceManager

# 创建管理器
manager = NamespaceManager()

# 为每个公司创建独立空间
company_a = manager.create_namespace("company_a", "Company A's documents")
company_b = manager.create_namespace("company_b", "Company B's documents")

# 向公司 A 添加文档
for i in range(10):
    manager.add_vector_to_namespace(f"company_a_doc_{i}", "company_a")

# 向公司 B 添加文档
for i in range(15):
    manager.add_vector_to_namespace(f"company_b_doc_{i}", "company_b")

# 获取每个公司的文档
a_docs = manager.get_namespace_vectors("company_a")
b_docs = manager.get_namespace_vectors("company_b")

print(f"Company A has {len(a_docs)} documents")
print(f"Company B has {len(b_docs)} documents")

# 设置权限（谁能访问什么）
company_a.set_access_control("admin@companya.com", ["read", "write", "delete"])
company_a.set_access_control("user@companya.com", ["read"])  # 只读

# 检查权限
print(f"\nAdmin can delete: {company_a.has_permission('admin@companya.com', 'delete')}")
print(f"User can delete: {company_a.has_permission('user@companya.com', 'delete')}")

7快速参考指南

我应该使用哪个索引？

# 小数据集（< 10,000 项）
index = adapter.create_index(index_type="flat", metric="L2")

# 中等数据集（10,000 - 1,000,000 项）✅ 推荐
index = adapter.create_index(index_type="hnsw", metric="L2", m=16)

# 大数据集（> 1,000,000 项）
index = adapter.create_index(index_type="ivf", metric="L2", nlist=100)

如何过滤结果？

# 单个条件
filter = MetadataFilter().eq("category", "Technology")

# 多个条件（AND）
filter = MetadataFilter() \
    .eq("category", "Technology") \
    .eq("year", 2024)

# 大于 / 小于
filter = MetadataFilter().gt("year", 2020)

如何合并结果？

# 公平组合
ranker = SearchRanker(strategy="reciprocal_rank_fusion")
combined = ranker.rank([results1, results2])

# 加权组合（偏好第一个来源）
ranker = SearchRanker(strategy="weighted_average")
combined = ranker.rank([results1, results2], weights=[0.7, 0.3])

如何分离用户数据？

# 为每个用户/公司创建命名空间
manager = NamespaceManager()
user_space = manager.create_namespace("user_123", "User 123's data")

# 向命名空间添加数据
manager.add_vector_to_namespace("doc_1", "user_123")

# 获取用户数据
user_docs = manager.get_namespace_vectors("user_123")

8小结

你已经学到了：

✅ 索引选择：大多数情况使用 HNSW
✅ 智能过滤：将向量搜索与元数据结合
✅ 结果融合：合并来自不同来源的搜索
✅ 数据隔离：将用户数据分开

下一步

用你自己的数据尝试这些示例
尝试不同的过滤器
构建一个多用户应用
探索入门 notebook以了解更多基础知识

需要帮助？ 查看我们的文档或在 GitHub 上提问。