Astra DB 矢量存储

本页提供了将 Astra DB 用作矢量存储的快速入门。

DataStax Astra DB 是一个无服务器基于 AI 就绪型数据库Apache Cassandra®并方便使用通过易于使用的 JSON API。

设置

依赖

使用集成需要langchain-astradb合作伙伴套餐：

!pip install \
    "langchain>=0.3.23,<0.4" \
    "langchain-core>=0.3.52,<0.4" \
    "langchain-astradb>=0.6,<0.7"

凭据

为了使用 AstraDB 向量存储，您必须先前往 AstraDB 网站，创建一个帐户，然后创建一个新数据库 - 初始化可能需要几分钟。

初始化数据库后，检索您暂时需要的连接密钥。这些是：

一API Endpoint如"https://01234567-89ab-cdef-0123-456789abcdef-us-east1.apps.astra.datastax.com/"
以及Database Token，例如"AstraCS:aBcD123......"

您可以选择提供keyspace（在 LangChain 组件中称为 “namespace”），您可以从Data Explorer选项卡。如果你愿意，你可以在下面的提示中将其留空，并回退到默认的键空间。

import getpass

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

desired_keyspace = input("(optional) ASTRA_DB_KEYSPACE = ").strip()
if desired_keyspace:
    ASTRA_DB_KEYSPACE = desired_keyspace
else:
    ASTRA_DB_KEYSPACE = None

ASTRA_DB_API_ENDPOINT =  https://01234567-89ab-cdef-0123-456789abcdef-us-east1.apps.astra.datastax.com
ASTRA_DB_APPLICATION_TOKEN =  ········
(optional) ASTRA_DB_KEYSPACE =

如果您想获得一流的模型调用自动跟踪，您还可以通过取消下面的注释来设置 LangSmith API 密钥：

# os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
# os.environ["LANGSMITH_TRACING"] = "true"

初始化

有多种方法可以创建 Astra DB 矢量存储：

方法 1：显式嵌入

您可以单独实例化langchain_core.embeddings.Embeddings类并将其传递给AstraDBVectorStoreconstructor 的 Intent 方法，就像大多数其他 LangChain 向量存储一样。

方法 2：服务器端嵌入（'vectorize'）

或者，您也可以使用 Astra DB 的服务器端嵌入计算功能（“vectorize”），并在为商店创建服务器基础架构时只需指定嵌入模型。然后，嵌入计算将在后续的读取和写入作中完全在数据库内处理。（要继续使用此方法，您必须为数据库启用所需的嵌入集成，如文档中所述。

方法 3：从预先存在的集合中自动检测

您的 Astra DB 中可能已经有一个集合，可能通过其他方式（例如通过 Astra UI 或第三方应用程序）预先填充了数据，并且只想开始在 LangChain 中查询它。在这种情况下，正确的方法是启用autodetect_collectionmode 并让类找出细节。（当然，如果您的集合没有 'vectorize'，您仍然需要提供Embeddings对象）。

关于 “hybrid search” 的说明

Astra DB 矢量存储支持矢量搜索中的元数据搜索;此外，0.6 版本通过 findAndRerank 数据库原语引入了对混合搜索的完全支持：从向量相似性和基于关键字的（“词法”）搜索中检索文档，然后通过 Reranker 模型进行合并。这种完全在服务器端处理的搜索策略可以提高结果的准确性，从而提高 RAG 应用程序的质量。只要可用，向量存储就会自动使用混合搜索（尽管如果您愿意，可以对其进行手动控制）。

其他信息：

这AstraDBVectorStore可以通过多种方式进行配置;请参阅 API 参考以获取涵盖异步初始化等的完整指南;非 Astra-DB 数据库;自定义索引允许/拒绝列表;手动混合搜索控制;以及更多。

显式嵌入初始化（方法 1）

使用显式嵌入类实例化我们的 vector store：

选择嵌入模型：

pip install -qU langchain-openai

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

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

from langchain_astradb import AstraDBVectorStore

vector_store_explicit_embeddings = AstraDBVectorStore(
    collection_name="astra_vector_langchain",
    embedding=embeddings,
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_KEYSPACE,
)

API 参考：AstraDBVectorStore

