Skip to main content

Backend Configuration

StateBackend (Ephemeral)

In-memory storage that persists only during the agent session.
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="openai/gpt-4o",
        filesystem_backend=backend
    )
    
    task = Task(description="Analyze the theoretical foundations of transformer attention mechanisms, comparing multi-head attention variants and their computational complexity trade-offs, then document your findings in /workspace/attention_analysis.txt")
    result = await agent.do_async(task)
    
    # Files exist only during this session
    exists = await backend.exists("/workspace/attention_analysis.txt")
    print(f"File exists: {exists}")

asyncio.run(main())

MemoryBackend (Persistent)

Persistent storage that survives process restarts.
import asyncio
from upsonic.agent.deepagent import DeepAgent
from upsonic.agent.deepagent.backends import MemoryBackend
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 = MemoryBackend(storage)
        
        agent = DeepAgent(
            model="openai/gpt-4o",
            filesystem_backend=backend
        )
        
        task = Task(description="Investigate the convergence properties of gradient-based optimization in non-convex landscapes, analyzing escape mechanisms from local minima and saddle point dynamics, then save your comprehensive analysis to /memory/optimization_analysis.txt")
        result = await agent.do_async(task)
        
        # Files persist across sessions
        content = await backend.read("/memory/optimization_analysis.txt")
        print(f"Persistent content: {content[:50]}...")
        
        await storage.disconnect_async()
    finally:
        import os
        os.unlink(db_path)

asyncio.run(main())

CompositeBackend (Hybrid)

Route different paths to different backends.
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 /tmp/
            routes={
                "/research/": MemoryBackend(storage),  # Persistent
                "/reports/": MemoryBackend(storage)   # Persistent
            }
        )
        
        agent = DeepAgent(
            model="openai/gpt-4o",
            filesystem_backend=backend
        )
        
        task = Task(description="Examine the information-theoretic limits of federated learning, investigating communication-efficient aggregation protocols and differential privacy guarantees, then save temporary calculations to /tmp/computations.txt, research findings to /research/federated_analysis.txt, and final synthesis to /reports/federated_summary.txt")
        
        result = await agent.do_async(task)
        print(result)
        
        await storage.disconnect_async()
    finally:
        import os
        os.unlink(db_path)

asyncio.run(main())

Monitoring

Get Current Plan

Track todo status during execution.
import asyncio
from upsonic.agent.deepagent import DeepAgent
from upsonic import Task

async def main():
    agent = DeepAgent(model="openai/gpt-4o")
    
    task = Task(description="Model the phase transitions in deep learning optimization landscapes, investigating critical phenomena in loss function geometry and gradient flow dynamics, then plan your analysis, execute the research, and save your findings to /analysis/phase_transitions.txt")
    
    result = await agent.do_async(task)
    
    # Get current plan
    plan = agent.get_current_plan()
    print(f"\nPlan Status:")
    for todo in plan:
        print(f"  [{todo['status']}] {todo['content']}")

asyncio.run(main())

Get Filesystem Stats

Monitor filesystem usage.
import asyncio
from upsonic.agent.deepagent import DeepAgent
from upsonic import Task

async def main():
    agent = DeepAgent(model="openai/gpt-4o")
    
    task = Task(description="Derive provable defense mechanisms against adaptive adversarial attacks in deep learning, analyzing the fundamental trade-offs between model accuracy and robustness certificates, then document your theoretical framework in /docs/adversarial_defense.txt, implementation strategies in /docs/implementation.txt, and experimental results in /docs/experiments.txt")
    result = await agent.do_async(task)
    
    # Get filesystem statistics
    stats = agent.get_filesystem_stats()
    print(f"\nFilesystem Stats: {stats}")

asyncio.run(main())