> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Workflows

> Create reusable multi-step workflows with context passing and per-step agents

# Workflows

Create and execute reusable multi-step workflows with advanced features like context passing between steps, per-step agent configuration, and async execution. Define complex task sequences in markdown files and execute them programmatically.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Workflow Definition"
        MD["📄 .praison/workflows/deploy.md"]
    end
    
    subgraph "WorkflowManager"
        DISCOVER["🔍 Discover"]
        PARSE["📝 Parse Steps"]
        VARS["🔄 Substitute Variables"]
        CTX["📋 Context Passing"]
    end
    
    subgraph "Execution"
        STEP1["Step 1: Research<br/>🤖 Researcher Agent"]
        STEP2["Step 2: Analyze<br/>🤖 Analyst Agent"]
        STEP3["Step 3: Write<br/>🤖 Writer Agent"]
    end
    
    MD --> DISCOVER
    DISCOVER --> PARSE
    PARSE --> VARS
    VARS --> CTX
    CTX --> STEP1
    STEP1 -->|"context"| STEP2
    STEP2 -->|"context"| STEP3
    
    classDef def fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef manager fill:#2E8B57,stroke:#7C90A0,color:#fff
    classDef exec fill:#189AB4,stroke:#7C90A0,color:#fff
    
    class MD def
    class DISCOVER,PARSE,VARS,CTX manager
    class STEP1,STEP2,STEP3 exec
```

## Key Features

| Feature                   | Description                                                        |
| ------------------------- | ------------------------------------------------------------------ |
| **Context Passing**       | Automatically pass outputs from previous steps to subsequent steps |
| **Per-Step Agents**       | Configure different agents with unique roles for each step         |
| **Per-Step Tools**        | Assign specific tools to each step                                 |
| **Async Execution**       | Execute workflows asynchronously with `aexecute()`                 |
| **Variable Substitution** | Use `{{previous_output}}` and `{{step_name_output}}`               |
| **Planning Mode**         | Enable planning mode at workflow level                             |

## Quick Start

<CodeGroup>
  ```python Agentic Workflow theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow

  # Create agents with specific roles
  researcher = Agent(
      name="Researcher",
      role="Research Analyst",
      goal="Research and provide information about topics",
      instructions="You are a research analyst. Provide concise, factual information."
  )

  writer = Agent(
      name="Writer", 
      role="Content Writer",
      goal="Write engaging content based on research",
      instructions="You are a content writer. Write clear, engaging content."
  )

  # Create workflow with agents as steps
  workflow = AgentFlow(steps=[researcher, writer])

  # Run workflow - agents process sequentially
  result = workflow.start("What are the key benefits of AI agents?")
  print(result["output"])
  ```

  ```python Multi-Agent Pipeline theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow

  # Create specialized agents
  analyst = Agent(
      name="Analyst",
      role="Data Analyst",
      goal="Analyze data and extract insights",
      instructions="Analyze the given topic and provide key insights."
  )

  strategist = Agent(
      name="Strategist",
      role="Strategy Expert", 
      goal="Develop strategies based on analysis",
      instructions="Based on the analysis, develop actionable strategies."
  )

  presenter = Agent(
      name="Presenter",
      role="Presentation Expert",
      goal="Create clear presentations",
      instructions="Summarize the strategies into a clear presentation format."
  )

  # Sequential workflow: Analyst -> Strategist -> Presenter
  workflow = AgentFlow(steps=[analyst, strategist, presenter])
  result = workflow.start("Market trends in AI industry 2024")
  print(result["output"])
  ```

  ```python With Callbacks theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow

  researcher = Agent(name="Researcher", role="Researcher", goal="Research topics")
  writer = Agent(name="Writer", role="Writer", goal="Write content")

  workflow = AgentFlow(
      steps=[researcher, writer],
      hooks={
          "on_step_complete": lambda name, r: print(f"✅ {name} completed")
      }
  )

  result = workflow.start("Explain quantum computing")
  ```
</CodeGroup>

## Workflow Class API

The `Workflow` class provides a powerful programmatic API for creating workflows with functions, agents, and pattern helpers.

### Basic Usage

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, WorkflowContext, StepResult
from praisonaiagents import route, parallel, loop, repeat

# Define step functions
def step1(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="result")

# Create workflow with callbacks using hooks= consolidated param
from praisonaiagents import AgentFlowHooksConfig

workflow = AgentFlow(
    steps=[step1, step2],
    hooks=WorkflowHooksConfig(
        on_workflow_start=lambda w, i: print(f"Starting: {i}"),
        on_workflow_complete=lambda w, r: print(f"Done: {r['status']}"),
        on_step_start=lambda name, ctx: print(f"Step: {name}"),
        on_step_complete=lambda name, r: print(f"{name}: {r.output}"),
        on_step_error=lambda name, e: print(f"Error in {name}: {e}")
    )
)

# Run
result = workflow.start("input")
```