服务器端嵌入初始化（“vectorize”，方法 2）

在此示例代码中，假定您具有

已在 Astra DB 组织中启用 OpenAI 集成，
添加了名为"OPENAI_API_KEY"添加到集成中，并将其范围限定为您正在使用的数据库。

有关更多详细信息，包括切换提供商/模型的说明，请参阅文档。

from astrapy.info import VectorServiceOptions

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

vector_store_integrated_embeddings = AstraDBVectorStore(
    collection_name="astra_vectorize_langchain",
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_KEYSPACE,
    collection_vector_service_options=openai_vectorize_options,
)

自动检测初始化（方法 3）

如果集合已存在于数据库中，并且AstraDBVectorStore需要使用它（用于读取和写入）。LangChain 组件将检查集合并找出详细信息。

如果已创建集合，并且最重要的是，已使用 LangChain 以外的工具填充，例如，如果数据是通过 Astra DB Web 界面摄取的，则建议使用此方法。

自动检测模式不能与集合设置（如相似性量度等）共存;另一方面，如果没有使用服务器端嵌入向量，则仍然需要传递一个Embeddingsobject 添加到构造函数中。

在以下示例代码中，我们将“自动检测”上述方法 2 创建的相同集合（“vectorize”）。因此，没有Embeddingsobject 的

vector_store_autodetected = AstraDBVectorStore(
    collection_name="astra_vectorize_langchain",
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    namespace=ASTRA_DB_KEYSPACE,
    autodetect_collection=True,
)

管理矢量存储

创建矢量存储后，通过添加和删除不同的项目来与之交互。

无论初始化方法如何，与 vector store 的所有交互都会继续进行：如果您愿意，请调整以下单元格，以选择您已创建并要进行测试的 vector store。

# If desired, uncomment a different line here:

# vector_store = vector_store_explicit_embeddings
vector_store = vector_store_integrated_embeddings
# vector_store = vector_store_autodetected

将项目添加到向量存储

使用add_documents方法。

“id” 字段可以在匹配的ids=[...]参数设置为add_documents，甚至完全省略让 store 生成 ID。

from langchain_core.documents import Document

documents_to_insert = [
    Document(
        page_content="ZYX, just another tool in the world, is actually my agent-based superhero",
        metadata={"source": "tweet"},
        id="entry_00",
    ),
    Document(
        page_content="I had chocolate chip pancakes and scrambled eggs "
        "for breakfast this morning.",
        metadata={"source": "tweet"},
        id="entry_01",
    ),
    Document(
        page_content="The weather forecast for tomorrow is cloudy and "
        "overcast, with a high of 62 degrees.",
        metadata={"source": "news"},
        id="entry_02",
    ),
    Document(
        page_content="Building an exciting new project with LangChain "
        "- come check it out!",
        metadata={"source": "tweet"},
        id="entry_03",
    ),
    Document(
        page_content="Robbers broke into the city bank and stole "
        "$1 million in cash.",
        metadata={"source": "news"},
        id="entry_04",
    ),
    Document(
        page_content="Thanks to her sophisticated language skills, the agent "
        "managed to extract strategic information all right.",
        metadata={"source": "tweet"},
        id="entry_05",
    ),
    Document(
        page_content="Is the new iPhone worth the price? Read this "
        "review to find out.",
        metadata={"source": "website"},
        id="entry_06",
    ),
    Document(
        page_content="The top 10 soccer players in the world right now.",
        metadata={"source": "website"},
        id="entry_07",
    ),
    Document(
        page_content="LangGraph is the best framework for building stateful, "
        "agentic applications!",
        metadata={"source": "tweet"},
        id="entry_08",
    ),
    Document(
        page_content="The stock market is down 500 points today due to "
        "fears of a recession.",
        metadata={"source": "news"},
        id="entry_09",
    ),
    Document(
        page_content="I have a bad feeling I am going to get deleted :(",
        metadata={"source": "tweet"},
        id="entry_10",
    ),
]


vector_store.add_documents(documents=documents_to_insert)

API 参考：文档

['entry_00',
 'entry_01',
 'entry_02',
 'entry_03',
 'entry_04',
 'entry_05',
 'entry_06',
 'entry_07',
 'entry_08',
 'entry_09',
 'entry_10']

