SQL Storage

SQL storage provides a flexible and reliable solution for storing conversation history in the Multi-Agent Orchestrator System. This implementation supports both local SQLite databases and remote Turso databases, making it suitable for various deployment scenarios from development to production.

Features

• Persistent storage across application restarts
• Support for both local and remote databases
• Built-in connection pooling and retry mechanisms
• Compatible with edge and serverless deployments
• Transaction support for data consistency
• Efficient indexing for quick data retrieval

When to Use SQL Storage

• When you need a balance between simplicity and scalability
• For applications requiring persistent storage without complex infrastructure
• In both development and production environments
• When working with edge or serverless deployments
• When you need local-first development with remote deployment options

Python Package Installation

To use SQL storage in your Python application, make sure to install them:

pip install "multi-agent-orchestrator[sql]"

This will install the libsql-client package required for SQL storage functionality.

Implementation

To use SQL storage in your Multi-Agent Orchestrator:

TypeScript

import { SqlChatStorage, MultiAgentOrchestrator } from 'multi-agent-orchestrator';

// For local SQLite database
const localStorage = new SqlChatStorage('file:local.db');
await localStorage.waitForInitialization();

// For remote database
const remoteStorage = new SqlChatStorage(
  'libsql://your-database-url.example.com',
  'your-auth-token'
);
await remoteStorage.waitForInitialization();

const orchestrator = new MultiAgentOrchestrator({
  storage: localStorage // or remoteStorage
});


// Close the database connections when done
await localStorage.close();
await remoteStorage.close();

Python

from multi_agent_orchestrator.storage import SqlChatStorage
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator

# For local SQLite database
local_storage = SqlChatStorage('file:local.db')
await local_storage.initialize()  # Must be called before use

# For remote Turso database
remote_storage = SqlChatStorage(
    url='libsql://your-database-url.turso.io',
    auth_token='your-auth-token'
)
await remote_storage.initialize()

# Create orchestrator with storage
orchestrator = MultiAgentOrchestrator(storage=local_storage)  # or remote_storage

# Example usage
messages = await local_storage.save_chat_message(
    user_id="user123",
    session_id="session456",
    agent_id="agent789",
    new_message=ConversationMessage(
        role="user",
        content=[{"text": "Hello!"}]
    )
)
# messages will contain the updated conversation history

# Don't forget to close connections when done
await local_storage.close()

Configuration

Local DB

For local development, simply provide a file URL:

Local DB

For local development, simply provide a file URL:

TypeScript

const storage = new SqlChatStorage('file:local.db');

Python

storage = SqlChatStorage('file:local.db')
await storage.initialize()  # Must be called before use

Remote DB

For production with Turso:

1. Create a Turso database through their platform
2. Obtain your database URL and authentication token
3. Configure your storage:

TypeScript

const storage = new SqlChatStorage(
  'libsql://your-database-url.turso.io',
  'your-auth-token'
);

Python

storage = SqlChatStorage(
    url='libsql://your-database-url.turso.io',
    auth_token='your-auth-token'
)
await storage.initialize()  # Required initialization

Database Schema

The SQL storage implementation uses the following schema:

CREATE TABLE conversations (
  user_id TEXT NOT NULL,
  session_id TEXT NOT NULL,
  agent_id TEXT NOT NULL,
  message_index INTEGER NOT NULL,
  role TEXT NOT NULL,
  content TEXT NOT NULL,
  timestamp INTEGER NOT NULL,
  PRIMARY KEY (user_id, session_id, agent_id, message_index)
);

CREATE INDEX idx_conversations_lookup
ON conversations(user_id, session_id, agent_id);

Considerations

• Automatic table and index creation on initialization
• Built-in transaction support for data consistency
• Efficient query performance through proper indexing
• Support for message history size limits
• Automatic JSON serialization/deserialization of message content

Best Practices (Python)

1. Initialization:

   storage = SqlChatStorage('file:local.db')
   await storage.initialize()  # Always call initialize after creation

2. Error Handling:

   try:
       messages = await storage.save_chat_message(...)
   except Exception as e:
       logger.error(f"Storage error: {e}")

3. Resource Cleanup:

   try:
       storage = SqlChatStorage('file:local.db')
       await storage.initialize()
       # ... use storage ...
   finally:
       await storage.close()  # Always close when done

4. Message History Management:

   # Limit conversation history
   messages = await storage.save_chat_message(
       ...,
       max_history_size=50  # Keep last 50 messages
   )

5. Batch Operations:

   # Save multiple messages efficiently
   messages = await storage.save_chat_messages(
       user_id="user123",
       session_id="session456",
       agent_id="agent789",
       new_messages=[message1, message2, message3]
   )

SQL storage provides a robust and flexible solution for managing conversation history in the Multi-Agent Orchestrator System. It offers a good balance between simplicity and features, making it suitable for both development and production environments.