Release v2.3.25

Release v2.3.24
Release v2.3.23
2026-11-10 11:41:32 -10:00 · 2025-12-17 08:17:15 +00:00 · 2025-12-17 08:04:24 +00:00 · 2025-12-17 00:41:48 +00:00 · 2025-12-17 00:25:14 +00:00 · 2025-12-17 00:05:40 +00:00
28 changed files with 5665 additions and 112 deletions
@@ -40,7 +40,17 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🔄 Self Reflection AI Agents | [Example](examples/python/concepts/self-reflection-details.py) | [📖](https://docs.praison.ai/features/selfreflection) |
 | 🧠 Reasoning AI Agents | [Example](examples/python/concepts/reasoning-extraction.py) | [📖](https://docs.praison.ai/features/reasoning) |
 | 👁️ Multi Modal AI Agents | [Example](examples/python/general/multimodal.py) | [📖](https://docs.praison.ai/features/multimodal) |
-| 🎭 AI Agent Workflow (Context Passing, Per-Step Agents) | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
+| **🔄 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) |
+| ↳ Agentic Routing (`route()`) | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/routing) |
+| ↳ Parallel Execution (`parallel()`) | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/parallelisation) |
+| ↳ Loop over List/CSV (`loop()`) | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/repetitive) |
+| ↳ Evaluator-Optimizer (`repeat()`) | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/evaluator-optimiser) |
+| ↳ Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/features/workflows) |
 | 📚 Add Custom Knowledge | [Example](examples/python/concepts/knowledge-agents.py) | [📖](https://docs.praison.ai/features/knowledge) |
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
@@ -103,15 +113,6 @@ 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](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) |

 ## Supported Providers

@@ -456,6 +457,17 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 - **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 | `Workflow` + `loop()` |
+| Complex task routing | `Workflow` + `route()` |
+| Markdown templates | `WorkflowManager` |
+| Early stop / conditional | `Workflow` class |
+
 ### Simple Workflow (Recommended)

 The easiest way to create workflows - just pass functions as steps:
@@ -506,33 +518,218 @@ loop_step = WorkflowStep(
 )
 ```

-### Task Callbacks & Variables (process="workflow")
+### Workflow Patterns (route, parallel, loop, repeat)

 ```python
-from praisonaiagents import Agent, Task, PraisonAIAgents
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+# Or use Pipeline (alias for Workflow)
+from praisonaiagents import Pipeline
+from praisonaiagents.workflows import route, parallel, loop, repeat

-# 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"}
+# 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
+    )
+])
+
+# 5. CALLBACKS - Monitor workflow execution
+workflow = Workflow(
+    steps=[step1, step2],
+    on_workflow_start=lambda w, i: print(f"Starting: {i}"),
+    on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
+    on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
 )

-# Callbacks for workflow execution
-def on_start(task, task_id):
-    print(f"Starting: {task.name}")
+# 6. GUARDRAILS - Validate and retry
+def validate(result):
+    return ("error" not in result.output, "Fix the error")

-def on_complete(task, output):
-    print(f"Completed: {task.name}")
+workflow = Workflow(steps=[
+    WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
+])
+```

-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
+### Workflow with Agents
+
+Use Agent objects directly as workflow steps:
+
+```python
+from praisonaiagents import Agent, Workflow
+from praisonaiagents.workflows import route, parallel
+
+# 1. SEQUENTIAL AGENTS
+researcher = Agent(name="Researcher", role="Research expert", tools=[tavily_search])
+writer = Agent(name="Writer", role="Content writer")
+editor = Agent(name="Editor", role="Editor")
+
+workflow = Workflow(steps=[researcher, writer, editor])
+result = workflow.start("Research and write about AI")
+
+# 2. PARALLEL AGENTS
+workflow = Workflow(steps=[
+    parallel([researcher1, researcher2, researcher3]),
+    aggregator_agent
+])
+
+# 3. ROUTE TO AGENTS
+workflow = Workflow(steps=[
+    classifier_function,
+    route({
+        "technical": [tech_agent],
+        "creative": [creative_agent],
+        "default": [general_agent]
+    })
+])
+
+# 4. WITH PLANNING & REASONING
+workflow = Workflow(
+    steps=[researcher, writer, editor],
+    planning=True,           # Create execution plan
+    planning_llm="gpt-4o",   # LLM for planning
+    reasoning=True,          # Chain-of-thought reasoning
+    verbose=True
 )
