ZeusDB 集成

ZeusDB 是一个由 Rust 驱动的高性能向量数据库，提供乘积量化、持久化存储和企业级日志记录等高级功能。

本文档展示如何使用 ZeusDB 为你的 LangChain 应用带来企业级向量搜索能力。

快速入门

安装

pip install langchain-zeusdb

入门示例

本示例使用 OpenAIEmbeddings，需要 OpenAI API 密钥——在此获取你的 OpenAI API 密钥如果你愿意，也可以将此包与任何其他嵌入提供商（Hugging Face、Cohere、自定义函数等）一起使用。

pip install langchain-openai

import os
import getpass

os.environ['OPENAI_API_KEY'] = getpass.getpass('OpenAI API Key:')

基本用法

from langchain_zeusdb import ZeusDBVectorStore
from langchain_openai import OpenAIEmbeddings
from zeusdb import VectorDatabase

# 初始化嵌入
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建 ZeusDB 索引
vdb = VectorDatabase()
index = vdb.create(
    index_type="hnsw",
    dim=1536,
    space="cosine"
)

# 创建向量存储
vector_store = ZeusDBVectorStore(
    zeusdb_index=index,
    embedding=embeddings
)

# 添加文档
from langchain_core.documents import Document

docs = [
    Document(page_content="ZeusDB is fast", metadata={"source": "docs"}),
    Document(page_content="LangChain is powerful", metadata={"source": "docs"}),
]

vector_store.add_documents(docs)

# 搜索
results = vector_store.similarity_search("fast database", k=2)
print(f"Found the following {len(results)} results:")
print(results)

预期结果：

Found the following 2 results:
[Document(id='ea2b4f13-b0b7-4cef-bb91-0fc4f4c41295', metadata={'source': 'docs'}, page_content='ZeusDB is fast'), Document(id='33dc1e87-a18a-4827-a0df-6ee47eabc7b2', metadata={'source': 'docs'}, page_content='LangChain is powerful')]

工厂方法

为方便起见，你可以在一步中创建并填充向量存储：

示例 1：从文本创建（一步创建索引并添加文本）

vector_store_texts = ZeusDBVectorStore.from_texts(
    texts=["Hello world", "Goodbye world"],
    embedding=embeddings,
    metadatas=[{"source": "text1"}, {"source": "text2"}]
)

print("texts store count:", vector_store_texts.get_vector_count())         # -> 2
print("texts store peek:", vector_store_texts.zeusdb_index.list(2))        # [('id1', {...}), ('id2', {...})]

# 在基于文本的存储中搜索
results = vector_store_texts.similarity_search("Hello", k=1)
print(f"Found in texts store: {results[0].page_content}")                  # -> "Hello world"

预期结果：

texts store count: 2
texts store peek: [('e9c39b44-b610-4e00-91f3-bf652e9989ac', {'source': 'text1', 'text': 'Hello world'}), ('d33f210c-ed53-4006-a64a-a9eee397fec9', {'source': 'text2', 'text': 'Goodbye world'})]
Found in texts store: Hello world

示例 2：从文档创建（一步创建索引并添加文档）

new_docs = [
    Document(page_content="Python is great", metadata={"source": "python"}),
    Document(page_content="JavaScript is flexible", metadata={"source": "js"}),
]

vector_store_docs = ZeusDBVectorStore.from_documents(
    documents=new_docs,
    embedding=embeddings
)

print("docs store count:", vector_store_docs.get_vector_count())           # -> 2
print("docs store peek:", vector_store_docs.zeusdb_index.list(2))          # [('id3', {...}), ('id4', {...})]

# 在基于文档的存储中搜索
results = vector_store_docs.similarity_search("Python", k=1)
print(f"Found in docs store: {results[0].page_content}")                   # -> "Python is great"

预期结果：

docs store count: 2
docs store peek: [('aab2d1c1-7e02-4817-8dd8-6fb03570bb6f', {'text': 'Python is great', 'source': 'python'}), ('9a8a82cb-0e70-456c-9db2-556e464de14e', {'text': 'JavaScript is flexible', 'source': 'js'})]
Found in docs store: Python is great

高级功能

ZeusDB 的企业级功能已完全集成到 LangChain 生态系统中，提供量化、持久化、高级搜索功能及众多其他企业能力。

节省内存的量化设置

对于大型数据集，使用乘积量化来减少内存使用：

# 创建量化索引以节省内存
quantization_config = {
    'type': 'pq',
    'subvectors': 8,
    'bits': 8,
    'training_size': 10000
}