### Workflow with Agents

Use Agent objects directly as workflow steps:

<CodeGroup>
  ```python Sequential Agents theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow

  researcher = Agent(name="Researcher", role="Research expert", tools=[tavily_search])
  writer = Agent(name="Writer", role="Content writer")
  editor = Agent(name="Editor", role="Editor")

  workflow = AgentFlow(steps=[researcher, writer, editor])
  result = workflow.start("Research and write about AI")
  ```

  ```python Parallel Agents theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow
  from praisonaiagents import parallel

  agent1 = Agent(name="Researcher1", role="Market researcher")
  agent2 = Agent(name="Researcher2", role="Competitor researcher")
  agent3 = Agent(name="Researcher3", role="Customer researcher")
  aggregator = Agent(name="Aggregator", role="Summarizer")

  workflow = AgentFlow(steps=[
      parallel([agent1, agent2, agent3], max_workers=3),
      aggregator
  ])
  result = workflow.start("Research the market")
  ```
</CodeGroup>

### Controlling concurrency with `max_workers`

`max_workers` caps how many steps run concurrently in a `parallel()` or a `loop(parallel=True)` block.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, AgentFlow, parallel

researcher = Agent(name="Researcher", role="Market researcher")
competitor = Agent(name="Competitor", role="Competitor researcher")
customer   = Agent(name="Customer",   role="Customer researcher")
aggregator = Agent(name="Aggregator", role="Summarizer")

workflow = AgentFlow(steps=[
    parallel([researcher, competitor, customer], max_workers=3),
    aggregator,
])
result = workflow.start("Research the market")
```

| `max_workers` value | Effective concurrency               | Notes                                         |
| ------------------- | ----------------------------------- | --------------------------------------------- |
| `None` (default)    | `min(3, len(steps))`                | Safe default, protects against rate limits    |
| `1`                 | Sequential                          | Useful for serializing rate-limited providers |
| `N ≤ len(steps)`    | `N`                                 | User value wins                               |
| `N > 3`             | Honored, but an info log is emitted | Consider adding a rate limiter                |

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    Q{How many parallel steps?} --> A[1-3 steps<br/>LLM-backed]
    Q --> B[4-10 steps<br/>LLM-backed]
    Q --> C[Many steps<br/>pure-compute]
    A --> A1[Leave max_workers unset]
    B --> B1[max_workers=N<br/>+ add rate limiter]
    C --> C1[max_workers=N<br/>no rate limiter needed]

    classDef q fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef opt fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef ans fill:#10B981,stroke:#7C90A0,color:#fff
    class Q q
    class A,B,C opt
    class A1,B1,C1 ans
```

<CodeGroup>
  ```python Route to Agents theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import Agent, AgentFlow
  from praisonaiagents import route

  tech_agent = Agent(name="TechExpert", role="Technical expert")
  creative_agent = Agent(name="Creative", role="Creative writer")
  general_agent = Agent(name="General", role="General assistant")

  workflow = AgentFlow(steps=[
      classifier_function,
      route({
          "technical": [tech_agent],
          "creative": [creative_agent],
          "default": [general_agent]
      })
  ])
  result = workflow.start("Help me with this task")
  ```
</CodeGroup>

### Planning & Reasoning

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlowPlanningConfig

workflow = AgentFlow(
    steps=[researcher, writer, editor],
    planning=WorkflowPlanningConfig(
        enabled=True,        # Create execution plan before running
        llm="gpt-4o",        # LLM for planning
        reasoning=True       # Enable chain-of-thought reasoning
    )
)
result = workflow.start("Research and write about AI trends")
```

### Tools per Step

<CodeGroup>
  ```python Via Task theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Task

  workflow = AgentFlow(steps=[
      Task(
          name="research",
          action="Research {{topic}}",
          tools=[tavily_search, web_scraper]
      ),
      Task(
          name="write",
          action="Write article based on: {{previous_output}}",
          tools=[file_writer]
      )
  ])
  ```

  ```python Via agent_config theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Task

  workflow = AgentFlow(steps=[
      Task(
          name="research",
          action="Research {{topic}}",
          agent_config={
              "name": "Researcher",
              "role": "Expert",
              "tools": [tavily_search]
          }
      )
  ])
  ```
</CodeGroup>

### Guardrails & Validation

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, Task

def validate_output(result):
    """Returns (is_valid, feedback_message)"""
    if "error" in result.output.lower():
        return (False, "Please fix the error and try again")
    return (True, None)

workflow = AgentFlow(steps=[
    Task(
        name="generator",
        handler=generate_content,
        guardrails=validate_output,  # Validation function
        max_retries=3               # Auto-retry on failure
    )
])
```

