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.
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.
How It Works
Steps run one after another without validation
HierarchicalManager validates each step before proceeding
Quick Start
- 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 - With Configuration
Define the same workflow in YAML with process: hierarchical and a manager_llm — see the YAML tab.
Parameters
stringWorkflow execution mode: – sequential – Steps run without validation (default) – hierarchical – Manager validates each step
stringLLM 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:
- Task Completion
Does the output address the task?
- Quality Check
Is the output meaningful (not an error)?
- Expected Output
Does it meet the step’s expected output criteria?
Handling Failures
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!")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
{
"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
YAML
agents:
researcher:
role: Research Specialist
goal: Find latest information
tools:
- search_web
tool_choice: required # Forces tool usage
llm: gpt-4o-miniPython
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"stringControls 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
| Feature | Sequential | Hierarchical |
|---|---|---|
| Validation | None | After each step |
| Failure Handling | Continues on error | Stops with reason |
| Performance | Faster | Slightly slower |
| Use Case | Simple pipelines | Quality-critical flows |
Real-World Examples
Research & Writing Pipeline
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
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
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.