Praison AI

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

bash
pip install praisonaiagents

Features

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

Quick Start

python
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.

python
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.

python
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.

python
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.

python
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

python
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

python
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

python
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

python
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

python
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