Cassandra 数据库工具包集成

Apache Cassandra® 是一种广泛使用的数据库，用于存储事务性应用数据。大型语言模型中函数和工具的引入，为生成式 AI 应用中现有数据开辟了一些令人兴奋的用例。

Cassandra 数据库工具包使 AI 工程师能够高效地将代理与 Cassandra 数据集成，提供以下功能：

通过优化查询实现快速数据访问。大多数查询应在个位数毫秒或更短时间内运行。

模式自省以增强 LLM 推理能力

兼容各种 Cassandra 部署，包括 Apache Cassandra®、DataStax Enterprise™ 和 DataStax Astra™

目前，该工具包仅限于 SELECT 查询和模式自省操作。（安全第一）

有关创建 Cassandra DB 代理的更多信息，请参阅 CQL 代理手册

快速开始

安装 cassio 库
为要连接的 Cassandra 数据库设置环境变量
初始化 CassandraDatabase
使用 toolkit.get_tools() 将工具传递给您的代理
坐下来，看着它为您完成所有工作

工作原理

Cassandra 查询语言 (CQL) 是与 Cassandra 数据库交互的主要以人为中心的方式。虽然在生成查询时提供了一定的灵活性，但它需要了解 Cassandra 数据建模的最佳实践。LLM 函数调用赋予代理推理并选择工具以满足请求的能力。使用 LLM 的代理在选择适当的工具包或工具链时，应使用 Cassandra 特定的逻辑进行推理。这减少了当 LLM 被迫提供自上而下解决方案时引入的随机性。您希望 LLM 拥有完全不受限制的数据库访问权限吗？嗯，可能不希望。为此，我们提供了一个提示，用于在为代理构建问题时使用：您是一个 Apache Cassandra 专家查询分析机器人，具有以下特性和规则：

您将从最终用户那里获取一个关于在数据库中查找特定数据的问题。
您将检查数据库的模式并创建查询路径。
您将为用户提供正确的查询以找到他们正在寻找的数据，显示查询路径提供的步骤。
您将使用使用分区键和聚簇列查询 Apache Cassandra 的最佳实践。
避免在查询中使用 ALLOW FILTERING。
目标是找到查询路径，因此可能需要查询其他表才能得到最终答案。

以下是 JSON 格式的查询路径示例：

 {
  "query_paths": [
    {
      "description": "使用电子邮件直接查询用户表",
      "steps": [
        {
          "table": "user_credentials",
          "query":
             "SELECT userid FROM user_credentials WHERE email = 'example@example.com';"
        },
        {
          "table": "users",
          "query": "SELECT * FROM users WHERE userid = ?;"
        }
      ]
    }
  ]
}

提供的工具

`cassandra_db_schema`

收集已连接数据库或特定模式的所有模式信息。对于代理确定操作至关重要。

`cassandra_db_select_table_data`

从特定键空间和表中选择数据。代理可以传递谓词参数和返回记录数量的限制。

`cassandra_db_query`

cassandra_db_select_table_data 的实验性替代方案，它接受由代理完全构建的查询字符串，而不是参数。警告：这可能导致不寻常的查询，这些查询可能性能不佳（甚至无法工作）。这可能会在未来的版本中移除。如果它做了什么很酷的事情，我们也想了解。您永远不知道！

环境设置

安装以下 Python 模块：

pip install ipykernel python-dotenv cassio langchain-openai langchain langchain-community langchainhub

.env 文件

连接通过 cassio 使用 auto=True 参数进行，笔记本使用 OpenAI。您应相应地创建一个 .env 文件。对于 Cassandra，设置：

CASSANDRA_CONTACT_POINTS
CASSANDRA_USERNAME
CASSANDRA_PASSWORD
CASSANDRA_KEYSPACE

对于 Astra，设置：

ASTRA_DB_APPLICATION_TOKEN
ASTRA_DB_DATABASE_ID
ASTRA_DB_KEYSPACE

例如：

# 连接到 Astra：
ASTRA_DB_DATABASE_ID=a1b2c3d4-...
ASTRA_DB_APPLICATION_TOKEN=AstraCS:...
ASTRA_DB_KEYSPACE=notebooks

# 同时设置
OPENAI_API_KEY=sk-....

（您也可以修改下面的代码以直接使用 cassio 连接。）

from dotenv import load_dotenv

load_dotenv(override=True)

# 导入必要的库
import os

