Storage Backends

import asyncio
from upsonic.agent.deepagent import DeepAgent
from upsonic.agent.deepagent.backends import StateBackend
from upsonic import Task

async def main():
    backend = StateBackend()
    agent = DeepAgent(
        model="anthropic/claude-sonnet-4-5",
        filesystem_backend=backend
    )
    
    task = Task(description="Create /tmp/scratch.txt with temporary data")
    result = await agent.do_async(task)
    
    # Files exist only during this session
    exists = await backend.exists("/tmp/scratch.txt")
    print(f"File exists: {exists}")

asyncio.run(main())

import asyncio
import os
import tempfile
from upsonic.agent.deepagent import DeepAgent
from upsonic.agent.deepagent.backends import MemoryBackend
from upsonic.storage import SqliteStorage
from upsonic import Task

async def main():
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
        db_path = tmp.name
    
    try:
        storage = SqliteStorage(db_file=db_path)
        backend = MemoryBackend(storage)
        
        agent = DeepAgent(
            model="anthropic/claude-sonnet-4-5",
            filesystem_backend=backend
        )
        
        task = Task(description="Research latest AI trends. Save important research findings to /memory/research.txt")
        await agent.do_async(task)
        
        # Files persist across sessions
        content = await backend.read("/memory/research.txt")
        print(f"Persistent content: {content[:50]}...")
        
        storage.close()
    finally:
        if os.path.exists(db_path):
            os.unlink(db_path)

asyncio.run(main())

import asyncio
from upsonic.agent.deepagent import DeepAgent
from upsonic.agent.deepagent.backends import StateBackend, MemoryBackend, CompositeBackend
from upsonic.storage.providers.sqlite import SqliteStorage
from upsonic import Task
import tempfile

async def main():
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
        db_path = tmp.name
    
    try:
        storage = SqliteStorage(db_file=db_path)
        backend = CompositeBackend(
            default=StateBackend(),  # Ephemeral for default paths
            routes={
                "/research/": MemoryBackend(storage),  # Persistent
                "/reports/": MemoryBackend(storage)   # Persistent
            }
        )
        
        agent = DeepAgent(
            model="anthropic/claude-sonnet-4-5",
            filesystem_backend=backend
        )
        
        task = Task(description="""
        Create /tmp/temp.txt (ephemeral)
        Create /research/findings.txt (persistent)
        Create /reports/summary.txt (persistent)
        """)
        
        result = await agent.do_async(task)
        print(result)
        
        await storage.disconnect_async()
    finally:
        import os
        os.unlink(db_path)

asyncio.run(main())