Release v2.3.21

2026-11-12 18:47:44 -10:00 · 2025-12-17 00:05:40 +00:00
24 changed files with 2783 additions and 40 deletions
@@ -112,10 +112,11 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🤖 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) |
+| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
+| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
+| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
+| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
+| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |

 ## Supported Providers

@@ -525,7 +526,9 @@ loop_step = WorkflowStep(

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

 # 1. ROUTING - Decision-based branching
 workflow = Workflow(steps=[
@@ -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.20" \
+    "praisonai>=2.3.21" \
    "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.20" \
+    "praisonai>=2.3.21" \
    "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.20" \
+    "praisonai>=2.3.21" \
    "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.20" \
+    "praisonai>=2.3.21" \
    "praisonai[ui]" \
    "praisonai[crewai]"

@@ -5,7 +5,7 @@ Demonstrates iterating over a CSV file, processing each row.
 """

 from praisonaiagents import Workflow, WorkflowContext, StepResult
-from praisonaiagents.memory.workflows import loop
+from praisonaiagents.workflows import loop
 import tempfile
 import os

@@ -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')}")
@@ -6,7 +6,7 @@ combining their results.
 """

 from praisonaiagents import Workflow, WorkflowContext, StepResult
-from praisonaiagents.memory.workflows import parallel
+from praisonaiagents.workflows import parallel
 import time

 # Parallel workers - each does independent work
@@ -6,7 +6,7 @@ This is useful for iterative improvement patterns.
 """

 from praisonaiagents import Workflow, WorkflowContext, StepResult
-from praisonaiagents.memory.workflows import repeat
+from praisonaiagents.workflows import repeat

 # Simulated content generator that improves each iteration
 class ContentGenerator:
@@ -6,7 +6,7 @@ takes different paths based on the output of a decision step.
 """

 from praisonaiagents import Workflow, WorkflowContext, StepResult
-from praisonaiagents.memory.workflows import route
+from praisonaiagents.workflows import route

 # Decision maker - determines which route to take
 def classify_request(ctx: WorkflowContext) -> StepResult:
@@ -54,10 +54,12 @@ except ImportError:
    MCP = None
 from .session import Session
 from .memory.memory import Memory
-from .memory.workflows import (
+# Workflows - import from dedicated workflows module
+from .workflows import (
    Workflow, WorkflowStep, WorkflowContext, StepResult,
    Route, Parallel, Loop, Repeat,
-    route, parallel, loop, repeat
+    route, parallel, loop, repeat,
+    Pipeline  # Alias for Workflow
 )
 from .guardrails import GuardrailResult, LLMGuardrail

@@ -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",
@@ -779,6 +779,10 @@ class Workflow:
        return self.run(input, **kwargs)


+# Alias: Pipeline = Workflow (they are the same concept)
+Pipeline = Workflow
+
+
 class WorkflowManager:
    """
    Manages workflow discovery, loading, and execution.
@@ -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.198"
+version = "0.0.199"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 requires-python = ">=3.10"
 authors = [
@@ -0,0 +1,544 @@
+"""
+Unit tests for Workflow Patterns: route, parallel, loop, repeat
+
+Test-Driven Development tests for all workflow pattern features.
+"""
+
+import pytest
+import tempfile
+import os
+from praisonaiagents import Workflow, WorkflowContext, StepResult, Pipeline
+from praisonaiagents.workflows import (
+    route, parallel, loop, repeat,
+    Route, Parallel, Loop, Repeat,
+    WorkflowStep, WorkflowManager
+)
+
+
+# =============================================================================
+# Test Fixtures
+# =============================================================================
+
+@pytest.fixture
+def simple_step():
+    """A simple step that returns its input."""
+    def step(ctx: WorkflowContext) -> StepResult:
+        return StepResult(output=f"Processed: {ctx.input}")
+    return step
+
+
+@pytest.fixture
+def csv_file():
+    """Create a temporary CSV file for testing."""
+    content = """name,email,task
+Alice,[email protected],Write docs
+Bob,[email protected],Review code
+Charlie,[email protected],Test features"""
+    
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
+        f.write(content)
+        path = f.name
+    
+    yield path
+    
+    # Cleanup
+    os.unlink(path)
+
+
+# =============================================================================
+# Test: Basic Workflow
+# =============================================================================
+
+class TestBasicWorkflow:
+    """Test basic workflow functionality."""
+    
+    def test_sequential_steps(self):
+        """Test sequential step execution."""
+        def step1(ctx): return StepResult(output="Step 1")
+        def step2(ctx): return StepResult(output="Step 2")
+        def step3(ctx): return StepResult(output="Step 3")
+        
+        workflow = Workflow(steps=[step1, step2, step3])
+        result = workflow.start("test")
+        
+        assert result["output"] == "Step 3"
+        assert len(result["steps"]) == 3
+    
+    def test_context_passing(self):
+        """Test that context is passed between steps."""
+        def step1(ctx): 
+            return StepResult(output="Hello", variables={"greeting": "Hello"})
+        
+        def step2(ctx):
+            greeting = ctx.variables.get("greeting", "")
+            prev = ctx.previous_result
+            return StepResult(output=f"{greeting} from step2, prev={prev}")
+        
+        workflow = Workflow(steps=[step1, step2])
+        result = workflow.start("test")
+        
+        assert "Hello" in result["output"]
+        assert "prev=Hello" in result["output"]
+    
+    def test_early_stop(self):
+        """Test early workflow termination."""
+        def stopper(ctx): 
+            return StepResult(output="Stopped", stop_workflow=True)
+        def never_runs(ctx): 
+            return StepResult(output="Should not run")
+        
+        workflow = Workflow(steps=[stopper, never_runs])
+        result = workflow.start("test")
+        
+        assert result["output"] == "Stopped"
+        assert len(result["steps"]) == 1
+    
+    def test_should_run_condition(self):
+        """Test conditional step execution with should_run."""
+        def always(ctx): return StepResult(output="Always")
+        def conditional(ctx): return StepResult(output="Conditional")
+        def is_special(ctx): return "special" in ctx.input.lower()
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="always", handler=always),
+            WorkflowStep(name="conditional", handler=conditional, should_run=is_special)
+        ])
+        
+        # Without "special" - conditional should be skipped
+        result = workflow.start("normal input")
+        assert len([s for s in result["steps"] if s["output"]]) == 1
+        
+        # With "special" - both should run
+        result = workflow.start("special input")
+        assert len([s for s in result["steps"] if s["output"]]) == 2
+
+
+# =============================================================================
+# Test: Route Pattern
+# =============================================================================
+
+class TestRoutePattern:
+    """Test decision-based routing."""
+    
+    def test_route_creation(self):
+        """Test Route class creation."""
+        r = route({"a": ["step_a"], "b": ["step_b"]})
+        assert isinstance(r, Route)
+        assert "a" in r.routes
+        assert "b" in r.routes
+    
+    def test_route_with_default(self):
+        """Test Route with default fallback."""
+        r = route({"a": ["step_a"]}, default=["fallback"])
+        assert r.default == ["fallback"]
+    
+    def test_route_matching(self):
+        """Test route matching based on output."""
+        def classifier(ctx): 
+            return StepResult(output="approve this request")
+        def approve(ctx): 
+            return StepResult(output="Approved!")
+        def reject(ctx): 
+            return StepResult(output="Rejected!")
+        
+        workflow = Workflow(steps=[
+            classifier,
+            route({
+                "approve": [approve],
+                "reject": [reject]
+            })
+        ])
+        
+        result = workflow.start("test")
+        assert "Approved" in result["output"]
+    
+    def test_route_default_fallback(self):
+        """Test route falls back to default when no match."""
+        def classifier(ctx): 
+            return StepResult(output="unknown category")
+        def fallback(ctx): 
+            return StepResult(output="Fallback executed")
+        
+        workflow = Workflow(steps=[
+            classifier,
+            route({
+                "approve": [lambda ctx: StepResult(output="Approved")],
+                "default": [fallback]
+            })
+        ])
+        
+        result = workflow.start("test")
+        assert "Fallback" in result["output"]
+    
+    def test_route_multi_step(self):
+        """Test route with multiple steps in a branch."""
+        def classifier(ctx): 
+            return StepResult(output="process")
+        def step1(ctx): 
+            return StepResult(output="Step1", variables={"step1": True})
+        def step2(ctx): 
+            return StepResult(output="Step2")
+        
+        workflow = Workflow(steps=[
+            classifier,
+            route({"process": [step1, step2]})
+        ])
+        
+        result = workflow.start("test")
+        assert result["output"] == "Step2"
+        assert result["variables"].get("step1") == True
+
+
+# =============================================================================
+# Test: Parallel Pattern
+# =============================================================================
+
+class TestParallelPattern:
+    """Test concurrent execution."""
+    
+    def test_parallel_creation(self):
+        """Test Parallel class creation."""
+        p = parallel(["step1", "step2", "step3"])
+        assert isinstance(p, Parallel)
+        assert len(p.steps) == 3
+    
+    def test_parallel_execution(self):
+        """Test parallel step execution."""
+        def task_a(ctx): return StepResult(output="Result A")
+        def task_b(ctx): return StepResult(output="Result B")
+        def task_c(ctx): return StepResult(output="Result C")
+        
+        workflow = Workflow(steps=[
+            parallel([task_a, task_b, task_c])
+        ])
+        
+        result = workflow.start("test")
+        
+        # All outputs should be in combined result
+        assert "Result A" in result["output"]
+        assert "Result B" in result["output"]
+        assert "Result C" in result["output"]
+        
+        # parallel_outputs variable should have all results
+        assert len(result["variables"]["parallel_outputs"]) == 3
+    
+    def test_parallel_with_aggregator(self):
+        """Test parallel followed by aggregator."""
+        def task_a(ctx): return StepResult(output="A")
+        def task_b(ctx): return StepResult(output="B")
+        
+        def aggregator(ctx):
+            outputs = ctx.variables.get("parallel_outputs", [])
+            return StepResult(output=f"Combined: {len(outputs)} results")
+        
+        workflow = Workflow(steps=[
+            parallel([task_a, task_b]),
+            aggregator
+        ])
+        
+        result = workflow.start("test")
+        assert "Combined: 2 results" in result["output"]
+
+
+# =============================================================================
+# Test: Loop Pattern
+# =============================================================================
+
+class TestLoopPattern:
+    """Test iteration over data."""
+    
+    def test_loop_creation(self):
+        """Test Loop class creation."""
+        l = loop("step", over="items")
+        assert isinstance(l, Loop)
+        assert l.over == "items"
+        assert l.var_name == "item"
+    
+    def test_loop_over_list(self):
+        """Test loop over list variable."""
+        def processor(ctx):
+            item = ctx.variables.get("item", "unknown")
+            return StepResult(output=f"Processed: {item}")
+        
+        workflow = Workflow(
+            steps=[loop(processor, over="items")],
+            variables={"items": ["apple", "banana", "cherry"]}
+        )
+        
+        result = workflow.start("test")
+        
+        assert len(result["variables"]["loop_outputs"]) == 3
+        assert "Processed: apple" in result["output"]
+        assert "Processed: cherry" in result["output"]
+    
+    def test_loop_with_index(self):
+        """Test loop provides index variable."""
+        def processor(ctx):
+            item = ctx.variables.get("item")
+            index = ctx.variables.get("loop_index")
+            return StepResult(output=f"[{index}] {item}")
+        
+        workflow = Workflow(
+            steps=[loop(processor, over="items")],
+            variables={"items": ["a", "b", "c"]}
+        )
+        
+        result = workflow.start("test")
+        
+        assert "[0] a" in result["output"]
+        assert "[1] b" in result["output"]
+        assert "[2] c" in result["output"]
+    
+    def test_loop_custom_var_name(self):
+        """Test loop with custom variable name."""
+        def processor(ctx):
+            fruit = ctx.variables.get("fruit", "unknown")
+            return StepResult(output=f"Fruit: {fruit}")
+        
+        workflow = Workflow(
+            steps=[loop(processor, over="fruits", var_name="fruit")],
+            variables={"fruits": ["apple", "banana"]}
+        )
+        
+        result = workflow.start("test")
+        assert "Fruit: apple" in result["output"]
+    
+    def test_loop_from_csv(self, csv_file):
+        """Test loop over CSV file."""
+        def processor(ctx):
+            row = ctx.variables.get("item", {})
+            name = row.get("name", "unknown")
+            return StepResult(output=f"Name: {name}")
+        
+        workflow = Workflow(steps=[
+            loop(processor, from_csv=csv_file)
+        ])
+        
+        result = workflow.start("test")
+        
+        assert len(result["variables"]["loop_outputs"]) == 3
+        assert "Name: Alice" in result["output"]
+        assert "Name: Bob" in result["output"]
+        assert "Name: Charlie" in result["output"]
+    
+    def test_loop_empty_list(self):
+        """Test loop with empty list."""
+        def processor(ctx):
+            return StepResult(output="Should not run")
+        
+        workflow = Workflow(
+            steps=[loop(processor, over="items")],
+            variables={"items": []}
+        )
+        
+        result = workflow.start("test")
+        assert len(result["variables"]["loop_outputs"]) == 0
+
+
+# =============================================================================
+# Test: Repeat Pattern
+# =============================================================================
+
+class TestRepeatPattern:
+    """Test evaluator-optimizer pattern."""
+    
+    def test_repeat_creation(self):
+        """Test Repeat class creation."""
+        r = repeat("step", until=lambda ctx: True, max_iterations=5)
+        assert isinstance(r, Repeat)
+        assert r.max_iterations == 5
+    
+    def test_repeat_until_condition(self):
+        """Test repeat stops when condition is met."""
+        counter = [0]
+        
+        def generator(ctx):
+            counter[0] += 1
+            return StepResult(
+                output=f"Count: {counter[0]}",
+                variables={"count": counter[0]}
+            )
+        
+        def is_done(ctx):
+            return ctx.variables.get("count", 0) >= 3
+        
+        workflow = Workflow(steps=[
+            repeat(generator, until=is_done, max_iterations=10)
+        ])
+        
+        result = workflow.start("test")
+        
+        assert result["variables"]["repeat_iterations"] == 3
+        assert counter[0] == 3
+    
+    def test_repeat_max_iterations(self):
+        """Test repeat respects max_iterations."""
+        counter = [0]
+        
+        def generator(ctx):
+            counter[0] += 1
+            return StepResult(output=f"Count: {counter[0]}")
+        
+        # Never-true condition
+        def never_done(ctx):
+            return False
+        
+        workflow = Workflow(steps=[
+            repeat(generator, until=never_done, max_iterations=5)
+        ])
+        
+        result = workflow.start("test")
+        
+        assert result["variables"]["repeat_iterations"] == 5
+        assert counter[0] == 5
+    
+    def test_repeat_without_condition(self):
+        """Test repeat without until condition runs max_iterations times."""
+        counter = [0]
+        
+        def generator(ctx):
+            counter[0] += 1
+            return StepResult(output=f"Count: {counter[0]}")
+        
+        workflow = Workflow(steps=[
+            repeat(generator, max_iterations=3)
+        ])
+        
+        result = workflow.start("test")
+        
+        assert counter[0] == 3
+    
+    def test_repeat_with_early_stop(self):
+        """Test repeat stops on stop_workflow=True."""
+        counter = [0]
+        
+        def generator(ctx):
+            counter[0] += 1
+            if counter[0] >= 2:
+                return StepResult(output="Stopping", stop_workflow=True)
+            return StepResult(output=f"Count: {counter[0]}")
+        
+        workflow = Workflow(steps=[
+            repeat(generator, max_iterations=10)
+        ])
+        
+        result = workflow.start("test")
+        
+        assert counter[0] == 2
+
+
+# =============================================================================
+# Test: Pipeline Alias
+# =============================================================================
+
+class TestPipelineAlias:
+    """Test Pipeline is an alias for Workflow."""
+    
+    def test_pipeline_is_workflow(self):
+        """Test Pipeline is the same class as Workflow."""
+        assert Pipeline is Workflow
+    
+    def test_pipeline_works(self):
+        """Test Pipeline can be used like Workflow."""
+        def step1(ctx): return StepResult(output="Pipeline works!")
+        
+        pipeline = Pipeline(steps=[step1])
+        result = pipeline.start("test")
+        
+        assert result["output"] == "Pipeline works!"
+
+
+# =============================================================================
+# Test: Backward Compatibility
+# =============================================================================
+
+class TestBackwardCompatibility:
+    """Test backward compatibility with old import paths."""
+    
+    def test_import_from_memory_workflows(self):
+        """Test imports from praisonaiagents.memory.workflows still work."""
+        from praisonaiagents.memory.workflows import (
+            Workflow, WorkflowStep, WorkflowContext, StepResult,
+            route, parallel, loop, repeat
+        )
+        
+        assert Workflow is not None
+        assert route is not None
+    
+    def test_import_from_memory_module(self):
+        """Test imports from praisonaiagents.memory still work."""
+        from praisonaiagents.memory import Workflow, Pipeline, route, parallel
+        
+        assert Workflow is not None
+        assert Pipeline is Workflow
+    
+    def test_step_input_output_aliases(self):
+        """Test StepInput/StepOutput backward compatibility aliases."""
+        from praisonaiagents.workflows import StepInput, StepOutput
+        
+        # These should be aliases for WorkflowContext and StepResult
+        assert StepInput is WorkflowContext
+        assert StepOutput is StepResult
+
+
+# =============================================================================
+# Test: Combined Patterns
+# =============================================================================
+
+class TestCombinedPatterns:
+    """Test combining multiple patterns."""
+    
+    def test_parallel_then_route(self):
+        """Test parallel followed by routing."""
+        def task_a(ctx): return StepResult(output="success from A")
+        def task_b(ctx): return StepResult(output="success from B")
+        
+        def classifier(ctx):
+            outputs = ctx.variables.get("parallel_outputs", [])
+            if all("success" in o for o in outputs):
+                return StepResult(output="all success")
+            return StepResult(output="some failed")
+        
+        def success_handler(ctx): return StepResult(output="All tasks succeeded!")
+        def failure_handler(ctx): return StepResult(output="Some tasks failed!")
+        
+        workflow = Workflow(steps=[
+            parallel([task_a, task_b]),
+            classifier,
+            route({
+                "success": [success_handler],
+                "failed": [failure_handler]
+            })
+        ])
+        
+        result = workflow.start("test")
+        assert "succeeded" in result["output"]
+    
+    def test_loop_with_aggregation(self):
+        """Test loop followed by aggregation step."""
+        def processor(ctx):
+            item = ctx.variables.get("item")
+            return StepResult(output=f"Processed: {item}")
+        
+        def aggregator(ctx):
+            outputs = ctx.variables.get("loop_outputs", [])
+            return StepResult(output=f"Total: {len(outputs)} items processed")
+        
+        workflow = Workflow(
+            steps=[
+                loop(processor, over="items"),
+                aggregator
+            ],
+            variables={"items": ["a", "b", "c", "d"]}
+        )
+        
+        result = workflow.start("test")
+        assert "Total: 4 items processed" in result["output"]
+
+
+# =============================================================================
+# Run Tests
+# =============================================================================
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])
@@ -2422,7 +2422,7 @@ wheels = [

 [[package]]
 name = "praisonaiagents"
-version = "0.0.198"
+version = "0.0.199"
 source = { editable = "." }
 dependencies = [
    { name = "aiohttp" },
@@ -112,10 +112,11 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🤖 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) |
+| 🔀 Agentic Routing | [Example](examples/python/workflows/workflow_routing.py) | [📖](https://docs.praison.ai/features/workflow-routing) |
+| ⚡ Parallel Execution | [Example](examples/python/workflows/workflow_parallel.py) | [📖](https://docs.praison.ai/features/workflow-parallel) |
+| 🔁 Loop over List | [Example](examples/python/workflows/workflow_loop_list.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
+| 📄 Loop over CSV | [Example](examples/python/workflows/workflow_loop_csv.py) | [📖](https://docs.praison.ai/features/workflow-loop) |
+| 🔄 Evaluator-Optimizer | [Example](examples/python/workflows/workflow_repeat.py) | [📖](https://docs.praison.ai/features/workflow-repeat) |

 ## Supported Providers

@@ -525,7 +526,9 @@ loop_step = WorkflowStep(

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

 # 1. ROUTING - Decision-based branching
 workflow = Workflow(steps=[
@@ -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.20.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.20.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.21.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.21.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.20 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.21 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.20"
+__version__ = "2.3.21"
@@ -12,7 +12,7 @@ dependencies = [
    "rich>=13.7",
    "markdown>=3.5",
    "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.198",
+    "praisonaiagents>=0.0.199",
    "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.198" },
+    { name = "praisonaiagents", specifier = ">=0.0.199" },
    { 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.198"
+version = "0.0.199"
 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/68/bb/a3daf58206327a4cc92abed7295b41c03f9c4a8f3411d9435c633eb45b7f/praisonaiagents-0.0.198.tar.gz", hash = "sha256:d02180d476dd3ea260ade633764f6f8831dddc550b17ffb21eed79809166d315", size = 523691, upload-time = "2025-12-16T23:46:10.247Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/8b/5f/fffbe2473e579edc0bc00022179a896e847128056b6b51905fc497b3cb7d/praisonaiagents-0.0.199.tar.gz", hash = "sha256:276a224350d1a975bdb69f5f07898a60ac9fce0b2d45edbab14e8b755a66853c", size = 538316, upload-time = "2025-12-17T00:01:47.411Z" }
 wheels = [
-    { 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" },
+    { url = "https://files.pythonhosted.org/packages/c9/25/489052a4ed1c8a65a08ae48fabe94b1a399cd82ed2e9c47264a1f82d066c/praisonaiagents-0.0.199-py3-none-any.whl", hash = "sha256:5bc887e097d2e444d63b755d356c537ed30e5f519c2e21adc819160ff4ed5697", size = 515315, upload-time = "2025-12-17T00:01:45.602Z" },
 ]

 [[package]]