### Output Modes

Control how workflow execution is displayed. Workflows use the same output presets as Agent for consistency.

<CodeGroup>
  ```python Status Output (Recommended for CLI) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Agent

  # Shows tool calls and final output inline
  workflow = AgentFlow(
      steps=[Agent(instructions="Research AI trends")],
      output="status"  # Shows: ▸ tool → result ✓
  )
  result = workflow.start("Research AI")
  ```

  ```python Trace Output (With Timestamps) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Agent

  # Shows timestamped execution trace
  workflow = AgentFlow(
      steps=[Agent(instructions="Research AI trends")],
      output="trace"  # Shows: [HH:MM:SS] ▸ tool → result [0.2s] ✓
  )
  result = workflow.start("Research AI")
  ```

  ```python Verbose Output (Rich Panels) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Agent

  # Shows full Rich panels with formatting
  workflow = AgentFlow(
      steps=[Agent(instructions="Research AI trends")],
      output="verbose"  # Full interactive display
  )
  result = workflow.start("Research AI")
  ```

  ```python Silent Output (Default) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Agent

  # No output - best for programmatic use
  workflow = AgentFlow(
      steps=[Agent(instructions="Research AI trends")],
      output="silent"  # Default - zero overhead
  )
  result = workflow.start("Research AI")
  ```
</CodeGroup>

**Available Output Presets:**

| Preset    | Description                      |
| --------- | -------------------------------- |
| `silent`  | No output (default, fastest)     |
| `status`  | Tool calls + final output inline |
| `trace`   | Timestamped execution trace      |
| `verbose` | Full Rich panels                 |
| `debug`   | Trace + metrics                  |
| `json`    | JSONL output for piping          |

### Output to File

<CodeGroup>
  ```python Output to File theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Task

  workflow = AgentFlow(steps=[
      Task(
          name="generator",
          action="Generate report",
          output_file="output/{{name}}_report.txt"
      )
  ])
  ```

  ```python With Images (Vision) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  workflow = AgentFlow(steps=[
      Task(
          name="analyzer",
          action="Analyze this image",
          images=["image.jpg", "diagram.png"]
      )
  ])
  ```

  ```python Pydantic Output theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from pydantic import BaseModel

  class Report(BaseModel):
      title: str
      content: str
      score: float

  workflow = AgentFlow(steps=[
      Task(
          name="generator",
          action="Generate structured report",
          output_pydantic=Report
      )
  ])
  ```
</CodeGroup>

### Memory Integration

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlowMemoryConfig

workflow = AgentFlow(
    steps=[researcher, writer],
    memory=WorkflowMemoryConfig(
        backend="chroma",
        persist=True,
        collection="my_workflow"
    )
)

# First run
result1 = workflow.start("Research AI")

# Second run - remembers first run
result2 = workflow.start("Continue the research")
```

### Async Execution

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import asyncio
from praisonaiagents import AgentFlow

workflow = AgentFlow(steps=[step1, step2])

async def main():
    result = await workflow.astart("input")
    print(result)

asyncio.run(main())
```

### Status Tracking

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[step1, step2, step3])
result = workflow.start("input")

# Check workflow status
print(workflow.status)  # "not_started" | "running" | "completed"

# Check individual step statuses
print(workflow.step_statuses)  # {"step1": "completed", "step2": "completed", "step3": "skipped"}
```

### Pattern Helpers Reference

| Pattern      | Description                                    | Example                                                     |
| ------------ | ---------------------------------------------- | ----------------------------------------------------------- |
| `route()`    | Decision-based branching                       | `route({"yes": [step_a], "no": [step_b]})`                  |
| `parallel()` | Concurrent execution (configurable worker cap) | `parallel([step1, step2, step3], max_workers=5)`            |
| `loop()`     | Iterate over list/CSV                          | `loop(handler, over="items", parallel=True, max_workers=3)` |
| `repeat()`   | Evaluator-optimizer                            | `repeat(gen, until=condition, max_iterations=5)`            |
| `when()`     | Conditional branching                          | `when(condition="{{score}} > 80", then_steps=[...])`        |
| `include()`  | Workflow composition                           | `include(workflow=sub_workflow)`                            |

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow
from praisonaiagents import route, parallel, loop, repeat, when, include

workflow = AgentFlow(steps=[
    classifier,
    route({"tech": [tech_agent], "creative": [creative_agent]}),
    parallel([worker1, worker2, worker3]),
    loop(processor, over="items"),
    repeat(generator, until=lambda r: "done" in r, max_iterations=5),
    when(condition="{{score}} > 80", then_steps=[approve], else_steps=[reject]),
    include(workflow=sub_workflow)  # Or include(recipe="recipe-name")
])
```