+
+# 5. TOOLS PER STEP
+workflow = Workflow(steps=[
+    WorkflowStep(
+        name="research",
+        action="Research {{topic}}",
+        tools=[tavily_search, web_scraper],
+        agent_config={"name": "Researcher", "role": "Expert"}
+    )
+])
+
+# 6. OUTPUT TO FILE / IMAGES / PYDANTIC
+from pydantic import BaseModel
+
+class Report(BaseModel):
+    title: str
+    content: str
+
+workflow = Workflow(steps=[
+    WorkflowStep(name="analyze", action="Analyze image", images=["image.jpg"]),
+    WorkflowStep(name="report", action="Generate report", output_pydantic=Report),
+    WorkflowStep(name="save", action="Save results", output_file="output/report.txt")
+])
+
+# 7. ASYNC EXECUTION
+import asyncio
+
+async def main():
+    result = await workflow.astart("input")
+    print(result)
+
+asyncio.run(main())
+
+# 8. STATUS TRACKING
+workflow.status  # "not_started" | "running" | "completed"
+workflow.step_statuses  # {"step1": "completed", "step2": "skipped"}
+
+# 9. MEMORY CONFIG
+workflow = Workflow(
+    steps=[researcher, writer],
+    memory_config={"provider": "chroma", "persist": True, "collection": "my_workflow"}
+)
+result1 = workflow.start("Research AI")
+result2 = workflow.start("Continue the research")  # Remembers first run
+```
+
+### YAML Workflow Template
+
+```yaml
+# .praison/workflows/research.yaml
+name: Research Workflow
+description: Research and write content with all patterns
+
+agents:
+  researcher:
+    role: Research Expert
+    goal: Find accurate information
+    tools: [tavily_search, web_scraper]
+  writer:
+    role: Content Writer
+    goal: Write engaging content
+  editor:
+    role: Editor
+    goal: Polish content
+
+steps:
+  # Sequential
+  - agent: researcher
+    action: Research {{topic}}
+    output_variable: research_data
+
+  # Routing
+  - name: classifier
+    action: Classify content type
+    route:
+      technical: [tech_handler]
+      creative: [creative_handler]
+      default: [general_handler]
+
+  # Parallel
+  - name: parallel_research
+    parallel:
+      - agent: researcher
+        action: Research market
+      - agent: researcher
+        action: Research competitors
+
+  # Loop
+  - agent: writer
+    action: Write about {{item}}
+    loop_over: topics
+    loop_var: item
+
+  # Repeat (evaluator-optimizer)
+  - agent: editor
+    action: Review and improve
+    repeat:
+      until: "quality > 8"
+      max_iterations: 3
+
+  # Output to file
+  - agent: writer
+    action: Write final report
+    output_file: output/{{topic}}_report.md
+
+variables:
+  topic: AI trends
+  topics: [ML, NLP, Vision]
+
+planning: true
+planning_llm: gpt-4o
+memory_config:
+  provider: chroma
+  persist: true
 ```

 ### 9. Hooks
@@ -678,6 +875,12 @@ praisonai workflow run deploy --workflow-var environment=staging --workflow-var
 # Execute with planning mode (AI creates sub-steps for each workflow step)
 praisonai workflow run "Research Blog" --planning --verbose

+# Execute with reasoning mode (chain-of-thought)
+praisonai workflow run "Analysis" --reasoning --verbose
+
+# Execute with memory enabled
+praisonai workflow run "Research" --memory
+
 # Show workflow details
 praisonai workflow show deploy

@@ -689,8 +892,23 @@ praisonai "What is AI?" --workflow "Research,Summarize" --save

 # Inline workflow with step actions
 praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
+
+# Workflow CLI help
+praisonai workflow help
 ```

