Documentation Index
Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
Use this file to discover all available pages before exploring further.
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
- Set appropriate timeouts - Prevent tasks from running indefinitely
- Handle failures - Always provide error callbacks
- Limit concurrency - Don’t overwhelm resources with too many workers
- Clean up - Call
shutdown() when done with the runner
- Monitor progress - Track long-running tasks
- Agent - Agent configuration
- Task - Task definition