### Loop with Parallel Execution

`loop()` also supports `max_workers` when `parallel=True` is enabled:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import loop

# Iterate in parallel with an explicit worker cap
step = loop(handler, over="items", parallel=True, max_workers=5)
```

### Robustness Features

Enable execution tracing and graceful degradation for production workflows:

<CodeGroup>
  ```python Workflow History theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents import AgentFlow, Agent

  # Enable execution trace for debugging
  workflow = AgentFlow(
      steps=[agent1, agent2],
      history=True  # Track step execution
  )

  result = workflow.start("input")

  # Get execution trace
  history = workflow.get_history()
  # Returns: [{"step": "agent1", "timestamp": "...", "success": True, "output": "..."}, ...]
  ```

  ```python Conditional Branching (when) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents.workflows import when

  # Use when() for cleaner conditional syntax (preferred over if_)
  workflow = AgentFlow(steps=[
      scorer_agent,
      when(
          condition="{{score}} > 80",
          then_steps=[approve_agent],
          else_steps=[reject_agent]
      )
  ])
  ```

  ```python Workflow Composition (include) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  from praisonaiagents.workflows import include

  # Include another workflow for modular composition
  sub_workflow = AgentFlow(name="sub", steps=[...])
  main_workflow = AgentFlow(steps=[
      agent1,
      include(workflow=sub_workflow),  # Embed sub-workflow
      include(recipe="wordpress-publisher"),  # Or include a recipe
  ])
  ```
</CodeGroup>

> \[!NOTE]
> `if_()` is deprecated in favor of `when()` for cleaner syntax.

````

### YAML Workflow Configuration

Define workflows in YAML format with agents and all patterns:

```yaml
# .praison/workflows/research.yaml
name: Research Workflow
description: Research and write content with multiple patterns

agents:
  researcher:
    role: Research Expert
    goal: Find accurate information
    tools: [tavily_search, web_scraper]
  
  writer:
    role: Content Writer
    goal: Write engaging content
    tools: [file_writer]
  
  editor:
    role: Editor
    goal: Polish and refine content

  tech_expert:
    role: Technical Expert
    goal: Handle technical queries

  creative_writer:
    role: Creative Writer
    goal: Handle creative content

steps:
  # Step 1: Sequential agent execution
  - agent: researcher
    action: Research {{topic}}
    output_variable: research_data

  # Step 2: Routing based on content type
  - name: classifier
    action: Classify this content as technical or creative
    route:
      technical: [tech_handler]
      creative: [creative_handler]
      default: [general_handler]

  # Step 3: Parallel execution
  - name: parallel_research
    parallel:
      - agent: researcher
        action: Research market trends
      - agent: researcher
        action: Research competitors
      - agent: researcher
        action: Research customers

  # Step 4: Loop over items
  - agent: writer
    action: Write article about {{item}}
    loop_over: topics
    loop_var: item

  # Step 5: Repeat until condition (evaluator-optimizer)
  - agent: editor
    action: Review and improve the content
    repeat:
      until: "quality score > 8"
      max_iterations: 3

  # Step 6: Final output with file save
  - agent: writer
    action: Write final report based on {{previous_output}}
    output_file: output/{{topic}}_report.md

variables:
  topic: AI trends
  topics:
    - Machine Learning
    - Neural Networks
    - Natural Language Processing

# Workflow-level settings
planning: true
planning_llm: gpt-4o
verbose: true
memory_config:
  provider: chroma
  persist: true
````

### Complete Python Example with All Patterns

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, AgentFlow, Task, WorkflowContext, StepResult
from praisonaiagents import route, parallel, loop, repeat
from pydantic import BaseModel

# Define output schema
class Report(BaseModel):
    title: str
    content: str
    score: float

