Skip to main content
本页面提供了将 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 标签页中管理。如果您希望，可以在下方提示中留空，使用默认 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 Key： 
os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
os.environ["LANGSMITH_TRACING"] = "true"

​
初始化

创建 Astra DB 向量存储有多种方式：

​
方式一：显式嵌入

您可以单独实例化一个 langchain_core.embeddings.Embeddings 类，并将其传递给 AstraDBVectorStore 构造函数，与大多数其他 LangChain 向量存储的用法相同。

​
方式二：服务端嵌入（‘vectorize’）

或者，您可以使用 Astra DB 的服务端嵌入计算功能（‘vectorize’），在创建存储的服务端基础设施时直接指定嵌入模型。后续的读写操作中，嵌入计算将完全在数据库内部处理。（使用此方式前，您必须已为数据库启用所需的嵌入集成，详见文档。）

​
方式三：从已有集合自动检测

您的 Astra DB 中可能已有一个集合（collection），可能通过其他方式（如 Astra UI 或第三方应用）预先填充了数据，您只需在 LangChain 中开始查询。这种情况下，正确的做法是在向量存储构造函数中启用 autodetect_collection 模式，让类自动识别相关细节。（当然，如果您的集合没有使用 ‘vectorize’，仍然需要提供一个 Embeddings 对象。）

​
关于”混合搜索”的说明

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

​
补充说明

AstraDBVectorStore 可以通过多种方式配置；完整指南请参阅 API 参考，涵盖异步初始化、非 Astra DB 数据库、自定义索引允许/拒绝列表、手动混合搜索控制等更多内容。

​
显式嵌入初始化（方式一）

使用显式嵌入类实例化我们的向量存储： 
# | output: false
# | echo: false
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,
)

​
服务端嵌入初始化（“vectorize”，方式二）

以下示例代码假设您已：
  • 在 Astra DB 组织中启用了 OpenAI 集成，
  • 向集成中添加了名为 "OPENAI_API_KEY" 的 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,
)

​
自动检测初始化（方式三）

如果集合已存在于数据库中，且您的 AstraDBVectorStore 需要使用它（进行读写操作），可以使用此模式。LangChain 组件将检查集合并自动识别相关细节。 如果集合是由 LangChain 以外的工具创建并填充的（例如通过 Astra DB Web 界面导入数据），推荐使用此方式。 自动检测模式不能与_集合_设置（如相似度指标等）共存；另一方面，如果没有使用服务端嵌入，仍然需要向构造函数传递一个 Embeddings 对象。 以下示例代码将”自动检测”方式二（“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” 字段可以单独通过 add_documents 的 ids=[...] 参数提供，也可以完全省略，让存储自动生成 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)

['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 命令且向量存储已知晓此事实时才可运行。
如果向量存储使用了支持混合搜索的集合并已检测到这一点，默认情况下在执行搜索时会使用该功能。 在这种情况下，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 参考）。

​
其他搜索方法

本 notebook 未涵盖多种其他搜索方法，例如 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 模板 here。

​
清理向量存储

如果您想从 Astra DB 实例中完全删除该集合，请运行此操作。 （您存储在其中的数据将会丢失。） 
vector_store.delete_collection()

​
API 参考

有关 AstraDBVectorStore 所有功能和配置的详细文档，请参阅 API 参考。
Edit this page on GitHub or file an issue.
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.