---
title: "DuckDB Agent"
description: "DuckDB database tools for AI agents."
icon: "database"
---

<Note>
  **Prerequisites**
  - Python 3.10 or higher
  - PraisonAI Agents package installed
  - `duckdb` package installed
  - Basic understanding of SQL and databases
</Note>

## DuckDB Tools

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

<Steps>
  <Step title="Install Dependencies">
    First, install the required packages:
    ```bash
    pip install praisonaiagents duckdb
    ```
  </Step>

  <Step title="Import Components">
    Import the necessary components:
    ```python
    from praisonaiagents import Agent, Task, PraisonAIAgents
    from praisonaiagents.tools import execute_query, load_csv, export_csv
    ```
  </Step>

  <Step title="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],
        self_reflect=False
    )
    ```
  </Step>

  <Step title="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"
    )
    ```
  </Step>

  <Step title="Run Agent">
    Initialize and run the agent:
    ```python
    agents = PraisonAIAgents(
        agents=[db_agent],
        tasks=[db_task],
        process="sequential"
    )
    agents.start()
    ```
  </Step>
</Steps>

## Available Functions

```python
from praisonaiagents.tools import execute_query
from praisonaiagents.tools import load_csv
from praisonaiagents.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
# 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]]
# Example: [{"id": 1, "name": "Alice", "department": "Engineering"}, ...]
```

### 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
# 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
# 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

<Card title="What are DuckDB Tools?" icon="question">
  DuckDB Tools provide database capabilities for AI agents:
  - SQL query execution
  - Data import/export
  - Schema management
  - Data analysis
  - Performance optimization
</Card>

## Key Components

<CardGroup cols={2}>
  <Card title="DB Agent" icon="user-robot">
    Create specialized database agents:
    ```python
    Agent(tools=[execute_query, load_csv, export_csv])
    ```
  </Card>
  <Card title="DB Task" icon="list-check">
    Define database tasks:
    ```python
    Task(description="db_operation")
    ```
  </Card>
  <Card title="Process Types" icon="arrows-split-up-and-left">
    Sequential or parallel processing:
    ```python
    process="sequential"
    ```
  </Card>
  <Card title="DB Options" icon="sliders">
    Customize database parameters:
    ```python
    read_only=True, memory=True
    ```
  </Card>
</CardGroup>

## Examples

### Basic Database Agent

```python
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents.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],
    self_reflect=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 = PraisonAIAgents(
    agents=[db_agent],
    tasks=[db_task],
    process="sequential"
)
agents.start()
```

### Advanced Database Operations with Multiple Agents

```python
# Create query agent
query_agent = Agent(
    name="QueryProcessor",
    role="SQL Query Specialist",
    goal="Execute SQL queries efficiently.",
    tools=[execute_query],
    self_reflect=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],
    self_reflect=False
)
```

## Dependencies

The DuckDB tools require the following Python packages:
- duckdb: For database operations
- pandas: For data manipulation and CSV handling

These will be automatically installed when needed.

## Example Agent Configuration

```python
from praisonaiagents import Agent
from praisonaiagents.tools import execute_query, load_csv, export_csv

agent = Agent(
    name="DBProcessor",
    description="An agent that works with DuckDB databases",
    tools=[execute_query, load_csv, export_csv]
)
```

## Error Handling

All functions include comprehensive error handling:
- Database connection errors
- Query syntax errors
- File I/O errors
- Schema validation errors
- Missing dependency errors

Errors are logged and returned in a consistent format:
- Success cases return the expected data type
- Error cases return a dict with an "error" key containing the error message

## Common Use Cases

1. Data Analysis:
```python
# Load data
load_csv("sales", "sales_data.csv")

# Analyze with SQL
results = execute_query("""
    SELECT 
        department,
        COUNT(*) as count,
        AVG(salary) as avg_salary
    FROM employees
    GROUP BY department
    HAVING count > 5
    ORDER BY avg_salary DESC
""")
```

2. Data Export:
```python
# Export filtered and aggregated data
export_csv("""
    SELECT 
        date_trunc('month', date) as month,
        SUM(amount) as total_sales
    FROM sales
    GROUP BY 1
    ORDER BY 1
""", "monthly_sales.csv")
```

3. Data Transformation:
```python
# Load and transform data
load_csv("raw_data", "input.csv")
execute_query("""
    CREATE TABLE transformed AS
    SELECT 
        id,
        UPPER(name) as name,
        ROUND(amount, 2) as amount,
        DATE_TRUNC('day', timestamp) as date
    FROM raw_data
""")
export_csv("SELECT * FROM transformed", "transformed.csv")
```

## Best Practices

<AccordionGroup>
  <Accordion title="Agent Configuration">
    Configure agents with clear database focus:
    ```python
    Agent(
        name="DBProcessor",
        role="Database Specialist",
        goal="Process queries accurately and efficiently",
        tools=[execute_query, load_csv, export_csv]
    )
    ```
  </Accordion>

  <Accordion title="Task Definition">
    Define specific database operations:
    ```python
    Task(
        description="Query sales data and generate reports",
        expected_output="Sales performance report"
    )
    ```
  </Accordion>
</AccordionGroup>

## Common Patterns

### Database Operation Pipeline
```python
# 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 = PraisonAIAgents(
    agents=[querier, analyzer],
    tasks=[query_task, analyze_task]
)