+**Workflow CLI Options:**
+| Flag | Description |
+|------|-------------|
+| `--workflow-var key=value` | Set workflow variable (can be repeated) |
+| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
+| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
+| `--planning` | Enable planning mode |
+| `--reasoning` | Enable reasoning mode |
+| `--memory` | Enable memory |
+| `--verbose` | Enable verbose output |
+| `--save` | Save output to file |
+
 ### Hooks CLI:
 ```bash
 # List configured hooks
@@ -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.19" \
+    "praisonai>=2.3.25" \
    "praisonai[api]" \
    gunicorn \
    markdown
@@ -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.19" \
+    "praisonai>=2.3.25" \
    "praisonai[chat]" \
    "embedchain[github,youtube]"

@@ -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.19" \
+    "praisonai>=2.3.25" \
    "praisonai[ui]" \
    "praisonai[chat]" \
    "praisonai[realtime]" \
@@ -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.19" \
+    "praisonai>=2.3.25" \
    "praisonai[ui]" \
    "praisonai[crewai]"

@@ -1,36 +1,38 @@
 """
-Simple Workflow Example
+Simple Agentic Workflow Example

-The easiest way to create and run workflows in PraisonAI.
-Just pass functions or agents as steps - no complex configuration needed!
+Demonstrates using Agent instances directly as workflow steps.
+Each agent processes the input and passes output to the next agent.
 """

-from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents import Agent, Workflow

-# 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}")
+# 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."
+)

-def process(ctx: WorkflowContext) -> StepResult:
-    """Process the validated data."""
-    return StepResult(output=f"Processed: {ctx.previous_result}")
+writer = Agent(
+    name="Writer", 
+    role="Content Writer",
+    goal="Write engaging content based on research",
+    instructions="You are a content writer. Write clear, engaging content based on the research provided."
+)

-def format_output(ctx: WorkflowContext) -> StepResult:
-    """Format the final output."""
-    return StepResult(output=f"✅ Final: {ctx.previous_result}")
-
-# Create workflow - just list your functions!
+# Create workflow with agents as steps
 workflow = Workflow(
-    name="Simple Pipeline",
-    steps=[validate, process, format_output]
+    name="Simple Agentic Pipeline",
+    steps=[researcher, writer]
 )

 if __name__ == "__main__":
-    # Run the workflow - PraisonAI style uses .start()
-    result = workflow.start("Hello, World!", verbose=True)
+    # Run workflow - agents process sequentially
+    result = workflow.start(
+        "What are the key benefits of AI agents?",
+        verbose=True
+    )
    
-    print(f"\nFinal output: {result['output']}")
-    print(f"Steps completed: {len(result['steps'])}")
+    print(f"\nFinal output:\n{result['output']}")
@@ -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'])}")
@@ -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.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)
@@ -0,0 +1,52 @@
+"""
+Workflow Loop with List Example
+
+Demonstrates iterating over a list of items, processing each one.
+"""
+
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+from praisonaiagents.workflows import loop
+
+# Sample data
+fruits = ["apple", "banana", "cherry", "date", "elderberry"]
+
+# Processor for each item
+def process_fruit(ctx: WorkflowContext) -> StepResult:
+    """Process a single fruit from the list."""
+    fruit = ctx.variables.get("item", "unknown")
+    index = ctx.variables.get("loop_index", 0)
+    
+    # Simulate processing
+    result = f"🍎 [{index + 1}] {fruit.upper()} - processed successfully"
+    
+    return StepResult(
+        output=result,
+        variables={"last_fruit": fruit}
+    )
+
+# Aggregator to summarize results
+def summarize(ctx: WorkflowContext) -> StepResult:
+    """Summarize all processed fruits."""
+    outputs = ctx.variables.get("loop_outputs", [])
+    summary = f"📊 Processed {len(outputs)} fruits:\n"
+    summary += "\n".join(f"  {o}" for o in outputs)
+    return StepResult(output=summary)
+
+# Create workflow
+workflow = Workflow(
+    name="Fruit Processor",
+    steps=[
+        loop(process_fruit, over="fruits"),
+        summarize
+    ],
+    variables={"fruits": fruits}
+)
+
+if __name__ == "__main__":
+    print("=== Testing Loop with List ===\n")
+    
+    result = workflow.start("Process all fruits", verbose=True)
+    
+    print(f"\nFinal Output:\n{result['output']}")
+    print(f"\nLoop outputs count: {len(result['variables']['loop_outputs'])}")
+    print(f"Last processed: {result['variables'].get('last_fruit')}")
@@ -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.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'])}")
@@ -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.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}")
@@ -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.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']}")
@@ -54,7 +54,13 @@ except ImportError:
    MCP = None
 from .session import Session
 from .memory.memory import Memory
-from .memory.workflows import Workflow, WorkflowStep, WorkflowContext, StepResult
+# Workflows - import from dedicated workflows module
+from .workflows import (
+    Workflow, WorkflowStep, WorkflowContext, StepResult,
+    Route, Parallel, Loop, Repeat,
+    route, parallel, loop, repeat,
+    Pipeline  # Alias for Workflow
+)
 from .guardrails import GuardrailResult, LLMGuardrail

 # Fast Context support (lazy loaded to avoid performance impact)
@@ -48,30 +48,58 @@ def __getattr__(name):
    if name == "create_auto_memory":
        from .auto_memory import create_auto_memory
        return create_auto_memory
+    # Workflows - now in dedicated module, re-export for backward compatibility
    if name == "WorkflowManager":
-        from .workflows import WorkflowManager
+        from ..workflows import WorkflowManager
        return WorkflowManager
    if name == "Workflow":
-        from .workflows import Workflow
+        from ..workflows import Workflow
        return Workflow
    if name == "WorkflowStep":
-        from .workflows import WorkflowStep
+        from ..workflows import WorkflowStep
        return WorkflowStep
    if name == "WorkflowContext":
-        from .workflows import WorkflowContext
+        from ..workflows import WorkflowContext
        return WorkflowContext
    if name == "StepResult":
-        from .workflows import StepResult
+        from ..workflows import StepResult
        return StepResult
+    if name == "Pipeline":
+        from ..workflows import Pipeline
+        return Pipeline
+    if name == "Route":
+        from ..workflows import Route
+        return Route
+    if name == "Parallel":
+        from ..workflows import Parallel
+        return Parallel
+    if name == "Loop":
+        from ..workflows import Loop
+        return Loop
+    if name == "Repeat":
+        from ..workflows import Repeat
+        return Repeat
+    if name == "route":
+        from ..workflows import route
+        return route
+    if name == "parallel":
+        from ..workflows import parallel
+        return parallel
+    if name == "loop":
+        from ..workflows import loop
+        return loop
+    if name == "repeat":
+        from ..workflows import repeat
+        return repeat
    # Backward compatibility aliases
    if name == "StepInput":
-        from .workflows import WorkflowContext
+        from ..workflows import WorkflowContext
        return WorkflowContext
    if name == "StepOutput":
-        from .workflows import StepResult
+        from ..workflows import StepResult
        return StepResult
    if name == "create_workflow_manager":
-        from .workflows import create_workflow_manager
+        from ..workflows.workflows import create_workflow_manager
        return create_workflow_manager
    if name == "HooksManager":
        from .hooks import HooksManager
@@ -98,13 +126,23 @@ __all__ = [
    "AutoMemory",
    "AutoMemoryExtractor",
    "create_auto_memory",
-    # Workflows
+    # Workflows (re-exported from praisonaiagents.workflows for backward compatibility)
    "WorkflowManager",
    "Workflow",
+    "Pipeline",
    "WorkflowStep",
    "WorkflowContext",
    "StepResult",
    "create_workflow_manager",
+    # Workflow patterns
+    "Route",
+    "Parallel",
+    "Loop",
+    "Repeat",
+    "route",
+    "parallel",
+    "loop",
+    "repeat",
    # Hooks
    "HooksManager",
    "HookResult",
@@ -49,6 +49,137 @@ 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."""
@@ -173,9 +304,6 @@ class Workflow:
        Returns:
            Dict with 'output' (final result) and 'steps' (all step results)
        """
-        # Normalize steps - convert Agents/functions to WorkflowStep
-        normalized_steps = self._normalize_steps()
-        
        # Use default LLM if not specified
        model = llm or self.default_llm or "gpt-4o-mini"
        
@@ -185,7 +313,59 @@ class Workflow:
        results = []
        previous_output = None
        
-        for i, step in enumerate(normalized_steps):
+        # 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,
@@ -200,6 +380,7 @@ class Workflow:
                    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}")
@@ -270,6 +451,8 @@ class Workflow:
            # 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,
@@ -306,11 +489,300 @@ class Workflow:
        
        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)


+# Alias: Pipeline = Workflow (they are the same concept)
+Pipeline = Workflow
+
+
 class WorkflowManager:
    """
    Manages workflow discovery, loading, and execution.
@@ -826,7 +826,34 @@ Provide a JSON with the structure:
        logging.info("Hierarchical task execution finished")

    def workflow(self):
-        """Synchronous version of workflow method"""
+        """Synchronous version of workflow method.
+        
+        DEPRECATED: This method is deprecated. Use the Workflow class instead:
+        
+        ```python
+        from praisonaiagents import Workflow, StepResult
+        from praisonaiagents.workflows import route, parallel, loop, repeat
+        
+        workflow = Workflow(steps=[step1, step2])
+        result = workflow.start("input")
+        ```
+        
+        The Workflow class provides a simpler API with more features:
+        - route() for decision-based branching
+        - parallel() for concurrent execution
+        - loop() for iteration over lists/CSV
+        - repeat() for evaluator-optimizer patterns
+        - Callbacks (on_step_start, on_step_complete, etc.)
+        - Guardrails with validation feedback
+        - Status tracking
+        """
+        import warnings
+        warnings.warn(
+            "process='workflow' is deprecated. Use the Workflow class instead: "
+            "from praisonaiagents import Workflow; workflow = Workflow(steps=[...]); workflow.start()",
+            DeprecationWarning,
+            stacklevel=2
+        )
        current_iter = 0  # Track how many times we've looped
        # Build workflow relationships first
        for task in self.tasks.values():
@@ -0,0 +1,55 @@
+"""
+Workflows module for PraisonAI Agents.
+
+Provides workflow/pipeline patterns for orchestrating agents and functions.
+"""
+
+from .workflows import (
+    # Core classes
+    Workflow,
+    Pipeline,  # Alias for Workflow
+    WorkflowStep,
+    WorkflowContext,
+    StepResult,
+    WorkflowManager,
+    
+    # Pattern helpers
+    Route,
+    Parallel,
+    Loop,
+    Repeat,
+    
+    # Convenience functions
+    route,
+    parallel,
+    loop,
+    repeat,
+    
+    # Backward compatibility aliases
+    StepInput,
+    StepOutput,
+)
+
+__all__ = [
+    # Core
+    "Workflow",
+    "Pipeline",
+    "WorkflowStep",
+    "WorkflowContext",
+    "StepResult",
+    "WorkflowManager",
+    
+    # Patterns
+    "Route",
+    "Parallel",
+    "Loop",
+    "Repeat",
+    "route",
+    "parallel",
+    "loop",
+    "repeat",
+    
+    # Aliases
+    "StepInput",
+    "StepOutput",
+]
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "praisonaiagents"
-version = "0.0.197"
+version = "0.1.3"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 requires-python = ">=3.10"
 authors = [
@@ -2422,7 +2422,7 @@ wheels = [

 [[package]]
 name = "praisonaiagents"
-version = "0.0.197"
+version = "0.1.3"
 source = { editable = "." }
 dependencies = [
    { name = "aiohttp" },
@@ -40,7 +40,17 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🔄 Self Reflection AI Agents | [Example](examples/python/concepts/self-reflection-details.py) | [📖](https://docs.praison.ai/features/selfreflection) |
 | 🧠 Reasoning AI Agents | [Example](examples/python/concepts/reasoning-extraction.py) | [📖](https://docs.praison.ai/features/reasoning) |
 | 👁️ Multi Modal AI Agents | [Example](examples/python/general/multimodal.py) | [📖](https://docs.praison.ai/features/multimodal) |
-| 🎭 AI Agent Workflow (Context Passing, Per-Step Agents) | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
+| **🔄 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) |
+| ↳ Agentic Routing (`route()`) | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/routing) |
+| ↳ Parallel Execution (`parallel()`) | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/parallelisation) |
+| ↳ Loop over List/CSV (`loop()`) | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/repetitive) |
+| ↳ Evaluator-Optimizer (`repeat()`) | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/evaluator-optimiser) |
+| ↳ Conditional Steps | [Example](examples/python/workflows/workflow_conditional.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Branching | [Example](examples/python/workflows/workflow_branching.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Early Stop | [Example](examples/python/workflows/workflow_early_stop.py) | [📖](https://docs.praison.ai/features/workflows) |
+| ↳ Workflow Checkpoints | [Example](examples/python/workflows/workflow_checkpoints.py) | [📖](https://docs.praison.ai/features/workflows) |
 | 📚 Add Custom Knowledge | [Example](examples/python/concepts/knowledge-agents.py) | [📖](https://docs.praison.ai/features/knowledge) |
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
@@ -103,15 +113,6 @@ 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](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) |

 ## Supported Providers

@@ -456,6 +457,17 @@ result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 - **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 | `Workflow` + `loop()` |
+| Complex task routing | `Workflow` + `route()` |
+| Markdown templates | `WorkflowManager` |
+| Early stop / conditional | `Workflow` class |
+
 ### Simple Workflow (Recommended)

 The easiest way to create workflows - just pass functions as steps:
@@ -506,33 +518,218 @@ loop_step = WorkflowStep(
 )
 ```

