Philosophy with Vector Embeddings, OpenAI and Cassandra / Astra DB through CQL

CassIO version

在本快速入门教程中，您将学习如何使用 OpenAI 的向量嵌入和 Apache Cassandra® 或等效的 DataStax Astra DB through CQL 作为数据持久化的向量存储来构建一个“哲学名言查找器和生成器”。

此笔记本的基本工作流程概述如下。您将评估并存储多位著名哲学家的名言的向量嵌入，利用它们构建一个强大的搜索引擎，然后甚至可以生成新的名言！

该笔记本举例说明了一些向量搜索的标准用法模式——同时展示了 Cassandra / Astra DB through CQL 的向量功能入门是多么容易。

有关使用向量搜索和文本嵌入构建问答系统的背景信息，请参阅这篇出色的实践笔记本：Question answering using embeddings。

选择您的框架

请注意，此笔记本使用 CassIO 库，但我们也涵盖了其他技术选择来完成相同的任务。请查看此文件夹的 README 以了解其他选项。此笔记本可以作为 Colab 笔记本或常规 Jupyter 笔记本运行。

目录：

• 设置
• 获取数据库连接
• 连接到 OpenAI
• 将名言加载到向量存储中
• 用例 1：名言搜索引擎
• 用例 2：名言生成器
• （可选）利用向量存储中的分区

工作原理

索引

每条名言都通过 OpenAI 的 Embedding 转换为嵌入向量。这些向量被保存在向量存储中以供以后搜索使用。一些元数据，包括作者姓名和其他一些预先计算的标签，与向量一起存储，以便进行搜索定制。

搜索

要查找与提供的搜索名言相似的名言，后者会被即时转换为嵌入向量，然后该向量用于查询存储以查找相似向量……即之前索引的相似名言。搜索还可以选择性地受其他元数据约束（“找到一些斯宾诺莎的关于此名言的引述……”）。

关键点在于，“内容相似的名言”在向量空间中转化为彼此距离接近的向量：因此，向量相似性搜索有效地实现了语义相似性。这是向量嵌入如此强大的关键原因。

下图试图传达这个想法。每条名言一旦被转换为向量，就成为空间中的一个点。在这个例子中，它在一个球体上，因为 OpenAI 的嵌入向量，像大多数其他向量一样，被归一化为_单位长度_。哦，而且这个球体实际上不是三维的，而是 1536 维的！

所以，本质上，向量空间中的相似性搜索返回最接近查询向量的向量：

设置

首先安装一些必需的包：

!pip install --quiet "cassio>=0.1.3" "openai>=1.0.0" datasets

from getpass import getpass
from collections import Counter

import cassio
from cassio.table import MetadataVectorCassandraTable

import openai
from datasets import load_dataset

获取数据库连接

为了通过 CQL 连接到您的 Astra DB，您需要两样东西：

• 一个具有“数据库管理员”角色的令牌（看起来像 AstraCS:...）
• 数据库 ID（看起来像 3df2a5b6-...）

确保您同时拥有这两个字符串——它们是在您登录后在 Astra UI 中获得的。有关更多信息，请参见此处：database ID 和 Token。

如果您想_连接到 Cassandra 集群_（但该集群必须支持向量搜索），请将 cassio.init(session=..., keyspace=...) 替换为适合您集群的 Session 和 keyspace 名称。

astra_token = getpass("Please enter your Astra token ('AstraCS:...')")
database_id = input("Please enter your database id ('3df2a5b6-...')")

Please enter your Astra token ('AstraCS:...') ········
Please enter your database id ('3df2a5b6-...') 01234567-89ab-dcef-0123-456789abcdef

cassio.init(token=astra_token, database_id=database_id)

创建数据库连接

这是通过 CQL 创建 Astra DB 连接的方法：

（顺便说一句，您也可以使用任何 Cassandra 集群（只要它提供向量功能），只需通过更改参数来实例化下面的 Cluster。）

通过 CassIO 创建向量存储

您需要一个支持向量并配备了元数据的表。称其为“philosophers_cassio”：

v_table = MetadataVectorCassandraTable(table="philosophers_cassio", vector_dimension=1536)

连接到 OpenAI

设置您的密钥

OPENAI_API_KEY = getpass("Please enter your OpenAI API Key: ")

