Workflows

Orchestrate agents in sequential, parallel, routed, and looped pipelines with a single AgentFlow call

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.

Quick Start

1. 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"])
   
2. 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"])
   
3. 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

silent (default)

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

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

Async

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"})
)

---

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.

---

Related

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