Workflow Patterns

This section demonstrates common workflow patterns and best practices.

Parallel Processing Pattern

from kailash.workflow import Workflow
from kailash.runtime.parallel import ParallelRuntime

# Create workflow with parallel branches
workflow = Workflow("parallel_pattern", name="Parallel Processing")

# Add nodes that can run in parallel
# ... (node setup)

# Execute with parallel runtime
runtime = ParallelRuntime(max_workers=4)
results, run_id = runtime.execute(workflow)

Error Handling Pattern

from kailash.workflow import Workflow
from kailash.nodes.logic import SwitchNode

# Create workflow with error handling
workflow = Workflow("error_handling", name="Error Handling Pattern")

# Add switch for error routing
error_router = SwitchNode()
workflow.add_node("error_handler", error_router)

# ... (error handling logic)

State Management Pattern

from kailash.workflow import Workflow
from kailash.workflow.state import WorkflowStateWrapper

# Create workflow with state management
workflow = Workflow("stateful_pattern", name="Stateful Workflow")

# Initialize state
state = {"counter": 0, "results": []}
state_wrapper = workflow.create_state_wrapper(state)

# ... (stateful operations)

Cyclic Workflow Patterns

Cyclic workflows enable iterative processing, optimization, and retry patterns.

Retry Pattern with Backoff

from kailash import Workflow
from kailash.nodes.base_cycle_aware import CycleAwareNode
import time

class RetryNode(CycleAwareNode):
    def run(self, context, **kwargs):
        iteration = self.get_iteration(context)
        max_retries = kwargs.get("max_retries", 3)

        try:
            # Attempt operation
            result = self.perform_operation(kwargs["data"])
            return {"success": True, "result": result}
        except Exception as e:
            if iteration < max_retries:
                # Exponential backoff
                delay = 2 ** iteration
                time.sleep(delay)
                self.log_cycle_info(context, f"Retry {iteration + 1} after {delay}s")
                return {"success": False, "retry": True}
            else:
                return {"success": False, "error": str(e)}

# Create workflow with retry cycle
workflow = Workflow("retry_pattern")
workflow.add_node("retry", RetryNode())
workflow.create_cycle("retry_cycle") \
        .connect("retry", "retry") \
        .max_iterations(5) \
        .converge_when("success == True") \
        .build()

Iterative Optimization Pattern

from kailash.nodes.base_cycle_aware import CycleAwareNode

class GradientDescentNode(CycleAwareNode):
    def run(self, context, **kwargs):
        iteration = self.get_iteration(context)
        prev_state = self.get_previous_state(context)

        # Get current parameters
        params = prev_state.get("params", kwargs.get("initial_params"))
        loss = self.calculate_loss(params)

        # Track loss history
        self.accumulate_values(context, "loss", loss)
        trend = self.detect_convergence_trend(context, "loss")

        # Update parameters
        learning_rate = 0.01 * (0.95 ** iteration)  # Decay
        gradient = self.calculate_gradient(params)
        new_params = params - learning_rate * gradient

        # Save state
        self.set_cycle_state({"params": new_params, "loss": loss})

        # Check convergence
        converged = (
            trend["converging"] and trend["stability"] > 0.99 or
            loss < 0.001 or
            iteration >= 1000
        )

        return {
            "params": new_params,
            "loss": loss,
            "converged": converged,
            "iteration": iteration
        }

# Create optimization workflow
workflow = Workflow("optimization")
workflow.add_node("optimizer", GradientDescentNode())
workflow.create_cycle("optimization_cycle") \
        .connect("optimizer", "optimizer") \
        .max_iterations(1000) \
        .converge_when("converged == True") \
        .build()

Data Quality Improvement Pattern

class DataQualityNode(CycleAwareNode):
    def run(self, context, **kwargs):
        data = kwargs["data"]
        iteration = self.get_iteration(context)

        # Calculate quality metrics
        completeness = self.calculate_completeness(data)
        accuracy = self.calculate_accuracy(data)
        quality_score = (completeness + accuracy) / 2

        # Track improvement
        self.accumulate_values(context, "quality", quality_score)
        trend = self.detect_convergence_trend(context, "quality")

        # Apply improvements
        if completeness < 0.95:
            data = self.fill_missing_values(data)
        if accuracy < 0.95:
            data = self.fix_outliers(data)

        # Log progress
        self.log_cycle_info(context,
            f"Iteration {iteration}: Quality={quality_score:.2%}")

        # Check if we've plateaued
        converged = (
            quality_score > 0.95 or
            trend["plateau_detected"] or
            iteration >= 10
        )

        return {
            "data": data,
            "quality_score": quality_score,
            "converged": converged
        }

Stream Processing Pattern

class StreamProcessorNode(CycleAwareNode):
    def run(self, context, **kwargs):
        stream = kwargs["stream"]
        batch_size = kwargs.get("batch_size", 100)
        iteration = self.get_iteration(context)

        # Calculate batch indices
        start = iteration * batch_size
        end = min(start + batch_size, len(stream))

        # Process batch
        batch = stream[start:end]
        results = self.process_batch(batch)

        # Accumulate results
        all_results = self.get_previous_state(context).get("results", [])
        all_results.extend(results)
        self.set_cycle_state({"results": all_results})

        # Check if more data exists
        more_data = end < len(stream)

        return {
            "batch_results": results,
            "total_processed": end,
            "more_data": more_data,
            "all_results": all_results if not more_data else None
        }

