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

# Workflows Module

> Workflow and pipeline patterns for orchestrating agents and functions

# Workflows

Workflow = "Agents in an organized way". Multi-step pipelines with routing, parallel execution, and loops.

## Quick Start

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

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

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

## Workflow Parameters

| Parameter     | Type   | Default      | Description                      |
| ------------- | ------ | ------------ | -------------------------------- |
| `name`        | `str`  | `"Workflow"` | Workflow name                    |
| `steps`       | `list` | `[]`         | Steps (Agent, function, or Task) |
| `variables`   | `dict` | `{}`         | Initial variables                |
| `default_llm` | `str`  | `None`       | Default LLM for all steps        |

### Consolidated Feature Params (Workflow)

| Parameter    | Type                                   | Default | Description         |
| ------------ | -------------------------------------- | ------- | ------------------- |
| `output`     | `str \| WorkflowOutputConfig`          | `None`  | Output config       |
| `planning`   | `bool \| WorkflowPlanningConfig`       | `False` | Planning mode       |
| `memory`     | `bool \| WorkflowMemoryConfig`         | `None`  | Memory config       |
| `hooks`      | `WorkflowHooksConfig`                  | `None`  | Lifecycle callbacks |
| `context`    | `bool \| ContextConfig`                | `False` | Context management  |
| `autonomy`   | `bool \| AutonomyConfig`               | `None`  | Autonomy settings   |
| `knowledge`  | `bool \| List[str] \| KnowledgeConfig` | `None`  | Knowledge/RAG       |
| `guardrails` | `bool \| Callable \| GuardrailConfig`  | `None`  | Validation          |
| `web`        | `bool \| WebConfig`                    | `None`  | Web search/fetch    |
| `reflection` | `bool \| ReflectionConfig`             | `None`  | Self-reflection     |

## Task Parameters

| Parameter   | Type    | Default  | Description              |
| ----------- | ------- | -------- | ------------------------ |
| `name`      | `str`   | required | Step name                |
| `action`    | `str`   | `""`     | Action/prompt to execute |
| `agent`     | `Agent` | `None`   | Agent for this step      |
| `tools`     | `list`  | `None`   | Tools for this step      |
| `condition` | `str`   | `None`   | Execution condition      |
| `loop_over` | `str`   | `None`   | Variable to iterate over |

### Consolidated Feature Params (Task)

| Parameter    | Type                                   | Default | Description        |
| ------------ | -------------------------------------- | ------- | ------------------ |
| `context`    | `List[str] \| TaskContextConfig`       | `None`  | Context from steps |
| `output`     | `str \| TaskOutputConfig`              | `None`  | Output handling    |
| `execution`  | `str \| TaskExecutionConfig`           | `None`  | Execution control  |
| `routing`    | `List[str] \| TaskRoutingConfig`       | `None`  | Branching          |
| `guardrails` | `Callable \| str \| GuardrailConfig`   | `None`  | Validation         |
| `autonomy`   | `bool \| AutonomyConfig`               | `None`  | Autonomy settings  |
| `knowledge`  | `bool \| List[str] \| KnowledgeConfig` | `None`  | Knowledge/RAG      |
| `web`        | `bool \| WebConfig`                    | `None`  | Web search/fetch   |
| `reflection` | `bool \| ReflectionConfig`             | `None`  | Self-reflection    |

## Precedence Ladder

<Info>
  **Resolution Order**: Instance > Config > Array > Dict > String > Bool > Default

  Same precedence as Agent. Workflow params propagate to steps, step params override workflow defaults.
</Info>

## Where to Set Params

| Param        | Workflow-level | Step-level | Agent-level | Notes                   |
| ------------ | -------------- | ---------- | ----------- | ----------------------- |
| `memory`     | ✅              | ❌          | ✅           | Workflow or Agent       |
| `planning`   | ✅              | ❌          | ✅           | Workflow or Agent       |
| `output`     | ✅              | ✅          | ✅           | Step overrides Workflow |
| `execution`  | ❌              | ✅          | ✅           | Step or Agent           |
| `guardrails` | ✅              | ✅          | ✅           | Step overrides Workflow |
| `knowledge`  | ✅              | ✅          | ✅           | Step overrides Workflow |
| `web`        | ✅              | ✅          | ✅           | Step overrides Workflow |
| `reflection` | ✅              | ✅          | ✅           | Step overrides Workflow |
| `context`    | ✅              | ✅          | ✅           | Step overrides Workflow |

## Presets & Options

### Workflow Output Presets

| Preset      | Description                          |
| ----------- | ------------------------------------ |
| `"silent"`  | Zero output (default)                |
| `"status"`  | Tool calls + response, no timestamps |
| `"trace"`   | Full trace with timestamps           |
| `"debug"`   | trace + metrics (no boxes)           |
| `"verbose"` | Rich panels with Markdown            |

### Step Execution Presets

| Preset       | max\_retries | quality\_check |
| ------------ | ------------ | -------------- |
| `"fast"`     | 1            | ❌              |
| `"balanced"` | 3            | ✅              |
| `"thorough"` | 5            | ✅              |

## Methods

| Method          | Description                    |
| --------------- | ------------------------------ |
| `start(input)`  | Execute workflow synchronously |
| `run(input)`    | Alias for start()              |
| `astart(input)` | Execute asynchronously         |
| `arun(input)`   | Alias for astart()             |

## Common Recipes

### Basic Workflow

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