-### Task Callbacks & Variables (process="workflow")
+### Workflow Patterns (route, parallel, loop, repeat)

 ```python
-from praisonaiagents import Agent, Task, PraisonAIAgents
+from praisonaiagents import Workflow, WorkflowContext, StepResult
+# Or use Pipeline (alias for Workflow)
+from praisonaiagents import Pipeline
+from praisonaiagents.workflows import route, parallel, loop, repeat

-# 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"}
+# 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
+    )
+])
+
+# 5. CALLBACKS - Monitor workflow execution
+workflow = Workflow(
+    steps=[step1, step2],
+    on_workflow_start=lambda w, i: print(f"Starting: {i}"),
+    on_step_complete=lambda name, r: print(f"{name}: {r.output[:50]}"),
+    on_workflow_complete=lambda w, r: print(f"Done: {r['status']}")
 )

-# Callbacks for workflow execution
-def on_start(task, task_id):
-    print(f"Starting: {task.name}")
+# 6. GUARDRAILS - Validate and retry
+def validate(result):
+    return ("error" not in result.output, "Fix the error")

-def on_complete(task, output):
-    print(f"Completed: {task.name}")
+workflow = Workflow(steps=[
+    WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
+])
+```

-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
+### Workflow with Agents
+
+Use Agent objects directly as workflow steps:
+
+```python
+from praisonaiagents import Agent, Workflow
+from praisonaiagents.workflows import route, parallel
+
+# 1. SEQUENTIAL AGENTS
+researcher = Agent(name="Researcher", role="Research expert", tools=[tavily_search])
+writer = Agent(name="Writer", role="Content writer")
+editor = Agent(name="Editor", role="Editor")
+
+workflow = Workflow(steps=[researcher, writer, editor])
+result = workflow.start("Research and write about AI")
+
+# 2. PARALLEL AGENTS
+workflow = Workflow(steps=[
+    parallel([researcher1, researcher2, researcher3]),
+    aggregator_agent
+])
+
+# 3. ROUTE TO AGENTS
+workflow = Workflow(steps=[
+    classifier_function,
+    route({
+        "technical": [tech_agent],
+        "creative": [creative_agent],
+        "default": [general_agent]
+    })
+])
+
+# 4. WITH PLANNING & REASONING
+workflow = Workflow(
+    steps=[researcher, writer, editor],
+    planning=True,           # Create execution plan
+    planning_llm="gpt-4o",   # LLM for planning
+    reasoning=True,          # Chain-of-thought reasoning
+    verbose=True
 )
+
+# 5. TOOLS PER STEP
+workflow = Workflow(steps=[
+    WorkflowStep(
+        name="research",
+        action="Research {{topic}}",
+        tools=[tavily_search, web_scraper],
+        agent_config={"name": "Researcher", "role": "Expert"}
+    )
+])
+
+# 6. OUTPUT TO FILE / IMAGES / PYDANTIC
+from pydantic import BaseModel
+
+class Report(BaseModel):
+    title: str
+    content: str
+
+workflow = Workflow(steps=[
+    WorkflowStep(name="analyze", action="Analyze image", images=["image.jpg"]),
+    WorkflowStep(name="report", action="Generate report", output_pydantic=Report),
+    WorkflowStep(name="save", action="Save results", output_file="output/report.txt")
+])
+
+# 7. ASYNC EXECUTION
+import asyncio
+
+async def main():
+    result = await workflow.astart("input")
+    print(result)
+
+asyncio.run(main())
+
+# 8. STATUS TRACKING
+workflow.status  # "not_started" | "running" | "completed"
+workflow.step_statuses  # {"step1": "completed", "step2": "skipped"}
+
+# 9. MEMORY CONFIG
+workflow = Workflow(
+    steps=[researcher, writer],
+    memory_config={"provider": "chroma", "persist": True, "collection": "my_workflow"}
+)
+result1 = workflow.start("Research AI")
+result2 = workflow.start("Continue the research")  # Remembers first run
+```
+
+### YAML Workflow Template
+
+```yaml
+# .praison/workflows/research.yaml
+name: Research Workflow
+description: Research and write content with all patterns
+
+agents:
+  researcher:
+    role: Research Expert
+    goal: Find accurate information
+    tools: [tavily_search, web_scraper]
+  writer:
+    role: Content Writer
+    goal: Write engaging content
+  editor:
+    role: Editor
+    goal: Polish content
+
+steps:
+  # Sequential
+  - agent: researcher
+    action: Research {{topic}}
+    output_variable: research_data
+
+  # Routing
+  - name: classifier
+    action: Classify content type
+    route:
+      technical: [tech_handler]
+      creative: [creative_handler]
+      default: [general_handler]
+
+  # Parallel
+  - name: parallel_research
+    parallel:
+      - agent: researcher
+        action: Research market
+      - agent: researcher
+        action: Research competitors
+
+  # Loop
+  - agent: writer
+    action: Write about {{item}}
+    loop_over: topics
+    loop_var: item
+
+  # Repeat (evaluator-optimizer)
+  - agent: editor
+    action: Review and improve
+    repeat:
+      until: "quality > 8"
+      max_iterations: 3
+
+  # Output to file
+  - agent: writer
+    action: Write final report
+    output_file: output/{{topic}}_report.md
+
+variables:
+  topic: AI trends
+  topics: [ML, NLP, Vision]
+
+planning: true
+planning_llm: gpt-4o
+memory_config:
+  provider: chroma
+  persist: true
 ```

 ### 9. Hooks
