Snowflake Search Tool

`SnowflakeSearchTool`

描述

SnowflakeSearchTool 专为连接 Snowflake 数据仓库并执行 SQL 查询而设计，具备连接池、重试逻辑和异步执行等高级特性。这个工具让 CrewAI agents 能与 Snowflake 数据库交互，非常适合需要访问存储在 Snowflake 中的企业数据的数据分析、报表和商业智能任务。

安装

要使用这个工具，需要安装所需依赖：

uv add cryptography snowflake-connector-python snowflake-sqlalchemy

或者：

uv sync --extra snowflake

入门步骤

要有效使用 SnowflakeSearchTool，请按以下步骤操作：

安装依赖：使用上面的命令之一安装所需包。
配置 Snowflake 连接：使用你的 Snowflake 凭据创建 SnowflakeConfig 对象。
初始化工具：使用所需配置创建工具实例。
执行查询：使用该工具在 Snowflake 数据库上运行 SQL 查询。

示例

下面的示例展示了如何使用 SnowflakeSearchTool 查询 Snowflake 数据库中的数据：

from crewai import Agent, Task, Crew
from crewai_tools import SnowflakeSearchTool, SnowflakeConfig

# Create Snowflake configuration
config = SnowflakeConfig(
    account="your_account",
    user="your_username",
    password="your_password",
    warehouse="COMPUTE_WH",
    database="your_database",
    snowflake_schema="your_schema"
)

# Initialize the tool
snowflake_tool = SnowflakeSearchTool(config=config)

# Define an agent that uses the tool
data_analyst_agent = Agent(
    role="Data Analyst",
    goal="Analyze data from Snowflake database",
    backstory="An expert data analyst who can extract insights from enterprise data.",
    tools=[snowflake_tool],
    verbose=True,
)

# Example task to query sales data
query_task = Task(
    description="Query the sales data for the last quarter and summarize the top 5 products by revenue.",
    expected_output="A summary of the top 5 products by revenue for the last quarter.",
    agent=data_analyst_agent,
)

# Create and run the crew
crew = Crew(agents=[data_analyst_agent],
            tasks=[query_task])
result = crew.kickoff()

你也可以用额外参数自定义该工具：

# Initialize the tool with custom parameters
snowflake_tool = SnowflakeSearchTool(
    config=config,
    pool_size=10,
    max_retries=5,
    retry_delay=2.0,
    enable_caching=True
)

参数

SnowflakeConfig 参数

SnowflakeConfig 类接受以下参数：

account：必填。Snowflake account 标识符。
user：必填。Snowflake 用户名。
password：可选*。Snowflake 密码。
private_key_path：可选*。私钥文件路径（可替代 password）。
warehouse：必填。Snowflake warehouse 名称。
database：必填。默认数据库。
snowflake_schema：必填。默认 schema。
role：可选。Snowflake 角色。
session_parameters：可选。自定义 session 参数字典。

*必须提供 password 或 private_key_path 二者之一。

SnowflakeSearchTool 参数

SnowflakeSearchTool 在初始化时接受以下参数：

config：必填。包含连接详情的 SnowflakeConfig 对象。
pool_size：可选。连接池中的连接数量。默认值为 5。
max_retries：可选。查询失败时的最大重试次数。默认值为 3。
retry_delay：可选。重试之间的延迟时间（秒）。默认值为 1.0。
enable_caching：可选。是否启用查询结果缓存。默认值为 True。

用法

使用 SnowflakeSearchTool 时，你需要提供以下参数：

query：必填。要执行的 SQL 查询。
database：可选。覆盖 config 中指定的默认数据库。
snowflake_schema：可选。覆盖 config 中指定的默认 schema。
timeout：可选。查询超时时间（秒）。默认值为 300。

工具会把查询结果作为字典列表返回，其中每个字典代表一行，列名作为 key。

# Example of using the tool with an agent
data_analyst = Agent(
    role="Data Analyst",
    goal="Analyze sales data from Snowflake",
    backstory="An expert data analyst with experience in SQL and data visualization.",
    tools=[snowflake_tool],
    verbose=True
)

# The agent will use the tool with parameters like:
# query="SELECT product_name, SUM(revenue) as total_revenue FROM sales GROUP BY product_name ORDER BY total_revenue DESC LIMIT 5"
# timeout=600

# Create a task for the agent
analysis_task = Task(
    description="Query the sales database and identify the top 5 products by revenue for the last quarter.",
    expected_output="A detailed analysis of the top 5 products by revenue.",
    agent=data_analyst
)

# Run the task
crew = Crew(
    agents=[data_analyst],
    tasks=[analysis_task]
)
result = crew.kickoff()

高级特性

连接池

SnowflakeSearchTool 通过复用数据库连接来提升性能。你可以使用 pool_size 参数控制连接池大小。

自动重试

该工具会使用指数退避自动重试失败的查询。你可以通过 max_retries 和 retry_delay 参数配置重试行为。

查询结果缓存

为了提升重复查询的性能，该工具可以缓存查询结果。该功能默认启用，但可以通过设置 enable_caching=False 禁用。

密钥对认证

除密码认证外，该工具还支持密钥对认证以增强安全性：

config = SnowflakeConfig(
    account="your_account",
    user="your_username",
    private_key_path="/path/to/your/private/key.p8",
    warehouse="COMPUTE_WH",
    database="your_database",
    snowflake_schema="your_schema"
)

错误处理

SnowflakeSearchTool 对常见的 Snowflake 问题提供了完整的错误处理：

连接失败
查询超时
认证错误
数据库和 schema 错误

发生错误时，工具会尝试重试该操作（如果已配置），并提供详细错误信息。

结论

SnowflakeSearchTool 提供了一种强大的方式，可将 Snowflake 数据仓库与 CrewAI agents 集成。凭借连接池、自动重试和查询缓存等功能，它能够高效且可靠地访问企业数据。这个工具对于需要访问存储在 Snowflake 中的结构化数据的数据分析、报表和商业智能任务尤其有用。