# Create agents
researcher = Agent(name="Researcher", role="Research expert", tools=[tavily_search])
writer = Agent(name="Writer", role="Content writer")
editor = Agent(name="Editor", role="Editor")
tech_agent = Agent(name="TechExpert", role="Technical expert")
creative_agent = Agent(name="Creative", role="Creative writer")

# Define handler functions
def classifier(ctx: WorkflowContext) -> StepResult:
    content = ctx.previous_result or ctx.input
    if "code" in content.lower() or "technical" in content.lower():
        return StepResult(output="technical")
    return StepResult(output="creative")

def validate_quality(result: StepResult) -> tuple[bool, str]:
    """Guardrail: Returns (is_valid, feedback)"""
    if len(result.output) < 100:
        return (False, "Content too short, please expand")
    return (True, None)

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

# Create comprehensive workflow
workflow = AgentFlow(
    steps=[
        # 1. Sequential agents
        researcher,
        writer,
        
        # 2. Routing based on classifier output
        classifier,
        route({
            "technical": [tech_agent],
            "creative": [creative_agent],
            "default": [writer]
        }),
        
        # 3. Parallel execution
        parallel([
            Task(name="market", action="Research market"),
            Task(name="competitors", action="Research competitors"),
            Task(name="customers", action="Research customers")
        ]),
        
        # 4. Loop over items
        loop(process_item, over="items"),
        
        # 5. Repeat until condition (evaluator-optimizer)
        repeat(
            Task(name="refine", handler=editor.chat),
            until=lambda ctx: "excellent" in ctx.previous_result.lower(),
            max_iterations=3
        ),
        
        # 6. Final step with guardrail and output options
        Task(
            name="final_report",
            handler=lambda ctx: StepResult(output=writer.chat(f"Write report: {ctx.previous_result}")),
            guardrails=validate_quality,
            execution={"max_retries": 2},
            output={"file": "output/report.md", "pydantic_model": Report}
        )
    ],
    
    # Workflow configuration
    variables={"items": ["AI", "ML", "NLP"]},
    planning=WorkflowPlanningConfig(enabled=True, llm="gpt-4o", reasoning=True),
    
    # Callbacks using hooks= consolidated param
    hooks=WorkflowHooksConfig(
        on_workflow_start=lambda w, i: print(f"🚀 Starting workflow: {i}"),
        on_step_start=lambda name, ctx: print(f"▶️ Step: {name}"),
        on_step_complete=lambda name, r: print(f"✅ {name}: {str(r.output)[:50]}..."),
        on_step_error=lambda name, e: print(f"❌ Error in {name}: {e}"),
        on_workflow_complete=lambda w, r: print(f"🎉 Workflow completed: {r['status']}")
    )
)

# Execute
result = workflow.start("Research and write about AI trends")

# Check status
print(f"Status: {workflow.status}")
print(f"Step statuses: {workflow.step_statuses}")
print(f"Final output: {result['output']}")
```

## Workflow File Format

Workflows are defined in markdown files with YAML frontmatter:

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
---
name: Research Pipeline
description: Multi-agent research and writing workflow
default_llm: gpt-4o-mini
planning: true
planning_llm: gpt-4o
variables:
  topic: AI trends
---

## Step 1: Research
Research the topic thoroughly.

```agent
role: Researcher
goal: Find comprehensive information
instructions:  # Canonical: use 'instructions' instead of 'backstory' Expert researcher with 10 years experience
```

```tools
tavily_search
web_browser
```

output_variable: research_data

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

## Step 2: Analyze
Analyze the research findings.

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

context_from: [Research]
retain_full_context: false

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

## Step 3: Write Report
Write the final report.

```agent
role: Writer
goal: Write engaging content
```

```action
Write a comprehensive report based on {{previous_output}}
```
````

### Frontmatter Options

| Option         | Type    | Description               |
| -------------- | ------- | ------------------------- |
| `name`         | string  | Workflow name             |
| `description`  | string  | Workflow description      |
| `default_llm`  | string  | Default LLM for all steps |
| `planning`     | boolean | Enable planning mode      |
| `planning_llm` | string  | LLM for planning          |
| `variables`    | object  | Default variables         |

### Step Options

| Option                | Type    | Description                                  |
| --------------------- | ------- | -------------------------------------------- |
| `context_from`        | list    | Specific steps to include context from       |
| `retain_full_context` | boolean | Include all previous outputs (default: true) |
| `output_variable`     | string  | Store output in custom variable name         |
| `output_file`         | string  | Save step output to file                     |
| `loop_over`           | string  | Variable name to iterate over                |
| `loop_var`            | string  | Variable name for current item in loop       |

### Pattern Blocks

Use code blocks to define workflow patterns:

| Block           | Description                      |
| --------------- | -------------------------------- |
| ` ```route `    | Define routing conditions        |
| ` ```parallel ` | Define parallel execution steps  |
| ` ```images `   | Define images for vision tasks   |
| ` ```repeat `   | Define repeat/iteration settings |

### Route Pattern Example

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 1: Classifier
Classify the request.

```action
Classify this request
```

```route
technical: [Tech Handler]
creative: [Creative Handler]
default: [General Handler]
```
````

### Parallel Pattern Example

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 1: Research
Research in parallel.

```parallel
- Market Research
- Competitor Analysis
- Customer Survey
```

```action
Research the topic
```
````

### Loop Pattern Example

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 1: Process Items
Process each item.

loop_over: items
loop_var: current_item

```action
Process {{current_item}}
```
````

### Images Pattern Example

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 1: Analyze Image
Analyze the provided images.

```images
image1.jpg
image2.png
```

```action
Analyze these images
```
````

### Output File Example

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 1: Generate Report
Generate and save a report.

output_file: output/report.txt

```action
Generate the report
```
````

## Storage Structure

```
project/
├── .praison/
│   └── workflows/
│       ├── deploy.md        # Deployment workflow
│       ├── test.md          # Testing workflow
│       ├── review.md        # Code review workflow
│       └── release.md       # Release workflow
```

## Variable Substitution

Use `{{variable}}` syntax for dynamic values:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlowManager

manager = WorkflowManager()

# Variables defined in workflow file are defaults
# Override at execution time
result = manager.execute(
    "deploy",
    default_agent=agent,
    variables={
        "environment": "staging",  # Override default
        "branch": "feature/new-ui",
        "version": "1.2.3"  # Additional variable
    }
)
```