vdb = VectorDatabase()
index = vdb.create(
    index_type="hnsw",
    dim=1536,
    space="cosine",
    quantization_config=quantization_config
)

vector_store = ZeusDBVectorStore(
    zeusdb_index=index,
    embedding=embeddings
)

请参阅我们的文档获取有关量化配置的有用指南和建议。

持久化

ZeusDB 持久化功能让你可以将完整填充的索引保存到磁盘，并在之后以完整状态恢复加载。这包括向量、元数据、HNSW 图和（如果启用）乘积量化模型。保存内容：

向量和 ID
元数据
HNSW 图结构
量化配置、中心点和训练状态（如果启用了 PQ）

保存向量存储

# 保存索引
vector_store.save_index("my_index.zdb")

加载向量存储

# 加载索引
loaded_store = ZeusDBVectorStore.load_index(
    path="my_index.zdb",
    embedding=embeddings
)

# 加载后验证
print("vector count:", loaded_store.get_vector_count())
print("index info:", loaded_store.info())
print("store peek:", loaded_store.zeusdb_index.list(2))

注意事项

路径是一个目录，而不是单个文件。确保目标目录可写。
保存的索引是跨平台的，包含格式/版本信息用于兼容性检查。
如果你使用了 PQ，压缩模型和状态都会被保留——加载后无需重新训练。
你可以继续在 loaded_store 上使用所有向量存储 API（similarity_search、retrievers 等）。

有关更多详情（包括文件结构和更全面的示例），请参阅文档。

高级搜索选项

使用这些选项来控制评分、多样性、元数据过滤和检索器集成以进行搜索。

带分数的相似性搜索

返回来自 ZeusDB 的 (Document, raw_distance) 对——距离越低 = 越相似。如果你希望在 [0, 1] 范围内获得归一化相关性，请使用 similarity_search_with_relevance_scores。

# 带分数的相似性搜索
results_with_scores = vector_store.similarity_search_with_score(
    query="machine learning",
    k=5
)

print(results_with_scores)

预期结果：

[
  (Document(id='ac0eaf5b-9f02-4ce2-8957-c369a7262c61', metadata={'source': 'docs'}, page_content='LangChain is powerful'), 0.8218843340873718),
  (Document(id='faae3adf-7cf3-463c-b282-3790b096fa23', metadata={'source': 'docs'}, page_content='ZeusDB is fast'), 0.9140053391456604)
]

MMR 搜索以增加多样性

MMR（最大边际相关性）在两个力量之间取得平衡：与查询的相关性和所选结果的多样性，减少近似重复答案。通过 lambda_mult 控制权衡（1.0 = 全相关性，0.0 = 全多样性）。

# MMR 搜索以增加多样性
mmr_results = vector_store.max_marginal_relevance_search(
    query="AI applications",
    k=5,
    fetch_k=20,
    lambda_mult=0.7  # 平衡相关性与多样性
)

print(mmr_results)

带元数据过滤的搜索

使用添加文档时存储的文档元数据过滤结果

# 带元数据过滤的搜索
results = vector_store.similarity_search(
    query="database performance",
    k=3,
    filter={"source": "documentation"}
)

有关支持的元数据查询类型和运算符，请参阅文档。

作为检索器使用

将向量存储转换为检索器提供了一个标准的 LangChain 接口，链（例如 RetrievalQA）可以调用该接口来获取上下文。在底层，它使用你选择的搜索类型（similarity 或 mmr）和 search_kwargs。

# 转换为检索器以用于链
retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "lambda_mult": 0.8}
)

# 与 LangChain 表达式语言（LCEL）配合使用——仅需 langchain-core
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI()

# 使用 LCEL 创建链
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 使用链
answer = chain.invoke("What is ZeusDB?")
print(answer)

预期结果：

ZeusDB is a fast database management system.

异步支持

ZeusDB 支持异步操作，用于非阻塞、并发的向量操作。 何时使用异步： Web 服务器（FastAPI/Starlette）、进行并行搜索的 agent/流水线，或希望非阻塞/并发检索的笔记本。如果你编写简单脚本，同步方法即可。这些是异步操作——常规同步方法的异步/await 版本。以下是每个方法的作用：

await vector_store.aadd_documents(documents) - 异步向向量存储添加文档（add_documents() 的异步版本）
await vector_store.asimilarity_search("query", k=5) - 异步执行相似性搜索（similarity_search() 的异步版本）
await vector_store.adelete(ids=["doc1", "doc2"]) - 通过 ID 异步删除文档（delete() 的异步版本）

