Skip to main content

​
Display System

The Display System in PraisonAI Agents provides flexible output formatting and custom display callbacks for rich terminal output, logging, and integration with external systems.

​
Overview

The display system allows you to customize how agent outputs, task results, and system messages are displayed. It supports both synchronous and asynchronous callbacks, rich formatting, and integration with various output destinations.

​
Core Components

​
TaskOutput

The TaskOutput class represents the result of a task execution: 
from praisonaiagents import TaskOutput

# TaskOutput contains:
# - raw: String output from the task
# - json_output: Parsed JSON (if applicable)
# - pydantic_output: Pydantic model instance (if applicable)
# - task_id: Unique identifier for the task
# - metadata: Rich metadata about execution

​
ReflectionOutput

Used for agent self-reflection results: 
from praisonaiagents import ReflectionOutput

# ReflectionOutput contains:
# - reflection: The reflection content
# - satisfactory: Boolean indicating if reflection passed
# - improvement_suggestions: List of suggested improvements

​
Display Callbacks

​
Basic Display Callback

from praisonaiagents import Agent, Task

def custom_display(output):
    """Custom display function for task outputs"""
    print(f"🎯 Task: {output.task_id}")
    print(f"📝 Result: {output.raw}")
    print(f"⏱️  Time: {output.metadata.get('execution_time', 'N/A')}")

# Use with agent
agent = Agent(
    role="Assistant",
    goal="Help with tasks",
    display_callback=custom_display
)

# Use with task
task = Task(
    description="Generate report",
    agent=agent,
    callback=custom_display
)

​
Rich Terminal Display

from rich.console import Console
from rich.panel import Panel
from rich.syntax import Syntax
from rich.table import Table

console = Console()

def rich_display(output):
    """Display output using Rich library"""
    # Create a panel for the output
    panel = Panel(
        output.raw,
        title=f"Task: {output.task_id}",
        border_style="green"
    )
    console.print(panel)
    
    # Display metadata in a table
    if output.metadata:
        table = Table(title="Execution Details")
        table.add_column("Metric", style="cyan")
        table.add_column("Value", style="magenta")
        
        for key, value in output.metadata.items():
            table.add_row(key, str(value))
        
        console.print(table)

# Use with agent
agent = Agent(
    role="Developer",
    goal="Write code",
    display_callback=rich_display
)

​
Async Display Callbacks

import asyncio
from datetime import datetime

async def async_display(output):
    """Async display callback for non-blocking operations"""
    # Log to async service
    await log_to_service(output)
    
    # Update dashboard
    await update_dashboard({
        'task_id': output.task_id,
        'completed_at': datetime.now(),
        'result': output.raw,
        'metrics': output.metadata
    })
    
    # Send notifications
    if output.metadata.get('priority') == 'high':
        await send_notification(f"High priority task {output.task_id} completed")

# Use with async task
task = Task(
    description="Async operation",
    agent=agent,
    async_execution=True,
    callback=async_display
)

​
Global Display Configuration

​
Setting Default Display

from praisonaiagents import set_default_display

