53：Agentic状态机设计：实现可控的复杂任务执行

class State:
    """状态类"""
    def __init__(self, name, description):
        self.name = name
        self.description = description
        self.actions = []
        self.transitions = {}
    
    def add_action(self, action):
        """添加动作"""
        self.actions.append(action)
    
    def add_transition(self, event, next_state):
        """添加转换规则"""
        self.transitions[event] = next_state
    
    def execute(self, context):
        """执行状态动作"""
        for action in self.actions:
            action.execute(context)
        return self

class Event:
    """事件类"""
    def __init__(self, name, data=None):
        self.name = name
        self.data = data

class Action:
    """动作类"""
    def execute(self, context):
        """执行动作"""
        pass

class StateMachine:
    """状态机类"""
    def __init__(self, initial_state):
        self.initial_state = initial_state
        self.current_state = initial_state
        self.context = {}
        self.history = []
    
    def set_context(self, context):
        """设置上下文"""
        self.context = context
    
    def send_event(self, event):
        """发送事件"""
        # 记录当前状态和事件
        self.history.append({
            "state": self.current_state.name,
            "event": event.name,
            "timestamp": self._get_current_time()
        })
        
        # 检查是否有对应的转换规则
        if event.name in self.current_state.transitions:
            # 执行当前状态的退出动作
            self._execute_exit_actions()
            
            # 转换到下一状态
            next_state_name = self.current_state.transitions[event.name]
            self.current_state = self._get_state_by_name(next_state_name)
            
            # 执行下一状态的进入动作
            self.current_state.execute(self.context)
            
            return True
        else:
            # 没有对应的转换规则
            return False
    
    def _execute_exit_actions(self):
        """执行退出动作"""
        # 执行当前状态的退出动作
        pass
    
    def _get_state_by_name(self, state_name):
        """根据名称获取状态"""
        # 这里应该实现状态查找逻辑
        pass
    
    def _get_current_time(self):
        """获取当前时间"""
        import datetime
        return datetime.datetime.now().isoformat()
    
    def get_current_state(self):
        """获取当前状态"""
        return self.current_state
    
    def get_history(self):
        """获取历史记录"""
        return self.history

class TaskStateMachine(StateMachine):
    """任务状态机"""
    def __init__(self):
        # 定义状态
        initial = State("initial", "初始状态")
        planning = State("planning", "规划状态")
        executing = State("executing", "执行状态")
        reflecting = State("reflecting", "反思状态")
        completed = State("completed", "完成状态")
        failed = State("failed", "失败状态")
        
        # 定义动作
        class PlanningAction(Action):
            def execute(self, context):
                print("执行规划动作")
                # 实现规划逻辑
                context["plan"] = {"steps": ["step1", "step2", "step3"]}
        
        class ExecutingAction(Action):
            def execute(self, context):
                print("执行执行动作")
                # 实现执行逻辑
                context["execution_result"] = {"success": True, "data": "执行结果"}
        
        class ReflectingAction(Action):
            def execute(self, context):
                print("执行反思动作")
                # 实现反思逻辑
                context["reflection"] = {"evaluation": "成功", "improvements": []}
        
        # 添加动作到状态
        planning.add_action(PlanningAction())
        executing.add_action(ExecutingAction())
        reflecting.add_action(ReflectingAction())
        
        # 定义转换规则
        initial.add_transition("start", "planning")
        planning.add_transition("plan_complete", "executing")
        planning.add_transition("plan_failed", "failed")
        executing.add_transition("execute_complete", "reflecting")
        executing.add_transition("execute_failed", "failed")
        reflecting.add_transition("reflect_complete", "completed")
        reflecting.add_transition("reflect_failed", "failed")
        failed.add_transition("retry", "planning")
        completed.add_transition("restart", "planning")
        
        # 初始化状态机
        super().__init__(initial)
        
        # 存储所有状态
        self.states = {
            "initial": initial,
            "planning": planning,
            "executing": executing,
            "reflecting": reflecting,
            "completed": completed,
            "failed": failed
        }
    
    def _get_state_by_name(self, state_name):
        """根据名称获取状态"""
        return self.states.get(state_name, self.initial_state)

# 状态机配置示例
state_machine_config = {
    "states": [
        {
            "name": "initial",
            "description": "初始状态",
            "actions": [],
            "transitions": {
                "start": "planning"
            }
        },
        {
            "name": "planning",
            "description": "规划状态",
            "actions": ["PlanningAction"],
            "transitions": {
                "plan_complete": "executing",
                "plan_failed": "failed"
            }
        },
        {
            "name": "executing",
            "description": "执行状态",
            "actions": ["ExecutingAction"],
            "transitions": {
                "execute_complete": "reflecting",
                "execute_failed": "failed"
            }
        },
        {
            "name": "reflecting",
            "description": "反思状态",
            "actions": ["ReflectingAction"],
            "transitions": {
                "reflect_complete": "completed",
                "reflect_failed": "failed"
            }
        },
        {
            "name": "completed",
            "description": "完成状态",
            "actions": [],
            "transitions": {
                "restart": "planning"
            }
        },
        {
            "name": "failed",
            "description": "失败状态",
            "actions": [],
            "transitions": {
                "retry": "planning"
            }
        }
    ],
    "initial_state": "initial"
}

