Hierarchical Workflows

Manager-based validation for graceful workflow failure handling

Hierarchical workflows add a manager agent that validates each step’s output before the next step runs, stopping gracefully with a clear reason if a step fails.

python
from praisonaiagents import Agent, AgentFlow

researcher = Agent(name="researcher", role="Research topics")
writer = Agent(name="writer", role="Write content")
workflow = AgentFlow(
    steps=[researcher, writer],
    process="hierarchical",
    manager_llm="gpt-4o-mini",
)
workflow.start("Summarise quantum computing for executives")

The user submits a multi-step brief; the manager validates each agent output before the workflow continues.

Diagram

Hierarchical Workflow

Approved

Rejected

Approved

Rejected

📄 User Brief

🤖 Agent 1

Manager
Validation

🤖 Agent 2

❌ Failed + Reason

Manager
Validation

✅ Result

How It Works

Diagram
AgentManagerAgentFlowUserAgentManagerAgentFlowUseralt[Approved][Rejected]Submit taskRun step 1OutputValidate outputProceed to step 2Stop with failure_reason{status: failed, failure_reason: …}{status: completed, output: …}
Diagram

Hierarchical Workflow

Approved

Rejected

Approved

Rejected

Input

Step 1: Agent

Manager
Validation

Step 2: Agent

Failed + Reason

Manager
Validation

Completed

Quick Start

  1. Simple Usage

    Python

    python
    from praisonaiagents import AgentFlow, Agent
    
    researcher = Agent(name="researcher", role="Research topics")
    writer = Agent(name="writer", role="Write content")
    
    workflow = AgentFlow(
        steps=[researcher, writer],
        process="hierarchical",  # Enable manager validation
        manager_llm="gpt-4o-mini"  # LLM for manager agent
    )
    
    result = workflow.start("Write about AI trends")
    
    if result["status"] == "failed":
        print(f"Workflow failed: {result['failure_reason']}")
    else:
        print(f"Output: {result['output']}")

    YAML

    yaml
    name: research-workflow
    process: hierarchical
    manager_llm: gpt-4o-mini
    
    agents:
      researcher:
        role: Research Analyst
        goal: Research topics thoroughly
        backstory: Expert researcher
    
      writer:
        role: Content Writer
        goal: Write engaging content
        backstory: Professional writer
    
    steps:
      - agent: researcher
        action: Research the topic
    
      - agent: writer
        action: Write article based on research
  2. With Configuration

    Define the same workflow in YAML with process: hierarchical and a manager_llm — see the YAML tab.

Parameters

process string

Workflow execution mode: – sequential – Steps run without validation (default) – hierarchical – Manager validates each step

manager_llm string

LLM model for the manager agent. If not specified, uses the workflow’s default_llm.

Manager Validation

The manager agent evaluates each step’s output based on:

  1. Task Completion

    Does the output address the task?

  2. Quality Check

    Is the output meaningful (not an error)?

  3. Expected Output

    Does it meet the step’s expected output criteria?

Handling Failures

Check Status
result = workflow.start("My task")

if result["status"] == "failed":
    # Workflow was rejected by manager
    print(f"Reason: {result['failure_reason']}")

    # Check which step failed
    for step in result["steps"]:
        if step["status"] == "failed":
            print(f"Failed step: {step['step']}")
elif result["status"] == "completed":
    print("All steps validated successfully!")
With Callbacks
def on_step_error(workflow, step, error):
    print(f"Step '{step.name}' failed: {error}")

workflow = AgentFlow(
    steps=[agent1, agent2],
    process="hierarchical",
    hooks={"on_step_error": on_step_error}
)

Result Structure

python
{
    "output": "Final output from last step",
    "steps": [
        {"step": "researcher", "output": "...", "status": "completed"},
        {"step": "writer", "output": "...", "status": "failed", "failure_reason": "..."}
    ],
    "variables": {...},
    "status": "failed",  # or "completed"
    "failure_reason": "Manager rejected step 'writer': Output lacks detail"
}

When to Use

Quality-Critical Workflows

When each step must meet quality standards before proceeding.

Multi-Agent Pipelines

When agents depend on validated output from previous agents.

Production Workflows

When you need graceful failure handling with clear reasons.

