04_workflow_engine.py

"""
生产级工作流引擎案例：企业级任务编排系统
=======================================

本案例展示如何使用 LangGraph 构建一个生产级的工作流引擎，
具备以下企业级特性：
1. 可配置的工作流定义
2. 任务依赖管理
3. 错误处理和重试机制
4. 监控和日志记录
5. 并行执行和资源管理
6. 工作流版本控制
"""

from typing import TypedDict, Annotated, List, Dict, Any, Optional, Union
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import operator
from datetime import datetime, timedelta
from enum import Enum
import json
import yaml
import time
import random
import threading
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed

# 定义工作流状态
class WorkflowStatus(Enum):
    DRAFT = "draft"  # 草稿
    READY = "ready"  # 就绪
    RUNNING = "running"  # 运行中
    PAUSED = "paused"  # 暂停
    COMPLETED = "completed"  # 完成
    FAILED = "failed"  # 失败
    CANCELLED = "cancelled"  # 取消

# 定义任务状态
class TaskStatus(Enum):
    PENDING = "pending"  # 待执行
    QUEUED = "queued"  # 排队中
    RUNNING = "running"  # 执行中
    SUCCESS = "success"  # 成功
    FAILED = "failed"  # 失败
    RETRYING = "retrying"  # 重试中
    SKIPPED = "skipped"  # 跳过
    TIMEOUT = "timeout"  # 超时

# 定义任务类型
class TaskType(Enum):
    HTTP_REQUEST = "http_request"  # HTTP请求
    DATABASE_QUERY = "database_query"  # 数据库查询
    SCRIPT_EXECUTION = "script_execution"  # 脚本执行
    DATA_TRANSFORMATION = "data_transformation"  # 数据转换
    NOTIFICATION = "notification"  # 通知
    WAIT = "wait"  # 等待
    CONDITION = "condition"  # 条件判断
    PARALLEL = "parallel"  # 并行执行

# 定义错误处理策略
class ErrorHandlingStrategy(Enum):
    RETRY = "retry"  # 重试
    CONTINUE = "continue"  # 继续
    STOP = "stop"  # 停止
    ROLLBACK = "rollback"  # 回滚

# 工作流配置
@dataclass
class WorkflowConfig:
    """工作流配置"""
    name: str
    version: str = "1.0.0"
    description: str = ""
    max_retries: int = 3
    retry_delay: int = 5  # 秒
    timeout: int = 300  # 秒
    concurrent_tasks: int = 5
    error_strategy: ErrorHandlingStrategy = ErrorHandlingStrategy.RETRY
    notification_email: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "name": self.name,
            "version": self.version,
            "description": self.description,
            "max_retries": self.max_retries,
            "retry_delay": self.retry_delay,
            "timeout": self.timeout,
            "concurrent_tasks": self.concurrent_tasks,
            "error_strategy": self.error_strategy.value,
            "notification_email": self.notification_email,
            "tags": self.tags
        }

# 任务定义
@dataclass
class TaskDefinition:
    """任务定义"""
    id: str
    name: str
    type: TaskType
    config: Dict[str, Any]
    dependencies: List[str] = field(default_factory=list)
    timeout: int = 60
    retries: int = 3
    retry_delay: int = 5
    on_success: Optional[str] = None
    on_failure: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            "id": self.id,
            "name": self.name,
            "type": self.type.value,
            "config": self.config,
            "dependencies": self.dependencies,
            "timeout": self.timeout,
            "retries": self.retries,
            "retry_delay": self.retry_delay,
            "on_success": self.on_success,
            "on_failure": self.on_failure,
            "metadata": self.metadata
        }

# 任务执行器基类
class TaskExecutor(ABC):
    """任务执行器基类"""
    
    def __init__(self, task_def: TaskDefinition):
        self.task_def = task_def
        self.start_time: Optional[datetime] = None
        self.end_time: Optional[datetime] = None
    
    @abstractmethod
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行任务"""
        pass
    
    def validate(self) -> bool:
        """验证任务配置"""
        return True
    
    def get_execution_time(self) -> Optional[float]:
        """获取执行时间（秒）"""
        if self.start_time and self.end_time:
            return (self.end_time - self.start_time).total_seconds()
        return None

# 具体任务执行器实现
class HttpRequestExecutor(TaskExecutor):
    """HTTP请求执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行HTTP请求"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        url = config.get("url", "")
        method = config.get("method", "GET")
        headers = config.get("headers", {})
        body = config.get("body", {})
        
        # 模拟HTTP请求
        print(f"[HTTP Task] {method} {url}")
        
        # 模拟网络延迟
        time.sleep(random.uniform(0.1, 0.5))
        
        # 模拟响应
        response = {
            "status_code": 200,
            "headers": {"Content-Type": "application/json"},
            "body": {"message": "Request successful", "data": body},
            "elapsed_time": random.uniform(0.1, 1.0)
        }
        
        # 模拟偶尔的失败
        if random.random() < 0.1:  # 10%失败率
            raise Exception(f"HTTP request failed: {url}")
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "response": response,
            "execution_time": self.get_execution_time()
        }

