Release v2.3.22

README.md

@@ -562,35 +562,22 @@ workflow = Workflow(steps=[
         max_iterations=5
     )
 ])
-```
 
-### Task Callbacks & Variables (process="workflow")
-
-```python
-from praisonaiagents import Agent, Task, PraisonAIAgents
-
-# Per-task agent config (no need to create agent separately)
-task = Task(
-    description="Research {{topic}}",
-    agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
-    variables={"topic": "AI trends"}
+# 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}")
-
-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 = Workflow(steps=[
+    WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
+])
 ```
 
 ### 9. Hooks

docker/Dockerfile

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

docker/Dockerfile.chat

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

docker/Dockerfile.dev

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

docker/Dockerfile.ui

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

src/praisonai-agents/praisonaiagents/process/process.py

@@ -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():

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

@@ -23,7 +23,7 @@ import os
 import re
 import logging
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Callable, Literal
+from typing import Any, Dict, List, Optional, Callable, Literal, Tuple
 from dataclasses import dataclass, field
 
 logger = logging.getLogger(__name__)
@@ -190,7 +190,7 @@ class WorkflowStep:
     should_run: Optional[Callable[['WorkflowContext'], bool]] = None  # Function to check if step should run
     handler: Optional[Callable[['WorkflowContext'], 'StepResult']] = None  # Custom function instead of agent
     on_error: Literal["stop", "continue", "retry"] = "stop"
-    max_retries: int = 1
+    max_retries: int = 3
     
     # Context passing fields
     context_from: Optional[List[str]] = None  # Which previous steps to include context from
@@ -210,6 +210,13 @@ class WorkflowStep:
     loop_over: Optional[str] = None  # Variable name to iterate over (e.g., "items")
     loop_var: str = "item"  # Variable name for current item in loop
     
+    # Guardrail (like process="workflow")
+    guardrail: Optional[Callable[['StepResult'], Tuple[bool, Any]]] = None  # Returns (is_valid, feedback)
+    
+    # Status tracking
+    status: str = "pending"  # pending, running, completed, failed, skipped
+    retry_count: int = 0
+    
     # Backward compatibility aliases
     @property
     def evaluator(self):
@@ -258,6 +265,17 @@ class Workflow:
     planning: bool = False  # Enable planning mode
     planning_llm: Optional[str] = None  # LLM for planning
     
+    # Callbacks (like process="workflow")
+    on_workflow_start: Optional[Callable[['Workflow', str], None]] = None  # (workflow, input)
+    on_workflow_complete: Optional[Callable[['Workflow', Dict[str, Any]], None]] = None  # (workflow, result)
+    on_step_start: Optional[Callable[[str, 'WorkflowContext'], None]] = None  # (step_name, context)
+    on_step_complete: Optional[Callable[[str, 'StepResult'], None]] = None  # (step_name, result)
+    on_step_error: Optional[Callable[[str, Exception], None]] = None  # (step_name, error)
+    
+    # Status tracking
+    status: str = "not_started"  # not_started, running, completed, failed
+    step_statuses: Dict[str, str] = field(default_factory=dict)  # {step_name: status}
+    
     def to_dict(self) -> Dict[str, Any]:
         return {
             "name": self.name,
@@ -313,6 +331,16 @@ class Workflow:
         results = []
         previous_output = None
         
+        # Update workflow status
+        self.status = "running"
+        
+        # Call on_workflow_start callback
+        if self.on_workflow_start:
+            try:
+                self.on_workflow_start(self, input)
+            except Exception as e:
+                logger.error(f"on_workflow_start callback failed: {e}")
+        
         # Process steps (may include Route, Parallel, Loop, Repeat)
         i = 0
         while i < len(self.steps):
@@ -374,68 +402,136 @@ class Workflow:
                 variables=all_variables.copy()
             )
             
+            # Update step status
+            if hasattr(step, 'status'):
+                step.status = "running"
+            self.step_statuses[step.name] = "running"
+            
+            # Call on_step_start callback
+            if self.on_step_start:
+                try:
+                    self.on_step_start(step.name, context)
+                except Exception as e:
+                    logger.error(f"on_step_start callback failed: {e}")
+            
             # Check should_run condition
             if step.should_run:
                 try:
                     if not step.should_run(context):
                         if verbose:
                             print(f"⏭️ Skipped: {step.name}")
+                        if hasattr(step, 'status'):
+                            step.status = "skipped"
+                        self.step_statuses[step.name] = "skipped"
                         i += 1
                         continue
                 except Exception as e:
                     logger.error(f"should_run failed for {step.name}: {e}")
             
-            # Execute step
+            # Execute step with retry and guardrail support
             output = None
             stop = False
+            step_error = None
+            max_retries = getattr(step, 'max_retries', 3)
+            retry_count = 0
+            validation_feedback = None
             
-            if step.handler:
-                # Custom handler function
+            while retry_count <= max_retries:
+                step_error = None
+                
+                # Add validation feedback to context if retrying
+                if validation_feedback:
+                    context.variables["validation_feedback"] = validation_feedback
+                
                 try:
-                    result = step.handler(context)
-                    if isinstance(result, StepResult):
-                        output = result.output
-                        stop = result.stop_workflow
-                        if result.variables:
-                            all_variables.update(result.variables)
-                    else:
-                        output = str(result)
+                    if step.handler:
+                        # Custom handler function
+                        result = step.handler(context)
+                        if isinstance(result, StepResult):
+                            output = result.output
+                            stop = result.stop_workflow
+                            if result.variables:
+                                all_variables.update(result.variables)
+                        else:
+                            output = str(result)
+                            
+                    elif step.agent:
+                        # Direct agent
+                        output = step.agent.chat(step.action or input)
+                            
+                    elif step.action:
+                        # Action with agent_config - create temporary agent
+                        from ..agent.agent import Agent
+                        config = step.agent_config or self.default_agent_config or {}
+                        temp_agent = Agent(
+                            name=config.get("name", step.name),
+                            role=config.get("role", "Assistant"),
+                            goal=config.get("goal", "Complete the task"),
+                            llm=config.get("llm", model),
+                            verbose=verbose
+                        )
+                        # Substitute variables in action
+                        action = step.action
+                        for key, value in all_variables.items():
+                            action = action.replace(f"{{{{{key}}}}}", str(value))
+                        if previous_output:
+                            action = action.replace("{{previous_output}}", str(previous_output))
+                        if validation_feedback:
+                            action = f"{action}\n\nPrevious attempt feedback: {validation_feedback}"
+                        output = temp_agent.chat(action)
+                        
                 except Exception as e:
+                    step_error = e
                     output = f"Error: {e}"
-                    
-            elif step.agent:
-                # Direct agent
+                    if self.on_step_error:
+                        try:
+                            self.on_step_error(step.name, e)
+                        except Exception:
+                            pass
+                
+                # Check guardrail if present
+                guardrail = getattr(step, 'guardrail', None)
+                if guardrail and output and not step_error:
+                    try:
+                        is_valid, feedback = guardrail(StepResult(output=output))
+                        if not is_valid:
+                            validation_feedback = str(feedback)
+                            retry_count += 1
+                            if verbose:
+                                print(f"⚠️ {step.name} failed validation (attempt {retry_count}/{max_retries}): {feedback}")
+                            continue  # Retry
+                    except Exception as e:
+                        logger.error(f"Guardrail failed for {step.name}: {e}")
+                
+                # Success - break out of retry loop
+                break
+            
+            # Update step status
+            if step_error:
+                if hasattr(step, 'status'):
+                    step.status = "failed"
+                self.step_statuses[step.name] = "failed"
+            else:
+                if hasattr(step, 'status'):
+                    step.status = "completed"
+                self.step_statuses[step.name] = "completed"
+            
+            # Create step result for callback
+            step_result = StepResult(output=output or "", stop_workflow=stop)
+            
+            # Call on_step_complete callback
+            if self.on_step_complete:
                 try:
-                    output = step.agent.chat(step.action or input)
+                    self.on_step_complete(step.name, step_result)
                 except Exception as e:
-                    output = f"Error: {e}"
-                    
-            elif step.action:
-                # Action with agent_config - create temporary agent
-                try:
-                    from ..agent.agent import Agent
-                    config = step.agent_config or self.default_agent_config or {}
-                    temp_agent = Agent(
-                        name=config.get("name", step.name),
-                        role=config.get("role", "Assistant"),
-                        goal=config.get("goal", "Complete the task"),
-                        llm=config.get("llm", model),
-                        verbose=verbose
-                    )
-                    # Substitute variables in action
-                    action = step.action
-                    for key, value in all_variables.items():
-                        action = action.replace(f"{{{{{key}}}}}", str(value))
-                    if previous_output:
-                        action = action.replace("{{previous_output}}", str(previous_output))
-                    output = temp_agent.chat(action)
-                except Exception as e:
-                    output = f"Error: {e}"
+                    logger.error(f"on_step_complete callback failed: {e}")
             
             # Store result
             results.append({
                 "step": step.name,
-                "output": output
+                "output": output,
+                "status": self.step_statuses.get(step.name, "completed"),
+                "retries": retry_count
             })
             previous_output = output
             
@@ -454,11 +550,24 @@ class Workflow:
             
             i += 1
         
-        return {
+        # Update workflow status
+        self.status = "completed"
+        
+        final_result = {
             "output": previous_output,
             "steps": results,
-            "variables": all_variables
+            "variables": all_variables,
+            "status": self.status
         }
+        
+        # Call on_workflow_complete callback
+        if self.on_workflow_complete:
+            try:
+                self.on_workflow_complete(self, final_result)
+            except Exception as e:
+                logger.error(f"on_workflow_complete callback failed: {e}")
+        
+        return final_result
     
     def _normalize_steps(self) -> List['WorkflowStep']:
         """Convert mixed steps (Agent, function, WorkflowStep) to WorkflowStep list."""

src/praisonai-agents/pyproject.toml

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

src/praisonai-agents/tests/test_workflow_patterns.py

@@ -536,6 +536,487 @@ class TestCombinedPatterns:
         assert "Total: 4 items processed" in result["output"]
 
 
+# =============================================================================
+# Test: Callbacks
+# =============================================================================
+
+class TestCallbacks:
+    """Test workflow and step callbacks."""
+    
+    def test_on_workflow_start(self):
+        """Test on_workflow_start callback is called."""
+        called = []
+        
+        def on_start(workflow, input_text):
+            called.append(("start", input_text))
+        
+        def step1(ctx): return StepResult(output="Done")
+        
+        workflow = Workflow(
+            steps=[step1],
+            on_workflow_start=on_start
+        )
+        workflow.start("test input")
+        
+        assert len(called) == 1
+        assert called[0] == ("start", "test input")
+    
+    def test_on_workflow_complete(self):
+        """Test on_workflow_complete callback is called."""
+        called = []
+        
+        def on_complete(workflow, result):
+            called.append(("complete", result["output"]))
+        
+        def step1(ctx): return StepResult(output="Final output")
+        
+        workflow = Workflow(
+            steps=[step1],
+            on_workflow_complete=on_complete
+        )
+        workflow.start("test")
+        
+        assert len(called) == 1
+        assert called[0] == ("complete", "Final output")
+    
+    def test_on_step_start(self):
+        """Test on_step_start callback is called for each step."""
+        called = []
+        
+        def on_step_start(step_name, context):
+            called.append(("step_start", step_name))
+        
+        def step1(ctx): return StepResult(output="Step 1")
+        def step2(ctx): return StepResult(output="Step 2")
+        
+        workflow = Workflow(
+            steps=[step1, step2],
+            on_step_start=on_step_start
+        )
+        workflow.start("test")
+        
+        assert len(called) == 2
+        assert called[0][1] == "step1"
+        assert called[1][1] == "step2"
+    
+    def test_on_step_complete(self):
+        """Test on_step_complete callback is called for each step."""
+        called = []
+        
+        def on_step_complete(step_name, result):
+            called.append(("step_complete", step_name, result.output))
+        
+        def step1(ctx): return StepResult(output="Output 1")
+        def step2(ctx): return StepResult(output="Output 2")
+        
+        workflow = Workflow(
+            steps=[step1, step2],
+            on_step_complete=on_step_complete
+        )
+        workflow.start("test")
+        
+        assert len(called) == 2
+        assert called[0] == ("step_complete", "step1", "Output 1")
+        assert called[1] == ("step_complete", "step2", "Output 2")
+    
+    def test_on_step_error(self):
+        """Test on_step_error callback is called on error."""
+        called = []
+        
+        def on_error(step_name, error):
+            called.append(("error", step_name, str(error)))
+        
+        def failing_step(ctx):
+            raise ValueError("Test error")
+        
+        workflow = Workflow(
+            steps=[failing_step],
+            on_step_error=on_error
+        )
+        workflow.start("test")
+        
+        assert len(called) == 1
+        assert called[0][0] == "error"
+        assert "Test error" in called[0][2]
+
+
+# =============================================================================
+# Test: Guardrails
+# =============================================================================
+
+class TestGuardrails:
+    """Test step guardrails and validation."""
+    
+    def test_guardrail_passes(self):
+        """Test guardrail that passes."""
+        def validator(result):
+            return (True, None)  # Valid
+        
+        def step1(ctx): return StepResult(output="Valid output")
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="step1", handler=step1, guardrail=validator)
+        ])
+        result = workflow.start("test")
+        
+        assert result["output"] == "Valid output"
+        assert result["steps"][0]["retries"] == 0
+    
+    def test_guardrail_fails_and_retries(self):
+        """Test guardrail that fails triggers retry."""
+        attempt = [0]
+        
+        def validator(result):
+            if "attempt 3" in result.output:
+                return (True, None)  # Valid on 3rd attempt
+            return (False, "Not valid yet")
+        
+        def step1(ctx):
+            attempt[0] += 1
+            return StepResult(output=f"Output attempt {attempt[0]}")
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="step1", handler=step1, guardrail=validator, max_retries=5)
+        ])
+        result = workflow.start("test")
+        
+        assert "attempt 3" in result["output"]
+        assert attempt[0] == 3
+    
+    def test_guardrail_max_retries_exceeded(self):
+        """Test guardrail respects max_retries."""
+        attempt = [0]
+        
+        def validator(result):
+            return (False, "Always fails")  # Never valid
+        
+        def step1(ctx):
+            attempt[0] += 1
+            return StepResult(output=f"Attempt {attempt[0]}")
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="step1", handler=step1, guardrail=validator, max_retries=2)
+        ])
+        result = workflow.start("test")
+        
+        # Should have tried 3 times (initial + 2 retries)
+        assert attempt[0] == 3
+    
+    def test_validation_feedback_in_context(self):
+        """Test validation feedback is passed to context on retry."""
+        feedbacks = []
+        
+        def validator(result):
+            if "fixed" in result.output:
+                return (True, None)
+            return (False, "Please fix the issue")
+        
+        def step1(ctx):
+            feedback = ctx.variables.get("validation_feedback")
+            if feedback:
+                feedbacks.append(feedback)
+                return StepResult(output="fixed")
+            return StepResult(output="initial")
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="step1", handler=step1, guardrail=validator, max_retries=3)
+        ])
+        result = workflow.start("test")
+        
+        assert result["output"] == "fixed"
+        assert len(feedbacks) == 1
+        assert "Please fix" in feedbacks[0]
+
+
+# =============================================================================
+# Test: Status Tracking
+# =============================================================================
+
+class TestStatusTracking:
+    """Test workflow and step status tracking."""
+    
+    def test_workflow_status(self):
+        """Test workflow status is tracked."""
+        def step1(ctx): return StepResult(output="Done")
+        
+        workflow = Workflow(steps=[step1])
+        assert workflow.status == "not_started"
+        
+        workflow.start("test")
+        assert workflow.status == "completed"
+    
+    def test_step_statuses(self):
+        """Test step statuses are tracked."""
+        def step1(ctx): return StepResult(output="Step 1")
+        def step2(ctx): return StepResult(output="Step 2")
+        
+        workflow = Workflow(steps=[step1, step2])
+        workflow.start("test")
+        
+        assert workflow.step_statuses["step1"] == "completed"
+        assert workflow.step_statuses["step2"] == "completed"
+    
+    def test_skipped_step_status(self):
+        """Test skipped step has correct status."""
+        def step1(ctx): return StepResult(output="Step 1")
+        def step2(ctx): return StepResult(output="Step 2")
+        def never_run(ctx): return False
+        
+        workflow = Workflow(steps=[
+            step1,
+            WorkflowStep(name="step2", handler=step2, should_run=never_run)
+        ])
+        result = workflow.start("test")
+        
+        assert workflow.step_statuses["step1"] == "completed"
+        assert workflow.step_statuses["step2"] == "skipped"
+    
+    def test_result_includes_status(self):
+        """Test result includes status information."""
+        def step1(ctx): return StepResult(output="Done")
+        
+        workflow = Workflow(steps=[step1])
+        result = workflow.start("test")
+        
+        assert result["status"] == "completed"
+        assert result["steps"][0]["status"] == "completed"
+
+
+# =============================================================================
+# Test: Documentation Examples (Integration Tests)
+# =============================================================================
+
+class TestDocumentationExamples:
+    """Test that documentation examples work correctly."""
+    
+    def test_routing_doc_example(self):
+        """Test the routing example from docs/features/routing.mdx"""
+        # Classifier - determines which route to take
+        def classify_request(ctx: WorkflowContext) -> StepResult:
+            input_lower = ctx.input.lower()
+            if "urgent" in input_lower:
+                return StepResult(output="priority: high")
+            elif "question" in input_lower:
+                return StepResult(output="priority: support")
+            else:
+                return StepResult(output="priority: normal")
+
+        def handle_high_priority(ctx): return StepResult(output="HIGH PRIORITY")
+        def handle_support(ctx): return StepResult(output="SUPPORT")
+        def handle_normal(ctx): return StepResult(output="NORMAL")
+
+        workflow = Workflow(
+            steps=[
+                classify_request,
+                route({
+                    "high": [handle_high_priority],
+                    "support": [handle_support],
+                    "default": [handle_normal]
+                })
+            ]
+        )
+
+        # Test urgent
+        result = workflow.start("This is an urgent request!")
+        assert "HIGH PRIORITY" in result["output"]
+
+        # Test question
+        result = workflow.start("I have a question about billing")
+        assert "SUPPORT" in result["output"]
+
+        # Test normal
+        result = workflow.start("Please process my order")
+        assert "NORMAL" in result["output"]
+    
+    def test_parallel_doc_example(self):
+        """Test the parallel example from docs/features/parallelisation.mdx"""
+        def research_market(ctx): return StepResult(output="Market: Growth 15%")
+        def research_competitors(ctx): return StepResult(output="Competitors: 3 found")
+        def research_customers(ctx): return StepResult(output="Customers: 5 segments")
+
+        def summarize_research(ctx: WorkflowContext) -> StepResult:
+            outputs = ctx.variables.get("parallel_outputs", [])
+            return StepResult(output=f"Summary: {len(outputs)} items")
+
+        workflow = Workflow(
+            steps=[
+                parallel([research_market, research_competitors, research_customers]),
+                summarize_research
+            ]
+        )
+
+        result = workflow.start("Research the AI market")
+        assert "Summary: 3 items" in result["output"]
+    
+    def test_repeat_doc_example(self):
+        """Test the repeat example from docs/features/evaluator-optimiser.mdx"""
+        class ContentGenerator:
+            def __init__(self):
+                self.points = []
+            
+            def generate(self, ctx: WorkflowContext) -> StepResult:
+                self.points.append(f"Point {len(self.points) + 1}")
+                self.points.append(f"Point {len(self.points) + 1}")
+                return StepResult(
+                    output=f"Generated {len(self.points)} points",
+                    variables={"count": len(self.points)}
+                )
+
+        def is_complete(ctx: WorkflowContext) -> bool:
+            return ctx.variables.get("count", 0) >= 10
+
+        generator = ContentGenerator()
+
+        workflow = Workflow(
+            steps=[
+                repeat(generator.generate, until=is_complete, max_iterations=10)
+            ]
+        )
+
+        result = workflow.start("Generate 10 points")
+        assert result["variables"].get("count", 0) >= 10
+    
+    def test_loop_doc_example(self):
+        """Test the loop example from docs/features/repetitive.mdx"""
+        def process_item(ctx: WorkflowContext) -> StepResult:
+            item = ctx.variables.get("item", "unknown")
+            index = ctx.variables.get("loop_index", 0)
+            return StepResult(output=f"[{index + 1}] Processed: {item}")
+
+        workflow = Workflow(
+            steps=[loop(process_item, over="tasks")],
+            variables={"tasks": ["Task A", "Task B", "Task C", "Task D"]}
+        )
+
+        result = workflow.start("Process all tasks")
+        outputs = result["variables"].get("loop_outputs", [])
+        assert len(outputs) == 4
+        assert "[1] Processed: Task A" in outputs[0]
+        assert "[4] Processed: Task D" in outputs[3]
+    
+    def test_promptchaining_doc_example(self):
+        """Test the prompt chaining example from docs/features/promptchaining.mdx"""
+        def validate_input(ctx: WorkflowContext) -> StepResult:
+            if len(ctx.input) > 5:
+                return StepResult(output="valid", variables={"validated": True})
+            return StepResult(output="invalid", stop_workflow=True)
+
+        def process_data(ctx: WorkflowContext) -> StepResult:
+            return StepResult(output=f"Processed: {ctx.input.upper()}")
+
+        def analyze_results(ctx: WorkflowContext) -> StepResult:
+            return StepResult(output=f"Analysis: {ctx.previous_result}")
+
+        def generate_output(ctx: WorkflowContext) -> StepResult:
+            return StepResult(output=f"Final: {ctx.previous_result}")
+
+        workflow = Workflow(
+            steps=[validate_input, process_data, analyze_results, generate_output]
+        )
+
+        # Valid input
+        result = workflow.start("Hello World")
+        assert "Final:" in result["output"]
+        assert "HELLO WORLD" in result["output"]
+
+        # Invalid input (too short) - should stop early
+        result = workflow.start("Hi")
+        assert result["output"] == "invalid"
+    
+    def test_autonomous_workflow_doc_example(self):
+        """Test the autonomous workflow example from docs/features/autonomous-workflow.mdx"""
+        class EnvironmentMonitor:
+            def __init__(self):
+                self.iteration = 0
+            
+            def check_state(self, ctx: WorkflowContext) -> StepResult:
+                self.iteration += 1
+                states = ["normal", "critical", "optimal"]
+                state = states[self.iteration % 3]
+                return StepResult(
+                    output=f"state: {state}",
+                    variables={"state": state, "iteration": self.iteration}
+                )
+
+        def handle_normal(ctx): return StepResult(output="Maintaining")
+        def handle_critical(ctx): return StepResult(output="Fixing")
+        def handle_optimal(ctx): return StepResult(output="Enhancing", stop_workflow=True)
+
+        monitor = EnvironmentMonitor()
+
+        workflow = Workflow(
+            steps=[
+                repeat(
+                    monitor.check_state,
+                    until=lambda ctx: ctx.variables.get("state") == "optimal",
+                    max_iterations=5
+                ),
+                route({
+                    "normal": [handle_normal],
+                    "critical": [handle_critical],
+                    "optimal": [handle_optimal]
+                })
+            ]
+        )
+
+        result = workflow.start("Monitor environment")
+        # Should have run until optimal state
+        assert result["variables"].get("iteration", 0) >= 1
+    
+    def test_workflow_with_all_callbacks(self):
+        """Test workflow with all callback types."""
+        events = []
+        
+        def on_start(w, i): events.append(f"start:{i}")
+        def on_complete(w, r): events.append(f"complete:{r['status']}")
+        def on_step_start(n, c): events.append(f"step_start:{n}")
+        def on_step_complete(n, r): events.append(f"step_complete:{n}")
+        
+        def step1(ctx): return StepResult(output="Step 1 done")
+        def step2(ctx): return StepResult(output="Step 2 done")
+        
+        workflow = Workflow(
+            steps=[step1, step2],
+            on_workflow_start=on_start,
+            on_workflow_complete=on_complete,
+            on_step_start=on_step_start,
+            on_step_complete=on_step_complete
+        )
+        
+        result = workflow.start("test input")
+        
+        assert "start:test input" in events
+        assert "complete:completed" in events
+        assert "step_start:step1" in events
+        assert "step_complete:step1" in events
+        assert "step_start:step2" in events
+        assert "step_complete:step2" in events
+    
+    def test_workflow_with_guardrail_retry(self):
+        """Test workflow with guardrail that triggers retry."""
+        attempts = [0]
+        
+        def generator(ctx: WorkflowContext) -> StepResult:
+            attempts[0] += 1
+            feedback = ctx.variables.get("validation_feedback", "")
+            if "fix" in feedback.lower():
+                return StepResult(output="fixed output")
+            return StepResult(output="bad output")
+        
+        def validator(result):
+            if "fixed" in result.output:
+                return (True, None)
+            return (False, "Please fix the output")
+        
+        workflow = Workflow(steps=[
+            WorkflowStep(name="gen", handler=generator, guardrail=validator, max_retries=3)
+        ])
+        
+        result = workflow.start("test")
+        assert "fixed" in result["output"]
+        assert attempts[0] == 2  # First attempt + 1 retry
+
+
 # =============================================================================
 # Run Tests
 # =============================================================================

src/praisonai-agents/uv.lock

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

src/praisonai/README.md

@@ -562,35 +562,22 @@ workflow = Workflow(steps=[
         max_iterations=5
     )
 ])
-```
 
-### Task Callbacks & Variables (process="workflow")
-
-```python
-from praisonaiagents import Agent, Task, PraisonAIAgents
-
-# Per-task agent config (no need to create agent separately)
-task = Task(
-    description="Research {{topic}}",
-    agent_config={"name": "Researcher", "role": "Expert", "llm": "gpt-4o-mini"},
-    variables={"topic": "AI trends"}
+# 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}")
-
-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 = Workflow(steps=[
+    WorkflowStep(name="gen", handler=generator, guardrail=validate, max_retries=3)
+])
 ```
 
 ### 9. Hooks

src/praisonai/praisonai.rb

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

src/praisonai/praisonai/deploy.py

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

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.3.21"
+__version__ = "2.3.22"

src/praisonai/pyproject.toml

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

src/praisonai/uv.lock

@@ -4197,7 +4197,7 @@ requires-dist = [
     { name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
-    { name = "praisonaiagents", specifier = ">=0.0.199" },
+    { name = "praisonaiagents", specifier = ">=0.1.0" },
     { 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.199"
+version = "0.1.0"
 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/8b/5f/fffbe2473e579edc0bc00022179a896e847128056b6b51905fc497b3cb7d/praisonaiagents-0.0.199.tar.gz", hash = "sha256:276a224350d1a975bdb69f5f07898a60ac9fce0b2d45edbab14e8b755a66853c", size = 538316, upload-time = "2025-12-17T00:01:47.411Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/1a/69/062b8e0867ec673483d83a7529b15d17c15196460060cd7e1542f96c8036/praisonaiagents-0.1.0.tar.gz", hash = "sha256:9c87d8ef965997ab6df2f76badda7911889c6409333b8d2afa20a4f60c80fef9", size = 545876, upload-time = "2025-12-17T00:24:37.466Z" }
 wheels = [
-    { 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" },
+    { url = "https://files.pythonhosted.org/packages/42/11/9f9a8802208e50a44d2e105c6b9743fa08f02378829fb5fad1b3ee9c8b8d/praisonaiagents-0.1.0-py3-none-any.whl", hash = "sha256:bb39f79e92b8a3abe215fcb8f49be896dee539a5e8998f55b14fb8bbbeb86905", size = 516692, upload-time = "2025-12-17T00:24:35.719Z" },
 ]
 
 [[package]]