@@ -678,6 +875,12 @@ praisonai workflow run deploy --workflow-var environment=staging --workflow-var
 # Execute with planning mode (AI creates sub-steps for each workflow step)
 praisonai workflow run "Research Blog" --planning --verbose

+# Execute with reasoning mode (chain-of-thought)
+praisonai workflow run "Analysis" --reasoning --verbose
+
+# Execute with memory enabled
+praisonai workflow run "Research" --memory
+
 # Show workflow details
 praisonai workflow show deploy

@@ -689,8 +892,23 @@ praisonai "What is AI?" --workflow "Research,Summarize" --save

 # Inline workflow with step actions
 praisonai "GPT-5" --workflow "Research:Search for info,Write:Write blog" --tools tavily
+
+# Workflow CLI help
+praisonai workflow help
 ```

+**Workflow CLI Options:**
+| Flag | Description |
+|------|-------------|
+| `--workflow-var key=value` | Set workflow variable (can be repeated) |
+| `--llm <model>` | LLM model (e.g., openai/gpt-4o-mini) |
+| `--tools <tools>` | Tools (comma-separated, e.g., tavily) |
+| `--planning` | Enable planning mode |
+| `--reasoning` | Enable reasoning mode |
+| `--memory` | Enable memory |
+| `--verbose` | Enable verbose output |
+| `--save` | Save output to file |
+
 ### Hooks CLI:
 ```bash
 # List configured hooks
