Skip to main content
Prerequisites
  • Python 3.10 or higher
  • PraisonAI Agents package installed
  • PraisonAI Tools package installed
  • duckdb package installed

​
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, Agents
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 = Agents(
    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, Agents
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 = Agents(
    agents=[db_agent],
    tasks=[db_task],
    process="sequential"
)
agents.start()

​
Advanced Database Operations with Multiple Agents

from praisonaiagents import Agent, Task, Agents
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 = Agents(
    agents=[query_agent, data_agent],
    tasks=[query_task, data_task],
    process="sequential"
)
agents.start()

​
Best Practices

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]
)
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, Agents
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 = Agents(
    agents=[querier, analyzer],
    tasks=[query_task, analyze_task]
)