Praison AI

DuckDB Agent

DuckDB database tools for AI agents.

python
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.

Diagram

Input

Agent

Tool

Output

Diagram
DuckDB AgentAgentUserDuckDB AgentAgentUserRequestProcess requestResultResponse

DuckDB Tools

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

  1. Install Dependencies

    First, install the required packages:

    bash
    pip install praisonaiagents praisonai-tools duckdb
  2. Import Components

    Import the necessary components:

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

    Create a DuckDB database agent:

    python
    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:

    python
    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:

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

Available Functions

python
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
python
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
python
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
python
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:

python
Agent(tools=[execute_query, load_csv, export_csv])
 DB Task

Define database tasks:

python
Task(description="db_operation")
 Process Types

Sequential or parallel processing:

python
process="sequential"
 DB Options

Customize database parameters:

python
read_only=True, memory=True

Examples

Basic Database Agent

python
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

python
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:

python
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:

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

Common Patterns

Database Operation Pipeline

python
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