workflow = AgentFlow(
    steps=[
        Agent(instructions="Research the topic"),
        Agent(instructions="Write a summary"),
    ],
    output="verbose"
)
result = workflow.start("AI trends")
```

### Task with Context

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

workflow = AgentFlow(
    steps=[
        Task(name="research", action="Research {{input}}"),
        Task(name="write", action="Write based on research", context=["research"]),
    ]
)
```

### Step with Guardrails

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

step = Task(
    name="validate",
    action="Generate content",
    guardrails="Ensure content is safe and accurate"
)
```

## ⚠️ Gotchas

* **Step names must be unique** - Used for context references
* **`guardrail` (singular) is deprecated** - Use `guardrails` (plural)
* **Workflow-level params are defaults** - Step-level params override them
* **`Pipeline` is an alias** - `Workflow` and `Pipeline` are the same class

## See Also

* [Agent](/docs/sdk/praisonaiagents/agent/agent) - Core agent class
* [Agents](/docs/sdk/praisonaiagents/agents/agents) - Multi-agent orchestration
* [Workflow Manager](/docs/sdk/praisonaiagents/workflows/workflow-manager) - Advanced workflow management

### WorkflowContext

Context passed between workflow steps.

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

context = WorkflowContext(
    initial_data={"topic": "AI"},
    metadata={"user_id": "123"}
)

# Access data
context.set("key", "value")
value = context.get("key")
```

### StepResult

Result from a workflow step.

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

result = StepResult(
    success=True,
    output="Step completed",
    metadata={"duration": 1.5}
)
```

### WorkflowManager

Manager for complex workflow orchestration.

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

manager = WorkflowManager()
manager.register_workflow("content", content_workflow)
manager.register_workflow("analysis", analysis_workflow)

result = manager.run("content", input_data)
```

## Pattern Helpers

### Route

Conditional routing between steps.

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

# Class-based
routing = Route(
    condition=lambda ctx: ctx.get("type"),
    routes={
        "technical": technical_agent,
        "creative": creative_agent
    },
    default=general_agent
)

# Function-based
@route(condition=lambda ctx: ctx.get("type"))
def my_router(ctx):
    if ctx.get("type") == "technical":
        return technical_agent
    return general_agent
```

### Parallel

Execute steps in parallel.

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

# Class-based
parallel_steps = Parallel(
    steps=[agent1, agent2, agent3],
    merge_strategy="combine"  # or "first", "all"
)

# Function-based
@parallel(merge_strategy="combine")
def parallel_research():
    return [web_search, database_search, api_search]
```

### Loop

Loop until condition is met.

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

# Class-based
refinement_loop = Loop(
    step=refine_agent,
    condition=lambda ctx: ctx.get("quality_score", 0) < 0.9,
    max_iterations=5
)

# Function-based
@loop(max_iterations=5)
def refine_until_good(ctx):
    return ctx.get("quality_score", 0) < 0.9
```

### Repeat

Repeat a step N times.

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

# Class-based
repeated = Repeat(
    step=generate_agent,
    times=3
)

# Function-based
@repeat(times=3)
def generate_variations():
    return generate_agent
```

## YAML Workflows

Parse workflows from YAML files.

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

parser = YAMLWorkflowParser()
workflow = parser.parse("workflow.yaml")
result = workflow.run()
```

### YAML Format

```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
name: Content Pipeline
steps:
  - name: research
    agent: researcher
    task: Research the topic
    
  - name: write
    agent: writer
    task: Write article
    depends_on: research
    
  - name: review
    agent: reviewer
    task: Review and edit
    depends_on: write
```

## Usage Examples

### Sequential Workflow

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

researcher = Agent(name="Researcher")
writer = Agent(name="Writer")
editor = Agent(name="Editor")

workflow = AgentFlow(steps=[
    {"agent": researcher, "task": "Research {topic}"},
    {"agent": writer, "task": "Write article based on research"},
    {"agent": editor, "task": "Edit and polish the article"}
])

result = workflow.run(topic="Machine Learning")
```

### Conditional Workflow

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

workflow = AgentFlow(steps=[
    {"agent": classifier, "task": "Classify the request"},
    Route(
        condition=lambda ctx: ctx.get("category"),
        routes={
            "technical": technical_agent,
            "billing": billing_agent,
            "general": general_agent
        }
    )
])
```

### Parallel Processing

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

workflow = AgentFlow(steps=[
    {"agent": planner, "task": "Create research plan"},
    Parallel(steps=[
        {"agent": web_researcher, "task": "Search web"},
        {"agent": db_researcher, "task": "Search database"},
        {"agent": api_researcher, "task": "Query APIs"}
    ]),
    {"agent": synthesizer, "task": "Combine all research"}
])
```

### Loop with Refinement

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

workflow = AgentFlow(steps=[
    {"agent": generator, "task": "Generate initial draft"},
    Loop(
        step={"agent": refiner, "task": "Improve the draft"},
        condition=lambda ctx: ctx.get("score", 0) < 0.9,
        max_iterations=3
    ),
    {"agent": finalizer, "task": "Finalize output"}
])
```

## Related

* [WorkflowManager](/docs/sdk/praisonaiagents/workflows/workflow-manager) - Detailed workflow manager docs
* [Agent](/docs/sdk/praisonaiagents/agent/agent) - Agent configuration
* [Task](/docs/sdk/praisonaiagents/task/task) - Task definition