Skip to main content
Apache Cassandra® 是一种广泛用于存储事务性应用程序数据的数据库。大型语言模型中函数和工具的引入为生成式人工智能应用程序中的现有数据开辟了一些令人兴奋的用例。
Cassandra Database 工具包使 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 子句指定使用主键的查询谓词。空白谓词将返回所有行。如果可能，请避免这种情况。
    使用限制指定要返回的行数。空白限制将返回所有行。

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;"
        }
      ]
    }
  ]
}
\`\`\`

遵循此查询路径，我们发现用户 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;"
        }
      ]
    }
  ]
}
\`\`\`

遵循此查询路径，我们发现用户 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`。如果有更多视频，可以使用相同的查询来检索它们，如果需要，可以增加限制。
在 GitHub 上编辑此页面 或 提交问题。
连接这些文档 到 Claude、VSCode 等，通过 MCP 获取实时答案。