Workflows - PraisonAI

from praisonaiagents import Agent, AgentFlow

researcher = Agent(name="Researcher", instructions="Research the given topic thoroughly.")
writer     = Agent(name="Writer",     instructions="Write a concise summary of the research.")

workflow = AgentFlow(steps=[researcher, writer])
result   = workflow.start("Latest breakthroughs in AI agents")
print(result["output"])

AgentFlow runs agents as sequential steps, passing each output into the next as context.

AgentFlow(model="gpt-4o") is now accepted (canonical) as the default model for steps that did not name one. llm= still works as a deprecated alias — see Model Parameter.

Quick Start

Sequential pipeline

from praisonaiagents import Agent, AgentFlow

analyst    = Agent(name="Analyst",    instructions="Analyze the topic and extract key insights.")
strategist = Agent(name="Strategist", instructions="Develop actionable strategies from the analysis.")
presenter  = Agent(name="Presenter",  instructions="Summarize strategies into a clear presentation.")

workflow = AgentFlow(steps=[analyst, strategist, presenter])
result   = workflow.start("Market trends in AI industry 2024")
print(result["output"])

Parallel research

from praisonaiagents import Agent, AgentFlow, parallel

market     = Agent(name="Market",     instructions="Research market trends.")
competitor = Agent(name="Competitor", instructions="Research competitor landscape.")
customer   = Agent(name="Customer",   instructions="Research customer needs.")
aggregator = Agent(name="Aggregator", instructions="Synthesize all research into a unified report.")

workflow = AgentFlow(steps=[
    parallel([market, competitor, customer], max_workers=3),
    aggregator,
])
result = workflow.start("AI industry 2024")
print(result["output"])

Conditional routing

from praisonaiagents import Agent, AgentFlow, WorkflowContext, StepResult, route

tech      = Agent(name="TechExpert",  instructions="Answer technical questions in depth.")
creative  = Agent(name="Creative",    instructions="Write engaging creative content.")
general   = Agent(name="General",     instructions="Provide a helpful general-purpose answer.")

def classify(ctx: WorkflowContext) -> StepResult:
    text = ctx.input.lower()
    if any(k in text for k in ("code", "python", "api", "debug")):
        return StepResult(output="technical")
    if any(k in text for k in ("story", "poem", "creative")):
        return StepResult(output="creative")
    return StepResult(output="default")

workflow = AgentFlow(steps=[
    classify,
    route({"technical": [tech], "creative": [creative], "default": [general]}),
])
result = workflow.start("Write a Python function to sort a list")
print(result["output"])

How It Works

Each step receives the previous step’s output as context. Context passing is automatic — no extra code required.

Which Pattern to Use?

Pattern Helpers

Pattern	Import	Purpose
`parallel(steps, max_workers=N)`	`from praisonaiagents import parallel`	Run steps concurrently
`route({"label": [steps]})`	`from praisonaiagents import route`	Branch on step output
`loop(fn, over="var")`	`from praisonaiagents import loop`	Iterate over a list variable
`repeat(fn, until=cond)`	`from praisonaiagents import repeat`	Evaluator-optimizer loop
`when(condition, then_steps, else_steps)`	`from praisonaiagents import when`	Conditional block
`include(workflow=wf)`	`from praisonaiagents import include`	Embed a sub-workflow

from praisonaiagents import AgentFlow, Agent, WorkflowContext, StepResult
from praisonaiagents import route, parallel, loop, repeat, when, include

agent_a = Agent(name="A", instructions="Handle task A.")
agent_b = Agent(name="B", instructions="Handle task B.")
worker  = Agent(name="Worker", instructions="Process one item.")

sub_flow = AgentFlow(name="sub", steps=[agent_b])

def classify(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="a" if "a" in ctx.input.lower() else "b")

def process_item(ctx: WorkflowContext) -> StepResult:
    item = ctx.variables.get("item", "")
    return StepResult(output=f"Processed: {item}")

