Storage Backends

Learn about different storage options for persisting agent interactions, learning data, and system state.

Overview

OACP supports multiple storage backends to fit different deployment scenarios and performance requirements. All backends implement the same interface, making it easy to switch between them.

File Storage

The simplest storage backend that saves data to JSON files on the local filesystem.

from oacp.storage import FileStorage

# Basic file storage
storage = FileStorage("./data")

# With custom configuration
storage = FileStorage(
    directory="./logs",
    auto_backup=True,
    backup_interval=3600,  # Backup every hour
    max_file_size="100MB"
)

# Use with agent system
system = AgentSystem(agents=agents, storage=storage)

✓ Pros

• Simple setup, no dependencies
• Human-readable JSON format
• Easy to backup and version control
• Good for development and small deployments

⚠ Cons

• Not suitable for high-concurrency
• Limited querying capabilities
• No built-in data integrity checks
• Performance degrades with large files

SQLite Storage

A lightweight database backend that provides better performance and querying capabilities.

from oacp.storage import SQLiteStorage

# Basic SQLite storage
storage = SQLiteStorage("agents.db")

# With connection pooling and optimization
storage = SQLiteStorage(
    database_path="./data/agents.db",
    pool_size=10,
    enable_wal_mode=True,  # Write-Ahead Logging for better concurrency
    cache_size=1000,
    auto_vacuum=True
)

# Use with agent system
system = AgentSystem(agents=agents, storage=storage)

✓ Pros

• Better performance than file storage
• SQL querying capabilities
• ACID transactions
• Built-in data integrity
• No server setup required

⚠ Cons

• Single file can become large
• Limited concurrent write performance
• Not suitable for distributed systems

PostgreSQL Storage

A full-featured database backend for production deployments with high concurrency requirements.

from oacp.storage import PostgreSQLStorage

# Basic PostgreSQL storage
storage = PostgreSQLStorage(
    host="localhost",
    database="oacp",
    user="oacp_user",
    password="your_password"
)

# With connection pooling and advanced features
storage = PostgreSQLStorage(
    connection_string="postgresql://user:pass@localhost:5432/oacp",
    pool_size=20,
    max_overflow=30,
    enable_json_indexing=True,
    partition_by_date=True
)

# Use with agent system
system = AgentSystem(agents=agents, storage=storage)

✓ Pros

• Excellent concurrent performance
• Advanced querying and indexing
• Horizontal scaling support
• Rich data types and JSON support
• Production-ready reliability

⚠ Cons

• Requires server setup and maintenance
• More complex configuration
• Resource intensive
• Overkill for simple use cases

Custom Storage Backends

Create your own storage backend by implementing the StorageBackend interface.

from oacp.storage import StorageBackend
from typing import Dict, List, Any, Optional

class RedisStorage(StorageBackend):
    def __init__(self, redis_url: str):
        import redis
        self.redis = redis.from_url(redis_url)
    
    async def store_interaction(self, agent_id: str, interaction: Dict[str, Any]) -> str:
        interaction_id = f"{agent_id}:{int(time.time())}"
        await self.redis.hset(f"interactions:{interaction_id}", mapping=interaction)
        return interaction_id
    
    async def get_interactions(self, agent_id: str, limit: int = 100) -> List[Dict[str, Any]]:
        pattern = f"interactions:{agent_id}:*"
        keys = await self.redis.keys(pattern)
        interactions = []
        for key in keys[-limit:]:
            interaction = await self.redis.hgetall(key)
            interactions.append(interaction)
        return interactions
    
    async def store_agent_state(self, agent_id: str, state: Dict[str, Any]) -> None:
        await self.redis.hset(f"agent_state:{agent_id}", mapping=state)
    
    async def get_agent_state(self, agent_id: str) -> Optional[Dict[str, Any]]:
        state = await self.redis.hgetall(f"agent_state:{agent_id}")
        return state if state else None

# Usage
storage = RedisStorage("redis://localhost:6379")
system = AgentSystem(agents=agents, storage=storage)

Choosing the Right Backend

Use Case	Recommended Backend	Reason
Development & Testing	File Storage	Simple setup, easy debugging
Small Production	SQLite	Better performance, no server needed
High Concurrency	PostgreSQL	Excellent concurrent performance
Distributed Systems	Custom (Redis/MongoDB)	Horizontal scaling capabilities
Analytics & Reporting	PostgreSQL	Advanced querying and JSON support