Backend Configuration
StateBackend (Ephemeral)
In-memory storage that persists only during the agent session.Copy
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.Copy
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.Copy
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.Copy
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.Copy
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())