workflow = AgentFlow(
    steps=[
        classify,
        route({"a": [agent_a], "default": [agent_b]}),
        parallel([agent_a, agent_b], max_workers=2),
        loop(process_item, over="items"),
        repeat(worker, until=lambda ctx: len(ctx.previous_result or "") > 50, max_iterations=3),
        when(condition="{{score}} > 80", then_steps=[agent_a], else_steps=[agent_b]),
        include(workflow=sub_flow),
    ],
    variables={"items": ["AI", "ML", "NLP"]},
)

`parallel()` and `max_workers`

max_workers caps concurrent steps in parallel() and loop(parallel=True).

`max_workers` value	Effective concurrency
`None` (default)	`min(3, len(steps))`
`1`	Sequential — useful for rate-limited providers
`N ≤ len(steps)`	`N`
`N > 3`	Honored; an info log is emitted

Workflow Hooks

from praisonaiagents import Agent, AgentFlow, WorkflowHooksConfig

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Write a summary.")

workflow = AgentFlow(
    steps=[researcher, writer],
    hooks=WorkflowHooksConfig(
        on_workflow_start=lambda w, i: print(f"🚀 Starting: {i}"),
        on_step_start=lambda name, ctx: print(f"▶ {name}"),
        on_step_complete=lambda name, r: print(f"✅ {name}: {str(r.output)[:60]}"),
        on_step_error=lambda name, e: print(f"❌ {name}: {e}"),
        on_workflow_complete=lambda w, r: print(f"🎉 Done: {r['status']}"),
    ),
)
result = workflow.start("AI trends 2024")

WorkflowHooksConfig options:

Option	Type	Default	Description
`on_workflow_start`	`Callable`	`None`	Called once before the first step
`on_workflow_complete`	`Callable`	`None`	Called once after the last step
`on_step_start`	`Callable`	`None`	Called before each step
`on_step_complete`	`Callable`	`None`	Called after each step succeeds
`on_step_error`	`Callable`	`None`	Called when a step raises an exception

Planning & Reasoning

from praisonaiagents import Agent, AgentFlow

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Write a summary.")

workflow = AgentFlow(
    steps=[researcher, writer],
    planning=True,  # or a dict: {"llm": "gpt-4o", "reasoning": True}
)
result = workflow.start("Research and write about AI trends")

Pass planning=True to enable it, or a dict to tune the planner:

workflow = AgentFlow(
    steps=[researcher, writer],
    planning={"llm": "gpt-4o", "reasoning": True},
)

Option	Type	Default	Description
`enabled`	`bool`	`True`	Enable planning mode
`llm`	`str`	`None`	LLM for planning (defaults to workflow LLM)
`reasoning`	`bool`	`False`	Enable chain-of-thought reasoning

Advanced: WorkflowPlanningConfig class

Prefer the bool or dict form above. Use WorkflowPlanningConfig when you want typed, IDE-friendly configuration.

from praisonaiagents import Agent, AgentFlow, WorkflowPlanningConfig

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Write a summary.")

workflow = AgentFlow(
    steps=[researcher, writer],
    planning=WorkflowPlanningConfig(enabled=True, llm="gpt-4o", reasoning=True),
)
result = workflow.start("Research and write about AI trends")

Memory Integration

from praisonaiagents import Agent, AgentFlow

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Continue from previous research.")

workflow = AgentFlow(
    steps=[researcher, writer],
    memory={"backend": "chroma", "config": {"persist": True, "collection": "my_workflow"}},
)

result1 = workflow.start("Research AI trends")
result2 = workflow.start("Continue the research on LLMs")

Pass memory=True for the default file backend, or a dict to pick a backend:

Option	Type	Default	Description
`backend`	`str`	`"file"`	Memory backend (`"file"`, `"chroma"`, `"redis"`)
`user_id`	`str`	`None`	User identifier for memory scoping
`session_id`	`str`	`None`	Session identifier
`config`	`dict`	`None`	Backend-specific configuration