异步版本在以下情况下很有用：

构建异步应用（使用 asyncio、FastAPI 等）
希望可以并发运行的非阻塞操作
同时处理多个请求
需要在 I/O 密集型应用中获得更好性能

例如，与其阻塞等待添加文档：

# 同步（阻塞）
vector_store.add_documents(docs)  # 阻塞直到完成

# 异步（非阻塞）
await vector_store.aadd_documents(docs)  # 运行时可以做其他工作

所有操作都支持 async/await： 脚本版本（python my_script.py）：

import asyncio
from langchain_zeusdb import ZeusDBVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from zeusdb import VectorDatabase

# 设置
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vdb = VectorDatabase()
index = vdb.create(index_type="hnsw", dim=1536, space="cosine")
vector_store = ZeusDBVectorStore(zeusdb_index=index, embedding=embeddings)

docs = [
    Document(page_content="ZeusDB is fast", metadata={"source": "docs"}),
    Document(page_content="LangChain is powerful", metadata={"source": "docs"}),
]

async def main():
    # 异步添加文档
    ids = await vector_store.aadd_documents(docs)
    print("Added IDs:", ids)

    # 并发运行多个搜索
    results_fast, results_powerful = await asyncio.gather(
        vector_store.asimilarity_search("fast", k=2),
        vector_store.asimilarity_search("powerful", k=2),
    )
    print("Fast results:", [d.page_content for d in results_fast])
    print("Powerful results:", [d.page_content for d in results_powerful])

    # 异步删除文档
    deleted = await vector_store.adelete(ids=ids[:1])
    print("Deleted first doc:", deleted)

if __name__ == "__main__":
    asyncio.run(main())

Colab/Notebook/Jupyter 版本（顶层 await）：

from langchain_zeusdb import ZeusDBVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from zeusdb import VectorDatabase
import asyncio

# 设置
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vdb = VectorDatabase()
index = vdb.create(index_type="hnsw", dim=1536, space="cosine")
vector_store = ZeusDBVectorStore(zeusdb_index=index, embedding=embeddings)

docs = [
    Document(page_content="ZeusDB is fast", metadata={"source": "docs"}),
    Document(page_content="LangChain is powerful", metadata={"source": "docs"}),
]

# 异步添加文档
ids = await vector_store.aadd_documents(docs)
print("Added IDs:", ids)

# 并发运行多个搜索
results_fast, results_powerful = await asyncio.gather(
    vector_store.asimilarity_search("fast", k=2),
    vector_store.asimilarity_search("powerful", k=2),
)
print("Fast results:", [d.page_content for d in results_fast])
print("Powerful results:", [d.page_content for d in results_powerful])

# 异步删除文档
deleted = await vector_store.adelete(ids=ids[:1])
print("Deleted first doc:", deleted)

预期结果：

Added IDs: ['9c440918-715f-49ba-9b97-0d991d29e997', 'ad59c645-d3ba-4a4a-a016-49ed39514123']
Fast results: ['ZeusDB is fast', 'LangChain is powerful']
Powerful results: ['LangChain is powerful', 'ZeusDB is fast']
Deleted first doc: True

监控与可观测性

性能监控

# 获取索引统计信息
stats = vector_store.get_zeusdb_stats()
print(f"Index size: {stats.get('total_vectors', '0')} vectors")
print(f"Dimension: {stats.get('dimension')} | Space: {stats.get('space')} | Index type: {stats.get('index_type')}")

# 对搜索性能进行基准测试
performance = vector_store.benchmark_search_performance(
    query_count=100,
    max_threads=4
)
print(f"Search QPS: {performance.get('parallel_qps', 0):.0f}")

# 检查量化状态
if vector_store.is_quantized():
    progress = vector_store.get_training_progress()
    print(f"Quantization training: {progress:.1f}% complete")
else:
    print("Index is not quantized")

预期结果：

Index size: 2 vectors
Dimension: 1536 | Space: cosine | Index type: HNSW
Search QPS: 53807
Index is not quantized

企业级日志记录

ZeusDB 内置企业级结构化日志记录，通过智能环境检测自动工作：

import logging

# ZeusDB 自动检测你的环境并应用适当的日志记录：
# - 开发环境：人类可读的日志，WARNING 级别
# - 生产环境：JSON 结构化日志，ERROR 级别
# - 测试环境：最小输出，CRITICAL 级别
# - Jupyter：清晰可读的日志，INFO 级别

# 操作会自动记录性能指标
vector_store.add_documents(docs)
# 日志：{"operation":"vector_addition","total_inserted":2,"duration_ms":45}