Please enter your OpenAI API Key:  ········

嵌入的测试调用

快速检查如何获取输入文本列表的嵌入向量：

client = openai.OpenAI(api_key=OPENAI_API_KEY)
embedding_model_name = "text-embedding-3-small"

result = client.embeddings.create(
    input=[
        "This is a sentence",
        "A second sentence"
    ],
    model=embedding_model_name,
)

注意：以上是 OpenAI v1.0+ 的语法。如果使用早期版本，获取嵌入的代码将有所不同。

print(f"len(result.data)              = {len(result.data)}")
print(f"result.data[1].embedding      = {str(result.data[1].embedding)[:55]}...")
print(f"len(result.data[1].embedding) = {len(result.data[1].embedding)}")

len(result.data)              = 2
result.data[1].embedding      = [-0.010821706615388393, 0.001387271680869162, 0.0035479...
len(result.data[1].embedding) = 1536

将名言加载到向量存储中

注意：以上是 OpenAI v1.0+ 的语法。如果使用早期版本，获取嵌入的代码将有所不同。

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

快速检查：

print("An example entry:")
print(philo_dataset[16])

An example entry:
{'author': 'aristotle', 'quote': 'Love well, be loved and do something of value.', 'tags': 'love;ethics'}

检查数据集大小：

author_count = Counter(entry["author"] for entry in philo_dataset)
print(f"Total: {len(philo_dataset)} quotes. By author:")
for author, count in author_count.most_common():
    print(f"    {author:<20}: {count} quotes")

Total: 450 quotes. By author:
    aristotle           : 50 quotes
    schopenhauer        : 50 quotes
    spinoza             : 50 quotes
    hegel               : 50 quotes
    freud               : 50 quotes
    nietzsche           : 50 quotes
    sartre              : 50 quotes
    plato               : 50 quotes
    kant                : 50 quotes

将名言插入向量存储

您将为名言计算嵌入向量并将它们与文本本身以及计划用于后续使用的元数据一起保存到向量存储中。请注意，作者作为元数据字段与名言本身已有的“标签”一起添加。

为了优化速度和减少调用次数，您将执行批量调用嵌入 OpenAI 服务。

（注意：为了执行更快，Cassandra 和 CassIO 将允许您进行并发插入，但在此处我们不这样做，以使演示代码更直观。）

BATCH_SIZE = 50

num_batches = ((len(philo_dataset) + BATCH_SIZE - 1) // BATCH_SIZE)

quotes_list = philo_dataset["quote"]
authors_list = philo_dataset["author"]
tags_list = philo_dataset["tags"]

print("Starting to store entries:")
for batch_i in range(num_batches):
    b_start = batch_i * BATCH_SIZE
    b_end = (batch_i + 1) * BATCH_SIZE
    # compute the embedding vectors for this batch
    b_emb_results = client.embeddings.create(
        input=quotes_list[b_start : b_end],
        model=embedding_model_name,
    )
    # prepare the rows for insertion
    print("B ", end="")
    for entry_idx, emb_result in zip(range(b_start, b_end), b_emb_results.data):
        if tags_list[entry_idx]:
            tags = {
                tag
                for tag in tags_list[entry_idx].split(";")
            }
        else:
            tags = set()
        author = authors_list[entry_idx]
        quote = quotes_list[entry_idx]
        v_table.put(
            row_id=f"q_{author}_{entry_idx}",
            body_blob=quote,
            vector=emb_result.embedding,
            metadata={**{tag: True for tag in tags}, **{"author": author}},
        )
        print("*", end="")
    print(f" done ({len(b_emb_results.data)})")

print("\nFinished storing entries.")

Starting to store entries:
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)
B ************************************************** done (50)

Finished storing entries.

用例 1：名言搜索引擎

对于名言搜索功能，您首先需要将输入的引语转换为向量，然后使用它来查询存储（除了处理可选的元数据到搜索调用中）。

将搜索引擎功能封装到一个函数中以便于重用：

def find_quote_and_author(query_quote, n, author=None, tags=None):
    query_vector = client.embeddings.create(
        input=[query_quote],
        model=embedding_model_name,
    ).data[0].embedding
    metadata = {}
    if author:
        metadata["author"] = author
    if tags:
        for tag in tags:
            metadata[tag] = True
    #
    results = v_table.ann_search(
        query_vector,
        n=n,
        metadata=metadata,
    )
    return [
        (result["body_blob"], result["metadata"]["author"])
        for result in results
    ]

测试搜索

仅传递一句名言：

find_quote_and_author("We struggle all our life for nothing", 3)

[('Life to the great majority is only a constant struggle for mere existence, with the certainty of losing it at last.',
  'schopenhauer'),
 ('We give up leisure in order that we may have leisure, just as we go to war in order that we may have peace.',
  'aristotle'),
 ('Perhaps the gods are kind to us, by making life more disagreeable as we grow older. In the end death seems less intolerable than the manifold burdens we carry',
  'freud')]

限制作者的搜索：

find_quote_and_author("We struggle all our life for nothing", 2, author="nietzsche")

[('To live is to suffer, to survive is to find some meaning in the suffering.',
  'nietzsche'),
 ('What makes us heroic?--Confronting simultaneously our supreme suffering and our supreme hope.',
  'nietzsche')]

受标签约束的搜索（从之前与名言一起保存的标签中）：

find_quote_and_author("We struggle all our life for nothing", 2, tags=["politics"])

[('Mankind will never see an end of trouble until lovers of wisdom come to hold political power, or the holders of power become lovers of wisdom',
  'plato'),
 ('Everything the State says is a lie, and everything it has it has stolen.',
  'nietzsche')]

排除不相关结果

向量相似性搜索通常会返回最接近查询向量的结果，即使这意味着结果可能有些不相关，如果没有任何更好的结果的话。

为了控制这个问题，您可以获取查询与每个结果之间的实际“距离”，然后设置一个阈值，从而有效地丢弃超出该阈值的结果。 正确调整此阈值并非易事：在此，我们仅向您展示方法。

为了感受它是如何工作的，请尝试以下查询，并调整名言和阈值的选择以比较结果：

注意（对于有数学头脑的人）：这个“距离”正是向量之间的余弦相似度，即标量积除以两个向量范数的乘积。因此，它是一个从 -1 到 +1 的数字，其中 -1 表示完全相反的向量，+1 表示方向相同的向量。在其他地方（例如，此演示的“CQL”对应项）您将获得此数量的重新缩放以适合 [0, 1] 区间，这意味着结果数值和适当的阈值在那里会相应地转换。

quote = "Animals are our equals."
# quote = "Be good."
# quote = "This teapot is strange."

metric_threshold = 0.84

quote_vector = client.embeddings.create(
    input=[quote],
    model=embedding_model_name,
).data[0].embedding

results = list(v_table.metric_ann_search(
    quote_vector,
    n=8,
    metric="cos",
    metric_threshold=metric_threshold,
))

print(f"{len(results)} quotes within the threshold:")
for idx, result in enumerate(results):
    print(f"    {idx}. [distance={result['distance']:.3f}] \"{result['body_blob'][:70]}...\"")

3 quotes within the threshold:

    0. [distance=0.855] "The assumption that animals are without rights, and the illusion that our treatment of them has no moral significance, is a positively outrageous example of Western crudity and barbarity. Universal compassion is the only guarantee of morality. ..."
    1. [distance=0.843] "Animals are in possession of themselves; their soul is in possession of itself, and not in the possession of the body. ..."
    2. [distance=0.841] "At his best, man is the noblest of all animals; separated from law and justice, he is the worst."

用例 2：名言生成器

对于这项任务，您需要 OpenAI 的另一个组件，即 LLM，为我们生成名言（基于通过查询向量存储获得的输入）。

您还需要一个模板来填充用于生成名言的 LLM 完成任务的提示。

completion_model_name = "gpt-3.5-turbo"

generation_prompt_template = """"Generate a single short philosophical quote on the given topic,
similar in spirit and form to the provided actual example quotes.
Do not exceed 20-30 words in your quote.

REFERENCE TOPIC: "{topic}"

ACTUAL EXAMPLES:
{examples}
"""

与搜索一样，最好将此功能封装到一个方便的函数中（该函数内部使用搜索）：

def generate_quote(topic, n=2, author=None, tags=None):
    quotes = find_quote_and_author(query_quote=topic, n=n, author=author, tags=tags)
    if quotes:
        prompt = generation_prompt_template.format(
            topic=topic,
            examples="\n".join(f"  - {quote[0]}" for quote in quotes),
        )
        # a little logging:
        print("** quotes found:")
        for q, a in quotes:
            print(f"**    - {q} ({a})")
        print("** end of logging")
        #
        response = client.chat.completions.create(
            model=completion_model_name,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=320,
        )
        return response.choices[0].message.content.replace('"', '').strip()
    else:
        print("** no quotes found.")
        return None

注意：与嵌入计算一样，对于 OpenAI v1.0 之前的版本，Chat Completion API 的代码会略有不同。

测试名言生成

只需传递一段文本（一句“名言”，但实际上可以只建议一个主题，因为它的向量嵌入仍然会放在向量空间的正确位置）：

q_topic = generate_quote("politics and virtue")
print("\nA new generated quote:")
print(q_topic)

** quotes found:

**    - Happiness is the reward of virtue. (aristotle)
**    - Our moral virtues benefit mainly other people; intellectual virtues, on the other hand, benefit primarily ourselves; therefore the former make us universally popular, the latter unpopular. (schopenhauer)
** end of logging

A new generated quote:
Virtuous politics purifies society, while corrupt politics breeds chaos and decay.

从一位哲学家那里获得灵感：

q_topic = generate_quote("animals", author="schopenhauer")
print("\nA new generated quote:")
print(q_topic)

** quotes found:

**    - Because Christian morality leaves animals out of account, they are at once outlawed in philosophical morals; they are mere 'things,' mere means to any ends whatsoever. They can therefore be used for vivisection, hunting, coursing, bullfights, and horse racing, and can be whipped to death as they struggle along with heavy carts of stone. Shame on such a morality that is worthy of pariahs, and that fails to recognize the eternal essence that exists in every living thing, and shines forth with inscrutable significance from all eyes that see the sun! (schopenhauer)
**    - The assumption that animals are without rights, and the illusion that our treatment of them has no moral significance, is a positively outrageous example of Western crudity and barbarity. Universal compassion is the only guarantee of morality. (schopenhauer)
** end of logging

A new generated quote:
The true measure of humanity lies not in our dominion over animals, but in our ability to show compassion and respect for all living beings.

（可选）分区

在完成此快速入门之前，有一个有趣的主题需要探讨。虽然，通常情况下，标签和名言可以有任何关系（例如，一个名言有多个标签），但_作者_实际上是一个精确的分组（它们定义了名言集合上的“不相交分区”）：每条名言只有一个作者（至少对我们来说是这样）。

现在，假设您提前知道您的应用程序通常（或总是）会对_单个作者_执行查询。那么您可以充分利用底层数据库结构：如果您将名言按分区（每个作者一个）分组，那么仅针对作者的向量查询将消耗更少的资源并返回得更快。

我们在这里不深入细节，这些细节与 Cassandra 存储的内部结构有关：重要的信息是如果您的查询是在一个组内运行的，请考虑相应地进行分区以提高性能。

您现在将看到此选择的实际应用。

首先，您需要一个不同的 CassIO 表抽象：

from cassio.table import ClusteredMetadataVectorCassandraTable

v_table_partitioned = ClusteredMetadataVectorCassandraTable(table="philosophers_cassio_partitioned", vector_dimension=1536)

现在在新表上重复计算嵌入并插入的步骤。

与您之前看到的相比，一个关键的区别在于，现在名言的作者被存储为插入行的_分区 ID_，而不是添加到通用的“元数据”字典中。

在您进行此操作的同时，作为演示，您将_并发_插入给定作者的所有名言：使用 CassIO，这可以通过使用每个名言的异步 put_async 方法来完成，收集由此产生的 Future 对象列表，然后调用它们的 result() 方法，以确保它们都已执行。Cassandra / Astra DB 非常支持高并发的 I/O 操作。

（注意：可以缓存先前计算的嵌入以节省一些 API 令牌——但是，在这里我们希望代码更容易检查。）

BATCH_SIZE = 50

num_batches = ((len(philo_dataset) + BATCH_SIZE - 1) // BATCH_SIZE)

quotes_list = philo_dataset["quote"]
authors_list = philo_dataset["author"]
tags_list = philo_dataset["tags"]

print("Starting to store entries:")
for batch_i in range(num_batches):
    b_start = batch_i * BATCH_SIZE
    b_end = (batch_i + 1) * BATCH_SIZE
    # compute the embedding vectors for this batch
    b_emb_results = client.embeddings.create(
        input=quotes_list[b_start : b_end],
        model=embedding_model_name,
    )
    # prepare the rows for insertion
    futures = []
    print("B ", end="")
    for entry_idx, emb_result in zip(range(b_start, b_end), b_emb_results.data):
        if tags_list[entry_idx]:
            tags = {
                tag
                for tag in tags_list[entry_idx].split(";")
            }
        else:
            tags = set()
        author = authors_list[entry_idx]
        quote = quotes_list[entry_idx]
        futures.append(v_table_partitioned.put_async(
            partition_id=author,
            row_id=f"q_{author}_{entry_idx}",
            body_blob=quote,
            vector=emb_result.embedding,
            metadata={tag: True for tag in tags},
        ))
    #
    for future in futures:
        future.result()
    #
    print(f" done ({len(b_emb_results.data)})")

print("\nFinished storing entries.")

Starting to store entries:
B  done (50)
B  done (50)
B  done (50)
B  done (50)
B  done (50)
B  done (50)
B  done (50)
B  done (50)
B  done (50)

Finished storing entries.

使用这个新表，相似性搜索也会相应地改变（注意 ann_search 的参数）：

def find_quote_and_author_p(query_quote, n, author=None, tags=None):
    query_vector = client.embeddings.create(
        input=[query_quote],
        model=embedding_model_name,
    ).data[0].embedding
    metadata = {}
    partition_id = None
    if author:
        partition_id = author
    if tags:
        for tag in tags:
            metadata[tag] = True
    #
    results = v_table_partitioned.ann_search(
        query_vector,
        n=n,
        partition_id=partition_id,
        metadata=metadata,
    )
    return [
        (result["body_blob"], result["partition_id"])
        for result in results
    ]

就是这样：新表仍然可以很好地支持“通用”相似性搜索……

find_quote_and_author_p("We struggle all our life for nothing", 3)

[('Life to the great majority is only a constant struggle for mere existence, with the certainty of losing it at last.',
  'schopenhauer'),
 ('We give up leisure in order that we may have leisure, just as we go to war in order that we may have peace.',
  'aristotle'),
 ('Perhaps the gods are kind to us, by making life more disagreeable as we grow older. In the end death seems less intolerable than the manifold burdens we carry',
  'freud')]

……但当指定作者时，您会注意到巨大的性能优势：

find_quote_and_author_p("We struggle all our life for nothing", 2, author="nietzsche")

[('To live is to suffer, to survive is to find some meaning in the suffering.',
  'nietzsche'),
 ('What makes us heroic?--Confronting simultaneously our supreme suffering and our supreme hope.',
  'nietzsche')]

好吧，如果您有一个真实规模的数据集，您确实会注意到性能的提升：在这个演示中，只有几十个条目，没有明显的区别——但您明白了。

结论

恭喜！您已经学会了如何使用 OpenAI 进行向量嵌入，以及使用 Cassandra / Astra DB through CQL 进行存储，以构建一个复杂的哲学搜索引擎和名言生成器。

本示例使用 CassIO 与向量存储进行接口——但这并非唯一的选择。请查看 README 以了解其他选项和与流行框架的集成。

要了解有关 Astra DB 的向量搜索功能如何成为您 ML/GenAI 应用程序的关键要素的更多信息，请访问 Astra DB 上的相关主题网页。

清理

如果您想删除此演示使用的所有资源，请运行此单元格（警告：这将删除表和其中插入的数据！）：

# we peek at CassIO's config to get a direct handle to the DB connection
session = cassio.config.resolve_session()
keyspace = cassio.config.resolve_keyspace()

session.execute(f"DROP TABLE IF EXISTS {keyspace}.philosophers_cassio;")
session.execute(f"DROP TABLE IF EXISTS {keyspace}.philosophers_cassio_partitioned;")

<cassandra.cluster.ResultSet at 0x7fdcc42e8f10>