Advanced: WorkflowMemoryConfig class

Prefer the bool or dict form above. Use WorkflowMemoryConfig when you want typed, IDE-friendly configuration.

from praisonaiagents import Agent, AgentFlow, WorkflowMemoryConfig

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Continue from previous research.")

workflow = AgentFlow(
    steps=[researcher, writer],
    memory=WorkflowMemoryConfig(backend="chroma", config={"persist": True, "collection": "my_workflow"}),
)
result = workflow.start("Research AI trends")

Guardrails & Validation

from praisonaiagents import AgentFlow, Task, WorkflowContext, StepResult

def generate_content(ctx: WorkflowContext) -> StepResult:
    return StepResult(output=f"Article about: {ctx.input}")

def validate_output(result: StepResult):
    if len(result.output) < 50:
        return (False, "Content too short — please expand to at least 50 characters.")
    return (True, None)

workflow = AgentFlow(steps=[
    Task(
        name="generator",
        handler=generate_content,
        guardrails=validate_output,
        max_retries=3,
    )
])
result = workflow.start("AI breakthroughs 2024")

Output Modes

from praisonaiagents import Agent, AgentFlow

workflow = AgentFlow(
    steps=[Agent(name="A", instructions="Answer the question.")],
    output="silent",
)
result = workflow.start("What is AI?")

from praisonaiagents import Agent, AgentFlow

workflow = AgentFlow(
    steps=[Agent(name="A", instructions="Answer the question.")],
    output="status",
)
result = workflow.start("What is AI?")

from praisonaiagents import Agent, AgentFlow

workflow = AgentFlow(
    steps=[Agent(name="A", instructions="Answer the question.")],
    output="trace",
)
result = workflow.start("What is AI?")

from praisonaiagents import Agent, AgentFlow

workflow = AgentFlow(
    steps=[Agent(name="A", instructions="Answer the question.")],
    output="verbose",
)
result = workflow.start("What is AI?")

Preset	Description
`silent`	No output (default, lowest overhead)
`minimal`	Final answer only
`status`	Tool calls + final output inline
`trace`	Timestamped execution trace
`verbose`	Full Rich panels
`debug`	Trace + metrics

Async Execution

import asyncio
from praisonaiagents import Agent, AgentFlow

workflow = AgentFlow(steps=[
    Agent(name="Researcher", instructions="Research the topic."),
    Agent(name="Writer",     instructions="Summarize the research."),
])

async def main():
    result = await workflow.astart("AI trends 2024")
    print(result["output"])

asyncio.run(main())

Status Tracking

from praisonaiagents import Agent, AgentFlow, WorkflowContext, StepResult

def step1(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="step1 done")

def step2(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="step2 done")

workflow = AgentFlow(steps=[step1, step2])
result = workflow.start("input")

print(workflow.status)         # "completed"
print(workflow.step_statuses)  # {"step1": "completed", "step2": "completed"}

Variable Substitution

Every action string processes placeholders in this order:

Placeholder	Resolves to
`{{your_variable}}`	Value from `variables={}` or earlier step’s `output_variable`
`{{previous_output}}`	Previous step output — auto-appended if omitted
`{{step_name_output}}`	Named step output
`{{input}}`	Value passed to `workflow.start(...)` / `workflow.run(input=...)`, or the YAML top-level `input:` field (`workflow.default_input`) when no argument is provided.

For YAML workflows loaded via YAMLWorkflowParser, the top-level input: field is stored as workflow.default_input and used automatically when start() / run() is called with no argument. See Workflow Input Resolution for the full precedence ladder.

from praisonaiagents import Agent, AgentFlow, Task

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer     = Agent(name="Writer",     instructions="Write a one-line tweet.")

workflow = AgentFlow(steps=[
    Task(name="research", agent=researcher, action="Research {{input}}"),
    Task(name="summary",  agent=writer,     action="Summarise: {{previous_output}}"),
    Task(name="tweet",    agent=writer,     action="Write a one-line tweet."),
])
workflow.start("AI agents")