## Context Passing

Workflow steps automatically pass context to subsequent steps. Use special variables to access previous outputs:

| Variable               | Description                                               |
| ---------------------- | --------------------------------------------------------- |
| `{{previous_output}}`  | Output from the immediately previous step                 |
| `{{step_name_output}}` | Output from a specific step (e.g., `{{research_output}}`) |

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, Task
from praisonaiagents import AgentFlowManager, TaskContextConfig, TaskOutputConfig

workflow = AgentFlow(
    name="pipeline",
    steps=[
        Task(
            name="research",
            action="Research AI trends",
            output=TaskOutputConfig(variable="research_data")  # Store as custom variable
        ),
        Task(
            name="analyze",
            action="Analyze: {{research_data}}",  # Use custom variable
            context=TaskContextConfig(from_steps=["research"], retain_full=False)
        ),
        Task(
            name="write",
            action="Write based on {{previous_output}}"  # Use last step's output
        )
    ]
)
```

### Context Control Options

| Option                | Default              | Description                                    |
| --------------------- | -------------------- | ---------------------------------------------- |
| `context_from`        | All previous         | List of step names to include context from     |
| `retain_full_context` | `True`               | Include all previous outputs vs only specified |
| `output_variable`     | `{step_name}_output` | Custom variable name for step output           |

## Conditional Steps

Add conditions to skip steps based on context:

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
## Step 3: Deploy to Staging
Only deploy to staging for non-production.

```condition
{{environment}} != production
```

```action
Deploy to staging environment.
```
````

## Callbacks

Monitor workflow execution with callbacks:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlowManager

manager = WorkflowManager()

def on_step(step, index):
    print(f"Starting step {index + 1}: {step.name}")

def on_result(step, result):
    print(f"Completed {step.name}: {result[:100]}...")

result = manager.execute(
    "deploy",
    executor=lambda prompt: agent.chat(prompt),
    on_step=on_step,
    on_result=on_result
)
```

## Error Handling

Configure how steps handle errors:

````markdown theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
---
name: Resilient Workflow
---

## Step 1: Optional Cleanup
This step can fail without stopping the workflow.

```action
on_error: continue
max_retries: 2
```

Clean up temporary files.

## Step 2: Critical Build
This step must succeed.

```action
on_error: stop
```

Build the application.
````

| Error Mode | Behavior                                 |
| ---------- | ---------------------------------------- |
| `stop`     | Stop workflow on failure (default)       |
| `continue` | Continue to next step on failure         |
| `retry`    | Retry the step up to `max_retries` times |

## Async Execution

Use `aexecute()` for async workflow execution:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import asyncio
from praisonaiagents import Agent
from praisonaiagents import AgentFlowManager

manager = WorkflowManager()