class DatabaseQueryExecutor(TaskExecutor):
    """数据库查询执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行数据库查询"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        query = config.get("query", "")
        params = config.get("params", {})
        
        # 模拟数据库查询
        print(f"[Database Task] Executing query: {query[:50]}...")
        
        # 模拟查询延迟
        time.sleep(random.uniform(0.2, 1.0))
        
        # 模拟查询结果
        results = [
            {"id": 1, "name": "Item 1", "value": random.randint(1, 100)},
            {"id": 2, "name": "Item 2", "value": random.randint(1, 100)},
            {"id": 3, "name": "Item 3", "value": random.randint(1, 100)}
        ]
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "results": results,
            "row_count": len(results),
            "execution_time": self.get_execution_time()
        }

class ScriptExecutionExecutor(TaskExecutor):
    """脚本执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行脚本"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        script = config.get("script", "")
        args = config.get("args", [])
        
        # 模拟脚本执行
        print(f"[Script Task] Executing script with args: {args}")
        
        # 模拟执行时间
        time.sleep(random.uniform(0.5, 2.0))
        
        # 模拟输出
        output = f"Script executed successfully with args: {args}"
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "output": output,
            "exit_code": 0,
            "execution_time": self.get_execution_time()
        }

class DataTransformationExecutor(TaskExecutor):
    """数据转换执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行数据转换"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        input_data = config.get("input_data", {})
        transformation = config.get("transformation", {})
        
        # 模拟数据转换
        print(f"[Data Transformation Task] Transforming data")
        
        # 模拟转换逻辑
        time.sleep(random.uniform(0.1, 0.3))
        
        # 简单的转换示例
        transformed_data = {}
        for key, value in input_data.items():
            if isinstance(value, (int, float)):
                transformed_data[f"transformed_{key}"] = value * 2
            elif isinstance(value, str):
                transformed_data[f"transformed_{key}"] = value.upper()
            else:
                transformed_data[f"transformed_{key}"] = value
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "input_data": input_data,
            "transformed_data": transformed_data,
            "execution_time": self.get_execution_time()
        }

class NotificationExecutor(TaskExecutor):
    """通知执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """发送通知"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        channel = config.get("channel", "email")
        recipient = config.get("recipient", "")
        message = config.get("message", "")
        
        # 模拟发送通知
        print(f"[Notification Task] Sending {channel} notification to {recipient}")
        
        # 模拟发送延迟
        time.sleep(random.uniform(0.05, 0.2))
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "channel": channel,
            "recipient": recipient,
            "message_sent": True,
            "execution_time": self.get_execution_time()
        }

class WaitExecutor(TaskExecutor):
    """等待执行器"""
    
    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行等待"""
        self.start_time = datetime.now()
        
        config = self.task_def.config
        duration = config.get("duration", 1)  # 秒
        
        # 执行等待
        print(f"[Wait Task] Waiting for {duration} seconds")
        
        time.sleep(duration)
        
        self.end_time = datetime.now()
        
        return {
            "success": True,
            "duration": duration,
            "execution_time": self.get_execution_time()
        }

# 任务执行器工厂
class TaskExecutorFactory:
    """任务执行器工厂"""
    
    @staticmethod
    def create_executor(task_def: TaskDefinition) -> TaskExecutor:
        """创建任务执行器"""
        if task_def.type == TaskType.HTTP_REQUEST:
            return HttpRequestExecutor(task_def)
        elif task_def.type == TaskType.DATABASE_QUERY:
            return DatabaseQueryExecutor(task_def)
        elif task_def.type == TaskType.SCRIPT_EXECUTION:
            return ScriptExecutionExecutor(task_def)
        elif task_def.type == TaskType.DATA_TRANSFORMATION:
            return DataTransformationExecutor(task_def)
        elif task_def.type == TaskType.NOTIFICATION:
            return NotificationExecutor(task_def)
        elif task_def.type == TaskType.WAIT:
            return WaitExecutor(task_def)
        else:
            raise ValueError(f"Unsupported task type: {task_def.type}")

# 定义工作流状态
class WorkflowState(TypedDict):
    """工作流状态"""
    workflow_id: str  # 工作流ID
    execution_id: str  # 执行ID
    config: WorkflowConfig  # 配置
    tasks: Dict[str, TaskDefinition]  # 任务定义
    task_status: Dict[str, TaskStatus]  # 任务状态
    task_results: Dict[str, Dict[str, Any]]  # 任务结果
    task_errors: Dict[str, str]  # 任务错误
    task_retries: Dict[str, int]  # 任务重试次数
    dependencies: Dict[str, List[str]]  # 任务依赖关系
    execution_order: List[str]  # 执行顺序
    current_task: Optional[str]  # 当前任务
    context: Dict[str, Any]  # 执行上下文
    start_time: Optional[datetime]  # 开始时间
    end_time: Optional[datetime]  # 结束时间
    status: WorkflowStatus  # 工作流状态
    error_message: Optional[str]  # 错误信息
    logs: Annotated[List[Dict], operator.add]  # 日志记录
    metrics: Dict[str, Any]  # 执行指标

