Run Management¶

The Run system in Bedrock Swarm manages the lifecycle and state of individual execution runs within threads. Each run represents a single execution context with its own state, actions, and completion status.

Class Documentation¶

23
24
25
26
27
28
29
30
31
32
33

def __init__(self) -> None:
    """Initialize a new run."""
    self.id = str(uuid4())
    self.status: Literal[
        "queued", "in_progress", "requires_action", "completed", "failed"
    ] = "queued"
    self.started_at = datetime.now()
    self.completed_at: Optional[datetime] = None
    self.required_action: Optional[Dict] = None
    self.last_error: Optional[str] = None
    self.tool_calls: List[Dict] = []  # Track tool calls made during this run

35
36
37
38

def complete(self) -> None:
    """Mark the run as completed."""
    self.status = "completed"
    self.completed_at = datetime.now()

40
41
42
43
44

def fail(self, error: str) -> None:
    """Mark the run as failed."""
    self.status = "failed"
    self.last_error = error
    self.completed_at = datetime.now()

46
47
48
49
50
51

def require_action(self, action: Dict) -> None:
    """Set the run to require action."""
    self.status = "requires_action"
    self.required_action = action
    if action.get("type") == "tool_calls":
        self.tool_calls.extend(action["tool_calls"])

Run States¶

A run can be in one of these states:

1. Created: Initial state when run is created
2. Running: Run is actively executing
3. Waiting: Run is waiting for action
4. Completed: Run has successfully completed
5. Failed: Run has failed with error
6. Cancelled: Run was cancelled

Run Lifecycle¶

stateDiagram-v2
    [*] --> Created
    Created --> Running: start
    Running --> Waiting: require_action
    Waiting --> Running: action_complete
    Running --> Completed: complete
    Running --> Failed: fail
    Running --> Cancelled: cancel
    Waiting --> Cancelled: cancel
    Completed --> [*]
    Failed --> [*]
    Cancelled --> [*]

Usage Examples¶

from bedrock_swarm.agency import Run

# Create a new run
run = Run()

# Start execution
run.start()

# Require action
run.require_action({
    "type": "tool_call",
    "tool": "calculator",
    "args": {"expression": "2 + 2"}
})

# Complete run
run.complete()

# Handle failure
run.fail("Error: Invalid operation")

# Check status
print(run.status)  # Output: "completed" or "failed"

Run Actions¶

Runs can require different types of actions:

1. Tool Calls:

   run.require_action({
       "type": "tool_call",
       "tool": "tool_name",
       "args": {"param": "value"}
   })
   

2. Message Actions:

   run.require_action({
       "type": "send_message",
       "recipient": "agent_name",
       "content": "message"
   })
   

3. System Actions:

   run.require_action({
       "type": "system_action",
       "action": "pause",
       "duration": 5
   })
   

Error Handling¶

The run system handles various error scenarios:

1. State Transitions:

   try:
       run.complete()  # When not in running state
   except InvalidStateError:
       print("Cannot complete run in current state")
   

2. Action Validation:

   try:
       run.require_action(invalid_action)
   except ValidationError:
       print("Invalid action format")
   

3. Failure Handling:

   try:
       # Run operation
       result = perform_operation()
   except Exception as e:
       run.fail(str(e))
   

Implementation Details¶

The run implementation includes:

1. State Management:
2. State validation
3. Transition rules

4. History tracking

5. Action Handling:

6. Action validation
7. Action queueing

8. Action completion

9. Resource Management:

10. Resource allocation
11. Resource cleanup
12. Memory management

Run Configuration¶

Runs can be configured with:

run_config = {
    "timeout": 300,  # Maximum run duration
    "retry": {
        "max_attempts": 3,
        "delay": 1.0
    },
    "resources": {
        "memory": "1GB",
        "cpu": 2
    }
}

Best Practices¶

1. State Management:

   # Always check state before operations
   if run.can_complete():
       run.complete()
   else:
       print(f"Cannot complete in state: {run.status}")
   

2. Error Handling:

   try:
       # Run operations
       result = perform_operations()
       run.complete()
   except Exception as e:
       run.fail(str(e))
   finally:
       cleanup_resources()
   

3. Action Management:

   # Record action requirement
   action_id = run.require_action(action)
   
   # Complete action
   run.complete_action(action_id, result)
   
   # Handle action failure
   run.fail_action(action_id, error)