Skip to main content
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

PatternImportPurpose
parallel(steps, max_workers=N)from praisonaiagents import parallelRun steps concurrently
route({"label": [steps]})from praisonaiagents import routeBranch on step output
loop(fn, over="var")from praisonaiagents import loopIterate over a list variable
repeat(fn, until=cond)from praisonaiagents import repeatEvaluator-optimizer loop
when(condition, then_steps, else_steps)from praisonaiagents import whenConditional block
include(workflow=wf)from praisonaiagents import includeEmbed 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 valueEffective concurrency
None (default)min(3, len(steps))
1Sequential — useful for rate-limited providers
N ≤ len(steps)N
N > 3Honored; 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:
OptionTypeDefaultDescription
on_workflow_startCallableNoneCalled once before the first step
on_workflow_completeCallableNoneCalled once after the last step
on_step_startCallableNoneCalled before each step
on_step_completeCallableNoneCalled after each step succeeds
on_step_errorCallableNoneCalled 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},
)
OptionTypeDefaultDescription
enabledboolTrueEnable planning mode
llmstrNoneLLM for planning (defaults to workflow LLM)
reasoningboolFalseEnable chain-of-thought reasoning
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:
OptionTypeDefaultDescription
backendstr"file"Memory backend ("file", "chroma", "redis")
user_idstrNoneUser identifier for memory scoping
session_idstrNoneSession identifier
configdictNoneBackend-specific configuration
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?")
PresetDescription
silentNo output (default, lowest overhead)
minimalFinal answer only
statusTool calls + final output inline
traceTimestamped execution trace
verboseFull Rich panels
debugTrace + 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:
PlaceholderResolves 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(...)
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

Give each step a distinct agent with a focused role — Researcher, Analyst, Writer. Agents with clear instructions outperform generalist ones.
LLM-backed parallel steps hit rate limits fast. Start with max_workers=3 and increase only after testing.
Use Task(name="...") for every step. Named steps appear in workflow.step_statuses and execution history.
Add guardrails= to steps that produce structured output (reports, JSON, code). Short feedback loops via max_retries cost less than prompt engineering.
Use when(condition="{{score}} > 80", ...) for simple thresholds. Reserve route() for multi-branch decisions.
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