@@ -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.19.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.19.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.25.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.25.tar.gz | shasum -a 256`.split.first
    license "MIT"
  
    depends_on "[email protected]"
@@ -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.19 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.25 gunicorn markdown\n")
            file.write("EXPOSE 8080\n")
            file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
            
@@ -1 +1 @@
-__version__ = "2.3.19"
+__version__ = "2.3.25"
@@ -12,7 +12,7 @@ dependencies = [
    "rich>=13.7",
    "markdown>=3.5",
    "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.197",
+    "praisonaiagents>=0.1.3",
    "python-dotenv>=0.19.0",
    "instructor>=1.3.3",
    "PyYAML>=6.0",
@@ -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.197" },
+    { name = "praisonaiagents", specifier = ">=0.1.3" },
    { 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.197"
+version = "0.1.3"
 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/c6/00/342eae9c9b8bc25e812ad761ad91c0ad18adc533eaa2aa3c7c31a85cd7f8/praisonaiagents-0.0.197.tar.gz", hash = "sha256:9504dcfc861907201a10335ea7b74bfe2c6d189cc4f97951b782cd6ac46c8ee0", size = 520740, upload-time = "2025-12-16T23:26:31.078Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/6d/6d/4c350703c10a53c8efee4a3ba989916f9e802dcc142b1a7fef8b874a7a00/praisonaiagents-0.1.3.tar.gz", hash = "sha256:064abf379e5ed4535db7c32fbe4344cb7274ef23f91842de1b336b12143c418d", size = 551139, upload-time = "2025-12-17T08:14:38.557Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ee/5d/2aaa57e97d84b12dc75b66e083839eb3f9d3ef8d7721a9d8e4e7d2513ded/praisonaiagents-0.0.197-py3-none-any.whl", hash = "sha256:701b60a02d9e7147909a1311572e092846ef8247b7251b67b31dbcd581eec1d7", size = 497086, upload-time = "2025-12-16T23:26:29.208Z" },
+    { url = "https://files.pythonhosted.org/packages/15/ba/fd74ac9f0fef8e05c9665faf541898cc8d015e857fd8524501409453d592/praisonaiagents-0.1.3-py3-none-any.whl", hash = "sha256:37e7320cbd304bc901cd32abb7184ffecc9ddf5526c25204c47097f391954334", size = 518808, upload-time = "2025-12-17T08:14:36.864Z" },
 ]

 [[package]]
Author	SHA1	Message	Date
Mervin Praison	473fda653d	Release v2.3.25	2025-12-17 08:17:15 +00:00
Mervin Praison	30f1a440b6	Release v2.3.24	2025-12-17 08:04:24 +00:00
Mervin Praison	f991f37cfe	Release v2.3.23	2025-12-17 00:41:48 +00:00
Mervin Praison	a43b3ac72c	Release v2.3.22	2025-12-17 00:25:14 +00:00
Mervin Praison	06f82249e1	Release v2.3.21	2025-12-17 00:05:40 +00:00
Mervin Praison	d3aa78241a	Release v2.3.20	2025-12-16 23:46:47 +00:00