async def run_workflows():
    # Run multiple workflows concurrently
    results = await asyncio.gather(
        manager.aexecute("research", default_llm="gpt-4o-mini"),
        manager.aexecute("analysis", default_llm="gpt-4o-mini"),
    )
    return results

# With async executor
async def async_executor(prompt):
    # Your async logic here
    await asyncio.sleep(0.1)
    return f"Processed: {prompt}"

async def main():
    result = await manager.aexecute(
        "deploy",
        executor=async_executor,
        variables={"environment": "staging"}
    )
    print(result)

asyncio.run(main())
```

## Execute Parameters

| Parameter       | Type     | Description                            |
| --------------- | -------- | -------------------------------------- |
| `workflow_name` | str      | Name of workflow to execute            |
| `executor`      | callable | Optional function to execute steps     |
| `default_agent` | Agent    | Default agent for steps without config |
| `default_llm`   | str      | Default LLM model                      |
| `memory`        | Memory   | Shared memory instance                 |
| `planning`      | bool     | Enable planning mode                   |
| `stream`        | bool     | Enable streaming output                |
| `verbose`       | int      | Verbosity level (0-3)                  |
| `variables`     | dict     | Variables to substitute                |
| `on_step`       | callable | Callback before each step              |
| `on_result`     | callable | Callback after each step               |

## Programmatic API

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, Task
from praisonaiagents import AgentFlowManager

manager = WorkflowManager(workspace_path="/path/to/project")

# Get a specific workflow
workflow = manager.get_workflow("deploy")
print(f"Workflow: {workflow.name}")
print(f"Steps: {[s.name for s in workflow.steps]}")

# Get statistics
stats = manager.get_stats()
print(f"Total workflows: {stats['total_workflows']}")
print(f"Total steps: {stats['total_steps']}")

# Reload workflows from disk
manager.reload()
```

## Best Practices

<AccordionGroup>
  <Accordion title="Use per-step agents for specialized tasks">
    Configure different agents with specific roles for each step. A Researcher agent for gathering data, an Analyst for processing, and a Writer for output.
  </Accordion>

  <Accordion title="Control context passing">
    Use `context_from` to limit which previous outputs are included. This reduces token usage and keeps agents focused on relevant information.
  </Accordion>

  <Accordion title="Use output_variable for clarity">
    Name your outputs with `output_variable` for clearer variable substitution in subsequent steps.
  </Accordion>

  <Accordion title="Keep steps focused">
    Each step should do one thing well. Break complex tasks into multiple steps for better error handling and visibility.
  </Accordion>

  <Accordion title="Use async for parallel workflows">
    Use `aexecute()` with `asyncio.gather()` to run multiple independent workflows concurrently.
  </Accordion>

  <Accordion title="Set appropriate error handling">
    Use `on_error: continue` for optional steps and `on_error: stop` for critical steps that must succeed.
  </Accordion>
</AccordionGroup>

## CLI Usage

Execute workflows directly from the command line:

<CodeGroup>
  ```bash Template-Based theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # List workflows
  praisonai workflow list

  # Execute with tools and save
  praisonai workflow run "Research Blog" --tools tavily --save

  # With planning mode (AI creates sub-steps)
  praisonai workflow run "Research Blog" --planning --verbose

  # With variables
  praisonai workflow run deploy --workflow-var environment=staging
  ```

  ```bash Inline (No Template) theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # Quick workflow without a file
  praisonai "What is AI?" --workflow "Research,Summarize" --save

  # With step actions
  praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
  ```
</CodeGroup>

### CLI Options

| Flag                       | Description             |
| -------------------------- | ----------------------- |
| `--workflow-var key=value` | Set workflow variable   |
| `--llm <model>`            | LLM model               |
| `--tools <tools>`          | Tools (comma-separated) |
| `--planning`               | Enable planning mode    |
| `--memory`                 | Enable memory           |
| `--save`                   | Save output to file     |
| `--verbose`                | Verbose output          |

<Tip>
  For full CLI documentation, see [Workflow CLI](/cli/workflow).
</Tip>

## Architecture Patterns

Understanding how Workflow relates to other PraisonAI patterns:

### Class Hierarchy

| Class      | Contains                   | Purpose                        |
| ---------- | -------------------------- | ------------------------------ |
| `Agent`    | Self (LLM wrapper)         | Core execution unit            |
| `Task`     | Reference to agent         | Work item for agent            |
| `Agents`   | List of `Agent`            | Multi-agent orchestrator       |
| `Workflow` | List of steps              | Declarative step orchestrator  |
| `Task`     | Reference to `agent` field | Work item with agent reference |