# Create stream processing workflow
workflow = Workflow("stream_processor")
workflow.add_node("processor", StreamProcessorNode())
workflow.create_cycle("stream_cycle") \
        .connect("processor", "processor") \
        .converge_when("more_data == False") \
        .build()

Multi-Node Cycle Pattern

# Create a cycle with multiple nodes
workflow = Workflow("multi_node_cycle")

# Add nodes for different stages
workflow.add_node("collector", DataCollectorNode())
workflow.add_node("processor", ProcessorNode())
workflow.add_node("validator", ValidatorNode())
workflow.add_node("decider", DecisionNode())

# Connect in a cycle: collector → processor → validator → decider → collector
workflow.connect("collector", "processor")
workflow.connect("processor", "validator")
workflow.connect("validator", "decider")
workflow.create_cycle("processing_cycle") \
        .connect("decider", "collector") \
        .max_iterations(50) \
        .converge_when("should_continue == False") \
        .build()

Best Practices for Cyclic Workflows

1. Always Set Convergence Conditions: Prevent infinite loops

2. Use State Management: Leverage CycleAwareNode helpers

3. Track Metrics: Monitor convergence with accumulate_values()

4. Handle Edge Cases: Consider first/last iteration behavior

5. Set Reasonable Limits: Use max_iterations as a safety net

6. Log Progress: Use log_cycle_info() for debugging

7. Optimize Performance: Cycles add ~0.03ms overhead per iteration

Alert and Notification Patterns

from kailash import Workflow
from kailash.nodes.alerts import DiscordAlertNode
from kailash.nodes.logic import SwitchNode
from kailash.nodes.code import PythonCodeNode

# Error Alert Pattern
def create_error_alert_workflow():
    workflow = Workflow("error_alerts")

    # Process data (might fail)
    processor = PythonCodeNode.from_function(
        func=process_data_with_errors,
        name="processor"
    )
    workflow.add_node("process", processor)

    # Route based on success/failure
    router = SwitchNode(
        condition_mapping={
            "error": "status == 'error'",
            "success": "status == 'success'"
        }
    )
    workflow.add_node("router", router)

    # Critical error alert
    error_alert = DiscordAlertNode()
    workflow.add_node("error_alert", error_alert)

    # Success notification
    success_alert = DiscordAlertNode()
    workflow.add_node("success_alert", success_alert)

    # Connect the workflow
    workflow.connect("process", "router")
    workflow.connect("router", "error_alert", output_key="error")
    workflow.connect("router", "success_alert", output_key="success")

    return workflow

# Health Dashboard Pattern
def create_health_dashboard():
    workflow = Workflow("health_dashboard")

    # System health check
    health_checker = PythonCodeNode.from_function(
        func=check_system_health,
        name="health_checker"
    )
    workflow.add_node("health_check", health_checker)

    # Dashboard alert
    dashboard = DiscordAlertNode()
    workflow.add_node("dashboard", dashboard)

    workflow.connect("health_check", "dashboard")

    return workflow

# Business KPI Alert Pattern
def create_kpi_alerts():
    workflow = Workflow("kpi_alerts")

    # Calculate KPIs
    kpi_calculator = PythonCodeNode.from_function(
        func=calculate_business_kpis,
        name="kpi_calculator"
    )
    workflow.add_node("calculate", kpi_calculator)

    # Check thresholds
    threshold_checker = SwitchNode(
        condition_mapping={
            "critical": "revenue < target_revenue * 0.8",
            "warning": "revenue < target_revenue * 0.9",
            "normal": "default"
        }
    )
    workflow.add_node("threshold_check", threshold_checker)

    # Different alert types
    critical_alert = DiscordAlertNode()
    warning_alert = DiscordAlertNode()
    normal_report = DiscordAlertNode()

    workflow.add_node("critical_alert", critical_alert)
    workflow.add_node("warning_alert", warning_alert)
    workflow.add_node("normal_report", normal_report)

    # Connect workflow
    workflow.connect("calculate", "threshold_check")
    workflow.connect("threshold_check", "critical_alert", output_key="critical")
    workflow.connect("threshold_check", "warning_alert", output_key="warning")
    workflow.connect("threshold_check", "normal_report", output_key="normal")

    return workflow

# Execute with parameters
def execute_alert_workflows():
    # Error alert execution
    error_workflow = create_error_alert_workflow()
    runtime.execute(error_workflow, parameters={
        "error_alert": {
            "webhook_url": "${DISCORD_WEBHOOK}",
            "title": "🚨 Processing Error",
            "alert_type": "critical",
            "mentions": ["@here"]
        },
        "success_alert": {
            "webhook_url": "${DISCORD_WEBHOOK}",
            "title": "✅ Processing Complete",
            "alert_type": "success"
        }
    })

    # Health dashboard execution
    health_workflow = create_health_dashboard()
    runtime.execute(health_workflow, parameters={
        "dashboard": {
            "webhook_url": "${DISCORD_WEBHOOK}",
            "title": "📊 System Health Dashboard",
            "alert_type": "info",
            "username": "Health Monitor",
            "fields": [
                {"name": "CPU", "value": "{cpu_usage:.1f}%", "inline": True},
                {"name": "Memory", "value": "{memory_usage:.1f}%", "inline": True},
                {"name": "Disk", "value": "{disk_usage:.1f}%", "inline": True}
            ]
        }
    })

Alert Pattern Benefits:

• Real-time Notifications: Immediate alerts on failures or thresholds

• Rich Formatting: Color-coded embeds with structured data

• Escalation Support: Different alert types for different severity levels

• Integration Ready: Works with existing monitoring and error handling

For more patterns, see the workflow examples.