AI Agents with Callbacks

# 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)

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()

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()

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)

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())