Background Module

Run agents in the background for long-running tasks

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

Parameter	Type	Default	Description
max_workers	int	4	Maximum concurrent tasks

Methods

Method	Description
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

Attribute	Type	Description
id	str	Unique task ID
agent	Agent	Agent executing the task
prompt	str	Task prompt
status	TaskStatus	Current status
result	Any	Task result (when completed)
error	str	Error message (if failed)
created_at	datetime	Creation time
completed_at	datetime	Completion time

Methods

Method	Description
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

Related

• Agent - Agent configuration
• Task - Task definition