Skip to main content

​
Background Module

The background module provides the ability to run agents in the background, allowing long-running tasks without blocking the main thread.

​
Installation

pip install praisonaiagents

​
Features

  • Long-running tasks without blocking
  • Task queuing and management
  • Progress monitoring and notifications
  • Graceful cancellation

​
Quick Start

from praisonaiagents.background import BackgroundRunner, BackgroundTask

# Create a background runner
runner = BackgroundRunner()

# Submit a task
task = runner.submit(
    agent=my_agent,
    prompt="Research AI trends",
    callback=on_complete
)

# Check status
print(task.status)  # "running", "completed", "failed"

# Wait for completion
result = await task.wait()

​
Classes

​
BackgroundRunner

Main class for managing background tasks. 
from praisonaiagents.background import BackgroundRunner

runner = BackgroundRunner(max_workers=4)

​
Constructor

ParameterTypeDefaultDescription
max_workersint4Maximum concurrent tasks

​
Methods

MethodDescription
submit(agent, prompt, callback)Submit a task
cancel(task_id)Cancel a running task
get_status(task_id)Get task status
list_tasks()List all tasks
shutdown()Shutdown the runner

​
BackgroundTask

Represents a background task. 
from praisonaiagents.background import BackgroundTask

task = BackgroundTask(
    id="task_123",
    agent=my_agent,
    prompt="Research AI trends"
)

​
Attributes

AttributeTypeDescription
idstrUnique task ID
agentAgentAgent executing the task
promptstrTask prompt
statusTaskStatusCurrent status
resultAnyTask result (when completed)
errorstrError message (if failed)
created_atdatetimeCreation time
completed_atdatetimeCompletion time

​
Methods

MethodDescription
wait()Wait for task completion
cancel()Cancel the task
get_progress()Get progress info

​
TaskStatus

Enumeration of task statuses. 
from praisonaiagents.background import TaskStatus

TaskStatus.PENDING     # Task is queued
TaskStatus.RUNNING     # Task is executing
TaskStatus.COMPLETED   # Task completed successfully
TaskStatus.FAILED      # Task failed
TaskStatus.CANCELLED   # Task was cancelled

​
BackgroundConfig

Configuration for background runner. 
from praisonaiagents.background import BackgroundConfig

config = BackgroundConfig(
    max_workers=4,
    timeout=3600,  # 1 hour
    retry_on_failure=True,
    max_retries=3
)

​
Usage Examples

​
Basic Background Task

from praisonaiagents import Agent
from praisonaiagents.background import BackgroundRunner

agent = Agent(name="Researcher")
runner = BackgroundRunner()

# Submit task
task = runner.submit(
    agent=agent,
    prompt="Research quantum computing advances in 2024"
)

# Do other work while task runs
print(f"Task {task.id} is running...")

# Wait for result
result = await task.wait()
print(f"Result: {result}")

​
With Callback

from praisonaiagents.background import BackgroundRunner

def on_complete(task):
    print(f"Task {task.id} completed!")
    print(f"Result: {task.result}")

def on_error(task):
    print(f"Task {task.id} failed: {task.error}")

runner = BackgroundRunner()
task = runner.submit(
    agent=agent,
    prompt="Long research task",
    on_complete=on_complete,
    on_error=on_error
)

​
Multiple Tasks

from praisonaiagents.background import BackgroundRunner

runner = BackgroundRunner(max_workers=4)

# Submit multiple tasks
tasks = []
topics = ["AI", "Blockchain", "Quantum Computing", "Robotics"]

for topic in topics:
    task = runner.submit(
        agent=researcher,
        prompt=f"Research {topic}"
    )
    tasks.append(task)

# Wait for all tasks
results = await asyncio.gather(*[t.wait() for t in tasks])

​
Task Cancellation

from praisonaiagents.background import BackgroundRunner

runner = BackgroundRunner()
task = runner.submit(agent=agent, prompt="Long task")

# Cancel if taking too long
await asyncio.sleep(60)
if task.status == TaskStatus.RUNNING:
    task.cancel()
    print("Task cancelled")

​
Progress Monitoring

from praisonaiagents.background import BackgroundRunner

runner = BackgroundRunner()
task = runner.submit(agent=agent, prompt="Research task")

# Monitor progress
while task.status == TaskStatus.RUNNING:
    progress = task.get_progress()
    print(f"Progress: {progress.percent}%")
    await asyncio.sleep(5)

​
Best Practices

  1. Set appropriate timeouts - Prevent tasks from running indefinitely
  2. Handle failures - Always provide error callbacks
  3. Limit concurrency - Don’t overwhelm resources with too many workers
  4. Clean up - Call shutdown() when done with the runner
  5. Monitor progress - Track long-running tasks
  • Agent - Agent configuration
  • Task - Task definition