从 vector store 中删除项目

使用 ID 删除项目delete功能。

vector_store.delete(ids=["entry_10", "entry_02"])

True

查询向量存储

创建并填充 vector store 后，您可以查询它（例如，作为 chain 或 agent 的一部分）。

直接查询

相似性搜索

搜索与提供的文本类似的文档，并根据需要使用其他元数据过滤器：

results = vector_store.similarity_search(
    "LangChain provides abstractions to make working with LLMs easy",
    k=3,
    filter={"source": "tweet"},
)
for res in results:
    print(f'* "{res.page_content}", metadata={res.metadata}')

* "Building an exciting new project with LangChain - come check it out!", metadata={'source': 'tweet'}
* "LangGraph is the best framework for building stateful, agentic applications!", metadata={'source': 'tweet'}
* "Thanks to her sophisticated language skills, the agent managed to extract strategic information all right.", metadata={'source': 'tweet'}

带分数的相似性搜索

您也可以返回相似性分数：

results = vector_store.similarity_search_with_score(
    "LangChain provides abstractions to make working with LLMs easy",
    k=3,
    filter={"source": "tweet"},
)
for res, score in results:
    print(f'* [SIM={score:.2f}] "{res.page_content}", metadata={res.metadata}')

* [SIM=0.71] "Building an exciting new project with LangChain - come check it out!", metadata={'source': 'tweet'}
* [SIM=0.70] "LangGraph is the best framework for building stateful, agentic applications!", metadata={'source': 'tweet'}
* [SIM=0.61] "Thanks to her sophisticated language skills, the agent managed to extract strategic information all right.", metadata={'source': 'tweet'}

指定其他关键字查询（需要混合搜索）

注意：仅当集合支持 find-and-rerank 命令并且向量存储知道这一事实时，才能运行此单元格。

如果 vector store 正在使用支持混合的集合并检测到这一事实，则默认情况下，它将在运行搜索时使用该功能。

在这种情况下，在 find-and-rerank 过程中，相同的查询文本将用于向量相似性和基于词法的检索步骤，除非您为后者明确提供不同的查询：

results = vector_store_autodetected.similarity_search(
    "LangChain provides abstractions to make working with LLMs easy",
    k=3,
    filter={"source": "tweet"},
    lexical_query="agent",
)
for res in results:
    print(f'* "{res.page_content}", metadata={res.metadata}')

* "Building an exciting new project with LangChain - come check it out!", metadata={'source': 'tweet'}
* "LangGraph is the best framework for building stateful, agentic applications!", metadata={'source': 'tweet'}
* "ZYX, just another tool in the world, is actually my agent-based superhero", metadata={'source': 'tweet'}

上面的示例对 “autodetected” vector store 进行硬编码，它肯定已经检查了集合并确定了 hybrid 是否可用。另一种选择是显式地向构造函数提供 hybrid-search 参数（有关更多详细信息/示例，请参阅 API 参考）。

其他搜索方法

本笔记本中未涵盖各种其他搜索方法，例如 MMR 搜索和按向量搜索。

有关中可用的搜索模式的完整列表，请参阅AstraDBVectorStore请查看 API 参考。

通过转换为 retriever 进行查询

您还可以将 vector store 变成检索器，以便在您的链中更轻松地使用。

将向量存储转换为检索器，并使用简单的查询 + 元数据过滤器调用它：

retriever = vector_store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 1, "score_threshold": 0.5},
)
retriever.invoke("Stealing from the bank is a crime", filter={"source": "news"})

[Document(id='entry_04', metadata={'source': 'news'}, page_content='Robbers broke into the city bank and stole $1 million in cash.')]

用于检索增强生成

有关如何使用此向量存储进行检索增强生成（RAG）的指南，请参阅以下部分：

有关更多信息，请在此处查看使用 Astra DB 的完整 RAG 模板。

清理矢量存储

如果要从 Astra 数据库实例中完全删除集合，请运行以下命令。

（您将丢失存储在其中的数据。

vector_store.delete_collection()

API 参考

有关所有AstraDBVectorStore特性和配置，请参阅 API 参考。

矢量存储概念指南
Vector store 操作指南

设置

依赖

凭据

初始化