Snowflake 搜索工具

SnowflakeSearchTool 使 CrewAI 智能体能够在 Snowflake 数据仓库上执行 SQL 查询并进行语义搜索。

SnowflakeSearchTool

描述

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

安装

要使用此工具，你需要安装所需的依赖项：

uv add cryptography snowflake-connector-python snowflake-sqlalchemy

或者另一种方式：

uv sync --extra snowflake

入门步骤

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

1. 安装依赖：使用上述命令之一安装所需的包。
2. 配置 Snowflake 连接：使用你的 Snowflake 凭据创建 SnowflakeConfig 对象。
3. 初始化工具：使用必要的配置创建该工具的实例。
4. 执行查询：使用该工具对你的 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 账户标识符。
• user：必填。Snowflake 用户名。
• password：可选*。Snowflake 密码。
• private_key_path：可选*。私钥文件路径（替代密码）。
• warehouse：必填。Snowflake 仓库名称。
• database：必填。默认数据库。
• snowflake_schema：必填。默认 schema。
• role：可选。Snowflake 角色。
• session_parameters：可选。以字典形式提供的自定义会话参数。

*必须提供 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：可选。覆盖配置中指定的默认数据库。
• snowflake_schema：可选。覆盖配置中指定的默认 schema。
• timeout：可选。查询超时时间（秒）。默认为 300。

该工具会将查询结果作为字典列表返回，其中每个字典表示一行，列名作为键。

# 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 智能体集成。借助连接池、自动重试和查询缓存等功能，它可以高效、可靠地访问企业数据。对于需要访问存储在 Snowflake 中的结构化数据的分析、报告和商业智能任务，该工具尤其有用。