AI Agents with Callbacks - PraisonAI Documentation

Prerequisites

Python 3.10 or higher
PraisonAI Agents package installed
Basic understanding of Python functions

Quick Start

Install Package

First, install the PraisonAI Agents package:

pip install praisonaiagents

Import Required Functions

Import the necessary components:

from praisonaiagents import (
    register_display_callback,
    Agent, 
    Task, 
    PraisonAIAgents
)

Create a Simple Callback

Define your first callback function:

def simple_callback(message=None, response=None, **kwargs):
    print(f"Received message: {message}")
    print(f"Got response: {response}")

# Register as synchronous callback
register_display_callback('interaction', simple_callback, is_async=False)

# For async callbacks
async def async_simple_callback(message=None, response=None, **kwargs):
    await asyncio.sleep(0)  # Non-blocking pause
    print(f"Received message: {message}")
    print(f"Got response: {response}")

# Register as async callback
register_display_callback('interaction', async_simple_callback, is_async=True)

Use in Your Code

Create and run an agent with callbacks enabled:

# Create an agent
agent = Agent(
    name="MyAgent",
    role="Assistant",
    goal="Help with tasks",
    backstory="I am a helpful assistant",
    verbose=True  # Enable verbose mode to see callbacks in action
)

# Create a task
task = Task(
    name="simple_task",
    description="Say hello",
    agent=agent,
    expected_output="A greeting"
)

# Run the agent
agents = PraisonAIAgents(
    agents=[agent],
    tasks=[task]
)
agents.start()

PraisonAI Agents supports both synchronous and asynchronous callbacks. When registering a callback, specify is_async=True for async callbacks. Both sync and async callbacks for the same display type can be executed simultaneously.

Understanding Callbacks

What are Callbacks?

Callbacks are functions that get called automatically when specific events occur in your AI agents:

Interactions between user and agent
Error messages
Tool calls
Self-reflection moments
Task completion
Generation progress

Available Callback Types

Interaction Callback

Triggered when the agent interacts with users

Error Callback

Called when errors occur

Tool Call Callback

Activated when tools are used

Self Reflection Callback

Triggered during agent self-reflection

Instruction Callback

Called when instructions are processed

Generating Callback

Activated during content generation

Basic Implementation

1. Simple Logging Callback

2. Multiple Callback Types

# Error callback
def error_callback(message=None):
    logging.error(f"Error occurred: {message}")

# Tool call callback
def tool_callback(message=None):
    logging.info(f"Tool called: {message}")

# Register multiple callbacks
register_display_callback('error', error_callback)
register_display_callback('tool_call', tool_callback)

Best Practices

Always include error handling in your callbacks:

def safe_callback(message=None, **kwargs):
    try:
        # Process the callback
        logging.info(message)
    except Exception as e:
        logging.error(f"Callback error: {e}")

Keep callbacks lightweight:

# Good - Quick logging
def good_callback(message=None):
    logging.info(message)

# Bad - Heavy processing
def bad_callback(message=None):
    time.sleep(1)  # Delays agent execution
    process_heavy_task(message)

Complete Example

Here’s a full implementation showing all callback types and proper logging:

from praisonaiagents import Agent, Task, PraisonAIAgents, register_display_callback
import logging
from datetime import datetime

# Setup logging
logging.basicConfig(
    filename='ai_interactions.log',
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Interaction callback
def interaction_callback(message=None, response=None, markdown=None, generation_time=None):
    logging.info(f"""
    === INTERACTION ===
    Time: {datetime.now()}
    Generation Time: {generation_time}s
    Message: {message}
    Response: {response}
    Markdown: {markdown}
    """)

# Error callback
def error_callback(message=None):
    logging.error(f"""
    === ERROR ===
    Time: {datetime.now()}
    Message: {message}
    """)

# Tool call callback
def tool_call_callback(message=None):
    logging.info(f"""
    === TOOL CALL ===
    Time: {datetime.now()}
    Message: {message}
    """)

# Register callbacks
register_display_callback('interaction', interaction_callback)
register_display_callback('error', error_callback)
register_display_callback('tool_call', tool_call_callback)

agent = Agent(
    name="CallbackAgent",
    role="Assistant",
    goal="Demonstrate callbacks",
    backstory="I am a helpful assistant",
    verbose=True
)

task = Task(
    name="callback_task",
    description="Show how callbacks work",
    agent=agent,
    expected_output="Demonstration complete"
)

agents = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    verbose=True
)

agents.start()

Advanced Examples

All callback types
Comprehensive logging
Task callbacks
Tool integration
Multiple agents

from praisonaiagents import Agent, Task, PraisonAIAgents, error_logs, register_display_callback
from duckduckgo_search import DDGS
from rich.console import Console
import json
from datetime import datetime
import logging

