Cassandra 数据库工具包集成

使用 LangChain Python 与 Cassandra 数据库工具包集成。

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

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

• 通过优化查询实现快速数据访问。大多数查询应在个位数毫秒或更短时间内运行。
• 模式检查以增强 LLM 推理能力
• 兼容各种 Cassandra 部署，包括 Apache Cassandra®、DataStax Enterprise™ 和 DataStax Astra™
• 目前，该工具包仅限于 SELECT 查询和模式检查操作。（安全第一）

有关创建 Cassandra DB 代理的更多信息，请参阅 CQL 代理示例代码库

快速开始

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

运作原理

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

你是一位 Apache Cassandra 专家查询分析机器人，具有以下功能和规则：

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

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

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

提供的工具

cassandra_db_schema

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

cassandra_db_select_table_data

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

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 文件。

对于 Casssandra，设置：

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}, '[e-mail]');
"""
)

session.execute(
    f"""
    INSERT INTO langchain_agent_test.users (id, name, email)
    VALUES ({user_id}, 'Patrick McFadin', '[e-mail]');
"""
)

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 -
    此工具的输入是 keyspace 名称，输出是 Apache Cassandra 表的表描述
    如果查询不正确，将返回错误消息。
    如果返回错误，请向用户报告 keyspace
    不存在并停止。

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

cassandra_db_select_table_data -
    从 Apache Cassandra 数据库的表中获取数据的工具。
    使用 WHERE 子句指定使用主键的查询的谓词。空白谓词将返回所有行。如果可能请避免这样做。
    使用 limit 指定要返回的行数。空白 limit 将返回所有行。

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

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

input = (
    QUERY_PATH_PROMPT
    + "\n\nHere is your task: Find all the videos that the user with the email address '[e-mail]' has uploaded to the langchain_agent_test keyspace."
)

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
- Keyspace: langchain_agent_test
- 列
  - password (text)
  - user_email (text)
  - user_id (uuid)
- 分区键：(user_email)
- 聚簇键：

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


表名：users
- Keyspace: 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 = '[e-mail]'", 'limit': 1}`


Row(user_email='[e-mail]', 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')为了找到邮箱地址为 '[e-mail]' 的用户上传到 `langchain_agent_test` keyspace 的所有视频，我们可以遵循以下步骤：

1. 查询 `user_credentials` 表以找到与电子邮件 '[e-mail]' 关联的 `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 = '[e-mail]';"
        },
        {
          "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.' 的视频。该视频的视频 ID 为 `27066014-bad7-9f58-5a30-f63fe03718f6`。如果有更多视频，可以使用相同的查询来检索它们，如有必要，可以增加限制。

> 完成链。
为了找到邮箱地址为 '[e-mail]' 的用户上传到 `langchain_agent_test` keyspace 的所有视频，我们可以遵循以下步骤：

1. 查询 `user_credentials` 表以找到与电子邮件 '[e-mail]' 关联的 `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 = '[e-mail]';"
        },
        {
          "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.' 的视频。该视频的视频 ID 为 `27066014-bad7-9f58-5a30-f63fe03718f6`。如果有更多视频，可以使用相同的查询来检索它们，如有必要，可以增加限制。

---

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.