{{today}}, {{now}}, and {{uuid}} are provided by the separate Dynamic Variables mechanism, not the workflow template engine.

WorkflowManager

WorkflowManager discovers and executes markdown-defined workflows from .praisonai/workflows/.

AgentFlow is the primary workflow surface. WorkflowManager is a secondary API for running markdown-defined workflows.

from praisonaiagents import Agent, WorkflowManager

agent   = Agent(name="Assistant", instructions="Execute workflow steps.")
manager = WorkflowManager()

workflows = manager.list_workflows()
result    = manager.execute(
    workflow_name="deploy",
    default_agent=agent,
    variables={"environment": "staging"},
)
print(result)

Checkpoints

Save progress with checkpoint= (saved after each step), continue an interrupted run with resume=, and — only when a real edit happened — override a fingerprint mismatch with rebase_checkpoint=True.

# Save a checkpoint after each completed step
result = manager.execute("research-pipeline", default_agent=agent,
                         checkpoint="my-checkpoint")

# Resume from where it stopped
result = manager.execute("research-pipeline", default_agent=agent,
                         resume="my-checkpoint")
print(result.get("resumed_from_step"))   # 0-indexed int when resumed; absent otherwise

Definition fingerprint. Each checkpoint stores a stable content hash of the workflow’s steps (each step’s name, action, agent name, and full agent config). Whitespace- or comment-only edits keep the same fingerprint, so a resume still works. Adding, removing, or reordering steps — or changing an agent config value such as instructions, model, tools, condition, or routing — produces a new fingerprint. Fail-closed on a missing checkpoint. Resuming onto a name that does not exist refuses to run rather than silently restarting from step 1:

result = manager.execute("research-pipeline", default_agent=agent,
                         resume="does-not-exist")
print(result["success"])   # False
print(result["error"])     # checkpoint 'does-not-exist' not found; refusing to resume. ...

Fingerprint mismatch. After a real edit, resume refuses and the error names both fingerprints and points at the remediation:

workflow definition changed since the checkpoint (fingerprint mismatch:
checkpoint=a1b2c3d4e5f6, current=9f8e7d6c5b4a). Re-run without --resume to start
fresh, or pass --rebase-checkpoint to deliberately continue at the same step
index against the edited definition.

Pass rebase_checkpoint=True to continue at the same step index against the edited workflow (a warning is logged naming both fingerprints):

result = manager.execute("research-pipeline", default_agent=agent,
                         resume="my-checkpoint", rebase_checkpoint=True)

Situation	Behaviour
Resume, no checkpoint by that name	Refuses to run; error names the missing checkpoint
Resume, file edited (whitespace / comments only)	Runs; same fingerprint
Resume, file edited (steps added/reordered, agent config changed)	Refuses; error names both fingerprints and the `--rebase-checkpoint` remediation
Resume with `rebase_checkpoint=True` after a real edit	Runs at the same numeric step index; warning logged
Loop step (`loop_over`)	Checkpoint written once after the whole loop completes; resume skips the loop entirely

For a full walkthrough — CLI flags, the workflow checkpoints command, the checkpoint-file schema, and a decision diagram — see Workflow Checkpoint & Resume.

Async

aexecute() accepts the same checkpoint, resume, and rebase_checkpoint parameters (and returns the same resumed_from_step field) and supports loop_over and branch_condition routing with full parity to execute().

import asyncio
from praisonaiagents import Agent, WorkflowManager

agent   = Agent(name="Assistant", instructions="Execute workflow steps.")
manager = WorkflowManager()

result = asyncio.run(
    manager.aexecute("research-pipeline", default_agent=agent, variables={"topic": "AI"})
)

# Same checkpoint / resume as the sync path
result = asyncio.run(
    manager.aexecute("research-pipeline", default_agent=agent, checkpoint="my-checkpoint")
)
result = asyncio.run(
    manager.aexecute("research-pipeline", default_agent=agent, resume="my-checkpoint")
)

