DuckDB Agent

DuckDB database tools for AI agents.

Prerequisites

• Python 3.10 or higher
• PraisonAI Agents package installed
• PraisonAI Tools package installed
• duckdb package installed

from praisonaiagents import Agent
from praisonai_tools import execute_query

agent = Agent(name="Analyst", tools=[execute_query])
agent.start("How many rows are in the orders table?")

The user asks an analytics question; the agent runs DuckDB queries and returns results.

DuckDB Tools

Use DuckDB Tools to manage and query databases with AI agents.

1. Install Dependencies

   First, install the required packages:

   pip install praisonaiagents praisonai-tools duckdb
   
2. Import Components

   Import the necessary components:

   from praisonaiagents import Agent, Task, AgentTeam
   from praisonai_tools import execute_query, load_csv, export_csv
   
3. Create Agent

   Create a DuckDB database agent:

   db_agent = Agent(
       name="DBProcessor",
       role="Database Management Specialist",
       goal="Manage and query databases efficiently.",
       backstory="Expert in database operations and SQL.",
       tools=[execute_query, load_csv, export_csv],
       reflection=False
   )
   
4. Define Task

   Define the database task:

   db_task = Task(
       description="Query and analyze database data.",
       expected_output="Query results and analysis.",
       agent=db_agent,
       name="db_analysis"
   )
   
5. Run Agent

   Initialize and run the agent:

   agents = AgentTeam(
       agents=[db_agent],
       tasks=[db_task],
       process="sequential"
   )
   agents.start()
   

Available Functions

from praisonai_tools import execute_query
from praisonai_tools import load_csv
from praisonai_tools import export_csv

Function Details

execute_query(query: str, params: Optional[Union[tuple, dict]] = None, return_df: bool = True)

Executes SQL queries with advanced features:

• Supports parameterized queries
• Returns results as DataFrame records or raw tuples
• Full SQL query support (SELECT, INSERT, UPDATE, DELETE, etc.)
• Automatic connection management

from praisonai_tools import execute_query

# Basic SELECT query
results = execute_query("SELECT * FROM employees")

# Parameterized query
results = execute_query(
    "SELECT * FROM employees WHERE department = ? AND salary > ?",
    params=('Engineering', 75000)
)

# Named parameters
results = execute_query(
    "SELECT * FROM employees WHERE department = :dept",
    params={'dept': 'Engineering'}
)
# Returns: List[Dict[str, Any]]

load_csv(table_name: str, filepath: str, schema: Optional[Dict[str, str]] = None, if_exists: str = ‘replace’)

Loads CSV files into DuckDB tables:

• Optional schema definition
• Flexible table existence handling
• Automatic type inference
• Support for large files

from praisonai_tools import load_csv

# Basic usage - auto schema inference
success = load_csv("employees", "employees.csv")

# With custom schema
schema = {
    "id": "INTEGER PRIMARY KEY",
    "name": "VARCHAR",
    "salary": "DECIMAL(10,2)",
    "hire_date": "DATE"
}
success = load_csv(
    "employees",
    "employees.csv",
    schema=schema,
    if_exists='replace'
)
# Returns: bool (True if successful)

export_csv(query: str, filepath: str, params: Optional[Union[tuple, dict]] = None)

Exports query results to CSV files:

• Supports parameterized queries
• Automatic header generation
• Configurable output formatting
• Large result set handling

from praisonai_tools import export_csv

# Export simple query results
success = export_csv(
    "SELECT * FROM employees",
    "exported_employees.csv"
)

# Export filtered data with parameters
success = export_csv(
    "SELECT * FROM employees WHERE department = ? AND year = ?",
    "eng_2023.csv",
    params=('Engineering', 2023)
)
# Returns: bool (True if successful)

Understanding DuckDB Tools

What are DuckDB Tools?

DuckDB Tools provide database capabilities for AI agents:

• SQL query execution
• Data import/export
• Schema management
• Data analysis
• Performance optimization

Key Components

DB Agent

Create specialized database agents:

Agent(tools=[execute_query, load_csv, export_csv])

DB Task

Define database tasks:

Task(description="db_operation")

Process Types

Sequential or parallel processing:

process="sequential"

DB Options

Customize database parameters:

read_only=True, memory=True

Examples

Basic Database Agent

from praisonaiagents import Agent, Task, AgentTeam
from praisonai_tools import execute_query, load_csv, export_csv

# Create database agent
db_agent = Agent(
    name="DBExpert",
    role="Database Specialist",
    goal="Query and analyze data efficiently.",
    backstory="Expert in database management and SQL.",
    tools=[execute_query, load_csv, export_csv],
    reflection=False
)

# Define database task
db_task = Task(
    description="Analyze sales performance data.",
    expected_output="Sales analysis report.",
    agent=db_agent,
    name="sales_analysis"
)

# Run agent
agents = AgentTeam(
    agents=[db_agent],
    tasks=[db_task],
    process="sequential"
)
agents.start()

Advanced Database Operations with Multiple Agents

from praisonaiagents import Agent, Task, AgentTeam
from praisonai_tools import execute_query, load_csv, export_csv

# Create query agent
query_agent = Agent(
    name="QueryProcessor",
    role="SQL Query Specialist",
    goal="Execute SQL queries efficiently.",
    tools=[execute_query],
    reflection=False
)

# Create data import/export agent
data_agent = Agent(
    name="DataProcessor",
    role="Data Import/Export Specialist",
    goal="Handle data import and export operations efficiently.",
    tools=[load_csv, export_csv],
    reflection=False
)

# Define tasks
query_task = Task(
    description="Query sales data",
    agent=query_agent,
    name="query_task"
)

data_task = Task(
    description="Export query results",
    agent=data_agent,
    name="data_task"
)

# Run agents
agents = AgentTeam(
    agents=[query_agent, data_agent],
    tasks=[query_task, data_task],
    process="sequential"
)
agents.start()

Best Practices

Agent Configuration

Configure agents with clear database focus:

from praisonai_tools import execute_query, load_csv, export_csv

Agent(
    name="DBProcessor",
    role="Database Specialist",
    goal="Process queries accurately and efficiently",
    tools=[execute_query, load_csv, export_csv]
)

Task Definition

Define specific database operations:

Task(
    description="Query sales data and generate reports",
    expected_output="Sales performance report"
)

Common Patterns

Database Operation Pipeline

from praisonaiagents import Agent, Task, AgentTeam
from praisonai_tools import execute_query, load_csv, export_csv

# Query agent
querier = Agent(
    name="Querier",
    role="SQL Specialist",
    tools=[execute_query]
)

# Analysis agent
analyzer = Agent(
    name="Analyzer",
    role="Data Analyst"
)

# Define tasks
query_task = Task(
    description="Execute SQL queries",
    agent=querier
)

analyze_task = Task(
    description="Analyze query results",
    agent=analyzer
)

# Run workflow
agents = AgentTeam(
    agents=[querier, analyzer],
    tasks=[query_task, analyze_task]
)

Related

Custom Tools

Build your own agent tools

Tools Overview

Browse PraisonAI tool documentation