import cassio
from langchain_classic import hub
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_community.agent_toolkits.cassandra_database.toolkit import (
    CassandraDatabaseToolkit,
)
from langchain_community.tools.cassandra_database.prompt import QUERY_PATH_PROMPT
from langchain_community.utilities.cassandra_database import CassandraDatabase
from langchain_openai import ChatOpenAI

连接到 Cassandra 数据库

cassio.init(auto=True)
session = cassio.config.resolve_session()
if not session:
    raise Exception(
        "检查环境配置或手动配置 cassio 连接参数"
    )

# 测试数据准备

session = cassio.config.resolve_session()

session.execute("""DROP KEYSPACE IF EXISTS langchain_agent_test; """)

session.execute(
    """
CREATE KEYSPACE if not exists langchain_agent_test
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
"""
)

session.execute(
    """
    CREATE TABLE IF NOT EXISTS langchain_agent_test.user_credentials (
    user_email text PRIMARY KEY,
    user_id UUID,
    password TEXT
);
"""
)

session.execute(
    """
    CREATE TABLE IF NOT EXISTS langchain_agent_test.users (
    id UUID PRIMARY KEY,
    name TEXT,
    email TEXT
);"""
)

session.execute(
    """
    CREATE TABLE IF NOT EXISTS langchain_agent_test.user_videos (
    user_id UUID,
    video_id UUID,
    title TEXT,
    description TEXT,
    PRIMARY KEY (user_id, video_id)
);
"""
)

user_id = "522b1fe2-2e36-4cef-a667-cd4237d08b89"
video_id = "27066014-bad7-9f58-5a30-f63fe03718f6"

session.execute(
    f"""
    INSERT INTO langchain_agent_test.user_credentials (user_id, user_email)
    VALUES ({user_id}, 'patrick@datastax.com');
"""
)

session.execute(
    f"""
    INSERT INTO langchain_agent_test.users (id, name, email)
    VALUES ({user_id}, 'Patrick McFadin', 'patrick@datastax.com');
"""
)

session.execute(
    f"""
    INSERT INTO langchain_agent_test.user_videos (user_id, video_id, title)
    VALUES ({user_id}, {video_id}, 'Use Langflow to Build a LangChain LLM Application in 5 Minutes');
"""
)

session.set_keyspace("langchain_agent_test")

# 创建 CassandraDatabase 实例
# 使用 cassio 会话连接到数据库
db = CassandraDatabase()

# 选择将驱动代理的 LLM
# 只有某些模型支持此功能
llm = ChatOpenAI(temperature=0, model="gpt-4-1106-preview")
toolkit = CassandraDatabaseToolkit(db=db)

tools = toolkit.get_tools()

print("可用工具：")
for tool in tools:
    print(tool.name + "\t- " + tool.description)

可用工具：
cassandra_db_schema -
    此工具的输入是键空间名称，输出是 Apache Cassandra 表的表描述。
    如果查询不正确，将返回错误消息。
    如果返回错误，请向用户报告键空间不存在并停止。

cassandra_db_query -
    对数据库执行 CQL 查询并获取结果。
    如果查询不正确，将返回错误消息。
    如果返回错误，请重写查询，检查查询，然后重试。

cassandra_db_select_table_data -
    用于从 Apache Cassandra 数据库中的表获取数据的工具。
    使用 WHERE 子句指定使用主键的查询谓词。空白谓词将返回所有行。尽可能避免这种情况。
    使用 limit 指定要返回的行数。空白限制将返回所有行。

prompt = hub.pull("hwchase17/openai-tools-agent")

# 构建 OpenAI Tools 代理
agent = create_openai_tools_agent(llm, tools, prompt)

