跳到主要内容
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

# Initialize embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

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

# Create vector store
vector_store = ZeusDBVectorStore(
    zeusdb_index=index,
    embedding=embeddings
)

# Add documents
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)

# Search
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', {...})]

# Search the texts-based store
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', {...})]

# Search the documents-based store
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 生态系统中，提供量化、持久化、高级搜索功能和许多其他企业功能。

​
采用量化的内存高效设置

对于大型数据集，请使用产品量化来减少内存使用 
# Create quantized index for memory efficiency
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）

​
如何保存您的向量存储

# Save index
vector_store.save_index("my_index.zdb")

​
如何加载您的向量存储

# Load index
loaded_store = ZeusDBVectorStore.load_index(
    path="my_index.zdb",
    embedding=embeddings
)

# Verify after load
print("vector count:", loaded_store.get_vector_count())
print("index info:", loaded_store.info())
print("store peek:", loaded_store.zeusdb_index.list(2))

​
备注

  • 路径是一个目录，而不是单个文件。请确保目标可写。
  • 保存的索引是跨平台的，并包含格式/版本信息以进行兼容性检查。
  • 如果您使用了 PQ，则压缩模型和状态都将被保留——加载后无需重新训练。
  • 您可以继续在加载的存储上使用所有向量存储 API（similarity_search、retrievers 等）。
有关更多详细信息（包括文件结构和更全面的示例），请参阅文档。

​
高级搜索选项

使用这些来控制您的搜索的评分、多样性、元数据过滤和检索器集成。

​
带分数的相似性搜索

返回 ZeusDB 的 (Document, raw_distance) 对——距离越小 = 越相似。如果您喜欢 [0, 1] 范围内的归一化相关性，请使用 similarity_search_with_relevance_scores。 
# Similarity search with 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 search for diversity
mmr_results = vector_store.max_marginal_relevance_search(
    query="AI applications",
    k=5,
    fetch_k=20,
    lambda_mult=0.7  # Balance relevance vs diversity
)

print(mmr_results)

​
带元数据过滤的搜索

使用您在添加文档时存储的文档元数据过滤结果 
# Search with metadata filtering
results = vector_store.similarity_search(
    query="database performance",
    k=3,
    filter={"source": "documentation"}
)
有关支持的元数据查询类型和运算符，请参阅文档。

​
作为检索器

将向量存储转换为检索器为您提供了一个标准的 LangChain 接口，链（例如 RetrievalQA）可以调用该接口来获取上下文。在底层，它使用您选择的搜索类型（相似性或 mmr）和 search_kwargs。 
# Convert to retriever for use in chains
retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "lambda_mult": 0.8}
)

# Use with LangChain Expression Language (LCEL) - requires only 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()

# Create a chain using LCEL
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Use the chain
answer = chain.invoke("What is ZeusDB?")
print(answer)
预期结果 
ZeusDB is a fast database management system.

​
异步支持

ZeusDB 支持异步操作，实现非阻塞的并发向量操作。 何时使用异步： Web 服务器 (FastAPI/Starlette)、进行并行搜索的代理/管道，或需要非阻塞/并发检索的笔记本。如果您正在编写简单的脚本，同步方法即可。 这些是异步操作——常规同步方法的 async/await 版本。它们各自的作用如下：
  1. await vector_store.aadd_documents(documents) - 异步将文档添加到向量存储（add_documents() 的异步版本）
  2. await vector_store.asimilarity_search("query", k=5) - 异步执行相似性搜索（similarity_search() 的异步版本）
  3. await vector_store.adelete(ids=["doc1", "doc2"]) - 异步通过 ID 删除文档（delete() 的异步版本）
异步版本在以下情况中很有用：
  • 您正在构建异步应用程序（使用 asyncio、FastAPI 等）
  • 您希望进行可以并发运行的非阻塞操作
  • 您正在同时处理多个请求
  • 您希望在 I/O 密集型应用程序中获得更好的性能
例如，而不是在添加文档时阻塞 
# Synchronous (blocking)
vector_store.add_documents(docs)  # Blocks until complete

# Asynchronous (non-blocking)
await vector_store.aadd_documents(docs)  # Can do other work while this runs
所有操作都支持 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

# Setup
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():
    # Add documents asynchronously
    ids = await vector_store.aadd_documents(docs)
    print("Added IDs:", ids)

    # Run multiple searches concurrently
    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])

    # Delete documents asynchronously
    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

# Setup
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"}),
]

# Add documents asynchronously
ids = await vector_store.aadd_documents(docs)
print("Added IDs:", ids)

# Run multiple searches concurrently
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])

# Delete documents asynchronously
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

​
监控和可观察性

​
性能监控

# Get index statistics
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')}")

# Benchmark search performance
performance = vector_store.benchmark_search_performance(
    query_count=100,
    max_threads=4
)
print(f"Search QPS: {performance.get('parallel_qps', 0):.0f}")

# Check quantization status
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 automatically detects your environment and applies appropriate logging:
# - Development: Human-readable logs, WARNING level
# - Production: JSON structured logs, ERROR level
# - Testing: Minimal output, CRITICAL level
# - Jupyter: Clean readable logs, INFO level

# Operations are automatically logged with performance metrics
vector_store.add_documents(docs)
# Logs: {"operation":"vector_addition","total_inserted":2,"duration_ms":45}

# Control logging with environment variables if needed
# ZEUSDB_LOG_LEVEL=debug ZEUSDB_LOG_FORMAT=json python your_app.py
要了解 ZeusDB 企业级日志记录功能的全部特性，请阅读以下文档。

​
配置选项

​
索引参数

vdb = VectorDatabase()
index = vdb.create(
    index_type="hnsw",           # Index algorithm
    dim=1536,                    # Vector dimension
    space="cosine",              # Distance metric: cosine, l2, l1
    m=16,                        # HNSW connectivity
    ef_construction=200,         # Build-time search width
    expected_size=100000,        # Expected number of vectors
    quantization_config=None     # Optional quantization
)

​
搜索参数

results = vector_store.similarity_search(
    query="search query",
    k=5,                         # Number of results
    ef_search=None,              # Runtime search width (auto if None)
    filter={"key": "value"}      # Metadata filter
)

​
错误处理

该集成包括全面的错误处理 
try:
    results = vector_store.similarity_search("query")
    print(results)
except Exception as e:
    # Graceful degradation with logging
    print(f"Search failed: {e}")
    # Fallback logic here

​
要求

  • 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+

​
距离度量

  • 余弦：默认，归一化相似度
  • 欧几里得 (L2)：几何距离
  • 曼哈顿 (L1)：城市区块距离

​
嵌入模型

兼容任何嵌入提供商
  • OpenAI (text-embedding-3-small, text-embedding-3-large)
  • Hugging Face Transformers
  • Cohere Embeddings
  • 自定义嵌入函数

​
支持

使向量搜索更快、可扩展、开发者友好。
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.