Apache Cassandra

本頁面提供使用 Apache Cassandra® 作為向量儲存庫的快速入門指南。

Cassandra 是一個 NoSQL、面向列、高度可擴展和高度可用的資料庫。從 5.0 版本開始，該資料庫附帶了 向量搜尋功能。

注意：除了存取資料庫外，執行完整範例還需要 OpenAI API 金鑰。

設定和一般依賴項

整合的使用需要以下 Python 套件。

%pip install --upgrade --quiet langchain-community "cassio>=0.1.4"

注意：根據您的 LangChain 設定，您可能需要安裝/升級此示範所需的其他依賴項 （特別是，需要最新版本的 datasets、openai、pypdf 和 tiktoken，以及 langchain-community）。

import os
from getpass import getpass

from datasets import (
    load_dataset,
)
from langchain_community.document_loaders import PyPDFLoader
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

API 參考資料：PyPDFLoader | Document | StrOutputParser | ChatPromptTemplate | RunnablePassthrough | ChatOpenAI | OpenAIEmbeddings | RecursiveCharacterTextSplitter

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass("OPENAI_API_KEY = ")

embe = OpenAIEmbeddings()

匯入向量儲存庫

from langchain_community.vectorstores import Cassandra

API 參考資料：Cassandra

連線參數

本頁面中顯示的向量儲存庫整合可用於 Cassandra 以及其他衍生資料庫，例如 Astra DB，它們使用 CQL（Cassandra 查詢語言）協定。

DataStax Astra DB 是一個基於 Cassandra 建構的託管無伺服器資料庫，提供相同的介面和優勢。

根據您是透過 CQL 連接到 Cassandra 叢集還是 Astra DB，您將在建立向量儲存庫物件時提供不同的參數。

連接到 Cassandra 叢集

您首先需要建立一個 cassandra.cluster.Session 物件，如 Cassandra 驅動程式文件中所述。詳細資訊各不相同（例如，網路設定和身份驗證），但可能類似於

from cassandra.cluster import Cluster

cluster = Cluster(["127.0.0.1"])
session = cluster.connect()

您現在可以將會話與您想要的鍵空間名稱一起設定為全域 CassIO 參數

import cassio

CASSANDRA_KEYSPACE = input("CASSANDRA_KEYSPACE = ")

cassio.init(session=session, keyspace=CASSANDRA_KEYSPACE)

現在您可以建立向量儲存庫

vstore = Cassandra(
    embedding=embe,
    table_name="cassandra_vector_demo",
    # session=None, keyspace=None  # Uncomment on older versions of LangChain
)

注意：您也可以在建立向量儲存庫時直接將會話和鍵空間作為參數傳遞。但是，如果您的應用程式以多種方式使用 Cassandra（例如，用於向量儲存庫、聊天記憶體和 LLM 回應快取），則使用全域 cassio.init 設定會很方便，因為它允許將憑證和資料庫連線管理集中在一個位置。

透過 CQL 連接到 Astra DB

在這種情況下，您可以使用以下連線參數初始化 CassIO

• 資料庫 ID，例如 01234567-89ab-cdef-0123-456789abcdef
• 權杖，例如 AstraCS:6gBhNmsk135.... （它必須是「資料庫管理員」權杖）
• 可選的鍵空間名稱（如果省略，將使用資料庫的預設名稱）

ASTRA_DB_ID = input("ASTRA_DB_ID = ")
ASTRA_DB_APPLICATION_TOKEN = getpass("ASTRA_DB_APPLICATION_TOKEN = ")

desired_keyspace = input("ASTRA_DB_KEYSPACE (optional, can be left empty) = ")
if desired_keyspace:
    ASTRA_DB_KEYSPACE = desired_keyspace
else:
    ASTRA_DB_KEYSPACE = None

import cassio

cassio.init(
    database_id=ASTRA_DB_ID,
    token=ASTRA_DB_APPLICATION_TOKEN,
    keyspace=ASTRA_DB_KEYSPACE,
)

現在您可以建立向量儲存庫

vstore = Cassandra(
    embedding=embe,
    table_name="cassandra_vector_demo",
    # session=None, keyspace=None  # Uncomment on older versions of LangChain
)

載入資料集

將來源資料集中的每個條目轉換為 Document，然後將它們寫入向量儲存庫

philo_dataset = load_dataset("datastax/philosopher-quotes")["train"]