input = (
    QUERY_PATH_PROMPT
    + "\n\n以下是您的任务：查找电子邮件地址为 'patrick@datastax.com' 的用户上传到 langchain_agent_test 键空间的所有视频。"
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

response = agent_executor.invoke({"input": input})

print(response["output"])

> 进入新的 AgentExecutor 链...

调用：`cassandra_db_schema`，参数为 `{'keyspace': 'langchain_agent_test'}`


表名：user_credentials
- 键空间：langchain_agent_test
- 列
  - password (text)
  - user_email (text)
  - user_id (uuid)
- 分区键：(user_email)
- 聚簇键：

表名：user_videos
- 键空间：langchain_agent_test
- 列
  - description (text)
  - title (text)
  - user_id (uuid)
  - video_id (uuid)
- 分区键：(user_id)
- 聚簇键：(video_id asc)


表名：users
- 键空间：langchain_agent_test
- 列
  - email (text)
  - id (uuid)
  - name (text)
- 分区键：(id)
- 聚簇键：


调用：`cassandra_db_select_table_data`，参数为 `{'keyspace': 'langchain_agent_test', 'table': 'user_credentials', 'predicate': "user_email = 'patrick@datastax.com'", 'limit': 1}`


Row(user_email='patrick@datastax.com', password=None, user_id=UUID('522b1fe2-2e36-4cef-a667-cd4237d08b89'))
调用：`cassandra_db_select_table_data`，参数为 `{'keyspace': 'langchain_agent_test', 'table': 'user_videos', 'predicate': 'user_id = 522b1fe2-2e36-4cef-a667-cd4237d08b89', 'limit': 10}`


Row(user_id=UUID('522b1fe2-2e36-4cef-a667-cd4237d08b89'), video_id=UUID('27066014-bad7-9f58-5a30-f63fe03718f6'), description='DataStax Academy is a free resource for learning Apache Cassandra.', title='DataStax Academy')要查找电子邮件地址为 'patrick@datastax.com' 的用户上传到 `langchain_agent_test` 键空间的所有视频，我们可以按照以下步骤操作：

1. 查询 `user_credentials` 表以查找与电子邮件 'patrick@datastax.com' 关联的 `user_id`。
2. 使用第一步获得的 `user_id` 查询 `user_videos` 表以检索用户上传的所有视频。

以下是 JSON 格式的查询路径：

\`\`\`json
{
  "query_paths": [
    {
      "description": "从 user_credentials 查找 user_id，然后查询 user_videos 获取用户上传的所有视频",
      "steps": [
        {
          "table": "user_credentials",
          "query": "SELECT user_id FROM user_credentials WHERE user_email = 'patrick@datastax.com';"
        },
        {
          "table": "user_videos",
          "query": "SELECT * FROM user_videos WHERE user_id = 522b1fe2-2e36-4cef-a667-cd4237d08b89;"
        }
      ]
    }
  ]
}
\`\`\`

遵循此查询路径，我们发现 user_id 为 `522b1fe2-2e36-4cef-a667-cd4237d08b89` 的用户至少上传了一个标题为 'DataStax Academy'、描述为 'DataStax Academy is a free resource for learning Apache Cassandra.' 的视频。此视频的 video_id 为 `27066014-bad7-9f58-5a30-f63fe03718f6`。如果有更多视频，可以使用相同的查询来检索它们，必要时可能需要增加限制。

> 链接完成。
要查找电子邮件地址为 'patrick@datastax.com' 的用户上传到 `langchain_agent_test` 键空间的所有视频，我们可以按照以下步骤操作：

1. 查询 `user_credentials` 表以查找与电子邮件 'patrick@datastax.com' 关联的 `user_id`。
2. 使用第一步获得的 `user_id` 查询 `user_videos` 表以检索用户上传的所有视频。

以下是 JSON 格式的查询路径：

\`\`\`json
{
  "query_paths": [
    {
      "description": "从 user_credentials 查找 user_id，然后查询 user_videos 获取用户上传的所有视频",
      "steps": [
        {
          "table": "user_credentials",
          "query": "SELECT user_id FROM user_credentials WHERE user_email = 'patrick@datastax.com';"
        },
        {
          "table": "user_videos",
          "query": "SELECT * FROM user_videos WHERE user_id = 522b1fe2-2e36-4cef-a667-cd4237d08b89;"
        }
      ]
    }
  ]
}
\`\`\`

遵循此查询路径，我们发现 user_id 为 `522b1fe2-2e36-4cef-a667-cd4237d08b89` 的用户至少上传了一个标题为 'DataStax Academy'、描述为 'DataStax Academy is a free resource for learning Apache Cassandra.' 的视频。此视频的 video_id 为 `27066014-bad7-9f58-5a30-f63fe03718f6`。如果有更多视频，可以使用相同的查询来检索它们，必要时可能需要增加限制。

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

​快速开始

​工作原理

​提供的工具

​cassandra_db_schema

​cassandra_db_select_table_data

​cassandra_db_query

​环境设置

​.env 文件

​连接到 Cassandra 数据库

快速开始

工作原理

提供的工具

`cassandra_db_schema`

`cassandra_db_select_table_data`

`cassandra_db_query`

环境设置

.env 文件

连接到 Cassandra 数据库