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"
凭证
为了使用 AstraDB 向量存储,您必须首先访问AstraDB 网站,创建一个账户,然后创建一个新数据库——初始化可能需要几分钟。
数据库初始化完成后,请检索您的连接密钥,您很快就会需要它们。它们是
- 一个
API 端点
,例如"https://01234567-89ab-cdef-0123-456789abcdef-us-east1.apps.astra.datastax.com/"
- 以及一个
数据库令牌
,例如"AstraCS:aBcD123......"
您可以选择提供一个 keyspace
(在 LangChain 组件中称为“命名空间”),您可以通过数据库仪表板的 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 密钥
# 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:服务器端嵌入('向量化')
或者,您可以使用 Astra DB 的服务器端嵌入计算功能('向量化'),在为存储创建服务器基础设施时,只需指定一个嵌入模型。此后,所有嵌入计算将在后续的读写操作中完全在数据库内部处理。(要使用此方法,您必须已为数据库启用了所需的嵌入集成,如文档中所述。)
方法 3:从现有集合自动检测
您的 Astra DB 中可能已经存在一个集合,该集合可能已通过其他方式(例如通过 Astra UI 或第三方应用程序)预先填充了数据,而您只是想在 LangChain 中开始查询它。在这种情况下,正确的方法是在向量存储构造函数中启用 autodetect_collection
模式,并让该类自动识别细节。(当然,如果您的集合没有启用 '向量化',您仍然需要提供一个 Embeddings
对象。)
关于“混合搜索”的说明
Astra DB 向量存储支持向量搜索中的元数据搜索;此外,0.6 版本通过 findAndRerank 数据库原语引入了对混合搜索的全面支持:文档通过向量相似度搜索和基于关键词(“词汇”)搜索两种方式检索,然后通过重新排名模型进行合并。这种完全在服务器端处理的搜索策略可以提高您的结果准确性,从而改善您的 RAG 应用程序质量。只要可用,向量存储就会自动使用混合搜索(尽管如果您愿意,可以对其进行手动控制)。
附加信息
AstraDBVectorStore
可以通过多种方式进行配置;有关完整指南,例如异步初始化;非 Astra-DB 数据库;自定义索引允许/拒绝列表;手动混合搜索控制等等,请参阅API 参考。
显式嵌入初始化(方法 1)
使用显式嵌入类实例化我们的向量存储
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,
)
服务器端嵌入初始化(“向量化”,方法 2)
在此示例代码中,假定您已满足以下条件:
- 在您的 Astra DB 组织中启用了 OpenAI 集成,
- 向集成添加了名为
"OPENAI_API_KEY"
的 API 密钥,并将其限定为您正在使用的数据库。
有关更多详细信息,包括切换提供商/模型的说明,请查阅文档。
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 界面摄入的,则这是推荐的方法。
自动检测模式不能与集合设置(例如相似度指标等)共存;另一方面,如果未采用服务器端嵌入,仍然需要将 Embeddings
对象传递给构造函数。
在以下示例代码中,我们将“自动检测”与上述方法 2(“向量化”)创建的同一个集合。因此,无需提供 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 参考)。
其他搜索方法
本笔记中未涵盖多种其他搜索方法,例如 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) 的指南,请参阅以下部分
欲了解更多,请在此处查看使用 Astra DB 的完整 RAG 模板。
清理向量存储
如果您想从 Astra DB 实例中完全删除集合,请运行此操作。
(您将丢失其中存储的数据。)
vector_store.delete_collection()
API 参考
有关 AstraDBVectorStore
所有功能和配置的详细文档,请查阅API 参考。