# 从配置创建状态机
def create_state_machine_from_config(config):
    """从配置创建状态机"""
    states = {}
    
    # 创建状态
    for state_config in config["states"]:
        state = State(state_config["name"], state_config["description"])
        # 添加动作
        for action_name in state_config.get("actions", []):
            # 这里应该根据动作名称创建动作实例
            # 简化处理，实际应该有动作注册表
            action = Action()
            state.add_action(action)
        # 添加转换规则
        for event, next_state in state_config.get("transitions", {}).items():
            state.add_transition(event, next_state)
        states[state_config["name"]] = state
    
    # 创建状态机
    initial_state = states.get(config["initial_state"])
    state_machine = StateMachine(initial_state)
    state_machine.states = states
    
    return state_machine

class ConditionalTransition:
    """条件转移"""
    def __init__(self, event, condition, next_state):
        self.event = event
        self.condition = condition
        self.next_state = next_state
    
    def should_transition(self, context):
        """判断是否应该转移"""
        return self.condition(context)

class StateWithConditionalTransitions(State):
    """带条件转移的状态"""
    def __init__(self, name, description):
        super().__init__(name, description)
        self.conditional_transitions = []
    
    def add_conditional_transition(self, conditional_transition):
        """添加条件转移"""
        self.conditional_transitions.append(conditional_transition)
    
    def get_next_state(self, event, context):
        """获取下一状态"""
        # 首先检查普通转移
        if event.name in self.transitions:
            return self.transitions[event.name]
        
        # 然后检查条件转移
        for conditional_transition in self.conditional_transitions:
            if conditional_transition.event == event.name and conditional_transition.should_transition(context):
                return conditional_transition.next_state
        
        # 没有找到转移规则
        return None

class ExceptionHandler:
    """异常处理器"""
    def handle(self, exception, context):
        """处理异常"""
        pass

class RetryExceptionHandler(ExceptionHandler):
    """重试异常处理器"""
    def __init__(self, max_retries=3, retry_interval=1):
        self.max_retries = max_retries
        self.retry_interval = retry_interval
    
    def handle(self, exception, context):
        """处理异常"""
        retry_count = context.get("retry_count", 0)
        if retry_count < self.max_retries:
            context["retry_count"] = retry_count + 1
            import time
            time.sleep(self.retry_interval)
            return "retry"
        else:
            return "fail"

class FallbackExceptionHandler(ExceptionHandler):
    """降级异常处理器"""
    def __init__(self, fallback_state):
        self.fallback_state = fallback_state
    
    def handle(self, exception, context):
        """处理异常"""
        # 执行降级逻辑
        context["fallback"] = True
        return self.fallback_state

class StateMachineWithExceptionHandling(StateMachine):
    """带异常处理的状态机"""
    def __init__(self, initial_state):
        super().__init__(initial_state)
        self.exception_handlers = {}
    
    def add_exception_handler(self, exception_type, handler):
        """添加异常处理器"""
        self.exception_handlers[exception_type] = handler
    
    def execute_action(self, action, context):
        """执行动作，处理异常"""
        try:
            action.execute(context)
            return True
        except Exception as e:
            # 查找对应的异常处理器
            exception_type = type(e).__name__
            handler = self.exception_handlers.get(exception_type, None)
            
            if handler:
                result = handler.handle(e, context)
                if result == "retry":
                    # 重试动作
                    return self.execute_action(action, context)
                elif result == "fail":
                    # 失败
                    return False
                else:
                    # 转移到指定状态
                    self.current_state = self._get_state_by_name(result)
                    return False
            else:
                # 没有对应的异常处理器，默认失败
                return False

class StateMachineWithAntiLoop(StateMachine):
    """带防循环机制的状态机"""
    def __init__(self, initial_state, max_transitions=100, timeout=300):
        super().__init__(initial_state)
        self.max_transitions = max_transitions
        self.timeout = timeout
        self.start_time = self._get_current_time()
        self.transition_count = 0
        self.state_history = []
    
    def send_event(self, event):
        """发送事件，防止无限循环"""
        # 检查是否超过最大转移次数
        if self.transition_count >= self.max_transitions:
            raise Exception("Maximum number of transitions reached")
        
        # 检查是否超时
        current_time = self._get_current_time()
        if (current_time - self.start_time).total_seconds() > self.timeout:
            raise Exception("State machine timeout")
        
        # 检查是否出现状态循环
        current_state_name = self.current_state.name
        if current_state_name in self.state_history[-5:]:  # 检查最近5个状态
            raise Exception("State loop detected")
        
        # 记录当前状态
        self.state_history.append(current_state_name)
        
        # 增加转移计数
        self.transition_count += 1
        
        # 调用父类方法
        return super().send_event(event)
    
    def _get_current_time(self):
        """获取当前时间"""
        import datetime
        return datetime.datetime.now()

class OptimizedStateMachine(StateMachine):
    """优化的状态机"""
    def __init__(self, initial_state):
        super().__init__(initial_state)
        self.state_cache = {}
        self.action_executors = {}
    
    def _get_state_by_name(self, state_name):
        """优化的状态查找"""
        if state_name in self.state_cache:
            return self.state_cache[state_name]
        else:
            state = super()._get_state_by_name(state_name)
            self.state_cache[state_name] = state
            return state
    
    def execute_action(self, action, context):
        """优化的动作执行"""
        action_id = id(action)
        if action_id not in self.action_executors:
            # 创建动作执行器
            import concurrent.futures
            executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
            self.action_executors[action_id] = executor
        
        # 异步执行动作
        executor = self.action_executors[action_id]
        future = executor.submit(action.execute, context)
        return future.result()
    
    def shutdown(self):
        """关闭状态机"""
        # 关闭所有动作执行器
        for executor in self.action_executors.values():
            executor.shutdown()