Release v2.3.20

README.md

@@ -103,7 +103,19 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
 | 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
 | 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
-| ⏸️ Workflow Checkpoints | [Example](#workflow-checkpoints) | [📖](https://docs.praison.ai/cli/session) |
+| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
+| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 📄 CSV Loop Processing | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflows) |
 
 ## Supported Providers
 
@@ -445,6 +457,138 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 - **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
 - **Async Execution**: Use `aexecute()` for async workflows
 - **Planning Mode**: Enable at workflow level with `planning=True`
+- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
+- **Loops**: Use `loop_over` and `loop_var` to iterate over data
+
+### Choosing the Right Workflow System
+
+| Use Case | Recommended |
+|----------|-------------|
+| Simple function pipelines | `Workflow` class ⭐ |
+| Agent-only pipelines | `Workflow` class |
+| CSV batch processing | `process="workflow"` |
+| Complex task routing | `process="workflow"` |
+| Markdown templates | `WorkflowManager` |
+| Early stop / conditional | `Workflow` class |
+
+### Simple Workflow (Recommended)
+
+The easiest way to create workflows - just pass functions as steps:
+
+```python
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+
+# Define simple handler functions
+def validate(ctx: WorkflowContext) -> StepResult:
+    if not ctx.input:
+        return StepResult(output="No input", stop_workflow=True)
+    return StepResult(output=f"Valid: {ctx.input}")
+
+def process(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output=f"Processed: {ctx.previous_result}")
+
+# Create and run workflow
+workflow = Workflow(steps=[validate, process])
+result = workflow.start("Hello World", verbose=True)
+print(result["output"])  # "Processed: Valid: Hello World"
+```
+
+**Key Features:**
+- **Just pass functions** - No complex configuration needed
+- **Early stop** - Return `stop_workflow=True` to stop the workflow
+- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
+- **Verbose mode** - See step-by-step progress
+
+### Workflow Branching & Loops
+
+```python
+from praisonaiagents import WorkflowStep
+
+# Branching step
+decision_step = WorkflowStep(
+    name="decide",
+    action="Evaluate if task is complete",
+    next_steps=["success_step", "retry_step"],
+    branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
+)
+
+# Loop step
+loop_step = WorkflowStep(
+    name="process_items",
+    action="Process {{item}}",
+    loop_over="items",  # Variable containing list
+    loop_var="item"     # Current item variable name
+)
+```
+
+### Workflow Patterns (route, parallel, loop, repeat)
+
+```python
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import route, parallel, loop, repeat
+
+# 1. ROUTING - Decision-based branching
+workflow = Workflow(steps=[
+    classify_request,  # Returns "approve" or "reject"
+    route({
+        "approve": [approve_handler],
+        "reject": [reject_handler],
+        "default": [fallback_handler]
+    })
+])
+
+# 2. PARALLEL - Concurrent execution
+workflow = Workflow(steps=[
+    parallel([research_market, research_competitors, research_customers]),
+    summarize_results  # Gets all parallel outputs
+])
+
+# 3. LOOP - Iterate over list or CSV
+workflow = Workflow(
+    steps=[loop(process_item, over="items")],
+    variables={"items": ["a", "b", "c"]}
+)
+# Or from CSV file:
+workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
+
+# 4. REPEAT - Evaluator-Optimizer pattern
+workflow = Workflow(steps=[
+    repeat(
+        generator,
+        until=lambda ctx: "done" in ctx.previous_result,
+        max_iterations=5
+    )
+])
+```
+
+### Task Callbacks & Variables (process="workflow")
+
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Per-task agent config (no need to create agent separately)
+task = Task(
+    description="Research {{topic}}",
+    agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
+    variables={"topic": "AI trends"}
+)
+
+# Callbacks for workflow execution
+def on_start(task, task_id):
+    print(f"Starting: {task.name}")
+
+def on_complete(task, output):
+    print(f"Completed: {task.name}")
+
+agents = PraisonAIAgents(
+    agents=[agent],
+    tasks=[task],
+    process="workflow",
+    on_task_start=on_start,
+    on_task_complete=on_complete,
+    variables={"global_var": "shared_value"}  # Global variables for all tasks
+)
+```
 
 ### 9. Hooks
 

docker/Dockerfile

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     flask \
-    "praisonai>=2.3.16" \
+    "praisonai>=2.3.20" \
     "praisonai[api]" \
     gunicorn \
     markdown

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.16" \
+    "praisonai>=2.3.20" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.16" \
+    "praisonai>=2.3.20" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.16" \
+    "praisonai>=2.3.20" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/cookbooks/Finance_Market_Job_Agents/Ai_Job_Hunting_Assistant.ipynb

@@ -29,7 +29,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": null,
    "metadata": {
     "colab": {
      "base_uri": "https://localhost:8080/"
@@ -71,7 +71,6 @@
     "import pandas as pd\n",
     "\n",
     "print(\"📦 Dependencies installed!\")\n",
-    "print(\"🔄 Converting from AGNO → PraisonAI\")\n",
     "print(\"🎯 Building Job Hunting Agent System\")"
    ]
   },

examples/python/workflows/simple_workflow.py

@@ -0,0 +1,36 @@
+"""
+Simple Workflow Example
+
+The easiest way to create and run workflows in PraisonAI.
+Just pass functions or agents as steps - no complex configuration needed!
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+
+# Define simple handler functions
+def validate(ctx: WorkflowContext) -> StepResult:
+    """Validate the input data."""
+    if not ctx.input:
+        return StepResult(output="No input provided", stop_workflow=True)
+    return StepResult(output=f"Validated: {ctx.input}")
+
+def process(ctx: WorkflowContext) -> StepResult:
+    """Process the validated data."""
+    return StepResult(output=f"Processed: {ctx.previous_result}")
+
+def format_output(ctx: WorkflowContext) -> StepResult:
+    """Format the final output."""
+    return StepResult(output=f"✅ Final: {ctx.previous_result}")
+
+# Create workflow - just list your functions!
+workflow = Workflow(
+    name="Simple Pipeline",
+    steps=[validate, process, format_output]
+)
+
+if __name__ == "__main__":
+    # Run the workflow - PraisonAI style uses .start()
+    result = workflow.start("Hello, World!", verbose=True)
+    
+    print(f"\nFinal output: {result['output']}")
+    print(f"Steps completed: {len(result['steps'])}")

examples/python/workflows/task_callbacks.py

@@ -0,0 +1,57 @@
+"""
+Task Callbacks Example
+
+Demonstrates using on_task_start and on_task_complete callbacks
+with PraisonAIAgents for workflow monitoring and logging.
+"""
+
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Create an agent
+researcher = Agent(
+    name="Researcher",
+    role="Research Assistant",
+    goal="Find and summarize information",
+    llm="gpt-4o-mini"
+)
+
+# Create tasks with variables
+task1 = Task(
+    name="research",
+    description="Research the topic: {{topic}}",
+    expected_output="A brief summary of the topic",
+    agent=researcher,
+    variables={"topic": "artificial intelligence"}
+)
+
+task2 = Task(
+    name="summarize",
+    description="Create a one-paragraph summary of the research.",
+    expected_output="A concise summary",
+    agent=researcher
+)
+
+# Define callbacks
+def on_start(task, task_id):
+    """Called before each task starts."""
+    print(f"🚀 Starting task: {task.name} (ID: {task_id})")
+
+def on_complete(task, output):
+    """Called after each task completes."""
+    print(f"✅ Completed task: {task.name}")
+    print(f"   Output preview: {str(output)[:100]}...")
+
+if __name__ == "__main__":
+    # Create agents with callbacks
+    agents = PraisonAIAgents(
+        agents=[researcher],
+        tasks=[task1, task2],
+        process="workflow",
+        on_task_start=on_start,
+        on_task_complete=on_complete,
+        variables={"global_var": "shared_value"}  # Global variables
+    )
+    
+    # Run the workflow
+    result = agents.start()
+    print("\n🎉 Workflow completed!")

examples/python/workflows/workflow_branching.py

@@ -0,0 +1,48 @@
+"""
+Workflow Branching Example
+
+Demonstrates conditional branching in workflows where steps can route
+to different next steps based on output content.
+"""
+
+from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
+
+# Create a workflow with branching
+workflow = Workflow(
+    name="Decision Workflow",
+    description="A workflow that branches based on validation result",
+    steps=[
+        WorkflowStep(
+            name="validate",
+            action="Check if the number 42 is positive. Reply with 'valid' or 'invalid'.",
+            next_steps=["success_handler", "error_handler"],
+            branch_condition={
+                "valid": ["success_handler"],
+                "invalid": ["error_handler"]
+            }
+        ),
+        WorkflowStep(
+            name="success_handler",
+            action="The validation passed! Generate a success message."
+        ),
+        WorkflowStep(
+            name="error_handler",
+            action="The validation failed. Generate an error message."
+        )
+    ]
+)
+
+if __name__ == "__main__":
+    # Create manager and register workflow
+    manager = WorkflowManager()
+    manager.workflows["Decision Workflow"] = workflow
+    
+    # Execute - will branch to success_handler since 42 is positive
+    result = manager.execute(
+        "Decision Workflow",
+        default_llm="gpt-4o-mini"
+    )
+    
+    print("Workflow completed!")
+    for step_result in result["results"]:
+        print(f"  {step_result['step']}: {step_result['status']}")

examples/python/workflows/workflow_checkpoints.py

@@ -0,0 +1,63 @@
+"""
+Workflow Checkpoints Example
+
+Demonstrates saving and resuming workflow execution using checkpoints.
+Useful for long-running workflows that may be interrupted.
+"""
+
+from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
+
+# Create a multi-step workflow
+workflow = Workflow(
+    name="Long Process",
+    description="A workflow with checkpoints for resumability",
+    steps=[
+        WorkflowStep(
+            name="step1",
+            action="Initialize the process and prepare data."
+        ),
+        WorkflowStep(
+            name="step2", 
+            action="Process the first batch of data."
+        ),
+        WorkflowStep(
+            name="step3",
+            action="Process the second batch of data."
+        ),
+        WorkflowStep(
+            name="step4",
+            action="Finalize and generate report."
+        )
+    ]
+)
+
+if __name__ == "__main__":
+    manager = WorkflowManager()
+    manager.workflows["Long Process"] = workflow
+    
+    # Execute with checkpoint - saves after each step
+    print("=== Starting workflow with checkpoint ===")
+    result = manager.execute(
+        "Long Process",
+        default_llm="gpt-4o-mini",
+        checkpoint="my-checkpoint"  # Saves progress after each step
+    )
+    
+    print(f"Completed {len(result['results'])} steps")
+    
+    # List available checkpoints
+    print("\n=== Available checkpoints ===")
+    checkpoints = manager.list_checkpoints()
+    for cp in checkpoints:
+        print(f"  {cp['name']}: {cp['completed_steps']} steps completed")
+    
+    # Resume from checkpoint (if workflow was interrupted)
+    # result = manager.execute(
+    #     "Long Process",
+    #     default_llm="gpt-4o-mini",
+    #     resume="my-checkpoint"  # Resumes from last saved state
+    # )
+    
+    # Clean up checkpoint
+    manager.delete_checkpoint("my-checkpoint")
+    print("\n✅ Checkpoint deleted")

examples/python/workflows/workflow_conditional.py

@@ -0,0 +1,68 @@
+"""
+Workflow Conditional Execution Example
+
+Demonstrates using should_run to conditionally execute steps
+based on input or previous results.
+"""
+
+from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult
+
+# Condition functions - return True to run the step, False to skip
+def is_sensitive_content(ctx: WorkflowContext) -> bool:
+    """Only run compliance check for sensitive topics."""
+    sensitive_keywords = ["legal", "medical", "financial", "security"]
+    return any(keyword in ctx.input.lower() for keyword in sensitive_keywords)
+
+def needs_translation(ctx: WorkflowContext) -> bool:
+    """Only translate if content is not in English."""
+    return ctx.variables.get("language", "en") != "en"
+
+# Handler functions
+def process_content(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output=f"Processed: {ctx.input}")
+
+def compliance_check(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output=f"✅ Compliance verified for: {ctx.previous_result}")
+
+def translate(ctx: WorkflowContext) -> StepResult:
+    lang = ctx.variables.get("language", "unknown")
+    return StepResult(output=f"🌐 Translated to {lang}: {ctx.previous_result}")
+
+def finalize(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output=f"📄 Final: {ctx.previous_result}")
+
+# Create workflow with conditional steps
+workflow = Workflow(
+    name="Conditional Pipeline",
+    steps=[
+        WorkflowStep(name="process", handler=process_content),
+        WorkflowStep(
+            name="compliance",
+            handler=compliance_check,
+            should_run=is_sensitive_content  # Only runs for sensitive content
+        ),
+        WorkflowStep(
+            name="translate",
+            handler=translate,
+            should_run=needs_translation  # Only runs if language != "en"
+        ),
+        WorkflowStep(name="finalize", handler=finalize)
+    ]
+)
+
+if __name__ == "__main__":
+    # Test 1: Normal content (skips compliance)
+    print("=== Test 1: Normal content ===")
+    result = workflow.start("Hello world", verbose=True)
+    print(f"Output: {result['output']}\n")
+    
+    # Test 2: Sensitive content (runs compliance)
+    print("=== Test 2: Legal content ===")
+    result = workflow.start("Review this legal document", verbose=True)
+    print(f"Output: {result['output']}\n")
+    
+    # Test 3: Non-English content (runs translation)
+    print("=== Test 3: Non-English content ===")
+    workflow.variables = {"language": "es"}
+    result = workflow.start("Hola mundo", verbose=True)
+    print(f"Output: {result['output']}")

examples/python/workflows/workflow_early_stop.py

@@ -0,0 +1,69 @@
+"""
+Workflow Early Stop Example
+
+Demonstrates how to stop a workflow early using custom handler
+functions that return StepResult with stop_workflow=True.
+"""
+
+from praisonaiagents.memory.workflows import (
+    WorkflowManager, WorkflowStep, Workflow,
+    WorkflowContext, StepResult
+)
+
+# Custom validator that can stop the workflow
+def validate_data(context: WorkflowContext) -> StepResult:
+    """Check if data is valid and stop workflow if not."""
+    data = context.variables.get("data", {})
+    
+    if data.get("value", 0) < 0:
+        return StepResult(
+            output="❌ Validation failed: negative value detected. Stopping workflow.",
+            stop_workflow=True  # This stops the entire workflow
+        )
+    
+    return StepResult(
+        output="✅ Validation passed: data is valid.",
+        stop_workflow=False  # Continue to next step
+    )
+
+# Create workflow with early stop capability
+workflow = Workflow(
+    name="Data Processing",
+    description="Process data with validation gate",
+    variables={
+        "data": {"value": -5}  # Invalid data - will trigger early stop
+    },
+    steps=[
+        WorkflowStep(
+            name="validate",
+            handler=validate_data  # Custom function
+        ),
+        WorkflowStep(
+            name="process",
+            action="Process the validated data."  # Won't run if validation fails
+        ),
+        WorkflowStep(
+            name="report",
+            action="Generate final report."  # Won't run if validation fails
+        )
+    ]
+)
+
+if __name__ == "__main__":
+    manager = WorkflowManager()
+    manager.workflows["Data Processing"] = workflow
+    
+    # Test with invalid data (will stop early)
+    print("=== Testing with invalid data ===")
+    result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
+    
+    for step_result in result["results"]:
+        print(f"  {step_result['step']}: {step_result['output']}")
+    
+    # Test with valid data (will complete)
+    print("\n=== Testing with valid data ===")
+    workflow.variables["data"] = {"value": 42}
+    result = manager.execute("Data Processing", default_llm="gpt-4o-mini")
+    
+    for step_result in result["results"]:
+        print(f"  {step_result['step']}: {step_result['status']}")

examples/python/workflows/workflow_loop_csv.py

@@ -0,0 +1,55 @@
+"""
+Workflow Loop with CSV Example
+
+Demonstrates iterating over a CSV file, processing each row.
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import loop
+import tempfile
+import os
+
+# Create sample CSV file
+csv_content = """name,email,task
+Alice,[email protected],Write documentation
+Bob,[email protected],Review pull requests
+Charlie,[email protected],Fix critical bug
+Diana,[email protected],Deploy to production"""
+
+# Processor for each CSV row
+def process_task(ctx: WorkflowContext) -> StepResult:
+    """Process a single task from CSV."""
+    item = ctx.variables.get("item", {})
+    name = item.get("name", "Unknown")
+    task = item.get("task", "No task")
+    index = ctx.variables.get("loop_index", 0)
+    
+    return StepResult(
+        output=f"[{index+1}] {name}: {task} ✅",
+        variables={"last_processed": name}
+    )
+
+if __name__ == "__main__":
+    # Create temp CSV file
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
+        f.write(csv_content)
+        csv_path = f.name
+    
+    try:
+        print("=== Testing Loop with CSV ===\n")
+        print(f"CSV file: {csv_path}\n")
+        
+        # Create workflow
+        workflow = Workflow(
+            name="CSV Task Processor",
+            steps=[loop(process_task, from_csv=csv_path)]
+        )
+        
+        result = workflow.start("Process all tasks", verbose=True)
+        
+        print(f"\nAll outputs:\n{result['output']}")
+        print(f"\nProcessed {len(result['variables']['loop_outputs'])} items")
+        print(f"Last processed: {result['variables'].get('last_processed')}")
+        
+    finally:
+        os.unlink(csv_path)

examples/python/workflows/workflow_loops.py

@@ -0,0 +1,46 @@
+"""
+Workflow Loops Example
+
+Demonstrates iterating over a list of items in a workflow step.
+Each item is processed individually with the loop variable.
+"""
+
+from praisonaiagents.memory.workflows import WorkflowManager, WorkflowStep, Workflow
+
+# Create a workflow with loop
+workflow = Workflow(
+    name="Process Items",
+    description="Process a list of items using loop",
+    variables={
+        "items": ["apple", "banana", "cherry"]  # List to iterate over
+    },
+    steps=[
+        WorkflowStep(
+            name="process_each",
+            action="Describe the fruit: {{item}}",
+            loop_over="items",  # Variable name containing the list
+            loop_var="item"     # Variable name for current item
+        ),
+        WorkflowStep(
+            name="summarize",
+            action="Summarize all the fruits that were processed."
+        )
+    ]
+)
+
+if __name__ == "__main__":
+    # Create manager and register workflow
+    manager = WorkflowManager()
+    manager.workflows["Process Items"] = workflow
+    
+    # Execute - will process each fruit in the list
+    result = manager.execute(
+        "Process Items",
+        default_llm="gpt-4o-mini"
+    )
+    
+    print("Workflow completed!")
+    for step_result in result["results"]:
+        print(f"  {step_result['step']}: {step_result['status']}")
+        if isinstance(step_result.get("output"), list):
+            print(f"    Loop processed {len(step_result['output'])} items")

examples/python/workflows/workflow_mixed_steps.py

@@ -0,0 +1,68 @@
+"""
+Mixed Steps Workflow Example
+
+Demonstrates combining different step types in a single workflow:
+- Agent instances
+- Handler functions
+- WorkflowStep with action strings
+"""
+
+from praisonaiagents import Workflow, WorkflowStep, WorkflowContext, StepResult, Agent
+
+# Create an agent
+analyzer = Agent(
+    name="Analyzer",
+    role="Data Analyst",
+    goal="Analyze and summarize data",
+    llm="gpt-4o-mini"
+)
+
+# Custom handler functions
+def validate_input(ctx: WorkflowContext) -> StepResult:
+    """Validate the input before processing."""
+    if len(ctx.input) < 5:
+        return StepResult(
+            output="Input too short. Minimum 5 characters required.",
+            stop_workflow=True
+        )
+    return StepResult(output=f"✅ Valid input: {ctx.input}")
+
+def format_output(ctx: WorkflowContext) -> StepResult:
+    """Format the final output."""
+    return StepResult(
+        output=f"📊 Report:\n{ctx.previous_result}",
+        variables={"report_generated": True}
+    )
+
+# Create workflow with mixed step types
+workflow = Workflow(
+    name="Mixed Pipeline",
+    steps=[
+        # Step 1: Function - validates input
+        validate_input,
+        
+        # Step 2: Agent - analyzes content
+        analyzer,
+        
+        # Step 3: WorkflowStep with action string
+        WorkflowStep(
+            name="enhance",
+            action="Add a conclusion to this analysis: {{previous_output}}"
+        ),
+        
+        # Step 4: Function - formats output
+        format_output
+    ],
+    default_llm="gpt-4o-mini"  # Used for action-based steps
+)
+
+if __name__ == "__main__":
+    # Test with valid input
+    print("=== Running mixed workflow ===")
+    result = workflow.start(
+        "Analyze the trends in AI adoption for 2024",
+        verbose=True
+    )
+    
+    print(f"\nFinal output:\n{result['output']}")
+    print(f"\nVariables: {result['variables']}")

examples/python/workflows/workflow_parallel.py

@@ -0,0 +1,53 @@
+"""
+Workflow Parallel Execution Example
+
+Demonstrates running multiple steps concurrently and
+combining their results.
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import parallel
+import time
+
+# Parallel workers - each does independent work
+def research_market(ctx: WorkflowContext) -> StepResult:
+    """Research market trends."""
+    time.sleep(0.1)  # Simulate work
+    return StepResult(output="📊 Market: Growth expected 15% YoY")
+
+def research_competitors(ctx: WorkflowContext) -> StepResult:
+    """Research competitor analysis."""
+    time.sleep(0.1)  # Simulate work
+    return StepResult(output="🏢 Competitors: 3 major players identified")
+
+def research_customers(ctx: WorkflowContext) -> StepResult:
+    """Research customer feedback."""
+    time.sleep(0.1)  # Simulate work
+    return StepResult(output="👥 Customers: 85% satisfaction rate")
+
+# Aggregator - combines parallel results
+def summarize_research(ctx: WorkflowContext) -> StepResult:
+    """Summarize all research findings."""
+    outputs = ctx.variables.get("parallel_outputs", [])
+    summary = "📋 RESEARCH SUMMARY:\n" + "\n".join(f"  • {o}" for o in outputs)
+    return StepResult(output=summary)
+
+# Create workflow with parallel execution
+workflow = Workflow(
+    name="Parallel Research",
+    steps=[
+        parallel([research_market, research_competitors, research_customers]),
+        summarize_research
+    ]
+)
+
+if __name__ == "__main__":
+    print("=== Testing Parallel Workflow ===\n")
+    
+    start_time = time.time()
+    result = workflow.start("Analyze the business landscape", verbose=True)
+    elapsed = time.time() - start_time
+    
+    print(f"\nFinal Output:\n{result['output']}")
+    print(f"\nCompleted in {elapsed:.2f}s (parallel execution)")
+    print(f"Individual outputs: {len(result['variables']['parallel_outputs'])}")

examples/python/workflows/workflow_repeat.py

@@ -0,0 +1,69 @@
+"""
+Workflow Repeat (Evaluator-Optimizer) Example
+
+Demonstrates repeating a step until a condition is met.
+This is useful for iterative improvement patterns.
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import repeat
+
+# Simulated content generator that improves each iteration
+class ContentGenerator:
+    def __init__(self):
+        self.iteration = 0
+        self.content = []
+    
+    def generate(self, ctx: WorkflowContext) -> StepResult:
+        """Generate content, adding more each iteration."""
+        self.iteration += 1
+        
+        # Add new content each iteration
+        new_items = [
+            f"Point {self.iteration}.1: Key insight about the topic",
+            f"Point {self.iteration}.2: Supporting evidence"
+        ]
+        self.content.extend(new_items)
+        
+        output = f"Iteration {self.iteration}: Generated {len(self.content)} points total"
+        
+        return StepResult(
+            output=output,
+            variables={
+                "total_points": len(self.content),
+                "content": self.content.copy()
+            }
+        )
+
+# Condition: stop when we have enough content
+def has_enough_content(ctx: WorkflowContext) -> bool:
+    """Check if we have generated enough content."""
+    total = ctx.variables.get("total_points", 0)
+    return total >= 6  # Stop when we have 6+ points
+
+# Create generator instance
+generator = ContentGenerator()
+
+# Create workflow with repeat pattern
+workflow = Workflow(
+    name="Content Generator",
+    steps=[
+        repeat(
+            generator.generate,
+            until=has_enough_content,
+            max_iterations=10
+        )
+    ]
+)
+
+if __name__ == "__main__":
+    print("=== Testing Repeat (Evaluator-Optimizer) Pattern ===\n")
+    
+    result = workflow.start("Generate comprehensive content", verbose=True)
+    
+    print(f"\nFinal output: {result['output']}")
+    print(f"Total iterations: {result['variables']['repeat_iterations']}")
+    print(f"Total points generated: {result['variables']['total_points']}")
+    print("\nGenerated content:")
+    for point in result['variables']['content']:
+        print(f"  • {point}")

examples/python/workflows/workflow_routing.py

@@ -0,0 +1,61 @@
+"""
+Workflow Routing Example
+
+Demonstrates decision-based routing where the workflow
+takes different paths based on the output of a decision step.
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import route
+
+# Decision maker - determines which route to take
+def classify_request(ctx: WorkflowContext) -> StepResult:
+    """Classify the input and decide the route."""
+    input_lower = ctx.input.lower()
+    
+    if "urgent" in input_lower or "emergency" in input_lower:
+        return StepResult(output="priority: high")
+    elif "question" in input_lower or "help" in input_lower:
+        return StepResult(output="priority: support")
+    else:
+        return StepResult(output="priority: normal")
+
+# Route handlers
+def handle_high_priority(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output="🚨 HIGH PRIORITY: Escalating to senior team immediately!")
+
+def handle_support(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output="💬 SUPPORT: Routing to help desk for assistance.")
+
+def handle_normal(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output="📋 NORMAL: Added to standard processing queue.")
+
+# Create workflow with routing
+workflow = Workflow(
+    name="Request Router",
+    steps=[
+        classify_request,
+        route({
+            "high": [handle_high_priority],
+            "support": [handle_support],
+            "normal": [handle_normal],
+            "default": [handle_normal]
+        })
+    ]
+)
+
+if __name__ == "__main__":
+    # Test different inputs
+    print("=== Testing Routing Workflow ===\n")
+    
+    # Test 1: Urgent request
+    result = workflow.start("This is an URGENT matter!", verbose=True)
+    print(f"Result: {result['output']}\n")
+    
+    # Test 2: Support request
+    result = workflow.start("I have a question about my account", verbose=True)
+    print(f"Result: {result['output']}\n")
+    
+    # Test 3: Normal request
+    result = workflow.start("Please process my order", verbose=True)
+    print(f"Result: {result['output']}")

examples/python/workflows/workflow_with_agents.py

@@ -0,0 +1,45 @@
+"""
+Workflow with Agents Example
+
+Demonstrates using Agent instances directly as workflow steps.
+Agents are automatically wrapped and executed in sequence.
+"""
+
+from praisonaiagents import Workflow, Agent
+
+# Create agents
+researcher = Agent(
+    name="Researcher",
+    role="Research Specialist",
+    goal="Find accurate information",
+    llm="gpt-4o-mini"
+)
+
+writer = Agent(
+    name="Writer", 
+    role="Content Writer",
+    goal="Write engaging content",
+    llm="gpt-4o-mini"
+)
+
+editor = Agent(
+    name="Editor",
+    role="Content Editor", 
+    goal="Polish and improve content",
+    llm="gpt-4o-mini"
+)
+
+# Create workflow with agents as steps
+workflow = Workflow(
+    name="Content Pipeline",
+    steps=[researcher, writer, editor]
+)
+
+if __name__ == "__main__":
+    # Run the workflow
+    result = workflow.start(
+        "Write a short paragraph about artificial intelligence",
+        verbose=True
+    )
+    
+    print(f"\nFinal output:\n{result['output']}")

src/praisonai-agents/praisonaiagents/__init__.py

@@ -54,6 +54,11 @@ except ImportError:
     MCP = None
 from .session import Session
 from .memory.memory import Memory
+from .memory.workflows import (
+    Workflow, WorkflowStep, WorkflowContext, StepResult,
+    Route, Parallel, Loop, Repeat,
+    route, parallel, loop, repeat
+)
 from .guardrails import GuardrailResult, LLMGuardrail
 
 # Fast Context support (lazy loaded to avoid performance impact)

src/praisonai-agents/praisonaiagents/memory/__init__.py

@@ -57,6 +57,19 @@ def __getattr__(name):
     if name == "WorkflowStep":
         from .workflows import WorkflowStep
         return WorkflowStep
+    if name == "WorkflowContext":
+        from .workflows import WorkflowContext
+        return WorkflowContext
+    if name == "StepResult":
+        from .workflows import StepResult
+        return StepResult
+    # Backward compatibility aliases
+    if name == "StepInput":
+        from .workflows import WorkflowContext
+        return WorkflowContext
+    if name == "StepOutput":
+        from .workflows import StepResult
+        return StepResult
     if name == "create_workflow_manager":
         from .workflows import create_workflow_manager
         return create_workflow_manager
@@ -89,6 +102,8 @@ __all__ = [
     "WorkflowManager",
     "Workflow",
     "WorkflowStep",
+    "WorkflowContext",
+    "StepResult",
     "create_workflow_manager",
     # Hooks
     "HooksManager",

src/praisonai-agents/praisonaiagents/memory/workflows.py

@@ -29,13 +29,166 @@ from dataclasses import dataclass, field
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class WorkflowContext:
+    """Context passed to step handlers. Contains all information about the current workflow state."""
+    input: str = ""  # Original workflow input
+    previous_result: Optional[str] = None  # Output from previous step
+    current_step: str = ""  # Current step name
+    variables: Dict[str, Any] = field(default_factory=dict)  # All workflow variables
+    
+@dataclass
+class StepResult:
+    """Result returned from step handlers."""
+    output: str = ""  # Step output content
+    stop_workflow: bool = False  # If True, stop the entire workflow early
+    variables: Dict[str, Any] = field(default_factory=dict)  # Variables to add/update
+
+# Aliases for backward compatibility
+StepInput = WorkflowContext
+StepOutput = StepResult
+
+
+# =============================================================================
+# Workflow Pattern Helpers
+# =============================================================================
+
+@dataclass
+class Route:
+    """
+    Decision-based branching. Routes to different steps based on previous output.
+    
+    Usage:
+        workflow = Workflow(steps=[
+            decision_maker,
+            route({
+                "approve": [approve_step],
+                "reject": [reject_step],
+                "default": [fallback_step]
+            })
+        ])
+    """
+    routes: Dict[str, List] = field(default_factory=dict)
+    default: Optional[List] = None
+    
+    def __init__(self, routes: Dict[str, List], default: Optional[List] = None):
+        self.routes = routes
+        self.default = default or routes.get("default", [])
+
+
+@dataclass  
+class Parallel:
+    """
+    Execute multiple steps concurrently and combine results.
+    
+    Usage:
+        workflow = Workflow(steps=[
+            parallel([agent1, agent2, agent3]),
+            aggregator
+        ])
+    """
+    steps: List = field(default_factory=list)
+    
+    def __init__(self, steps: List):
+        self.steps = steps
+
+
+@dataclass
+class Loop:
+    """
+    Iterate over a list or CSV file, executing step for each item.
+    
+    Usage:
+        # Loop over list variable
+        workflow = Workflow(
+            steps=[loop(processor, over="items")],
+            variables={"items": ["a", "b", "c"]}
+        )
+        
+        # Loop over CSV file
+        workflow = Workflow(steps=[
+            loop(processor, from_csv="data.csv")
+        ])
+    """
+    step: Any = None
+    over: Optional[str] = None  # Variable name containing list
+    from_csv: Optional[str] = None  # CSV file path
+    from_file: Optional[str] = None  # Text file path (one item per line)
+    var_name: str = "item"  # Variable name for current item
+    
+    def __init__(
+        self, 
+        step: Any, 
+        over: Optional[str] = None,
+        from_csv: Optional[str] = None,
+        from_file: Optional[str] = None,
+        var_name: str = "item"
+    ):
+        self.step = step
+        self.over = over
+        self.from_csv = from_csv
+        self.from_file = from_file
+        self.var_name = var_name
+
+
+@dataclass
+class Repeat:
+    """
+    Repeat a step until a condition is met (evaluator-optimizer pattern).
+    
+    Usage:
+        workflow = Workflow(steps=[
+            repeat(
+                generator,
+                until=lambda ctx: "done" in ctx.previous_result.lower(),
+                max_iterations=5
+            )
+        ])
+    """
+    step: Any = None
+    until: Optional[Callable[[WorkflowContext], bool]] = None
+    max_iterations: int = 10
+    
+    def __init__(
+        self,
+        step: Any,
+        until: Optional[Callable[[WorkflowContext], bool]] = None,
+        max_iterations: int = 10
+    ):
+        self.step = step
+        self.until = until
+        self.max_iterations = max_iterations
+
+
+# Convenience functions for cleaner API
+def route(routes: Dict[str, List], default: Optional[List] = None) -> Route:
+    """Create a routing decision point."""
+    return Route(routes=routes, default=default)
+
+def parallel(steps: List) -> Parallel:
+    """Execute steps in parallel."""
+    return Parallel(steps=steps)
+
+def loop(step: Any, over: Optional[str] = None, from_csv: Optional[str] = None, 
+         from_file: Optional[str] = None, var_name: str = "item") -> Loop:
+    """Loop over items executing step for each."""
+    return Loop(step=step, over=over, from_csv=from_csv, from_file=from_file, var_name=var_name)
+
+def repeat(step: Any, until: Optional[Callable[[WorkflowContext], bool]] = None,
+           max_iterations: int = 10) -> Repeat:
+    """Repeat step until condition is met."""
+    return Repeat(step=step, until=until, max_iterations=max_iterations)
+
+
 @dataclass
 class WorkflowStep:
     """A single step in a workflow."""
     name: str
     description: str = ""
     action: str = ""  # The action/prompt to execute
-    condition: Optional[str] = None  # Optional condition for execution
+    condition: Optional[str] = None  # Optional condition for execution (string)
+    should_run: Optional[Callable[['WorkflowContext'], bool]] = None  # Function to check if step should run
+    handler: Optional[Callable[['WorkflowContext'], 'StepResult']] = None  # Custom function instead of agent
     on_error: Literal["stop", "continue", "retry"] = "stop"
     max_retries: int = 1
     
@@ -46,6 +199,7 @@ class WorkflowStep:
     
     # Agent configuration fields
     agent_config: Optional[Dict[str, Any]] = None  # Per-step agent config {role, goal, backstory, llm}
+    agent: Optional[Any] = None  # Direct agent instance for this step
     tools: Optional[List[Any]] = None  # Tools for this step
     
     # Branching fields
@@ -56,12 +210,24 @@ class WorkflowStep:
     loop_over: Optional[str] = None  # Variable name to iterate over (e.g., "items")
     loop_var: str = "item"  # Variable name for current item in loop
     
+    # Backward compatibility aliases
+    @property
+    def evaluator(self):
+        return self.should_run
+    
+    @property
+    def executor(self):
+        return self.handler
+    
     def to_dict(self) -> Dict[str, Any]:
         return {
             "name": self.name,
             "description": self.description,
             "action": self.action,
             "condition": self.condition,
+            "should_run": self.should_run is not None,
+            "handler": self.handler is not None,
+            "agent": self.agent is not None,
             "on_error": self.on_error,
             "max_retries": self.max_retries,
             "context_from": self.context_from,
@@ -79,9 +245,9 @@ class WorkflowStep:
 @dataclass
 class Workflow:
     """A complete workflow with multiple steps."""
-    name: str
+    name: str = "Workflow"
     description: str = ""
-    steps: List[WorkflowStep] = field(default_factory=list)
+    steps: List = field(default_factory=list)  # Can be WorkflowStep, Agent, or function
     variables: Dict[str, Any] = field(default_factory=dict)
     file_path: Optional[str] = None
     
@@ -105,6 +271,512 @@ class Workflow:
             "planning": self.planning,
             "planning_llm": self.planning_llm
         }
+    
+    def run(
+        self,
+        input: str = "",
+        llm: Optional[str] = None,
+        verbose: bool = False
+    ) -> Dict[str, Any]:
+        """
+        Run the workflow with the given input.
+        
+        This is the simplest way to execute a workflow:
+        
+        ```python
+        from praisonaiagents import Workflow, Agent
+        
+        workflow = Workflow(
+            steps=[
+                Agent(name="Writer", role="Write content"),
+                my_validator_function,
+                Agent(name="Editor", role="Edit content"),
+            ]
+        )
+        result = workflow.run("Write about AI")
+        ```
+        
+        Args:
+            input: The input text/prompt for the workflow
+            llm: LLM model to use (default: gpt-4o-mini)
+            verbose: Print step outputs
+            
+        Returns:
+            Dict with 'output' (final result) and 'steps' (all step results)
+        """
+        # Use default LLM if not specified
+        model = llm or self.default_llm or "gpt-4o-mini"
+        
+        # Add input to variables
+        all_variables = {**self.variables, "input": input}
+        
+        results = []
+        previous_output = None
+        
+        # Process steps (may include Route, Parallel, Loop, Repeat)
+        i = 0
+        while i < len(self.steps):
+            step = self.steps[i]
+            
+            # Handle special pattern types
+            if isinstance(step, Route):
+                # Decision-based routing
+                route_result = self._execute_route(
+                    step, previous_output, input, all_variables, model, verbose
+                )
+                results.extend(route_result["steps"])
+                previous_output = route_result["output"]
+                all_variables.update(route_result.get("variables", {}))
+                i += 1
+                continue
+                
+            elif isinstance(step, Parallel):
+                # Parallel execution
+                parallel_result = self._execute_parallel(
+                    step, previous_output, input, all_variables, model, verbose
+                )
+                results.extend(parallel_result["steps"])
+                previous_output = parallel_result["output"]
+                all_variables.update(parallel_result.get("variables", {}))
+                i += 1
+                continue
+                
+            elif isinstance(step, Loop):
+                # Loop over items
+                loop_result = self._execute_loop(
+                    step, previous_output, input, all_variables, model, verbose
+                )
+                results.extend(loop_result["steps"])
+                previous_output = loop_result["output"]
+                all_variables.update(loop_result.get("variables", {}))
+                i += 1
+                continue
+                
+            elif isinstance(step, Repeat):
+                # Repeat until condition
+                repeat_result = self._execute_repeat(
+                    step, previous_output, input, all_variables, model, verbose
+                )
+                results.extend(repeat_result["steps"])
+                previous_output = repeat_result["output"]
+                all_variables.update(repeat_result.get("variables", {}))
+                i += 1
+                continue
+            
+            # Normalize single step
+            step = self._normalize_single_step(step, i)
+            
+            # Create context for handlers
+            context = WorkflowContext(
+                input=input,
+                previous_result=str(previous_output) if previous_output else None,
+                current_step=step.name,
+                variables=all_variables.copy()
+            )
+            
+            # Check should_run condition
+            if step.should_run:
+                try:
+                    if not step.should_run(context):
+                        if verbose:
+                            print(f"⏭️ Skipped: {step.name}")
+                        i += 1
+                        continue
+                except Exception as e:
+                    logger.error(f"should_run failed for {step.name}: {e}")
+            
+            # Execute step
+            output = None
+            stop = False
+            
+            if step.handler:
+                # Custom handler function
+                try:
+                    result = step.handler(context)
+                    if isinstance(result, StepResult):
+                        output = result.output
+                        stop = result.stop_workflow
+                        if result.variables:
+                            all_variables.update(result.variables)
+                    else:
+                        output = str(result)
+                except Exception as e:
+                    output = f"Error: {e}"
+                    
+            elif step.agent:
+                # Direct agent
+                try:
+                    output = step.agent.chat(step.action or input)
+                except Exception as e:
+                    output = f"Error: {e}"
+                    
+            elif step.action:
+                # Action with agent_config - create temporary agent
+                try:
+                    from ..agent.agent import Agent
+                    config = step.agent_config or self.default_agent_config or {}
+                    temp_agent = Agent(
+                        name=config.get("name", step.name),
+                        role=config.get("role", "Assistant"),
+                        goal=config.get("goal", "Complete the task"),
+                        llm=config.get("llm", model),
+                        verbose=verbose
+                    )
+                    # Substitute variables in action
+                    action = step.action
+                    for key, value in all_variables.items():
+                        action = action.replace(f"{{{{{key}}}}}", str(value))
+                    if previous_output:
+                        action = action.replace("{{previous_output}}", str(previous_output))
+                    output = temp_agent.chat(action)
+                except Exception as e:
+                    output = f"Error: {e}"
+            
+            # Store result
+            results.append({
+                "step": step.name,
+                "output": output
+            })
+            previous_output = output
+            
+            if verbose:
+                print(f"✅ {step.name}: {str(output)[:100]}...")
+            
+            # Handle early stop
+            if stop:
+                if verbose:
+                    print(f"🛑 Workflow stopped at: {step.name}")
+                break
+            
+            # Store output in variables
+            var_name = step.output_variable or f"{step.name}_output"
+            all_variables[var_name] = output
+            
+            i += 1
+        
+        return {
+            "output": previous_output,
+            "steps": results,
+            "variables": all_variables
+        }
+    
+    def _normalize_steps(self) -> List['WorkflowStep']:
+        """Convert mixed steps (Agent, function, WorkflowStep) to WorkflowStep list."""
+        normalized = []
+        
+        for i, step in enumerate(self.steps):
+            if isinstance(step, WorkflowStep):
+                normalized.append(step)
+            elif callable(step):
+                # It's a function - wrap as handler
+                normalized.append(WorkflowStep(
+                    name=getattr(step, '__name__', f'step_{i+1}'),
+                    handler=step
+                ))
+            elif hasattr(step, 'chat'):
+                # It's an Agent - wrap with agent reference
+                normalized.append(WorkflowStep(
+                    name=getattr(step, 'name', f'agent_{i+1}'),
+                    agent=step,
+                    action="{{input}}"
+                ))
+            else:
+                # Unknown type - try to use as string action
+                normalized.append(WorkflowStep(
+                    name=f'step_{i+1}',
+                    action=str(step)
+                ))
+        
+        return normalized
+    
+    def _normalize_single_step(self, step: Any, index: int) -> 'WorkflowStep':
+        """Normalize a single step to WorkflowStep."""
+        if isinstance(step, WorkflowStep):
+            return step
+        elif callable(step):
+            return WorkflowStep(
+                name=getattr(step, '__name__', f'step_{index+1}'),
+                handler=step
+            )
+        elif hasattr(step, 'chat'):
+            return WorkflowStep(
+                name=getattr(step, 'name', f'agent_{index+1}'),
+                agent=step,
+                action="{{input}}"
+            )
+        else:
+            return WorkflowStep(
+                name=f'step_{index+1}',
+                action=str(step)
+            )
+    
+    def _execute_single_step_internal(
+        self, 
+        step: Any, 
+        previous_output: Optional[str],
+        input: str,
+        all_variables: Dict[str, Any],
+        model: str,
+        verbose: bool,
+        index: int = 0
+    ) -> Dict[str, Any]:
+        """Execute a single step and return result."""
+        normalized = self._normalize_single_step(step, index)
+        
+        context = WorkflowContext(
+            input=input,
+            previous_result=str(previous_output) if previous_output else None,
+            current_step=normalized.name,
+            variables=all_variables.copy()
+        )
+        
+        output = None
+        stop = False
+        
+        if normalized.handler:
+            try:
+                result = normalized.handler(context)
+                if isinstance(result, StepResult):
+                    output = result.output
+                    stop = result.stop_workflow
+                    if result.variables:
+                        all_variables.update(result.variables)
+                else:
+                    output = str(result)
+            except Exception as e:
+                output = f"Error: {e}"
+        elif normalized.agent:
+            try:
+                output = normalized.agent.chat(normalized.action or input)
+            except Exception as e:
+                output = f"Error: {e}"
+        elif normalized.action:
+            try:
+                from ..agent.agent import Agent
+                config = normalized.agent_config or self.default_agent_config or {}
+                temp_agent = Agent(
+                    name=config.get("name", normalized.name),
+                    role=config.get("role", "Assistant"),
+                    goal=config.get("goal", "Complete the task"),
+                    llm=config.get("llm", model),
+                    verbose=verbose
+                )
+                action = normalized.action
+                for key, value in all_variables.items():
+                    action = action.replace(f"{{{{{key}}}}}", str(value))
+                if previous_output:
+                    action = action.replace("{{previous_output}}", str(previous_output))
+                output = temp_agent.chat(action)
+            except Exception as e:
+                output = f"Error: {e}"
+        
+        if verbose:
+            print(f"✅ {normalized.name}: {str(output)[:100]}...")
+        
+        return {
+            "step": normalized.name,
+            "output": output,
+            "stop": stop,
+            "variables": all_variables
+        }
+    
+    def _execute_route(
+        self,
+        route_step: Route,
+        previous_output: Optional[str],
+        input: str,
+        all_variables: Dict[str, Any],
+        model: str,
+        verbose: bool
+    ) -> Dict[str, Any]:
+        """Execute routing based on previous output."""
+        results = []
+        output = previous_output
+        
+        # Find matching route
+        matched_route = None
+        prev_lower = str(previous_output).lower() if previous_output else ""
+        
+        for key in route_step.routes:
+            if key.lower() in prev_lower or key == "default":
+                if key != "default":
+                    matched_route = route_step.routes[key]
+                    break
+        
+        if matched_route is None:
+            matched_route = route_step.default or []
+        
+        if verbose:
+            route_name = next((k for k in route_step.routes if route_step.routes[k] == matched_route), "default")
+            print(f"🔀 Routing to: {route_name}")
+        
+        # Execute matched route steps
+        for idx, step in enumerate(matched_route):
+            step_result = self._execute_single_step_internal(
+                step, output, input, all_variables, model, verbose, idx
+            )
+            results.append({"step": step_result["step"], "output": step_result["output"]})
+            output = step_result["output"]
+            all_variables.update(step_result.get("variables", {}))
+            
+            if step_result.get("stop"):
+                break
+        
+        return {"steps": results, "output": output, "variables": all_variables}
+    
+    def _execute_parallel(
+        self,
+        parallel_step: Parallel,
+        previous_output: Optional[str],
+        input: str,
+        all_variables: Dict[str, Any],
+        model: str,
+        verbose: bool
+    ) -> Dict[str, Any]:
+        """Execute steps in parallel (simulated with sequential for now)."""
+        import concurrent.futures
+        
+        results = []
+        outputs = []
+        
+        if verbose:
+            print(f"⚡ Running {len(parallel_step.steps)} steps in parallel...")
+        
+        # Use ThreadPoolExecutor for parallel execution
+        with concurrent.futures.ThreadPoolExecutor(max_workers=len(parallel_step.steps)) as executor:
+            futures = []
+            for idx, step in enumerate(parallel_step.steps):
+                future = executor.submit(
+                    self._execute_single_step_internal,
+                    step, previous_output, input, all_variables.copy(), model, False, idx
+                )
+                futures.append((idx, future))
+            
+            for idx, future in futures:
+                try:
+                    step_result = future.result()
+                    results.append({"step": step_result["step"], "output": step_result["output"]})
+                    outputs.append(step_result["output"])
+                except Exception as e:
+                    results.append({"step": f"parallel_{idx}", "output": f"Error: {e}"})
+                    outputs.append(f"Error: {e}")
+        
+        # Combine outputs
+        combined_output = "\n---\n".join(str(o) for o in outputs)
+        all_variables["parallel_outputs"] = outputs
+        
+        if verbose:
+            print(f"✅ Parallel complete: {len(outputs)} results")
+        
+        return {"steps": results, "output": combined_output, "variables": all_variables}
+    
+    def _execute_loop(
+        self,
+        loop_step: Loop,
+        previous_output: Optional[str],
+        input: str,
+        all_variables: Dict[str, Any],
+        model: str,
+        verbose: bool
+    ) -> Dict[str, Any]:
+        """Execute step for each item in loop."""
+        import csv
+        
+        results = []
+        outputs = []
+        items = []
+        
+        # Get items from variable, CSV, or file
+        if loop_step.over:
+            items = all_variables.get(loop_step.over, [])
+            if isinstance(items, str):
+                items = [items]
+        elif loop_step.from_csv:
+            try:
+                with open(loop_step.from_csv, 'r', encoding='utf-8') as f:
+                    reader = csv.DictReader(f)
+                    items = list(reader)
+            except Exception as e:
+                logger.error(f"Failed to read CSV {loop_step.from_csv}: {e}")
+                items = []
+        elif loop_step.from_file:
+            try:
+                with open(loop_step.from_file, 'r', encoding='utf-8') as f:
+                    items = [line.strip() for line in f if line.strip()]
+            except Exception as e:
+                logger.error(f"Failed to read file {loop_step.from_file}: {e}")
+                items = []
+        
+        if verbose:
+            print(f"🔁 Looping over {len(items)} items...")
+        
+        for idx, item in enumerate(items):
+            # Add current item to variables
+            loop_vars = all_variables.copy()
+            loop_vars[loop_step.var_name] = item
+            loop_vars["loop_index"] = idx
+            
+            step_result = self._execute_single_step_internal(
+                loop_step.step, previous_output, input, loop_vars, model, verbose, idx
+            )
+            results.append({"step": f"{step_result['step']}_{idx}", "output": step_result["output"]})
+            outputs.append(step_result["output"])
+            previous_output = step_result["output"]
+        
+        all_variables["loop_outputs"] = outputs
+        combined_output = "\n".join(str(o) for o in outputs) if outputs else ""
+        
+        return {"steps": results, "output": combined_output, "variables": all_variables}
+    
+    def _execute_repeat(
+        self,
+        repeat_step: Repeat,
+        previous_output: Optional[str],
+        input: str,
+        all_variables: Dict[str, Any],
+        model: str,
+        verbose: bool
+    ) -> Dict[str, Any]:
+        """Repeat step until condition is met."""
+        results = []
+        output = previous_output
+        
+        if verbose:
+            print(f"🔄 Repeating up to {repeat_step.max_iterations} times...")
+        
+        for iteration in range(repeat_step.max_iterations):
+            step_result = self._execute_single_step_internal(
+                repeat_step.step, output, input, all_variables, model, verbose, iteration
+            )
+            results.append({"step": f"{step_result['step']}_{iteration}", "output": step_result["output"]})
+            output = step_result["output"]
+            all_variables.update(step_result.get("variables", {}))
+            
+            # Check until condition
+            if repeat_step.until:
+                context = WorkflowContext(
+                    input=input,
+                    previous_result=str(output) if output else None,
+                    current_step=f"repeat_{iteration}",
+                    variables=all_variables.copy()
+                )
+                try:
+                    if repeat_step.until(context):
+                        if verbose:
+                            print(f"✅ Repeat condition met at iteration {iteration + 1}")
+                        break
+                except Exception as e:
+                    logger.error(f"Repeat until condition failed: {e}")
+            
+            if step_result.get("stop"):
+                break
+        
+        all_variables["repeat_iterations"] = iteration + 1
+        return {"steps": results, "output": output, "variables": all_variables}
+    
+    def start(self, input: str = "", **kwargs) -> Dict[str, Any]:
+        """Alias for run() for consistency with PraisonAIAgents."""
+        return self.run(input, **kwargs)
 
 
 class WorkflowManager:
@@ -717,6 +1389,16 @@ class WorkflowManager:
                 on_result=on_result
             )
             
+            # Handle skipped steps
+            if step_result.get("skipped"):
+                results.append({
+                    "step": step.name,
+                    "status": "skipped",
+                    "output": None
+                })
+                current_step_idx += 1
+                continue
+            
             results.append({
                 "step": step.name,
                 "status": "success" if step_result["success"] else "failed",
@@ -724,6 +1406,11 @@ class WorkflowManager:
                 "error": step_result.get("error")
             })
             
+            # Handle early stop 
+            if step_result.get("stop"):
+                self._log(f"Workflow stopped early at step '{step.name}'")
+                break
+            
             # Save checkpoint after each step if enabled
             if checkpoint:
                 self._save_checkpoint(
@@ -784,9 +1471,64 @@ class WorkflowManager:
         planning: bool = False,
         verbose: int = 0,
         on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
-        on_result: Optional[Callable[[WorkflowStep, str], None]] = None
+        on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
+        original_input: str = ""
     ) -> Dict[str, Any]:
         """Execute a single workflow step."""
+        # Get previous step output
+        previous_output = results[-1].get("output") if results else None
+        
+        # Create context for step handlers
+        context = WorkflowContext(
+            input=original_input,
+            previous_result=str(previous_output) if previous_output else None,
+            current_step=step.name,
+            variables=all_variables.copy()
+        )
+        
+        # Check should_run condition if provided
+        if step.should_run:
+            try:
+                if not step.should_run(context):
+                    return {
+                        "success": True,
+                        "output": None,
+                        "skipped": True,
+                        "stop": False
+                    }
+            except Exception as e:
+                self._log(f"should_run check for step '{step.name}' failed: {e}")
+        
+        # If step has a custom handler function
+        if step.handler:
+            try:
+                result = step.handler(context)
+                # Handle StepResult
+                if isinstance(result, StepResult):
+                    # Update variables from result
+                    if result.variables:
+                        all_variables.update(result.variables)
+                    return {
+                        "success": True,
+                        "output": result.output,
+                        "stop": result.stop_workflow,  # Early termination flag
+                        "error": None
+                    }
+                else:
+                    return {
+                        "success": True,
+                        "output": str(result),
+                        "stop": False,
+                        "error": None
+                    }
+            except Exception as e:
+                return {
+                    "success": False,
+                    "output": None,
+                    "stop": False,
+                    "error": str(e)
+                }
+        
         # Build context from previous steps
         context = self._build_step_context(step, step_idx, results, all_variables)
         
@@ -800,15 +1542,20 @@ class WorkflowManager:
         # Get or create executor for this step
         step_executor = executor
         if step_executor is None:
-            # Create agent for this step
-            step_agent = self._create_step_agent(
-                step=step,
-                default_agent=default_agent,
-                default_llm=default_llm,
-                memory=memory,
-                planning=planning,
-                verbose=verbose
-            )
+            # Use step's direct agent if provided
+            step_agent = step.agent
+            
+            # Otherwise create agent from config
+            if step_agent is None:
+                step_agent = self._create_step_agent(
+                    step=step,
+                    default_agent=default_agent,
+                    default_llm=default_llm,
+                    memory=memory,
+                    planning=planning,
+                    verbose=verbose
+                )
+            
             if step_agent:
                 if planning and hasattr(step_agent, 'start'):
                     def agent_executor(prompt, agent=step_agent):
@@ -823,6 +1570,7 @@ class WorkflowManager:
             return {
                 "success": False,
                 "output": None,
+                "stop": False,
                 "error": f"No executor available for step '{step.name}'"
             }
         
@@ -853,6 +1601,7 @@ class WorkflowManager:
         return {
             "success": step_success,
             "output": output,
+            "stop": False,  # Normal execution doesn't request stop
             "error": error
         }
     

src/praisonai-agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "0.0.194"
+version = "0.0.198"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 requires-python = ">=3.10"
 authors = [

src/praisonai-agents/uv.lock

@@ -2422,7 +2422,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.194"
+version = "0.0.198"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai/README.md

@@ -103,7 +103,19 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | @ @Mentions in Prompts | [Example](#mentions-in-prompts) | [📖](https://docs.praison.ai/cli/mentions) |
 | 💾 Auto-Save Sessions | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
 | 📜 History in Context | [Example](#session-management-python) | [📖](https://docs.praison.ai/cli/session) |
-| ⏸️ Workflow Checkpoints | [Example](#workflow-checkpoints) | [📖](https://docs.praison.ai/cli/session) |
+| ⏸️ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/cli/session) |
+| 🔀 Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔁 Workflow Loops | [Example](examples/python/workflows/workflow_loops.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 📞 Task Callbacks | [Example](examples/python/workflows/task_callbacks.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🛑 Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ⚡ Simple Workflow | [Example](examples/python/workflows/simple_workflow.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🤖 Workflow with Agents | [Example](examples/python/workflows/workflow_with_agents.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔄 Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🎭 Mixed Steps | [Example](examples/python/workflows/workflow_mixed_steps.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 📄 CSV Loop Processing | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflows) |
 
 ## Supported Providers
 
@@ -445,6 +457,138 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 - **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
 - **Async Execution**: Use `aexecute()` for async workflows
 - **Planning Mode**: Enable at workflow level with `planning=True`
+- **Branching**: Use `next_steps` and `branch_condition` for conditional routing
+- **Loops**: Use `loop_over` and `loop_var` to iterate over data
+
+### Choosing the Right Workflow System
+
+| Use Case | Recommended |
+|----------|-------------|
+| Simple function pipelines | `Workflow` class ⭐ |
+| Agent-only pipelines | `Workflow` class |
+| CSV batch processing | `process="workflow"` |
+| Complex task routing | `process="workflow"` |
+| Markdown templates | `WorkflowManager` |
+| Early stop / conditional | `Workflow` class |
+
+### Simple Workflow (Recommended)
+
+The easiest way to create workflows - just pass functions as steps:
+
+```python
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+
+# Define simple handler functions
+def validate(ctx: WorkflowContext) -> StepResult:
+    if not ctx.input:
+        return StepResult(output="No input", stop_workflow=True)
+    return StepResult(output=f"Valid: {ctx.input}")
+
+def process(ctx: WorkflowContext) -> StepResult:
+    return StepResult(output=f"Processed: {ctx.previous_result}")
+
+# Create and run workflow
+workflow = Workflow(steps=[validate, process])
+result = workflow.start("Hello World", verbose=True)
+print(result["output"])  # "Processed: Valid: Hello World"
+```
+
+**Key Features:**
+- **Just pass functions** - No complex configuration needed
+- **Early stop** - Return `stop_workflow=True` to stop the workflow
+- **Context passing** - Access `ctx.input`, `ctx.previous_result`, `ctx.variables`
+- **Verbose mode** - See step-by-step progress
+
+### Workflow Branching & Loops
+
+```python
+from praisonaiagents import WorkflowStep
+
+# Branching step
+decision_step = WorkflowStep(
+    name="decide",
+    action="Evaluate if task is complete",
+    next_steps=["success_step", "retry_step"],
+    branch_condition={"success": ["success_step"], "failure": ["retry_step"]}
+)
+
+# Loop step
+loop_step = WorkflowStep(
+    name="process_items",
+    action="Process {{item}}",
+    loop_over="items",  # Variable containing list
+    loop_var="item"     # Current item variable name
+)
+```
+
+### Workflow Patterns (route, parallel, loop, repeat)
+
+```python
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.memory.workflows import route, parallel, loop, repeat
+
+# 1. ROUTING - Decision-based branching
+workflow = Workflow(steps=[
+    classify_request,  # Returns "approve" or "reject"
+    route({
+        "approve": [approve_handler],
+        "reject": [reject_handler],
+        "default": [fallback_handler]
+    })
+])
+
+# 2. PARALLEL - Concurrent execution
+workflow = Workflow(steps=[
+    parallel([research_market, research_competitors, research_customers]),
+    summarize_results  # Gets all parallel outputs
+])
+
+# 3. LOOP - Iterate over list or CSV
+workflow = Workflow(
+    steps=[loop(process_item, over="items")],
+    variables={"items": ["a", "b", "c"]}
+)
+# Or from CSV file:
+workflow = Workflow(steps=[loop(process_row, from_csv="data.csv")])
+
+# 4. REPEAT - Evaluator-Optimizer pattern
+workflow = Workflow(steps=[
+    repeat(
+        generator,
+        until=lambda ctx: "done" in ctx.previous_result,
+        max_iterations=5
+    )
+])
+```
+
+### Task Callbacks & Variables (process="workflow")
+
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Per-task agent config (no need to create agent separately)
+task = Task(
+    description="Research {{topic}}",
+    agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
+    variables={"topic": "AI trends"}
+)
+
+# Callbacks for workflow execution
+def on_start(task, task_id):
+    print(f"Starting: {task.name}")
+
+def on_complete(task, output):
+    print(f"Completed: {task.name}")
+
+agents = PraisonAIAgents(
+    agents=[agent],
+    tasks=[task],
+    process="workflow",
+    on_task_start=on_start,
+    on_task_complete=on_complete,
+    variables={"global_var": "shared_value"}  # Global variables for all tasks
+)
+```
 
 ### 9. Hooks
 