# Setup logging
logging.basicConfig(
    filename='ai_interactions.log',
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Callback functions for different display types
def interaction_callback(message=None, response=None, markdown=None, generation_time=None):
    """Callback for display_interaction"""
    logging.info(f"""
    === INTERACTION ===
    Time: {datetime.now()}
    Generation Time: {generation_time}s
    Message: {message}
    Response: {response}
    Markdown: {markdown}
    """)

def error_callback(message=None):
    """Callback for display_error"""
    logging.error(f"""
    === ERROR ===
    Time: {datetime.now()}
    Message: {message}
    """)

def tool_call_callback(message=None):
    """Callback for display_tool_call"""
    logging.info(f"""
    === TOOL CALL ===
    Time: {datetime.now()}
    Message: {message}
    """)

def instruction_callback(message=None):
    """Callback for display_instruction"""
    logging.info(f"""
    === INSTRUCTION ===
    Time: {datetime.now()}
    Message: {message}
    """)

def self_reflection_callback(message=None):
    """Callback for display_self_reflection"""
    logging.info(f"""
    === SELF REFLECTION ===
    Time: {datetime.now()}
    Message: {message}
    """)

def generating_callback(content=None, elapsed_time=None):
    """Callback for display_generating"""
    logging.info(f"""
    === GENERATING ===
    Time: {datetime.now()}
    Content: {content}
    Elapsed Time: {elapsed_time}
    """)

# Register all callbacks
register_display_callback('interaction', interaction_callback)
register_display_callback('error', error_callback)
register_display_callback('tool_call', tool_call_callback)
register_display_callback('instruction', instruction_callback)
register_display_callback('self_reflection', self_reflection_callback)
# register_display_callback('generating', generating_callback)

def task_callback(output):
    """Callback for task completion"""
    logging.info(f"""
    === TASK COMPLETED ===
    Time: {datetime.now()}
    Description: {output.description}
    Agent: {output.agent}
    Output: {output.raw[:200]}...
    """)

def internet_search_tool(query) -> list:
    """
    Perform a search using DuckDuckGo.

    Args:
        query (str): The search query.

    Returns:
        list: A list of search result titles and URLs.
    """
    try:
        results = []
        ddgs = DDGS()
        for result in ddgs.text(keywords=query, max_results=10):
            results.append({
                "title": result.get("title", ""),
                "url": result.get("href", ""),
                "snippet": result.get("body", "")
            })
        return results

    except Exception as e:
        print(f"Error during DuckDuckGo search: {e}")
        return []

def main():
    # Create agents
    researcher = Agent(
        name="Researcher",
        role="Senior Research Analyst",
        goal="Uncover cutting-edge developments in AI and data science",
        backstory="""You are an expert at a technology research group, 
        skilled in identifying trends and analyzing complex data.""",
        verbose=True,
        allow_delegation=False,
        tools=[internet_search_tool],
        llm="gpt-4o",
        markdown=True,
        reflect_llm="gpt-4o",
        min_reflect=2,
        max_reflect=4
    )
    
    writer = Agent(
        name="Writer",
        role="Tech Content Strategist",
        goal="Craft compelling content on tech advancements",
        backstory="""You are a content strategist known for 
        making complex tech topics interesting and easy to understand.""",
        verbose=True,
        allow_delegation=True,
        llm="gpt-4o",
        tools=[],
        markdown=True
    )

    # Create tasks with callbacks
    task1 = Task(
        name="research_task",
        description="""Analyze 2024's AI advancements. 
        Find major trends, new technologies, and their effects.""",
        expected_output="""A detailed report on 2024 AI advancements""",
        agent=researcher,
        tools=[internet_search_tool],
        callback=task_callback
    )

    task2 = Task(
        name="writing_task",
        description="""Create a blog post about major AI advancements using the insights you have.
        Make it interesting, clear, and suited for tech enthusiasts. 
        It should be at least 4 paragraphs long.""",
        expected_output="A blog post of at least 4 paragraphs",
        agent=writer,
        context=[task1],
        callback=task_callback,
        tools=[]
    )

    task3 = Task(
        name="json_task",
        description="""Create a json object with a title of "My Task" and content of "My content".""",
        expected_output="""JSON output with title and content""",
        agent=researcher,
        callback=task_callback
    )

    task4 = Task(
        name="save_output_task",
        description="""Save the AI blog post to a file""",
        expected_output="""File saved successfully""",
        agent=writer,
        context=[task2],
        output_file='test.txt',
        create_directory=True,
        callback=task_callback
    )

    # Create and run agents manager
    agents = PraisonAIAgents(
        agents=[researcher, writer],
        tasks=[task1, task2, task3, task4],
        verbose=True,
        process="sequential",
        manager_llm="gpt-4o"
    )

    agents.start()

if __name__ == "__main__":
    main()

## Async Callbacks

Async callbacks allow you to handle events asynchronously, which is particularly useful for long-running operations or when dealing with multiple agents simultaneously.

### Basic Async Callback Implementation

```python
import asyncio
from praisonaiagents import register_display_callback

async def async_interaction_callback(message=None, response=None, **kwargs):
    """Async callback for handling interactions"""
    await asyncio.sleep(0)  # Non-blocking pause
    print(f"Async processing - Message: {message}")
    print(f"Response: {response}")

# Register the async callback
register_display_callback('interaction', async_interaction_callback)

Complete Async Example

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents, register_display_callback
import logging
from datetime import datetime

# Setup async logging
async def setup_async_logging():
    logging.basicConfig(
        filename='async_ai_interactions.log',
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

# Async callbacks
async def async_interaction_callback(message=None, response=None, markdown=None, generation_time=None):
    await asyncio.sleep(0)
    logging.info(f"""
    === ASYNC INTERACTION ===
    Time: {datetime.now()}
    Generation Time: {generation_time}s
    Message: {message}
    Response: {response}
    """)

async def async_error_callback(message=None):
    await asyncio.sleep(0)
    logging.error(f"""
    === ASYNC ERROR ===
    Time: {datetime.now()}
    Message: {message}
    """)

# Register async callbacks
register_display_callback('interaction', async_interaction_callback, is_async=True)
register_display_callback('error', async_error_callback, is_async=True)

# Create async task callback
async def async_task_callback(output):
    await asyncio.sleep(0)
    logging.info(f"""
    === ASYNC TASK COMPLETED ===
    Time: {datetime.now()}
    Description: {output.description}
    Agent: {output.agent}
    Output: {output.raw[:200]}...
    """)

async def main():
    await setup_async_logging()
    
    # Create agent with async capabilities
    agent = Agent(
        name="AsyncAgent",
        role="Assistant",
        goal="Demonstrate async callbacks",
        backstory="I am an async-capable assistant",
        verbose=True
    )

    # Create task with async callback
    task = Task(
        name="async_task",
        description="Demonstrate async callbacks",
        agent=agent,
        expected_output="Async demonstration complete",
        callback=async_task_callback
    )

    # Create and run agents with async support
    agents = PraisonAIAgents(
        agents=[agent],
        tasks=[task],
        verbose=True
    )

    await agents.start_async()

if __name__ == "__main__":
    asyncio.run(main())

Available Async Functions

PraisonAI Agents provides several async versions of display functions, prefixed with ‘a’. Here’s the complete list:

adisplay_interaction

async def adisplay_interaction(
    message, 
    response, 
    markdown=True, 
    generation_time=None, 
    console=None
)

Async version of display_interaction for showing conversations.

adisplay_self_reflection

async def adisplay_self_reflection(
    message: str, 
    console=None
)

Async version for displaying agent’s self-reflection moments.

adisplay_instruction

async def adisplay_instruction(
    message: str, 
    console=None
)

Async version for showing instructions.

adisplay_tool_call

async def adisplay_tool_call(
    message: str, 
    console=None
)

Async version for displaying tool calls.

adisplay_error

async def adisplay_error(
    message: str, 
    console=None
)

Async version for error messages.

adisplay_generating

async def adisplay_generating(
    content: str = "", 
    start_time: Optional[float] = None
)

Async version for showing generation progress.

Example Usage

import asyncio
from praisonaiagents import (
    adisplay_interaction,
    adisplay_error,
    adisplay_tool_call
)

async def main():
    # Display an interaction
    await adisplay_interaction(
        message="What's the weather?",
        response="Let me check that for you.",
        generation_time=0.5
    )

    # Display a tool call
    await adisplay_tool_call(
        "Calling weather API for location data..."
    )

    # Handle an error
    try:
        raise Exception("API connection failed")
    except Exception as e:
        await adisplay_error(str(e))

if __name__ == "__main__":
    asyncio.run(main())

Best Practices for Async Callbacks

Implement proper error handling in async callbacks:

async def safe_async_callback(message=None, **kwargs):
    try:
        await asyncio.sleep(0)
        # Process the callback
        await process_message(message)
    except Exception as e:
        logging.error(f"Async callback error: {e}")

Optimize async callbacks for better performance:

async def optimized_async_callback(message=None, **kwargs):
    # Use asyncio.gather for concurrent operations
    results = await asyncio.gather(
        process_message(message),
        update_status(),
        log_interaction(message)
    )
    return results

Key Considerations

Concurrency

Use asyncio.gather for parallel operations
Avoid blocking calls in async callbacks
Handle concurrent access to shared resources

Resource Management

Implement proper cleanup in async contexts
Use async context managers when appropriate
Monitor memory usage in long-running callbacks

When using async callbacks, ensure all dependent operations are also async-compatible to prevent blocking the event loop.

Troubleshooting

Callback Not Firing

Verify callback registration
Check verbose mode is enabled
Ensure correct callback name

Logging Issues

Check file permissions
Verify logging configuration
Monitor log file size

Next Steps

Async Callbacks

Learn about asynchronous callbacks

Advanced Patterns

Explore advanced callback patterns

Remember to handle callbacks efficiently and implement proper error handling to ensure smooth agent operations.

Was this page helpful?

Structured AI Agents PraisonAI Train

On this page

Quick Start
Understanding Callbacks
Available Callback Types
Basic Implementation
1. Simple Logging Callback
2. Multiple Callback Types
Best Practices
Complete Example
Advanced Examples
Complete Async Example
Available Async Functions
Example Usage
Best Practices for Async Callbacks
Key Considerations
Troubleshooting
Next Steps

Getting Started (Code)

Core Concepts

Getting Started (No Code)

User Interface

Features

Models

Tools (No Code)

Monitoring

Developers

API Reference

​Quick Start

​Understanding Callbacks