Multi-Agent Hooks

Add lifecycle callbacks to multi-agent workflows for monitoring and custom logic

Multi-Agent Hooks let you run custom code when tasks start, complete, or when you need custom completion detection in multi-agent workflows.

from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

def on_task_start(task, task_id):
    print(f"Starting: {task.name} (id={task_id})")

def on_task_complete(task, task_output):
    print(f"Done: {task.name}")

agent = Agent(name="Researcher", instructions="Research topics thoroughly.")
task = Task(description="Research quantum computing", agent=agent)

team = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(
        on_task_start=on_task_start,
        on_task_complete=on_task_complete,
    ),
)
team.start()

The user registers hooks; the workflow fires them as each task starts and completes.

Quick Start

1. Task Lifecycle Logging

   Log every task transition with on_task_start and on_task_complete:

   from praisonaiagents import Agent, Task, PraisonAIAgents
   from praisonaiagents import MultiAgentHooksConfig
   import time
   
   start_times = {}
   
   def on_start(task, task_id):
       start_times[task.name] = time.time()
       print(f"[START] {task.name} (id={task_id})")
   
   def on_complete(task, task_output):
       duration = time.time() - start_times.get(task.name, time.time())
       print(f"[DONE]  {task.name} ({duration:.1f}s)")
   
   researcher = Agent(name="Researcher", instructions="Research topics thoroughly.")
   writer = Agent(name="Writer", instructions="Write clear summaries.")
   
   research_task = Task(description="Research quantum computing", agent=researcher)
   write_task = Task(description="Write a summary", agent=writer)
   
   team = PraisonAIAgents(
       agents=[researcher, writer],
       tasks=[research_task, write_task],
       hooks=MultiAgentHooksConfig(
           on_task_start=on_start,
           on_task_complete=on_complete,
       )
   )
   
   team.start()

2. Async Workflow with async def Hooks

   Run the team with astart() and use async def callbacks to await I/O directly:

   import asyncio
   from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig
   
   async def on_start(task, task_id):
       print(f"[START] {task.name} (id={task_id})")
   
   async def on_complete(task, task_output):
       # await your async I/O directly — no wrapper needed
       print(f"[DONE]  {task.name}: {task_output.raw[:80]}")
   
   agent = Agent(name="Researcher", instructions="Research topics thoroughly.")
   task  = Task(description="Research quantum computing", agent=agent)
   
   team = PraisonAIAgents(
       agents=[agent],
       tasks=[task],
       hooks=MultiAgentHooksConfig(
           on_task_start=on_start,       # async def — awaited
           on_task_complete=on_complete, # async def — awaited
       ),
   )
   
   asyncio.run(team.astart())  # fires the same hooks as .start()

3. Custom Completion Checker

   Override the default completion logic with your own validation:

   from praisonaiagents import Agent, Task, PraisonAIAgents
   from praisonaiagents import MultiAgentHooksConfig
   
   def quality_checker(task, task_output):
       """Return True if output meets quality standards."""
       if len(task_output) < 100:
           print(f"Output too short ({len(task_output)} chars), retrying...")
           return False
       if "error" in task_output.lower():
           print("Output contains error, retrying...")
           return False
       return True
   
   agent = Agent(name="Analyst", instructions="Provide detailed analysis.")
   task = Task(description="Analyze the market trends for 2025", agent=agent)
   
   team = PraisonAIAgents(
       agents=[agent],
       tasks=[task],
       hooks=MultiAgentHooksConfig(
           completion_checker=quality_checker,
           on_task_complete=lambda t, o: print(f"✅ {t.name} passed quality check"),
       ),
   )
   team.start()

How It Works

Async and Sync Callbacks

MultiAgentHooksConfig accepts either regular functions or async def coroutines for on_task_start and on_task_complete.

• In sync workflows (team.start()), only sync callables are meaningful — an async def here would return an un-awaited coroutine.
• In async workflows (team.astart() / arun_task()), the framework detects the callable type and does the right thing:
• async def callbacks are awaited inline.
• Sync callbacks are offloaded to the default executor via loop.run_in_executor(None, ...), so a blocking hook can’t stall the event loop.
• Exceptions raised by either callback are logged (logger.error) and swallowed — they never abort the task.

Global Variables Propagation

Pass variables= once at the team level and every task substitutes them into its description at runtime:

from praisonaiagents import Agent, Task, PraisonAIAgents

agent = Agent(name="Writer", instructions="Write a paragraph on {topic} for {audience}.")
task  = Task(description="Write on {topic}", agent=agent)

team = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    variables={"topic": "quantum computing", "audience": "beginners"},
)
team.start()

Values passed to variables= are shallow-copied into task.variables before each task runs (unless the task defines its own variables), then substituted into task.description. The shallow copy means a task mutating its own variables no longer leaks back into the team-level dict, so parallel tasks in a process="workflow" team with async_execution=True on each task don’t stomp on each other.

Configuration Options

Common Patterns

Pattern 1 — Logging workflow progress

from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig
import time

start_times = {}

def log_start(task, task_id):
    start_times[task.name] = time.time()
    print(f"[START] {task.name}")

def log_complete(task, task_output):
    elapsed = time.time() - start_times.get(task.name, 0)
    print(f"[DONE] {task.name} in {elapsed:.1f}s")

agent = Agent(name="Worker", instructions="Complete tasks efficiently.")
task = Task(description="Analyze quarterly sales data", agent=agent)

team = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(on_task_start=log_start, on_task_complete=log_complete),
)
team.start()

Pattern 2 — Quality gate with retry logic

from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

def length_checker(task, task_output):
    return len(task_output) >= 200

agent = Agent(name="Analyst", instructions="Write comprehensive analyses.")
task = Task(description="Analyze the impact of remote work on productivity", agent=agent)

response = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(completion_checker=length_checker),
).start()
print(response)

Pattern 3 — Async I/O in a completion hook

import asyncio
from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

async def notify(task, task_output):
    # await a DB write or HTTP call directly inside the hook
    print(f"Notifying downstream: {task.name}")

agent = Agent(name="Worker", instructions="Do the work.")
task = Task(description="Process batch #42", agent=agent)

team = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(on_task_complete=notify),
)
asyncio.run(team.astart())

Best Practices

Related