Skip to main content

Overview

AsyncSqliteStorage provides an asynchronous file-based SQLite storage backend using SQLAlchemy with aiosqlite. Ideal for async applications, local development, and single-node deployments.

Basic Usage

import asyncio
from upsonic import Agent, Task
from upsonic.storage.memory import Memory
from upsonic.storage.sqlite import AsyncSqliteStorage

async def main():
    storage = AsyncSqliteStorage(db_file="memory.db")

    memory = Memory(
        storage=storage,
        session_id="session_001",
        user_id="user_123",
        full_session_memory=True,
        summary_memory=True,
        user_analysis_memory=True,
        model="openai/gpt-4o"
    )

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

    result1 = await agent.ado(Task("My name is Alice"))
    result2 = await agent.ado(Task("What's my name?"))
    print(result2)  # "Your name is Alice"

asyncio.run(main())

Parameters

ParameterTypeDefaultDescription
db_filestrNonePath to the SQLite database file
db_urlstrNoneSQLAlchemy async database URL (e.g., sqlite+aiosqlite:///./data.db)
db_engineAsyncEngineNonePre-configured SQLAlchemy AsyncEngine
session_tablestrNoneCustom name for the session table
user_memory_tablestrNoneCustom name for the user memory table
idstrNoneUnique identifier for this storage instance
If no connection parameter is provided, the storage defaults to creating ./upsonic.db in the current directory.

Requirements

pip install sqlalchemy aiosqlite