Skip to main content

​
Basic Usage

Durable executions are created by attaching a DurableExecution instance to a Task. The system automatically handles checkpoint management throughout the execution lifecycle. 
from upsonic import Task, Agent
from upsonic.durable import DurableExecution, FileDurableStorage

# Create storage backend
storage = FileDurableStorage(path="./checkpoints")

# Create durable execution
durable = DurableExecution(storage=storage)

# Create task with durable execution
task = Task(
    "Process customer data and generate report",
    durable_execution=durable
)

# Get the unique execution ID
print(f"Execution ID: {task.durable_execution_id}")
# Output: Execution ID: 20251027123456-a1b2c3d4

# Execute the task
agent = Agent("openai/gpt-4o-mini", name="DataProcessor")
try:
    result = agent.do(task)
    print(f"Success: {result}")
except Exception as e:
    print(f"Failed: {e}")
    print(f"Checkpoint saved at: {task.durable_execution_id}")

​
Recovery from Failure

# Resume from a failed execution
execution_id = "20251027123456-a1b2c3d4"

# Create a new agent instance (simulating restart)
agent = Agent("openai/gpt-4o-mini", name="DataProcessor")

# Resume from the checkpoint
result = agent.continue_durable(execution_id, storage)
print(f"Recovered and completed: {result}")