Skip to main content

​
Custom Storage Backend

Chat supports multiple storage backends for persistent memory and session data.

​
Available Storage Providers

  • InMemoryStorage: Ephemeral in-memory storage (default)
  • SqliteStorage: SQLite database for local persistence
  • PostgresStorage: PostgreSQL for production
  • MongoStorage: MongoDB for scalable deployments
  • RedisStorage: Redis for high-performance caching
  • JSONStorage: File-based JSON storage
  • Mem0Storage: Mem0 platform for advanced memory management

​
Using SQLite Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import SqliteStorage


async def main():
    # Setup persistent SQLite storage
    storage = SqliteStorage(
        db_file="chat.db",
        sessions_table_name="sessions",
        profiles_table_name="profiles"
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using PostgreSQL Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import PostgresStorage


async def main():
    # PostgreSQL storage
    storage = PostgresStorage(
        sessions_table_name="sessions",
        profiles_table_name="profiles",
        db_url="postgresql://user:pass@localhost:5432/dbname",
        schema="public"
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using MongoDB Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import MongoStorage


async def main():
    # MongoDB storage
    storage = MongoStorage(
        db_url="mongodb://localhost:27017",
        database_name="chat_db",
        sessions_collection_name="sessions",
        profiles_collection_name="profiles"
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using Redis Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import RedisStorage


async def main():
    # Redis storage with TTL
    storage = RedisStorage(
        prefix="chat",
        host="localhost",
        port=6379,
        expire=3600  # 1 hour TTL
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using JSON Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import JSONStorage


async def main():
    # JSON file-based storage
    storage = JSONStorage(
        directory_path="./chat_data",
        pretty_print=True
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using Mem0 Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import Mem0Storage


async def main():
    # Mem0 Platform storage
    """
    storage = Mem0Storage(
        api_key="your-api-key",
        org_id="your-org-id",
        project_id="your-project-id",
        namespace="upsonic"
    )
    """

    # Or Mem0 Open Source
    storage = Mem0Storage(
        local_config={"vector_store": {"provider": "chroma"}},
        namespace="upsonic"
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Explain AI")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Using In-Memory Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import InMemoryStorage


async def main():
    # In-memory storage with LRU cache limits
    storage = InMemoryStorage(
        max_sessions=1000,
        max_profiles=100
    )

    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Invoke chat
    response = await chat.invoke("Hello")
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

​
Storage Connection Management

Chat automatically manages storage connections: 
import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import SqliteStorage


async def main():
    # Setup persistent SQLite storage
    storage = SqliteStorage(
        db_file="chat.db",
        sessions_table_name="sessions",
        profiles_table_name="profiles"
    )
    
    agent = Agent("openai/gpt-4o")

    chat = Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    )

    # Storage connection managed automatically
    response = await chat.invoke("Hello")
    print(response)

    # Close connection when done
    await chat.close()


if __name__ == "__main__":
    asyncio.run(main())