Forcing Tool Usage

Diagram

tool_choice: required

Agent with tools

Must call tool

Tool call → Response

tool_choice: auto (default)

May skip

May call

Agent with tools

LLM decides

Response without tool

Tool call → Response

YAML

yaml
agents:
  researcher:
    role: Research Specialist
    goal: Find latest information
    tools:
      - search_web
    tool_choice: required  # Forces tool usage
    llm: gpt-4o-mini

Python

python
from praisonaiagents import Agent

researcher = Agent(
    name="researcher",
    role="Research Specialist",
    tools=["search_web"],
    model="gpt-4o-mini"
)
# Set tool_choice for YAML workflows
researcher._yaml_tool_choice = "required"
tool_choice string

Controls when the LLM calls tools: – auto – LLM decides whether to call tools (default) – required – LLM must call a tool before responding – none – LLM cannot call tools

Comparison

FeatureSequentialHierarchical
ValidationNoneAfter each step
Failure HandlingContinues on errorStops with reason
PerformanceFasterSlightly slower
Use CaseSimple pipelinesQuality-critical flows

Real-World Examples

Research & Writing Pipeline
python
from praisonaiagents import AgentFlow, Agent

researcher = Agent(
    name="researcher",
    role="Research Analyst",
    goal="Find accurate, up-to-date information",
    backstory="Expert researcher with attention to detail",
    model="gpt-4o-mini"
)

writer = Agent(
    name="writer",
    role="Content Writer", 
    goal="Write engaging, clear content",
    backstory="Professional writer who creates compelling articles",
    model="gpt-4o-mini"
)

editor = Agent(
    name="editor",
    role="Editor",
    goal="Polish and improve content quality",
    backstory="Experienced editor with keen eye for detail",
    model="gpt-4o-mini"
)

workflow = AgentFlow(
    name="content-pipeline",
    steps=[researcher, writer, editor],
    process="hierarchical",
    manager_llm="gpt-4o-mini"
)

result = workflow.start("Write an article about AI in healthcare")
Data Analysis Pipeline
python
from praisonaiagents import AgentFlow, Agent

collector = Agent(
    name="collector",
    role="Data Collector",
    goal="Gather and prepare data",
    model="gpt-4o-mini"
)

analyst = Agent(
    name="analyst",
    role="Data Analyst",
    goal="Analyze data and find insights",
    model="gpt-4o-mini"
)

reporter = Agent(
    name="reporter",
    role="Report Writer",
    goal="Create clear reports from analysis",
    model="gpt-4o-mini"
)

workflow = AgentFlow(
    name="data-pipeline",
    steps=[collector, analyst, reporter],
    process="hierarchical",
    manager_llm="gpt-4o-mini"
)

result = workflow.start("Analyze Q4 sales trends")

if result["status"] == "failed":
    print(f"Pipeline failed: {result['failure_reason']}")
Code Review Pipeline
python
from praisonaiagents import AgentFlow, Agent

reader = Agent(
    name="code_reader",
    role="Code Reader",
    goal="Understand code structure",
    model="gpt-4o-mini"
)

reviewer = Agent(
    name="reviewer",
    role="Code Reviewer",
    goal="Find issues and improvements",
    model="gpt-4o-mini"
)

suggester = Agent(
    name="suggester",
    role="Improvement Suggester",
    goal="Suggest specific code improvements",
    model="gpt-4o-mini"
)

workflow = AgentFlow(
    name="code-review-pipeline",
    steps=[reader, reviewer, suggester],
    process="hierarchical",
    manager_llm="gpt-4o-mini",
    output="verbose"  # See validation results
)

result = workflow.start("Review this Python module for best practices")

Best Practices

Use a cheaper model for the manager

The manager only validates output, so a smaller manager_llm like gpt-4o-mini keeps cost low without hurting quality.

Force tool usage in tool-heavy steps

Set tool_choice="required" so the manager can confirm tools actually ran before approving a step.

Always check result status

Inspect result["status"] and result["failure_reason"] before using the output — hierarchical workflows fail gracefully rather than raising.

Keep steps single-purpose

Narrow, well-scoped steps make manager validation more reliable than broad, multi-goal steps.

Related