# 如需控制日志记录，可使用环境变量
# ZEUSDB_LOG_LEVEL=debug ZEUSDB_LOG_FORMAT=json python your_app.py

要了解 ZeusDB 企业级日志记录功能的完整特性，请阅读以下文档。

配置选项

索引参数

vdb = VectorDatabase()
index = vdb.create(
    index_type="hnsw",           # 索引算法
    dim=1536,                    # 向量维度
    space="cosine",              # 距离度量：cosine、l2、l1
    m=16,                        # HNSW 连通性
    ef_construction=200,         # 构建时搜索宽度
    expected_size=100000,        # 预期向量数量
    quantization_config=None     # 可选量化
)

搜索参数

results = vector_store.similarity_search(
    query="search query",
    k=5,                         # 结果数量
    ef_search=None,              # 运行时搜索宽度（None 为自动）
    filter={"key": "value"}      # 元数据过滤器
)

错误处理

该集成包含全面的错误处理：

try:
    results = vector_store.similarity_search("query")
    print(results)
except Exception as e:
    # 优雅降级并记录日志
    print(f"Search failed: {e}")
    # 备用逻辑在此

系统要求

Python：3.10 或更高版本
ZeusDB：0.0.8 或更高版本
LangChain Core：0.3.74 或更高版本

从源码安装

git clone https://github.com/zeusdb/langchain-zeusdb.git
cd langchain-zeusdb/libs/zeusdb
pip install -e .

使用场景

RAG 应用：用于问答的高性能检索
语义搜索：跨大型文档集合的快速相似性搜索
推荐系统：基于向量的内容和协同过滤
嵌入分析：高维嵌入空间的分析
实时应用：生产系统的低延迟向量搜索

兼容性

LangChain 版本

LangChain Core：0.3.74+

距离度量

余弦（Cosine）：默认，归一化相似性
欧氏距离（L2）：几何距离
曼哈顿距离（L1）：城市街区距离

嵌入模型

与任何嵌入提供商兼容：

OpenAI（text-embedding-3-small、text-embedding-3-large）
Hugging Face Transformers
Cohere Embeddings
自定义嵌入函数

支持

文档：docs.zeusdb.com
问题：GitHub Issues
邮件：contact@zeusdb.com

让向量搜索快速、可扩展且对开发者友好。

在 GitHub 上编辑此页面或提交问题。

通过 MCP 将这些文档连接到 Claude、VSCode 等以获取实时答案。

Popular Providers

Integrations by component

快速入门

安装

入门示例

基本用法

工厂方法

示例 1：从文本创建（一步创建索引并添加文本）

示例 2：从文档创建（一步创建索引并添加文档）

高级功能

节省内存的量化设置

持久化

保存向量存储

加载向量存储

注意事项

高级搜索选项

带分数的相似性搜索

MMR 搜索以增加多样性

带元数据过滤的搜索

作为检索器使用

异步支持

监控与可观测性

性能监控

企业级日志记录

配置选项

索引参数

搜索参数

错误处理

系统要求

从源码安装

使用场景

兼容性

LangChain 版本

距离度量

嵌入模型

支持

Popular Providers

Integrations by component

​快速入门

​安装

​入门示例

​基本用法

​工厂方法

​示例 1：从文本创建（一步创建索引并添加文本）

​示例 2：从文档创建（一步创建索引并添加文档）

​高级功能

​节省内存的量化设置

​持久化

​保存向量存储

​加载向量存储

​注意事项

​高级搜索选项

​带分数的相似性搜索

​MMR 搜索以增加多样性

​带元数据过滤的搜索

​作为检索器使用

​异步支持

​监控与可观测性

​性能监控

​企业级日志记录

​配置选项

​索引参数

​搜索参数

​错误处理

​系统要求

​从源码安装

​使用场景

​兼容性

​LangChain 版本

​距离度量

​嵌入模型

​支持

快速入门

安装

入门示例

基本用法

工厂方法

示例 1：从文本创建（一步创建索引并添加文本）

示例 2：从文档创建（一步创建索引并添加文档）

高级功能

节省内存的量化设置

持久化

保存向量存储

加载向量存储

注意事项

高级搜索选项

带分数的相似性搜索

MMR 搜索以增加多样性

带元数据过滤的搜索

作为检索器使用

异步支持

监控与可观测性

性能监控

企业级日志记录

配置选项

索引参数

搜索参数

错误处理

系统要求

从源码安装

使用场景

兼容性

LangChain 版本

距离度量

嵌入模型

支持