docs = []
for entry in philo_dataset:
    metadata = {"author": entry["author"]}
    doc = Document(page_content=entry["quote"], metadata=metadata)
    docs.append(doc)

inserted_ids = vstore.add_documents(docs)
print(f"\nInserted {len(inserted_ids)} documents.")

在上面，metadata 字典是從來源資料建立的，並且是 Document 的一部分。

再新增一些條目，這次使用 add_texts

texts = ["I think, therefore I am.", "To the things themselves!"]
metadatas = [{"author": "descartes"}, {"author": "husserl"}]
ids = ["desc_01", "huss_xy"]

inserted_ids_2 = vstore.add_texts(texts=texts, metadatas=metadatas, ids=ids)
print(f"\nInserted {len(inserted_ids_2)} documents.")

注意：您可能希望透過提高這些批次操作的並行級別來加速 add_texts 和 add_documents 的執行 - 查看這些方法的 batch_size 參數 以獲取更多詳細資訊。根據網路和客戶端機器的規格，您最佳效能的參數選擇可能會有所不同。

執行搜尋

本節示範元數據篩選並取回相似度分數

results = vstore.similarity_search("Our life is what we make of it", k=3)
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")

results_filtered = vstore.similarity_search(
    "Our life is what we make of it",
    k=3,
    filter={"author": "plato"},
)
for res in results_filtered:
    print(f"* {res.page_content} [{res.metadata}]")

results = vstore.similarity_search_with_score("Our life is what we make of it", k=3)
for res, score in results:
    print(f"* [SIM={score:3f}] {res.page_content} [{res.metadata}]")

MMR（最大邊際相關性）搜尋

results = vstore.max_marginal_relevance_search(
    "Our life is what we make of it",
    k=3,
    filter={"author": "aristotle"},
)
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")

刪除儲存的文件

delete_1 = vstore.delete(inserted_ids[:3])
print(f"all_succeed={delete_1}")  # True, all documents deleted

delete_2 = vstore.delete(inserted_ids[2:5])
print(f"some_succeeds={delete_2}")  # True, though some IDs were gone already

最小 RAG 鏈

接下來的單元格將實作一個簡單的 RAG 管道

• 下載範例 PDF 檔案並將其載入到儲存庫中；
• 使用 LCEL（LangChain 運算式語言）建立 RAG 鏈，向量儲存庫是其核心；
• 執行問答鏈。

!curl -L \
    "https://github.com/awesome-astra/datasets/blob/main/demo-resources/what-is-philosophy/what-is-philosophy.pdf?raw=true" \
    -o "what-is-philosophy.pdf"

pdf_loader = PyPDFLoader("what-is-philosophy.pdf")
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
docs_from_pdf = pdf_loader.load_and_split(text_splitter=splitter)

print(f"Documents from PDF: {len(docs_from_pdf)}.")
inserted_ids_from_pdf = vstore.add_documents(docs_from_pdf)
print(f"Inserted {len(inserted_ids_from_pdf)} documents.")

retriever = vstore.as_retriever(search_kwargs={"k": 3})

philo_template = """
You are a philosopher that draws inspiration from great thinkers of the past
to craft well-thought answers to user questions. Use the provided context as the basis
for your answers and do not make up new reasoning paths - just mix-and-match what you are given.
Your answers must be concise and to the point, and refrain from answering about other topics than philosophy.

CONTEXT:
{context}

QUESTION: {question}

YOUR ANSWER:"""

philo_prompt = ChatPromptTemplate.from_template(philo_template)

llm = ChatOpenAI()

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | philo_prompt
    | llm
    | StrOutputParser()
)

chain.invoke("How does Russel elaborate on Peirce's idea of the security blanket?")

如需更多資訊，請查看使用 CQL 的 Astra DB 的完整 RAG 範本 此處。

清理

以下內容本質上是從 CassIO 檢索 Session 物件，並使用它執行 CQL DROP TABLE 語句

（您將遺失儲存在其中的資料。）

cassio.config.resolve_session().execute(
    f"DROP TABLE {cassio.config.resolve_keyspace()}.cassandra_vector_demo;"
)

了解更多

如需更多資訊、擴展的快速入門指南和其他使用範例，請造訪 CassIO 文件，以了解更多關於使用 LangChain Cassandra 向量儲存庫的資訊。

歸屬聲明

Apache Cassandra、Cassandra 和 Apache 是 Apache Software Foundation 在美國和/或其他國家/地區的註冊商標或商標。

相關內容

• 向量儲存庫概念指南
• 向量儲存庫操作指南