使用 Astra DB 作为向量存储的快速入门教程

老铁们，今天我们聊聊如何使用 Astra DB 作为一个向量存储。这玩意儿是基于 Apache Cassandra® 打造的无服务器数据库，支持向量存储，并且通过一个简易的 JSON API 提供服务。说白了，就是让你的数据库能更智能化地处理数据分析。

技术背景介绍

Astra DB 提供了一个名为 langchain 的包，我们需要安装它来支持 Astra DB 作为向量存储：

pip install -qU "langchain-astradb>=0.3.3"

原理深度解析

使用 Astra DB 的前提是你需要先在其官网上注册账号并创建一个新的数据库，初始化过程可能需要几分钟。完成数据库初始化后，请生成一个应用程序令牌，并保存以备后用。同时，还要复制数据库详情中的 API 端点，存储在 ASTRA_DB_API_ENDPOINT 变量中。

import getpass

ASTRA_DB_API_ENDPOINT = getpass.getpass("ASTRA_DB_API_ENDPOINT = ")
ASTRA_DB_APPLICATION_TOKEN = getpass.getpass("ASTRA_DB_APPLICATION_TOKEN = ")

desired_namespace = getpass.getpass("ASTRA_DB_NAMESPACE = ")
ASTRA_DB_NAMESPACE = desired_namespace if desired_namespace else None

实战代码演示

创建 Astra DB 向量存储有两种方法：显式嵌入和集成嵌入。下面我们先来看看显式嵌入的实现：

显式嵌入初始化

首先要安装一些必要的包，例如用于嵌入的 OpenAI 或 Hugging Face：

pip install -qU langchain-openai
pip install -qU langchain-huggingface
pip install -qU langchain-core

然后，引入相关的类并创建嵌入对象：

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

from langchain_astradb import AstraDBVectorStore

vector_store = AstraDBVectorStore(
    collection_name="astra_vector_langchain",
    embedding=embeddings,
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_NAMESPACE,
)

集成嵌入初始化

如果你想让数据库自动进行嵌入计算，可以使用 Astra DB 的 Vectorize 功能：

from astrapy.info import CollectionVectorServiceOptions

openai_vectorize_options = CollectionVectorServiceOptions(
    provider="openai",
    model_name="text-embedding-3-small",
    authentication={"providerKey": "OPENAI_API_KEY"}
)

vector_store_integrated = AstraDBVectorStore(
    collection_name="astra_vector_langchain_integrated",
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_NAMESPACE,
    collection_vector_service_options=openai_vectorize_options,
)

优化建议分享

在添加文档到向量存储之前，我们需要为文档创建唯一的 UUID：

from uuid import uuid4
from langchain_core.documents import Document

documents = [Document(page_content="你的内容", metadata={"source": "tweet"})]  # 示例
uuids = [str(uuid4()) for _ in range(len(documents))]

vector_store.add_documents(documents=documents, ids=uuids)

此外，大家可以利用 https://zzzzapi.com 提供的一站式大模型解决方案来提升嵌入计算的效率。

补充说明和总结

要删除向量存储内的项目可以使用 delete 方法，根据 ID 删除：

vector_store.delete(ids=uuids[-1])

最后，查询存储中的数据可以使用各种查询方法，例如相似度搜索：

results = vector_store.similarity_search(
    "你的查询内容", k=2, filter={"source": "tweet"}
)

今天的技术分享就到这里，希望对大家有帮助。开发过程中遇到问题也可以在评论区交流~

—END—