AI Workflow Orchestration

from dataclasses import dataclass, field
from typing import Any
import time, json

@dataclass
class PipelineContext:
    data: dict = field(default_factory=dict)
    errors: list = field(default_factory=list)
    start_time: float = field(default_factory=time.time)
    steps_completed: list = field(default_factory=list)

class AIPipeline:
    def __init__(self, name: str):
        self.name = name
        self.steps = []

    def add_step(self, name: str, fn, required: bool = True):
        self.steps.append({"name": name, "fn": fn, "required": required})
        return self

    def run(self, initial_data: dict = None) -> PipelineContext:
        ctx = PipelineContext(data=initial_data or {})
        print(f"Pipeline '{self.name}' starting with {len(self.steps)} steps")

        for step in self.steps:
            try:
                print(f"  Running: {step['name']}...")
                result = step["fn"](ctx)
                ctx.data.update(result or {})
                ctx.steps_completed.append(step["name"])
            except Exception as e:
                error = {"step": step["name"], "error": str(e)}
                ctx.errors.append(error)
                print(f"  FAILED: {step['name']} — {e}")
                if step["required"]:
                    print("  Pipeline aborted (required step failed)")
                    break

        elapsed = time.time() - ctx.start_time
        print(f"Pipeline done in {elapsed:.1f}s. {len(ctx.steps_completed)}/{len(self.steps)} steps completed.")
        return ctx

# Usage:
# pipeline = AIPipeline("lead-enrichment")
# pipeline.add_step("fetch_website", fetch_step)
# pipeline.add_step("ai_analysis", analyze_step)
# pipeline.add_step("update_crm", crm_step, required=False)
# result = pipeline.run({"domain": "stripe.com"})

import time, random
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1.0, max_delay=60.0,
                       retryable_exceptions=(Exception,)):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return fn(*args, **kwargs)
                except retryable_exceptions as e:
                    if attempt == max_retries:
                        raise
                    delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
                    print(f"  Retry {attempt+1}/{max_retries} in {delay:.1f}s: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure = 0
        self.state = "closed"  # closed=normal, open=blocking, half_open=testing

    def call(self, fn, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.reset_timeout:
                self.state = "half_open"
            else:
                raise Exception("Circuit breaker is OPEN — service unavailable")

        try:
            result = fn(*args, **kwargs)
            if self.state == "half_open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "open"
                print(f"Circuit OPEN after {self.failures} failures")
            raise

breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

import time, json
from datetime import datetime
from collections import defaultdict

class PipelineMonitor:
    def __init__(self):
        self.runs = []
        self.step_stats = defaultdict(lambda: {"success": 0, "failure": 0, "total_time": 0})

    def record_step(self, pipeline: str, step: str, success: bool, duration: float, error: str = None):
        stat = self.step_stats[f"{pipeline}.{step}"]
        stat["success" if success else "failure"] += 1
        stat["total_time"] += duration

        if not success:
            self.runs.append({
                "timestamp": datetime.utcnow().isoformat(),
                "pipeline": pipeline, "step": step,
                "error": error, "duration": duration
            })

    def report(self) -> dict:
        report = {}
        for key, stats in self.step_stats.items():
            total = stats["success"] + stats["failure"]
            report[key] = {
                "success_rate": f"{stats['success']/total:.1%}" if total > 0 else "N/A",
                "avg_time": f"{stats['total_time']/total:.2f}s" if total > 0 else "N/A",
                "total_runs": total
            }
        return report

monitor = PipelineMonitor()
# In your pipeline steps:
# start = time.time()
# try:
#     result = do_work()
#     monitor.record_step("enrichment", "ai_analysis", True, time.time()-start)
# except Exception as e:
#     monitor.record_step("enrichment", "ai_analysis", False, time.time()-start, str(e))