Niger - Storage Manager Agent

Overview

Niger is the agent responsible for managing data storage and retrieval in the TKM AI Agency Platform. It provides a centralized storage solution using SingleStore, handling data persistence for all other agents in the system.

Directory Structure

Niger/
├── models/               # Data models and database schemas
├── niger.py             # Main agent implementation
├── niger_tools.py       # Storage utilities and connection management
├── niger_schema.py      # Data models and request/response schemas
├── niger_migrations.py  # Database migration scripts
├── niger_definitions.py # Constants and type definitions
└── test_niger.py       # Unit tests

Main Components

NigerAgent Class

The main class that handles storage operations:

SingleStore connection management
Data storage and retrieval
Batch processing capabilities
Query execution

Storage Connection

Singleton pattern implementation
Connection pooling
Migration management
Error handling

Key Features

Data Operations

Single record storage
Batch data processing
Flexible querying
Data retrieval with filtering

Database Management

Automated migrations
Schema versioning
Data integrity checks
Connection pooling

Integration Features

Multi-agent data storage
Organization-level data isolation
User-specific data management
Cross-agent data querying

API Operations

Storage Operations

# Store Data Request
{
    "action": "store",
    "agent": str,
    "data_type": str,
    "content": dict,
    "organization_id": str,
    "user_id": str
}
 
# Retrieve Data Request
{
    "action": "retrieve",
    "agent": str,
    "data_type": str,
    "content": dict
}

Batch Operations

Multiple storage requests processing
Success/failure tracking
Detailed error reporting
Transaction management

Query Operations

Filtered data retrieval
Pagination support
Result aggregation
Error handling

Integration

Agent Communication

Supports data storage for various agents:

Atta: Conversation data
Bala: Embedding data
Hova: Image processing data
Rufa: Audio processing data
Scalaris: Document data

Data Flow

Request validation
Connection management
Operation execution
Response formatting
Error handling
Result delivery

Error Handling

Connection error management
Transaction rollback
Query timeout handling
Data validation errors
Detailed error logging

Performance Features

Optimization

Connection pooling
Query optimization
Batch processing
Efficient data retrieval

Scalability

Horizontal scaling support
Load balancing
Resource management
Performance monitoring

Data Models

Storage Response

{
    "success": bool,
    "data": Optional[dict],
    "error": Optional[str]
}

Batch Response

{
    "success": bool,
    "total_processed": int,
    "successful": int,
    "failed": int,
    "errors": Optional[List[dict]]
}

Query Response

{
    "success": bool,
    "data": List[dict],
    "total": int,
    "page": int,
    "total_pages": int,
    "error": Optional[str]
}

Hova - Image Processing Rufa - Audio Processing