src/praisonai/praisonai.rb

@@ -3,8 +3,8 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.16.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.16.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.20.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.20.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/deploy.py

@@ -57,7 +57,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==2.3.16 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.20 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.3.16"
+__version__ = "2.3.20"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.194",
+    "praisonaiagents>=0.0.198",
     "python-dotenv>=0.19.0",
     "instructor>=1.3.3",
     "PyYAML>=6.0",

src/praisonai/uv.lock

@@ -4197,7 +4197,7 @@ requires-dist = [
     { name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
-    { name = "praisonaiagents", specifier = ">=0.0.194" },
+    { name = "praisonaiagents", specifier = ">=0.0.198" },
     { name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
     { name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
     { name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.194"
+version = "0.0.198"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
     { name = "pydantic" },
     { name = "rich" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/e9/5f/f572e88d4148e61da3d114f71357c9103a822099103b0b57adc7d88bedc2/praisonaiagents-0.0.194.tar.gz", hash = "sha256:e70c85203c9cf3792697535844a8bba4da158ecb9959abc5025967199eb86d36", size = 518340, upload-time = "2025-12-16T22:58:43.725Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/68/bb/a3daf58206327a4cc92abed7295b41c03f9c4a8f3411d9435c633eb45b7f/praisonaiagents-0.0.198.tar.gz", hash = "sha256:d02180d476dd3ea260ade633764f6f8831dddc550b17ffb21eed79809166d315", size = 523691, upload-time = "2025-12-16T23:46:10.247Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/01/18/d0d0d99ec828d0cc749eb642e0246abda45575c18646893786cdad01fd9c/praisonaiagents-0.0.194-py3-none-any.whl", hash = "sha256:a1f89d3169a4df8c0c96706afcc7c71f20560c4fd2a37aaa0b8193ee6445fdef", size = 494859, upload-time = "2025-12-16T22:58:41.237Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/cf/704504b3f5f3c9fc5674a0514ebea294fd4ffe7d040aae86d8af91dc0446/praisonaiagents-0.0.198-py3-none-any.whl", hash = "sha256:9011703eca1474fbef0a7208e0f875f1b62ff19bfaf7806fa25ccf7afb5c65a0", size = 499909, upload-time = "2025-12-16T23:46:08.677Z" },
 ]
 
 [[package]]