Loop Steps: Error Handling & Checkpoints

A loop_over step runs its step once per item. Two behaviors apply on both execute() and aexecute():

Stop-on-error aborts the workflow. When a loop_over step fails on any iteration and its on_error="stop" (the Task default), the entire workflow aborts — not just the loop. result["success"] becomes False and no later steps run.
Completed loop steps checkpoint once. With checkpoint= set, a loop_over step persists a checkpoint after the whole loop completes (matching single-step behavior). On resume=, the loop is not re-run — execution continues from the next step.

Error Handling Across Patterns

As of PraisonAI #4194, the same failure rules now apply inside every nested pattern (loop, parallel, when/if_, route, repeat), not just loop_over:

on_error="stop" (the Task default) inside any nested pattern halts the whole workflow — the stop signal propagates out of the pattern to the top level.
on_error="continue" keeps the workflow running, but the final result["status"] is "failed" if any step failed.
Downstream steps never receive the exception text as input; the error is recorded on the step, not folded into its output.

See Error Handling for the full guide, including parallel(on_failure=...) modes.

Workflow File Format

Markdown files with YAML frontmatter define multi-step workflows for WorkflowManager:

---
name: Research Pipeline
default_llm: gpt-4o-mini
planning: true
variables:
  topic: AI trends
---

## Step 1: Research
Research the topic thoroughly.

```agent
role: Researcher
goal: Find comprehensive information
instructions: Expert researcher with broad domain knowledge
```

```action
Search for information about {{topic}}
```

output_variable: research_data

## Step 2: Analyze

```agent
role: Analyst
goal: Analyze data patterns
```

```action
Analyze: {{research_data}}
```

if_() is deprecated. Use when() instead for conditional steps. if_() will be removed in a future release.

Best Practices

Use specialized agents per step

Give each step a distinct agent with a focused role — Researcher, Analyst, Writer. Agents with clear instructions outperform generalist ones.

Set max_workers for parallel steps

LLM-backed parallel steps hit rate limits fast. Start with max_workers=3 and increase only after testing.

Name your steps for debugging

Use Task(name="...") for every step. Named steps appear in workflow.step_statuses and execution history.

Guardrails over hope

Add guardrails= to steps that produce structured output (reports, JSON, code). Short feedback loops via max_retries cost less than prompt engineering.

Prefer when() over complex routing

Use when(condition="{{score}} > 80", ...) for simple thresholds. Reserve route() for multi-branch decisions.

Keep steps atomic

One step → one responsibility. Easier to debug, retry, and replace.

YAML Workflows

Define complex workflows in YAML files

Nested Workflows

Combine loops, parallel, and routing patterns

Hybrid Workflows

Mix deterministic shell steps with agent steps

Job Workflows

Ordered pipelines of shell commands and agent steps

​Quick Start

​How It Works

​Which Pattern to Use?

​Pattern Helpers

​parallel() and max_workers

​Workflow Hooks

​Planning & Reasoning

​Memory Integration

​Guardrails & Validation

​Output Modes

​Async Execution

​Status Tracking

​Variable Substitution

​WorkflowManager

​Checkpoints

​Async

​Loop Steps: Error Handling & Checkpoints

​Error Handling Across Patterns

​Workflow File Format

​Best Practices

​Related

YAML Workflows

Nested Workflows

Hybrid Workflows

Job Workflows

Quick Start

How It Works

Which Pattern to Use?

Pattern Helpers

`parallel()` and `max_workers`

Workflow Hooks

Planning & Reasoning

Memory Integration

Guardrails & Validation

Output Modes

Async Execution

Status Tracking

Variable Substitution

WorkflowManager

Checkpoints

Async

Loop Steps: Error Handling & Checkpoints

Error Handling Across Patterns

Workflow File Format

Best Practices

Related