def global_display(output):
    """Global display function for all outputs"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{timestamp}] {output.task_id}: {output.raw[:100]}...")

# Set as default for all agents/tasks
set_default_display(global_display)

​
Display Middleware

class DisplayMiddleware:
    """Chain multiple display callbacks"""
    
    def __init__(self, callbacks):
        self.callbacks = callbacks
    
    def __call__(self, output):
        for callback in self.callbacks:
            try:
                callback(output)
            except Exception as e:
                print(f"Display callback error: {e}")

# Create middleware
display_chain = DisplayMiddleware([
    console_display,
    file_logger,
    metrics_collector,
    notification_sender
])

agent = Agent(
    role="Worker",
    goal="Process tasks",
    display_callback=display_chain
)

​
Specialized Display Handlers

​
JSON Output Display

import json

def json_display(output):
    """Specialized display for JSON outputs"""
    if output.json_output:
        # Pretty print JSON
        formatted = json.dumps(output.json_output, indent=2)
        print(f"JSON Output for {output.task_id}:")
        print(formatted)
    else:
        # Fallback for non-JSON
        print(f"Text Output: {output.raw}")

# Use with JSON task
# from pydantic import BaseModel
# class ConfigSchema(BaseModel):
#     key: str
#     value: str
task = Task(
    description="Generate configuration",
    agent=agent,
    output_json=ConfigSchema,  # Define ConfigSchema as shown above
    callback=json_display
)

​
Code Output Display

from pygments import highlight
from pygments.lexers import get_lexer_by_name
from pygments.formatters import TerminalFormatter

def code_display(output):
    """Syntax highlighted code display"""
    # Detect language from metadata or content
    language = output.metadata.get('language', 'python')
    
    try:
        lexer = get_lexer_by_name(language)
        formatter = TerminalFormatter()
        highlighted = highlight(output.raw, lexer, formatter)
        print(highlighted)
    except:
        # Fallback to plain text
        print(output.raw)

# Use with code generation
coder = Agent(
    role="Programmer",
    goal="Write code",
    display_callback=code_display
)

​
Progress Display

from tqdm import tqdm
import threading

class ProgressDisplay:
    """Display progress for long-running tasks"""
    
    def __init__(self):
        self.progress_bars = {}
    
    def __call__(self, output):
        task_id = output.task_id
        
        if output.metadata.get('status') == 'started':
            # Create progress bar
            total = output.metadata.get('total_items', 100)
            self.progress_bars[task_id] = tqdm(
                total=total,
                desc=task_id,
                unit='items'
            )
        
        elif output.metadata.get('status') == 'progress':
            # Update progress
            if task_id in self.progress_bars:
                progress = output.metadata.get('completed', 1)
                self.progress_bars[task_id].update(progress)
        
        elif output.metadata.get('status') == 'completed':
            # Close progress bar
            if task_id in self.progress_bars:
                self.progress_bars[task_id].close()
                del self.progress_bars[task_id]
            print(f"✓ {task_id}: {output.raw}")

# Use with batch processing
progress_display = ProgressDisplay()

task = Task(
    description="Process large dataset",
    agent=processor,
    task_type="loop",
    loop_data="large_dataset.csv",
    callback=progress_display
)

​
Logging Integration

​
File Logging

import logging
from pathlib import Path

class FileLogger:
    """Log outputs to files"""
    
    def __init__(self, log_dir="logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)
        
        # Configure logger
        self.logger = logging.getLogger("task_outputs")
        handler = logging.FileHandler(self.log_dir / "tasks.log")
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def __call__(self, output):
        # Log task completion
        self.logger.info(
            f"Task {output.task_id} completed. "
            f"Execution time: {output.metadata.get('execution_time', 'N/A')}"
        )
        
        # Save full output to separate file
        output_file = self.log_dir / f"{output.task_id}.txt"
        output_file.write_text(output.raw)

# Use file logger
file_logger = FileLogger()

agent = Agent(
    role="Analyst",
    goal="Analyze data",
    display_callback=file_logger
)

​
Structured Logging

import structlog

# Configure structured logging
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

def structured_logger(output):
    """Log with structured data"""
    logger = structlog.get_logger()
    
    logger.info(
        "task_completed",
        task_id=output.task_id,
        execution_time=output.metadata.get('execution_time'),
        agent_role=output.metadata.get('agent_role'),
        tools_used=output.metadata.get('tools_used', []),
        quality_score=output.metadata.get('quality_score'),
        result_preview=output.raw[:200]
    )

# Use structured logging
agent = Agent(
    role="Worker",
    goal="Complete tasks",
    display_callback=structured_logger
)

​
Error Logging

​
Error Display Handler

import json
from datetime import datetime
from praisonaiagents import error_logger

class ErrorDisplay:
    """Enhanced error display and logging"""
    
    def __init__(self, log_file="errors.log"):
        self.log_file = log_file
    
    def __call__(self, output):
        # Check for errors in metadata
        if output.metadata.get('error'):
            error_info = {
                'task_id': output.task_id,
                'error': output.metadata['error'],
                'traceback': output.metadata.get('traceback', ''),
                'timestamp': datetime.now().isoformat()
            }
            
            # Display error
            print(f"❌ Error in {output.task_id}: {error_info['error']}")
            
            # Log to file
            with open(self.log_file, 'a') as f:
                f.write(json.dumps(error_info) + '\n')
            
            # Send alert for critical errors
            if output.metadata.get('severity') == 'critical':
                self.send_alert(error_info)
    
    def send_alert(self, error_info):
        """Send alerts for critical errors"""
        # Implement alerting logic
        pass

# Use error display
error_display = ErrorDisplay()

# Set as global error logger
error_logger.set_handler(error_display)

​
Integration Examples

​
Slack Integration

import os
from slack_sdk import WebClient

class SlackDisplay:
    """Send outputs to Slack"""
    
    def __init__(self, token, channel):
        self.client = WebClient(token=token)
        self.channel = channel
    
    def __call__(self, output):
        # Format message
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"Task Completed: {output.task_id}"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": output.raw[:500]
                }
            },
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Agent:* {output.metadata.get('agent_role', 'Unknown')}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Time:* {output.metadata.get('execution_time', 'N/A')}"
                    }
                ]
            }
        ]
        
        # Send to Slack
        self.client.chat_postMessage(
            channel=self.channel,
            blocks=blocks
        )

# Use Slack display
slack_display = SlackDisplay(
    token=os.getenv("SLACK_TOKEN"),
    channel="#task-updates"
)

agent = Agent(
    role="Reporter",
    goal="Generate reports",
    display_callback=slack_display
)

​
Dashboard Integration

import aiohttp
from datetime import datetime

class DashboardDisplay:
    """Update real-time dashboard"""
    
    def __init__(self, api_url):
        self.api_url = api_url
    
    async def __call__(self, output):
        """Async update to dashboard"""
        payload = {
            'task_id': output.task_id,
            'status': 'completed',
            'result': output.raw,
            'metadata': output.metadata,
            'timestamp': datetime.now().isoformat()
        }
        
        # Post to dashboard API
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.api_url}/tasks/update",
                json=payload
            ) as response:
                if response.status != 200:
                    print(f"Dashboard update failed: {response.status}")

# Use dashboard display
dashboard = DashboardDisplay("https://dashboard.example.com/api")

task = Task(
    description="Monitor system",
    agent=monitor,
    async_execution=True,
    callback=dashboard
)

​
Best Practices

  1. Error Handling: Always wrap display callbacks in try-except blocks
  2. Performance: Keep display callbacks lightweight for better performance
  3. Async Usage: Use async callbacks for I/O operations
  4. Composition: Combine multiple displays using middleware pattern
  5. Testing: Test display callbacks separately from agent logic

​
See Also