### Pattern Comparison

| Concept          | Agents Pattern              | Workflow Pattern            |
| ---------------- | --------------------------- | --------------------------- |
| **Orchestrator** | `Agents`                    | `Workflow`                  |
| **Work Item**    | `Task` (contains agent ref) | `Task` (contains agent ref) |
| **Executor**     | `Agent`                     | `Agent` (same!)             |

> **Key Insight:** Task is to Workflow what Task is to Agents — it's a work item that **references** an agent, not an agent itself.

### Orchestrator Diagram

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    subgraph "Agents Pattern (Multi-Agent)"
        AGENTS[Agents]
        AGENTS --> AGENT1[Agent]
        AGENTS --> TASK[Task]
        TASK -.->|agent ref| AGENT1
        AGENTS --> AGENT2[Agent]
    end
    
    subgraph "Workflow Pattern (Declarative Steps)"
        WORKFLOW[Workflow]
        WORKFLOW --> STEP[Task]
        STEP -.->|agent ref| AGENT3[Agent]
        WORKFLOW --> ROUTE[Route - branching]
        WORKFLOW --> PARALLEL[Parallel - fan-out]
        WORKFLOW --> LOOP[Loop - iteration]
    end
    
    classDef orchestrator fill:#2E8B57,stroke:#333,color:#fff
    classDef agent fill:#189AB4,stroke:#333,color:#fff
    classDef workitem fill:#8B0000,stroke:#333,color:#fff
    classDef pattern fill:#FF8C00,stroke:#333,color:#fff
    
    class AGENTS,WORKFLOW orchestrator
    class AGENT1,AGENT2,AGENT3 agent
    class TASK,STEP workitem
    class ROUTE,PARALLEL,LOOP pattern
```

## WorkflowManager API

`WorkflowManager` discovers, loads, and executes markdown-defined workflows from your project's `.praisonai/workflows/` directory.

### Discover and Execute

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import WorkflowManager, Agent

# Create manager (auto-discovers workflows in .praisonai/workflows/)
manager = WorkflowManager()

# List available workflows
workflows = manager.list_workflows()
print(workflows)  # ["deploy", "research-pipeline", ...]

# Execute with a default agent
agent = Agent(name="Assistant", instructions="Execute workflow steps")
result = manager.execute(
    workflow_name="deploy",
    default_agent=agent,
    variables={"environment": "staging"}
)
print(result)
```

### Checkpoints

Save and resume long-running workflows:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Save checkpoint at step 3
result = manager.execute(
    "research-pipeline",
    default_agent=agent,
    checkpoint="research-checkpoint"  # Name for save point
)

# Resume from checkpoint
result = manager.execute(
    "research-pipeline",
    default_agent=agent,
    resume="research-checkpoint"      # Resume from save point
)

# Manage checkpoints
checkpoints = manager.list_checkpoints()
manager.delete_checkpoint("research-checkpoint")
```

### Variable Substitution

Workflows support `{{variable}}` placeholders with dynamic providers:

| Variable               | Value                           |
| ---------------------- | ------------------------------- |
| `{{previous_output}}`  | Output of the previous step     |
| `{{step_name_output}}` | Output of a specific named step |
| `{{now}}`              | Current datetime                |
| `{{today}}`            | Today's date                    |
| `{{uuid}}`             | Unique identifier               |
| Custom variables       | Passed via `variables={}` dict  |

### Async Execution

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import asyncio

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

### Create Workflows Programmatically

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = manager.create_workflow(
    name="my-pipeline",
    description="Custom multi-step pipeline",
    steps=[
        {"name": "Research", "description": "Gather data", "action": "Search for {{topic}}"},
        {"name": "Analyze", "description": "Process findings", "action": "Analyze: {{previous_output}}"},
        {"name": "Report", "description": "Write summary", "action": "Write report on findings"},
    ],
    variables={"topic": "AI trends"}
)
```

## See Also

<CardGroup cols={2}>
  <Card title="Workflow CLI" icon="terminal" href="/cli/workflow">
    Command-line workflow execution
  </Card>

  <Card title="Hooks" icon="plug" href="/features/hooks">
    Pre/post operation hooks for custom actions
  </Card>

  <Card title="Rules & Instructions" icon="scroll" href="/features/rules">
    Auto-discover and apply persistent rules
  </Card>

  <Card title="Agent Memory" icon="memory" href="/features/memory">
    Persistent memory for agents
  </Card>
</CardGroup>