Astra DB 向量存储

此页面提供了将 Astra DB 作为向量存储使用的基本指南。

DataStax Astra DB 是一个基于 Apache Cassandra® 构建的无服务器 AI 就绪数据库，并通过易于使用的 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"

Credentials

要在使用 AstraDB 向量存储之前，您必须首先访问 AstraDB 网站，创建一个账户，然后创建一个新的数据库 - 初始化可能需要几分钟时间。

初始化数据库后，请获取您的连接密钥，您将很快需要它们。这些包括：

• an API Endpoint, such as "https://01234567-89ab-cdef-0123-456789abcdef-us-east1.apps.astra.datastax.com/"
• and a Database Token, e.g. "AstraCS:aBcD123......"

您可以在数据库仪表盘的Data Explorer标签页中管理名为“命名空间”（namespace）的keyspace。如果您愿意，可以在下面的提示中留空，并使用默认键空间作为替代。

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类，并将其传递给AstraDBVectorStore构造函数，就像大多数其他LangChain向量存储一样。

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

Alternatively, 您可以使用Astra DB的服务器端嵌入计算功能（'vectorize'），并在创建存储的服务基础设施时指定一个嵌入模型。随后在读写操作中，嵌入计算将完全由数据库处理。 （要继续此方法，请确保已按照文档^中所述为您的数据库启用了所需的嵌入集成。）

方法3：从已存在的集合中自动检测​

您可能已经在Astra DB中有一个集合，该集合可能是通过其他方式预先填充的数据（例如：通过Astra UI或第三方应用程序），而只是想在LangChain中开始查询它。在这种情况下，正确的做法是在向量存储构造函数中启用autodetect_collection模式，并让类自行处理详细信息。（当然，如果您集合没有'vectorize'，您仍然需要提供一个Embeddings对象）。

关于“混合搜索”的说明

Astra DB 向量存储支持在向量搜索中进行元数据搜索；此外，0.6 版本引入了通过 findAndRerank 数据库原语对 混合搜索 的全面支持：文档会从向量相似性和基于关键词（“词典”）的搜索中同时检索，并通过重新排序模型进行合并。这种完全在服务器端处理的搜索策略可以提高结果的准确性，从而提升您的 RAG 应用的质量。当可用时，向量存储将自动使用混合搜索（尽管您可以在需要时手动控制它）。

Additional information

The AstraDBVectorStore 可以通过多种方式进行配置；请参阅 API参考文档，了解包括异步初始化、非Astra-DB数据库、自定义索引允许/拒绝列表、手动混合搜索控制等在内的完整指南。

Explicit embedding 初始化（方法 1）

使用显式的嵌入类初始化我们的向量存储：

选择 嵌入模型：

OpenAI

Azure

Google

AWS

HuggingFace

Ollama

Cohere

Mistral AI

Nomic

英伟达

Voyage AI

IBM watsonx

假

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 参考:Astra DB 向量存储

Server-side embedding 初始化（"vectorize", 方法 2)

在以下代码示例中，假设您已经

• 已为您启用 Astra DB 组织中的 OpenAI 集成,
• 已将一个API密钥命名为"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,
)

Auto-detect 初始化（方法3）

您可以使用此模式，如果集合已经在数据库中存在并且您的AstraDBVectorStore需要使用它（用于读取和写入）。LangChain 组件将检查该集合并确定相关细节。

如果集合已经创建并且——最重要的是——由非 LangChain 的工具填充数据，则建议采用此方法，例如通过 Astra DB 网页界面导入数据的情况。

Auto-detect模式不能与收集设置（如相似度度量等）共存；另一方面，如果没有使用服务器端嵌入，则仍然需要向构造函数传递一个Embeddings对象。

在以下示例代码中，我们将“自动检测”与上方方法2（"vectorize"）创建的相同集合。因此，无需提供Embeddings对象。

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,
)

管理向量存储

创建好向量存储后，可以通过添加和删除不同的项目与其进行交互。

所有与向量存储的交互都将无视初始化方法进行：请根据需要 适应以下单元格，选择您创建并想测试的向量存储。

# 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中作为参数给出，甚至可以直接省略由存储自动生成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']

删除向量存储中的项

通过使用delete函数来按ID删除项。

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

True

查询向量存储库

一旦向量存储创建并填充完毕，您就可以对其进行查询（例如，作为链或代理的一部分）。

查询直接

相似度搜索

搜索与提供的文本相似的文档，如有需要可使用附加元数据过滤器：

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过程中，向量相似度检索和词法基础检索步骤将使用相同的查询文本，除非您明确为后者提供不同的查询：

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'}

上述示例将“自动检测”向量存储硬编码在其中，该向量存储肯定已经检查了集合，并确定是否可用混合搜索。另一种选择是显式地在构造函数中提供混合搜索参数（更多细节/示例请参阅API参考）。

其他搜索方法

这是本笔记本未涵盖的其他搜索方法，例如MMR搜索和向量搜索。

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

查询通过转换为检索器

您可以将向量存储转换为检索器，以便在您的链中更轻松地使用。

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

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）的指南，请参见以下部分：

• 教程
• 如何使用RAG进行问答
• 检索概念文档

要了解更多，请查看使用Astra DB的完整RAG模板 这里。

清理向量存储

如果您希望从Astra DB 实例中完全删除集合，请运行此操作。

(您将丢失其中存储的所有数据。)

vector_store.delete_collection()

API 参考

对于所有AstraDBVectorStore功能和配置的详细文档，请参阅API参考。

相关

• 向量存储 概念指南
• 向量存储 使用指南