# 工作流引擎
class WorkflowEngine:
    """工作流引擎"""
    
    def __init__(self, config: WorkflowConfig, tasks: List[TaskDefinition]):
        self.config = config
        self.tasks = {task.id: task for task in tasks}
        self.state_graph = None
        self.build_state_graph()
    
    def build_state_graph(self):
        """构建状态图"""
        workflow = StateGraph(WorkflowState)
        
        # 添加节点
        workflow.add_node("initialize", self.initialize_workflow)
        workflow.add_node("validate", self.validate_workflow)
        workflow.add_node("plan", self.plan_execution)
        workflow.add_node("execute", self.execute_tasks)
        workflow.add_node("handle_error", self.handle_error)
        workflow.add_node("complete", self.complete_workflow)
        workflow.add_node("cleanup", self.cleanup_resources)
        
        # 设置入口点
        workflow.set_entry_point("initialize")
        
        # 添加边
        workflow.add_edge("initialize", "validate")
        
        # 添加条件边
        workflow.add_conditional_edges(
            "validate",
            self.check_validation,
            {
                "valid": "plan",
                "invalid": "handle_error"
            }
        )
        
        workflow.add_edge("plan", "execute")
        
        workflow.add_conditional_edges(
            "execute",
            self.check_execution_status,
            {
                "completed": "complete",
                "failed": "handle_error",
                "continue": "execute"
            }
        )
        
        workflow.add_conditional_edges(
            "handle_error",
            self.check_error_handling,
            {
                "retry": "execute",
                "continue": "complete",
                "stop": "cleanup"
            }
        )
        
        workflow.add_edge("complete", "cleanup")
        workflow.add_edge("cleanup", END)
        
        # 编译图
        self.state_graph = workflow.compile()
    
    def initialize_workflow(self, state: WorkflowState) -> dict:
        """初始化工作流"""
        print(f"[{datetime.now()}] Initializing workflow: {state['workflow_id']}")
        
        execution_id = str(uuid.uuid4())[:8]
        
        return {
            "execution_id": execution_id,
            "start_time": datetime.now(),
            "status": WorkflowStatus.RUNNING,
            "logs": [{
                "timestamp": datetime.now(),
                "level": "INFO",
                "message": f"Workflow {state['workflow_id']} initialized with execution ID: {execution_id}"
            }],
            "metrics": {
                "total_tasks": len(state["tasks"]),
                "tasks_completed": 0,
                "tasks_failed": 0,
                "tasks_retried": 0,
                "start_time": datetime.now()
            }
        }
    
    def validate_workflow(self, state: WorkflowState) -> dict:
        """验证工作流"""
        print(f"[{datetime.now()}] Validating workflow")
        
        errors = []
        
        # 检查任务定义
        for task_id, task_def in state["tasks"].items():
            try:
                executor = TaskExecutorFactory.create_executor(task_def)
                if not executor.validate():
                    errors.append(f"Task {task_id} validation failed")
            except Exception as e:
                errors.append(f"Task {task_id} error: {str(e)}")
        
        # 检查依赖关系
        for task_id, task_def in state["tasks"].items():
            for dep_id in task_def.dependencies:
                if dep_id not in state["tasks"]:
                    errors.append(f"Task {task_id} depends on non-existent task: {dep_id}")
        
        # 检查循环依赖
        if self.has_cycle(state["tasks"]):
            errors.append("Workflow has cyclic dependencies")
        
        is_valid = len(errors) == 0
        
        return {
            "logs": [{
                "timestamp": datetime.now(),
                "level": "INFO" if is_valid else "ERROR",
                "message": f"Workflow validation {'passed' if is_valid else 'failed'}. Errors: {errors}"
            }],
            "error_message": None if is_valid else "; ".join(errors)
        }
    
    def has_cycle(self, tasks: Dict[str, TaskDefinition]) -> bool:
        """检查是否有循环依赖"""
        visited = set()
        recursion_stack = set()
        
        def dfs(task_id: str) -> bool:
            if task_id in recursion_stack:
                return True
            if task_id in visited:
                return False
            
            visited.add(task_id)
            recursion_stack.add(task_id)
            
            task = tasks.get(task_id)
            if task:
                for dep_id in task.dependencies:
                    if dfs(dep_id):
                        return True
            
            recursion_stack.remove(task_id)
            return False
        
        for task_id in tasks:
            if dfs(task_id):
                return True
        
        return False
    
    def plan_execution(self, state: WorkflowState) -> dict:
        """规划执行顺序"""
        print(f"[{datetime.now()}] Planning execution order")
        
        # 拓扑排序获取执行顺序
        execution_order = self.topological_sort(state["tasks"])
        
        # 初始化任务状态
        task_status = {}
        task_retries = {}
        dependencies = {}
        
        for task_id in state["tasks"]:
            task_status[task_id] = TaskStatus.PENDING
            task_retries[task_id] = 0
            dependencies[task_id] = state["tasks"][task_id].dependencies.copy()
        
        return {
            "execution_order": execution_order,
            "task_status": task_status,
            "task_retries": task_retries,
            "dependencies": dependencies,
            "logs": [{
                "timestamp": datetime.now(),
                "level": "INFO",
                "message": f"Execution plan created with {len(execution_order)} tasks in order"
            }]
        }
    
    def topological_sort(self, tasks: Dict[str, TaskDefinition]) -> List[str]:
        """拓扑排序"""
        in_degree = {task_id: 0 for task_id in tasks}
        graph = {task_id: [] for task_id in tasks}
        
        # 构建图
        for task_id, task_def in tasks.items():
            for dep_id in task_def.dependencies:
                graph[dep_id].append(task_id)
                in_degree[task_id] += 1
        
        # 找到入度为0的节点
        queue = [task_id for task_id in tasks if in_degree[task_id] == 0]
        result = []
        
        while queue:
            task_id = queue.pop(0)
            result.append(task_id)
            
            for neighbor in graph[task_id]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)
        
        if len(result) != len(tasks):
            # 有环，返回原始顺序
            return list(tasks.keys())
        
        return result
    
    def execute_tasks(self, state: WorkflowState) -> dict:
        """执行任务"""
        print(f"[{datetime.now()}] Executing tasks")
        
        # 找出可以执行的任务（依赖已满足且状态为PENDING）
        ready_tasks = []
        for task_id in state["execution_order"]:
            if state["task_status"][task_id] == TaskStatus.PENDING:
                # 检查依赖是否都已完成
                deps_met = all(
                    state["task_status"].get(dep_id) == TaskStatus.SUCCESS
                    for dep_id in state["dependencies"][task_id]
                )
                
                if deps_met or not state["dependencies"][task_id]:
                    ready_tasks.append(task_id)
        
        # 并行执行任务
        results = {}
        with ThreadPoolExecutor(max_workers=self.config.concurrent_tasks) as executor:
            future_to_task = {}
            
            for task_id in ready_tasks[:self.config.concurrent_tasks]:
                task_def = state["tasks"][task_id]
                future = executor.submit(self.execute_single_task, task_def, state["context"])
                future_to_task[future] = task_id
            
            for future in as_completed(future_to_task):
                task_id = future_to_task[future]
                try:
                    result = future.result()
                    results[task_id] = result
                except Exception as e:
                    results[task_id] = {
                        "success": False,
                        "error": str(e)
                    }
        
        # 更新状态
        task_status_updates = {}
        task_results_updates = {}
        task_errors_updates = {}
        task_retries_updates = {}
        metrics_updates = state["metrics"].copy()
        logs_updates = []
        
        for task_id, result in results.items():
            if result.get("success", False):
                task_status_updates[task_id] = TaskStatus.SUCCESS
                task_results_updates[task_id] = result
                metrics_updates["tasks_completed"] = metrics_updates.get("tasks_completed", 0) + 1
                
                logs_updates.append({
                    "timestamp": datetime.now(),
                    "level": "INFO",
                    "message": f"Task {task_id} completed successfully"
                })
            else:
                retry_count = state["task_retries"][task_id]
                max_retries = state["tasks"][task_id].retries
                
                if retry_count < max_retries:
                    task_status_updates[task_id] = TaskStatus.RETRYING
                    task_retries_updates[task_id] = retry_count + 1
                    metrics_updates["tasks_retried"] = metrics_updates.get("tasks_retried", 0) + 1
                    
                    logs_updates.append({
                        "timestamp": datetime.now(),
                        "level": "WARNING",
                        "message": f"Task {task_id} failed, retrying ({retry_count + 1}/{max_retries})"
                    })
                else:
                    task_status_updates[task_id] = TaskStatus.FAILED
                    task_errors_updates[task_id] = result.get("error", "Unknown error")
                    metrics_updates["tasks_failed"] = metrics_updates.get("tasks_failed", 0) + 1
                    
                    logs_updates.append({
                        "timestamp": datetime.now(),
                        "level": "ERROR",
                        "message": f"Task {task_id} failed after {max_retries} retries"
                    })
        
        # 检查是否所有任务都已完成
        all_completed = all(
            state["task_status"].get(task_id) in [TaskStatus.SUCCESS, TaskStatus.SKIPPED, TaskStatus.FAILED]
            for task_id in state["tasks"]
        )
        
        status_update = {}
        if all_completed:
            failed_tasks = [
                task_id for task_id, status in state["task_status"].items()
                if status == TaskStatus.FAILED
            ]
            
            if failed_tasks:
                status_update = {"status": WorkflowStatus.FAILED}
            else:
                status_update = {"status": WorkflowStatus.COMPLETED}
        
        return {
            "task_status": task_status_updates,
            "task_results": task_results_updates,
            "task_errors": task_errors_updates,
            "task_retries": task_retries_updates,
            "metrics": metrics_updates,
            "logs": logs_updates,
            **status_update
        }
    
    def execute_single_task(self, task_def: TaskDefinition, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行单个任务"""
        try:
            executor = TaskExecutorFactory.create_executor(task_def)
            result = executor.execute(context)
            return result
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "execution_time": None
            }
    
    def handle_error(self, state: WorkflowState) -> dict:
        """处理错误"""
        print(f"[{datetime.now()}] Handling error")
        
        error_strategy = self.config.error_strategy
        
        if error_strategy == ErrorHandlingStrategy.RETRY:
            # 找出失败的任务并重置状态
            task_status_updates = {}
            for task_id, status in state["task_status"].items():
                if status == TaskStatus.FAILED:
                    task_status_updates[task_id] = TaskStatus.PENDING
            
            return {
                "task_status": task_status_updates,
                "logs": [{
                    "timestamp": datetime.now(),
                    "level": "INFO",
                    "message": "Retrying failed tasks"
                }]
            }
        
        elif error_strategy == ErrorHandlingStrategy.CONTINUE:
            # 跳过失败的任务，继续执行
            task_status_updates = {}
            for task_id, status in state["task_status"].items():
                if status == TaskStatus.FAILED:
                    task_status_updates[task_id] = TaskStatus.SKIPPED
            
            return {
                "task_status": task_status_updates,
                "logs": [{
                    "timestamp": datetime.now(),
                    "level": "WARNING",
                    "message": "Skipping failed tasks and continuing"
                }]
            }
        
        else:  # STOP or ROLLBACK
            return {
                "status": WorkflowStatus.FAILED,
                "logs": [{
                    "timestamp": datetime.now(),
                    "level": "ERROR",
                    "message": f"Workflow stopped due to error handling strategy: {error_strategy.value}"
                }]
            }
    
    def complete_workflow(self, state: WorkflowState) -> dict:
        """完成工作流"""
        print(f"[{datetime.now()}] Completing workflow")
        
        metrics = state["metrics"].copy()
        metrics["end_time"] = datetime.now()
        metrics["total_duration"] = (
            metrics["end_time"] - metrics["start_time"]
        ).total_seconds() if metrics.get("start_time") else 0
        
        return {
            "end_time": datetime.now(),
            "status": WorkflowStatus.COMPLETED,
            "metrics": metrics,
            "logs": [{
                "timestamp": datetime.now(),
                "level": "INFO",
                "message": f"Workflow completed successfully in {metrics['total_duration']:.2f} seconds"
            }]
        }
    
    def cleanup_resources(self, state: WorkflowState) -> dict:
        """清理资源"""
        print(f"[{datetime.now()}] Cleaning up resources")
        
        return {
            "logs": [{
                "timestamp": datetime.now(),
                "level": "INFO",
                "message": "Resources cleaned up"
            }]
        }
    
    def check_validation(self, state: WorkflowState) -> str:
        """检查验证结果"""
        if state.get("error_message"):
            return "invalid"
        return "valid"
    
    def check_execution_status(self, state: WorkflowState) -> str:
        """检查执行状态"""
        if state["status"] == WorkflowStatus.COMPLETED:
            return "completed"
        elif state["status"] == WorkflowStatus.FAILED:
            return "failed"
        
        # 检查是否有任务需要执行
        pending_tasks = [
            task_id for task_id, status in state["task_status"].items()
            if status in [TaskStatus.PENDING, TaskStatus.RETRYING]
        ]
        
        if not pending_tasks:
            return "completed"
        
        return "continue"
    
    def check_error_handling(self, state: WorkflowState) -> str:
        """检查错误处理策略"""
        return self.config.error_strategy.value
    
    def run(self, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """运行工作流"""
        initial_state = {
            "workflow_id": f"workflow_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            "execution_id": "",
            "config": self.config,
            "tasks": self.tasks,
            "task_status": {},
            "task_results": {},
            "task_errors": {},
            "task_retries": {},
            "dependencies": {},
            "execution_order": [],
            "current_task": None,
            "context": context or {},
            "start_time": None,
            "end_time": None,
            "status": WorkflowStatus.DRAFT,
            "error_message": None,
            "logs": [],
            "metrics": {}
        }
        
        return self.state_graph.invoke(initial_state)

# 工作流定义加载器
class WorkflowLoader:
    """工作流定义加载器"""
    
    @staticmethod
    def from_yaml(yaml_content: str) -> tuple[WorkflowConfig, List[TaskDefinition]]:
        """从YAML加载工作流定义"""
        data = yaml.safe_load(yaml_content)
        
        # 加载配置
        config_data = data.get("config", {})
        config = WorkflowConfig(
            name=config_data.get("name", "Unnamed Workflow"),
            version=config_data.get("version", "1.0.0"),
            description=config_data.get("description", ""),
            max_retries=config_data.get("max_retries", 3),
            retry_delay=config_data.get("retry_delay", 5),
            timeout=config_data.get("timeout", 300),
            concurrent_tasks=config_data.get("concurrent_tasks", 5),
            error_strategy=ErrorHandlingStrategy(config_data.get("error_strategy", "retry")),
            notification_email=config_data.get("notification_email"),
            tags=config_data.get("tags", [])
        )
        
        # 加载任务
        tasks = []
        for task_data in data.get("tasks", []):
            task = TaskDefinition(
                id=task_data["id"],
                name=task_data["name"],
                type=TaskType(task_data["type"]),
                config=task_data.get("config", {}),
                dependencies=task_data.get("dependencies", []),
                timeout=task_data.get("timeout", 60),
                retries=task_data.get("retries", 3),
                retry_delay=task_data.get("retry_delay", 5),
                on_success=task_data.get("on_success"),
                on_failure=task_data.get("on_failure"),
                metadata=task_data.get("metadata", {})
            )
            tasks.append(task)
        
        return config, tasks
    
    @staticmethod
    def from_json(json_content: str) -> tuple[WorkflowConfig, List[TaskDefinition]]:
        """从JSON加载工作流定义"""
        data = json.loads(json_content)
        
        # 加载配置
        config_data = data.get("config", {})
        config = WorkflowConfig(
            name=config_data.get("name", "Unnamed Workflow"),
            version=config_data.get("version", "1.0.0"),
            description=config_data.get("description", ""),
            max_retries=config_data.get("max_retries", 3),
            retry_delay=config_data.get("retry_delay", 5),
            timeout=config_data.get("timeout", 300),
            concurrent_tasks=config_data.get("concurrent_tasks", 5),
            error_strategy=ErrorHandlingStrategy(config_data.get("error_strategy", "retry")),
            notification_email=config_data.get("notification_email"),
            tags=config_data.get("tags", [])
        )
        
        # 加载任务
        tasks = []
        for task_data in data.get("tasks", []):
            task = TaskDefinition(
                id=task_data["id"],
                name=task_data["name"],
                type=TaskType(task_data["type"]),
                config=task_data.get("config", {}),
                dependencies=task_data.get("dependencies", []),
                timeout=task_data.get("timeout", 60),
                retries=task_data.get("retries", 3),
                retry_delay=task_data.get("retry_delay", 5),
                on_success=task_data.get("on_success"),
                on_failure=task_data.get("on_failure"),
                metadata=task_data.get("metadata", {})
            )
            tasks.append(task)
        
        return config, tasks

# 工作流监控器
class WorkflowMonitor:
    """工作流监控器"""
    
    def __init__(self):
        self.workflow_history = []
    
    def record_execution(self, result: Dict[str, Any]):
        """记录执行结果"""
        self.workflow_history.append({
            "timestamp": datetime.now(),
            "workflow_id": result.get("workflow_id"),
            "execution_id": result.get("execution_id"),
            "status": result.get("status"),
            "metrics": result.get("metrics", {}),
            "error_message": result.get("error_message")
        })
    
    def get_statistics(self) -> Dict[str, Any]:
        """获取统计信息"""
        if not self.workflow_history:
            return {}
        
        total_executions = len(self.workflow_history)
        successful = sum(1 for r in self.workflow_history if r["status"] == WorkflowStatus.COMPLETED.value)
        failed = sum(1 for r in self.workflow_history if r["status"] == WorkflowStatus.FAILED.value)
        
        durations = [
            r["metrics"].get("total_duration", 0)
            for r in self.workflow_history
            if "total_duration" in r["metrics"]
        ]
        
        avg_duration = sum(durations) / len(durations) if durations else 0
        
        return {
            "total_executions": total_executions,
            "successful": successful,
            "failed": failed,
            "success_rate": successful / total_executions if total_executions > 0 else 0,
            "average_duration": avg_duration,
            "last_execution": self.workflow_history[-1] if self.workflow_history else None
        }
    
    def print_dashboard(self):
        """打印监控仪表板"""
        stats = self.get_statistics()
        
        print("\n" + "=" * 70)
        print("工作流监控仪表板")
        print("=" * 70)
        
        print(f"总执行次数: {stats.get('total_executions', 0)}")
        print(f"成功次数: {stats.get('successful', 0)}")
        print(f"失败次数: {stats.get('failed', 0)}")
        print(f"成功率: {stats.get('success_rate', 0):.1%}")
        print(f"平均执行时间: {stats.get('average_duration', 0):.2f}秒")
        
        if stats.get('last_execution'):
            last = stats['last_execution']
            print(f"\n最后执行:")
            print(f"  工作流ID: {last.get('workflow_id')}")
            print(f"  执行ID: {last.get('execution_id')}")
            print(f"  状态: {last.get('status')}")
            print(f"  时间: {last.get('timestamp')}")

# 运行示例
def run_workflow_example():
    """运行工作流引擎示例"""
    print("=" * 70)
    print("生产级工作流引擎示例")
    print("=" * 70)
    
    # 创建监控器
    monitor = WorkflowMonitor()
    
    # 示例1：简单数据处理工作流
    print("\n示例1：简单数据处理工作流")
    print("-" * 50)
    
    # 定义工作流配置
    config = WorkflowConfig(
        name="Data Processing Pipeline",
        version="1.0.0",
        description="Process and transform data from multiple sources",
        max_retries=3,
        concurrent_tasks=3,
        error_strategy=ErrorHandlingStrategy.RETRY
    )
    
    # 定义任务
    tasks = [
        TaskDefinition(
            id="fetch_data",
            name="Fetch Data from API",
            type=TaskType.HTTP_REQUEST,
            config={
                "url": "https://api.example.com/data",
                "method": "GET",
                "headers": {"Authorization": "Bearer token"}
            }
        ),
        TaskDefinition(
            id="query_database",
            name="Query Database",
            type=TaskType.DATABASE_QUERY,
            config={
                "query": "SELECT * FROM users WHERE status = 'active'",
                "params": {}
            },
            dependencies=["fetch_data"]
        ),
        TaskDefinition(
            id="transform_data",
            name="Transform Data",
            type=TaskType.DATA_TRANSFORMATION,
            config={
                "input_data": {"source": "api_and_db"},
                "transformation": {"method": "merge_and_clean"}
            },
            dependencies=["fetch_data", "query_database"]
        ),
        TaskDefinition(
            id="send_notification",
            name="Send Completion Notification",
            type=TaskType.NOTIFICATION,
            config={
                "channel": "email",
                "recipient": "admin@example.com",
                "message": "Data processing completed successfully"
            },
            dependencies=["transform_data"]
        )
    ]
    
    # 创建工作流引擎
    workflow = WorkflowEngine(config, tasks)
    
    # 运行工作流
    result = workflow.run({"environment": "production"})
    
    # 记录执行结果
    monitor.record_execution(result)
    
    print(f"工作流ID: {result['workflow_id']}")
    print(f"执行ID: {result['execution_id']}")
    print(f"状态: {result['status'].value}")
    print(f"开始时间: {result['start_time']}")
    print(f"结束时间: {result['end_time']}")
    
    metrics = result.get('metrics', {})
    print(f"总任务数: {metrics.get('total_tasks', 0)}")
    print(f"完成任务数: {metrics.get('tasks_completed', 0)}")
    print(f"失败任务数: {metrics.get('tasks_failed', 0)}")
    print(f"重试任务数: {metrics.get('tasks_retried', 0)}")
    print(f"总执行时间: {metrics.get('total_duration', 0):.2f}秒")
    
    # 示例2：从YAML文件加载工作流
    print("\n示例2：从YAML文件加载工作流")
    print("-" * 50)
    
    yaml_content = """
config:
  name: "E-commerce Order Processing"
  version: "2.0.0"
  description: "Process e-commerce orders from placement to delivery"
  max_retries: 3
  concurrent_tasks: 4
  error_strategy: "continue"

tasks:
  - id: "validate_order"
    name: "Validate Order"
    type: "http_request"
    config:
      url: "https://api.example.com/orders/validate"
      method: "POST"
    retries: 2

  - id: "process_payment"
    name: "Process Payment"
    type: "http_request"
    config:
      url: "https://api.example.com/payments/process"
      method: "POST"
    dependencies: ["validate_order"]
    retries: 3

  - id: "update_inventory"
    name: "Update Inventory"
    type: "database_query"
    config:
      query: "UPDATE inventory SET quantity = quantity - :quantity WHERE product_id = :product_id"
    dependencies: ["process_payment"]

  - id: "schedule_shipping"
    name: "Schedule Shipping"
    type: "http_request"
    config:
      url: "https://api.example.com/shipping/schedule"
      method: "POST"
    dependencies: ["update_inventory"]

  - id: "send_confirmation"
    name: "Send Confirmation Email"
    type: "notification"
    config:
      channel: "email"
      recipient: "{{customer_email}}"
      message: "Your order has been confirmed and will be shipped soon."
    dependencies: ["schedule_shipping"]
"""
    
    try:
        config2, tasks2 = WorkflowLoader.from_yaml(yaml_content)
        workflow2 = WorkflowEngine(config2, tasks2)
        result2 = workflow2.run({
            "customer_email": "customer@example.com",
            "order_id": "ORD123456"
        })
        
        monitor.record_execution(result2)
        
        print(f"工作流名称: {config2.name}")
        print(f"版本: {config2.version}")
        print(f"执行状态: {result2['status'].value}")
        print(f"任务数量: {len(tasks2)}")
        
    except Exception as e:
        print(f"YAML加载失败: {e}")
    
    # 示例3：复杂并行工作流
    print("\n示例3：复杂并行工作流")
    print("-" * 50)
    
    config3 = WorkflowConfig(
        name="Parallel Data Processing",
        description="Process multiple data sources in parallel",
        concurrent_tasks=5,
        error_strategy=ErrorHandlingStrategy.CONTINUE
    )
    
    tasks3 = []
    # 创建5个并行任务
    for i in range(5):
        tasks3.append(
            TaskDefinition(
                id=f"data_source_{i}",
                name=f"Process Data Source {i}",
                type=TaskType.DATA_TRANSFORMATION,
                config={
                    "input_data": {"source": f"data_source_{i}"},
                    "transformation": {"method": "normalize"}
                }
            )
        )
    
    # 汇总任务（依赖所有并行任务）
    tasks3.append(
        TaskDefinition(
            id="aggregate_results",
            name="Aggregate Results",
            type=TaskType.DATA_TRANSFORMATION,
            config={
                "input_data": {"sources": "all"},
                "transformation": {"method": "aggregate"}
            },
            dependencies=[f"data_source_{i}" for i in range(5)]
        )
    )
    
    workflow3 = WorkflowEngine(config3, tasks3)
    result3 = workflow3.run()
    
    monitor.record_execution(result3)
    
    print(f"工作流状态: {result3['status'].value}")
    print(f"并行任务数: 5")
    print(f"总任务数: {len(tasks3)}")
    
    # 显示监控仪表板
    print("\n" + "=" * 70)
    print("工作流执行统计")
    print("=" * 70)
    monitor.print_dashboard()
    
    return monitor

# 高级功能：工作流版本管理
class WorkflowVersionManager:
    """工作流版本管理器"""
    
    def __init__(self):
        self.versions = {}
        self.current_version = None
    
    def add_version(self, version: str, config: WorkflowConfig, tasks: List[TaskDefinition]):
        """添加版本"""
        self.versions[version] = {
            "config": config,
            "tasks": tasks,
            "created_at": datetime.now()
        }
        
        if not self.current_version:
            self.current_version = version
    
    def get_version(self, version: str) -> Optional[Dict[str, Any]]:
        """获取版本"""
        return self.versions.get(version)
    
    def list_versions(self) -> List[str]:
        """列出所有版本"""
        return sorted(self.versions.keys())
    
    def rollback(self, version: str) -> bool:
        """回滚到指定版本"""
        if version in self.versions:
            self.current_version = version
            return True
        return False
    
    def get_current_workflow(self) -> Optional[WorkflowEngine]:
        """获取当前版本的工作流"""
        if not self.current_version:
            return None
        
        version_data = self.versions.get(self.current_version)
        if not version_data:
            return None
        
        return WorkflowEngine(version_data["config"], version_data["tasks"])

# 工作流调度器
class WorkflowScheduler:
    """工作流调度器"""
    
    def __init__(self):
        self.scheduled_workflows = []
        self.running = False
    
    def schedule(self, workflow: WorkflowEngine, schedule_time: datetime, context: Optional[Dict] = None):
        """调度工作流"""
        self.scheduled_workflows.append({
            "workflow": workflow,
            "schedule_time": schedule_time,
            "context": context or {},
            "status": "scheduled"
        })
    
    def run_scheduled(self):
        """运行调度的任务"""
        self.running = True
        current_time = datetime.now()
        
        for job in self.scheduled_workflows:
            if job["status"] == "scheduled" and job["schedule_time"] <= current_time:
                print(f"[Scheduler] Running scheduled workflow at {current_time}")
                try:
                    result = job["workflow"].run(job["context"])
                    job["status"] = "completed"
                    job["result"] = result
                except Exception as e:
                    job["status"] = "failed"
                    job["error"] = str(e)
        
        self.running = False
    
    def get_schedule_status(self) -> List[Dict[str, Any]]:
        """获取调度状态"""
        return self.scheduled_workflows.copy()

if __name__ == "__main__":
    # 运行示例
    monitor = run_workflow_example()
    
    print("\n" + "=" * 70)
    print("生产级工作流引擎案例总结:")
    print("1. 可配置的工作流定义和任务")
    print("2. 任务依赖管理和拓扑排序")
    print("3. 并行执行和资源管理")
    print("4. 错误处理和重试机制")
    print("5. 监控、日志和指标收集")
    print("6. 工作流版本控制和调度")
    print("=" * 70)
    
    # 演示高级功能
    print("\n演示高级功能:")
    
    # 工作流版本管理
    print("\n1. 工作流版本管理:")
    version_manager = WorkflowVersionManager()
    
    # 添加不同版本
    config_v1 = WorkflowConfig(name="Data Pipeline", version="1.0.0")
    tasks_v1 = [
        TaskDefinition(
            id="task1",
            name="Data Extraction",
            type=TaskType.HTTP_REQUEST,
            config={"url": "https://api.example.com/data"}
        )
    ]
    version_manager.add_version("1.0.0", config_v1, tasks_v1)
    
    config_v2 = WorkflowConfig(name="Data Pipeline", version="2.0.0", description="Improved version")
    tasks_v2 = [
        TaskDefinition(
            id="task1",
            name="Data Extraction",
            type=TaskType.HTTP_REQUEST,
            config={"url": "https://api.example.com/v2/data"}
        ),
        TaskDefinition(
            id="task2",
            name="Data Validation",
            type=TaskType.DATA_TRANSFORMATION,
            config={"input_data": {}, "transformation": {"method": "validate"}},
            dependencies=["task1"]
        )
    ]
    version_manager.add_version("2.0.0", config_v2, tasks_v2)
    
    print(f"可用版本: {version_manager.list_versions()}")
    print(f"当前版本: {version_manager.current_version}")
    
    # 工作流调度
    print("\n2. 工作流调度:")
    scheduler = WorkflowScheduler()
    
    # 调度一个未来执行的工作流
    future_time = datetime.now() + timedelta(seconds=10)
    simple_config = WorkflowConfig(name="Scheduled Workflow")
    simple_tasks = [
        TaskDefinition(
            id="scheduled_task",
            name="Scheduled Task",
            type=TaskType.NOTIFICATION,
            config={
                "channel": "email",
                "recipient": "admin@example.com",
                "message": "This is a scheduled workflow execution"
            }
        )
    ]
    scheduled_workflow = WorkflowEngine(simple_config, simple_tasks)
    
    scheduler.schedule(scheduled_workflow, future_time)
    print(f"已调度工作流，执行时间: {future_time}")
    print(f"调度状态: {len(scheduler.get_schedule_status())} 个任务等待执行")
    
    # 演示工作流可视化（简化版）
    print("\n3. 工作流可视化（简化）:")
    print("""
    Data Processing Pipeline:
    
    [fetch_data] → [query_database] → [transform_data] → [send_notification]
                    ↗
          [wait_task]
    
    依赖关系:
    - transform_data 依赖 fetch_data 和 query_database
    - send_notification 依赖 transform_data
    - query_database 依赖 fetch_data
    """)
    
    print("\n生产级工作流引擎已成功演示！")

======================================================================
生产级工作流引擎示例
======================================================================

示例1：简单数据处理工作流
--------------------------------------------------
[2026-03-25 21:44:37.901011] Initializing workflow: workflow_20260325_214437
[2026-03-25 21:44:37.901365] Validating workflow
[2026-03-25 21:44:37.901744] Planning execution order
[2026-03-25 21:44:37.902033] Executing tasks
[HTTP Task] GET https://api.example.com/data
[2026-03-25 21:44:38.058005] Completing workflow
[2026-03-25 21:44:38.058294] Cleaning up resources
工作流ID: workflow_20260325_214437
执行ID: 6a20704a
状态: completed
开始时间: 2026-03-25 21:44:37.901099
结束时间: 2026-03-25 21:44:38.058084
总任务数: 4
完成任务数: 1
失败任务数: 0
重试任务数: 0
总执行时间: 0.16秒

示例2：从YAML文件加载工作流
--------------------------------------------------
[2026-03-25 21:44:38.072240] Initializing workflow: workflow_20260325_214438
[2026-03-25 21:44:38.072576] Validating workflow
[2026-03-25 21:44:38.072927] Planning execution order
[2026-03-25 21:44:38.073162] Executing tasks
[HTTP Task] POST https://api.example.com/orders/validate
[2026-03-25 21:44:38.522408] Completing workflow
[2026-03-25 21:44:38.522692] Cleaning up resources
工作流名称: E-commerce Order Processing
版本: 2.0.0
执行状态: completed
任务数量: 5

示例3：复杂并行工作流
--------------------------------------------------
[2026-03-25 21:44:38.534132] Initializing workflow: workflow_20260325_214438
[2026-03-25 21:44:38.534552] Validating workflow
[2026-03-25 21:44:38.535039] Planning execution order
[2026-03-25 21:44:38.535379] Executing tasks
[Data Transformation Task] Transforming data
[Data Transformation Task] Transforming data
[Data Transformation Task] Transforming data
[Data Transformation Task] Transforming data
[Data Transformation Task] Transforming data
[2026-03-25 21:44:38.743083] Completing workflow
[2026-03-25 21:44:38.743419] Cleaning up resources
工作流状态: completed
并行任务数: 5
总任务数: 6

======================================================================
工作流执行统计
======================================================================

======================================================================
工作流监控仪表板
======================================================================
总执行次数: 3
成功次数: 0
失败次数: 0
成功率: 0.0%
平均执行时间: 0.27秒

最后执行:
  工作流ID: workflow_20260325_214438
  执行ID: a6b21dab
  状态: WorkflowStatus.COMPLETED
  时间: 2026-03-25 21:44:38.743654

======================================================================
生产级工作流引擎案例总结:
1. 可配置的工作流定义和任务
2. 任务依赖管理和拓扑排序
3. 并行执行和资源管理
4. 错误处理和重试机制
5. 监控、日志和指标收集
6. 工作流版本控制和调度
======================================================================

演示高级功能:

1. 工作流版本管理:
可用版本: ['1.0.0', '2.0.0']
当前版本: 1.0.0

2. 工作流调度:
已调度工作流，执行时间: 2026-03-25 21:44:48.744856
调度状态: 1 个任务等待执行

3. 工作流可视化（简化）:

    Data Processing Pipeline:

    [fetch_data] → [query_database] → [transform_data] → [send_notification]
                    ↗
          [wait_task]

    依赖关系:
    - transform_data 依赖 fetch_data 和 query_database
    - send_notification 依赖 transform_data
    - query_database 依赖 fetch_data


生产级工作流引擎已成功演示！