python
# 基础版：状态快照+定向补偿（通用鲁棒性方案）
import redis  # 推荐用Redis缓存快照，支持快速回滚

# 1. 初始化快照缓存（通用配置）
snapshot_cache = redis.Redis(host='localhost', port=6379, db=0)
MAX_RETRY = 2  # 补偿最大重试次数

def agent_execution(sop_steps):
    """
    通用Agent执行函数，支持状态快照、定向补偿、断点续跑
    sop_steps: 列表，存储Agent的全流程SOP步骤（通用格式）
    """
    current_step = 0  # 当前执行步骤索引
    snapshot_data = {}  # 存储所有步骤的快照数据
    
    while current_step < len(sop_steps):
        step = sop_steps[current_step]
        step_name = step["name"]
        step_func = step["func"]  # 步骤执行函数（通用，可适配任意场景）
        step_params = step["params"]  # 步骤参数
        
        try:
            # 执行当前步骤
            step_result = step_func(** step_params)
            print(f"步骤{current_step+1} [{step_name}] 执行成功")
            
            # 生成当前步骤快照（留存中间成果，通用逻辑）
            snapshot_key = f"agent_snapshot_step_{current_step+1}"
            snapshot_data = {
                "step": current_step + 1,
                "result": step_result,  # 步骤执行成果（如处理后的数据）
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "params": step_params  # 步骤参数，用于回滚后复用
            }
            # 存入缓存（持久化，避免服务重启丢失）
            snapshot_cache.set(snapshot_key, json.dumps(snapshot_data))
            
            # 执行下一步
            current_step += 1
            
        except Exception as e:
            # 捕获异常，触发局部修正逻辑
            error_msg = str(e)
            print(f"步骤{current_step+1} [{step_name}] 执行失败：{error_msg}")
            
            # 1. 识别异常类型（通用判断逻辑，可扩展）
            if is_compensable_error(error_msg):
                # 可补偿异常（如内容违规、参数错误），执行定向补偿
                retry_count = 0
                while retry_count < MAX_RETRY:
                    try:
                        # 回滚至前一步快照（核心逻辑）
                        if current_step == 0:
                            # 第一步失败，无前置快照，直接重试当前步骤
                            step_params = adjust_params(step_params, error_msg)  # 定向调整参数
                            continue
                        else:
                            # 回滚至前一步快照，复用前序成果
                            prev_snapshot_key = f"agent_snapshot_step_{current_step}"
                            prev_snapshot = json.loads(snapshot_cache.get(prev_snapshot_key))
                            # 基于前序成果，定向修正异常
                            corrected_params = correct_error(prev_snapshot["result"], error_msg)
                            # 更新当前步骤参数，重新执行
                            step["params"] = corrected_params
                            retry_count += 1
                            continue
                    except:
                        retry_count += 1
                # 补偿失败，推送人工提醒
                if retry_count >= MAX_RETRY:
                    send_alert(f"步骤{current_step+1} 补偿失败，请人工介入：{error_msg}")
                    break
            else:
                # 不可补偿异常，直接推送人工提醒
                send_alert(f"步骤{current_step+1} 执行失败（不可补偿）：{error_msg}")
                break

# 通用辅助函数（可直接复用，适配所有场景）
def is_compensable_error(error_msg):
    """判断异常是否可补偿（通用逻辑，可扩展异常类型）"""
    compensable_errors = ["违规", "敏感词", "参数错误", "格式错误", "同义词替换"]
    return any(err in error_msg for err in compensable_errors)

def correct_error(prev_result, error_msg):
    """定向修正异常（通用逻辑，可根据具体场景扩展）"""
    # 示例：敏感词/违规内容修正（通用，适配文本类校验异常）
    if "敏感词" in error_msg or "违规" in error_msg:
        sensitive_words = extract_sensitive_words(error_msg)  # 提取异常关键词
        for word in sensitive_words:
            prev_result = prev_result.replace(word, get_synonym(word))  # 同义词替换修正
    # 示例：参数错误修正（通用）
    elif "参数错误" in error_msg:
        prev_result["params"] = adjust_params(prev_result["params"], error_msg)
    return prev_result

# 示例：辅助理解——跨境电商场景的函数适配（仅示例，非核心）
def extract_sensitive_words(error_msg):
    """提取敏感词（示例函数，可替换为任意场景的异常提取逻辑）"""
    return re.findall(r"敏感词：([^，。]+)", error_msg)
    

python
# 进阶版：分层反思机制+动态分支规划（通用鲁棒性方案）
import redis
from typing import List, Dict

# 1. 初始化：通用异常分类库（核心，可扩展至任意行业异常）
EXCEPTION_CLASSIFY = {
    "临时可恢复异常": {
        "keywords": ["接口超时", "网络波动", "临时维护", "卡顿"],
        "strategy": "retry",  # 处理策略：原地重试
        "params": {"max_retry": 3, "interval": 30}  # 重试参数（秒）
    },
    "可补偿异常": {
        "keywords": ["敏感词", "违规", "参数错误", "格式错误"],
        "strategy": "local_correction"  # 处理策略：局部修正
    },
    "前提变更异常": {
        "keywords": ["规则变更", "工具下线", "备货周期变更", "渠道停运"],
        "strategy": "branch_replan"  # 处理策略：分支重规划
    },
    "无法解决异常": {
        "keywords": ["停产", "严重侵权", "账号受限", "权限不足"],
        "strategy": "terminate"  # 处理策略：终止任务
    }
}

# 2. 反思模块（通用，核心决策逻辑）
class ReflectionModule:
    def __init__(self, exception_classify: Dict):
        self.exception_classify = exception_classify
    
    def classify_exception(self, error_msg: str) -> Dict:
        """通用异常分类：根据错误信息，匹配异常类型与处理策略"""
        for exception_type, config in self.exception_classify.items():
            if any(keyword in error_msg for keyword in config["keywords"]):
                return {"exception_type": exception_type, "strategy": config["strategy"], "params": config.get("params", {})}
        # 未匹配到分类，默认推送人工介入
        return {"exception_type": "未知异常", "strategy": "terminate", "params": {}}
    
    def decide_strategy(self, error_msg: str, current_step: int, snapshot_data: Dict) -> Dict:
        """通用决策逻辑：根据异常类型、当前步骤、快照数据，确定处理方案"""
        classify_result = self.classify_exception(error_msg)
        strategy = classify_result["strategy"]
        params = classify_result["params"]
        
        if strategy == "retry":
            return {"strategy": "retry", "max_retry": params["max_retry"], "interval": params["interval"]}
        elif strategy == "local_correction":
            # 定向修正，复用前序快照
            prev_step = current_step - 1 if current_step > 0 else 0
            prev_snapshot = snapshot_data.get(f"step_{prev_step}", {})
            return {
                "strategy": "local_correction",
                "prev_snapshot": prev_snapshot,
                "error_msg": error_msg
            }
        elif strategy == "branch_replan":
            # 分支重规划，仅重跑异常及后续关联步骤
            return {
                "strategy": "branch_replan",
                "current_step": current_step,
                "snapshot_data": snapshot_data,
                "error_msg": error_msg
            }
        else:
            return {"strategy": "terminate", "error_msg": error_msg}

# 3. 通用Agent执行函数（集成反思模块）
def advanced_agent_execution(sop_steps: List[Dict]):
    snapshot_cache = redis.Redis(host='localhost', port=6379, db=0)
    reflection_module = ReflectionModule(EXCEPTION_CLASSIFY)  # 初始化反思模块
    current_step = 0
    snapshot_data = {}  # 存储所有步骤快照
    
    while current_step < len(sop_steps):
        step = sop_steps[current_step]
        step_name = step["name"]
        step_func = step["func"]
        step_params = step["params"]
        
        try:
            # 执行当前步骤，生成快照（同基础版，通用逻辑）
            step_result = step_func(** step_params)
            snapshot_key = f"step_{current_step}"
            snapshot_data[snapshot_key] = {
                "result": step_result,
                "params": step_params
            }
            snapshot_cache.set(snapshot_key, json.dumps(snapshot_data[snapshot_key]))
            print(f"步骤{current_step+1} [{step_name}] 执行成功")
            current_step += 1
            
        except Exception as e:
            error_msg = str(e)
            print(f"步骤{current_step+1} [{step_name}] 执行失败：{error_msg}")
            
            # 反思模块决策，获取处理策略（通用核心）
            strategy = reflection_module.decide_strategy(error_msg, current_step, snapshot_data)
            
            # 执行对应处理策略（通用逻辑）
            if strategy["strategy"] == "retry":
                # 原地重试
                for i in range(strategy["max_retry"]):
                    time.sleep(strategy["interval"])
                    try:
                        step_result = step_func(** step_params)
                        print(f"重试{i+1}次，步骤{current_step+1}执行成功")
                        # 更新快照，执行下一步
                        snapshot_data[f"step_{current_step}"] = {"result": step_result, "params": step_params}
                        snapshot_cache.set(f"step_{current_step}", json.dumps(snapshot_data[f"step_{current_step}"]))
                        current_step += 1
                        break
                    except:
                        continue
                else:
                    send_alert(f"步骤{current_step+1} 重试{strategy['max_retry']}次失败，请人工介入")
                    break
            
            elif strategy["strategy"] == "local_correction":
                # 局部修正，断点续跑
                prev_snapshot = strategy["prev_snapshot"]
                corrected_params = correct_error(prev_snapshot["result"], error_msg)  # 通用修正函数
                step["params"] = corrected_params
                continue  # 重新执行当前步骤
            
            elif strategy["strategy"] == "branch_replan":
                # 分支重规划，生成新的后续步骤（通用逻辑）
                current_step_idx = strategy["current_step"]
                # 基于前序快照，重新规划当前及后续步骤
                new_sop_steps = replan_steps(
                    sop_steps, current_step_idx, strategy["snapshot_data"], error_msg
                )
                # 替换原SOP，从当前步骤重新执行
                sop_steps = new_sop_steps
                continue
            
            else:
                # 终止任务，推送人工介入
                send_alert(f"步骤{current_step+1} 执行失败，无法解决：{error_msg}")
                break

# 通用辅助函数（可扩展）
def replan_steps(original_sop: List[Dict], current_step: int, snapshot_data: Dict, error_msg: str) -> List[Dict]:
    """通用分支重规划函数，根据异常调整后续步骤（可适配任意场景）"""
    new_sop = original_sop[:current_step]  # 保留前序成功步骤
    # 基于异常信息，重新生成当前及后续步骤（示例逻辑，可扩展）
    current_step_info = original_sop[current_step]
    if "规则变更" in error_msg:
        # 规则变更，更新校验步骤参数
        current_step_info["params"]["rule_version"] = "latest"
    elif "工具下线" in error_msg:
        # 工具下线，替换核心执行函数
        current_step_info["func"] = alternative_tool_func  # 备选工具函数
    # 追加重新规划的步骤
    new_sop.append(current_step_info)
    new_sop.extend(original_sop[current_step+1:])
    return new_sop
    

python
# 高级版：PEV闭环+修正边界控制（通用鲁棒性方案）
import redis
import time

# 1. 通用修正边界配置（核心，可适配所有场景）
CORRECTION_BOUNDARY = {
    "max_correction_times": 2,  # 单步骤最大修正次数（避免过度修正）
    "max_resource_cost": 100,   # 最大资源消耗阈值（Token/时间，单位：Token）
    "abandon_threshold": 0.5,   # 修正成本/重新执行成本 > 0.5，直接弃用局部修正
}

# 2. PEV闭环核心类（通用，可复用）
class PEVAgent:
    def __init__(self, exception_classify: Dict, correction_boundary: Dict):
        self.reflection_module = ReflectionModule(exception_classify)  # 复用反思模块
        self.boundary = correction_boundary
        self.snapshot_cache = redis.Redis(host='localhost', port=6379, db=0)
        self.resource_cost = 0  # 累计资源消耗（Token）
        self.correction_record = {}  # 修正记录，key：步骤索引，value：修正次数
    
    def plan(self, base_sop: List[Dict]) -> List[Dict]:
        """Plan：生成初始SOP+通用分支预案（基于历史异常数据）"""
        # 通用分支预案（可扩展，适配所有场景）
        branch_plans = {
            "tool_error": self._tool_error_plan,  # 工具异常预案
            "rule_error": self._rule_error_plan,  # 规则异常预案
            "param_error": self._param_error_plan  # 参数异常预案
        }
        # 为初始SOP添加预案关联
        for step in base_sop:
            step["branch_plans"] = branch_plans
        return base_sop
    
    def execute(self, step: Dict) -> tuple[bool, any]:
        """Execute：执行步骤，返回执行结果（成功/失败）、执行成果"""
        step_func = step["func"]
        step_params = step["params"]
        try:
            self.resource_cost += 10  # 模拟消耗Token（每步固定消耗）
            result = step_func(** step_params)
            return True, result
        except Exception as e:
            self.resource_cost += 5  # 异常执行额外消耗Token
            return False, str(e)
    
    def verify(self, step: Dict, execute_result: any, step_index: int) -> tuple[bool, Dict]:
        """Verify：双重校验（系统校验+反思校验），通用逻辑"""
        # 1. 系统校验：执行结果是否符合预期
        system_verify = self._system_verify(step, execute_result)
        if not system_verify:
            error_msg = f"系统校验失败：执行结果不符合预期"
            return False, {"error_msg": error_msg, "verify_type": "system"}
        
        # 2. 反思校验：是否存在潜在异常（如资源消耗过高、参数隐患）
        reflection_verify = self._reflection_verify(step, execute_result, step_index)
        if not reflection_verify:
            error_msg = f"反思校验失败：存在潜在异常"
            return False, {"error_msg": error_msg, "verify_type": "reflection"}
        
        return True, {"status": "success"}
    
    def _system_verify(self, step: Dict, result: any) -> bool:
        """通用系统校验逻辑（可扩展，适配任意场景）"""
        # 示例：校验执行结果非空、格式正确
        if result is None:
            return False
        if step["name"] in ["数据处理/转换", "规则/风控校验"]:
            return isinstance(result, (dict, str, list))  # 校验数据格式
        return True
    
    def _reflection_verify(self, step: Dict, result: any, step_index: int) -> bool:
        """通用反思校验逻辑（可扩展）"""
        # 1. 校验资源消耗是否超出阈值
        if self.resource_cost > self.boundary["max_resource_cost"]:
            return False
        # 2. 校验当前步骤修正次数是否超出上限
        correction_times = self.correction_record.get(step_index, 0)
        if correction_times > self.boundary["max_correction_times"]:
            return False
        return True
    
    def control_boundary(self, step_index: int, error_msg: str) -> str:
        """通用修正边界控制：判断是否继续修正（核心逻辑）"""
        correction_times = self.correction_record.get(step_index, 0)
        # 1. 修正次数超出上限，终止修正
        if correction_times >= self.boundary["max_correction_times"]:
            return "terminate"
        # 2. 计算修正成本与重新执行成本的比例
        correction_cost = 10  # 模拟单次修正成本
        retry_all_cost = (step_index + 1) * 10  # 重新执行前序所有步骤的成本
        if correction_cost / retry_all_cost > self.boundary["abandon_threshold"]:
            # 修正成本过高，放弃局部修正
            return "abandon"
        # 3. 边界内，可继续修正
        return "continue"
    
    # 通用分支预案（可扩展）
    def _tool_error_plan(self, step: Dict) -> Dict:
        """工具异常通用预案：替换备选工具"""
        step["func"] = alternative_tool_func  # 备选工具函数
        return step
    
    def _rule_error_plan(self, step: Dict) -> Dict:
        """规则异常通用预案：更新至最新规则"""
        step["params"]["rule_version"] = "latest"
        return step
    
    def _param_error_plan(self, step: Dict) -> Dict:
        """参数异常通用预案：自动校准参数"""
        step["params"] = adjust_params(step["params"])
        return step
    
    def run(self, base_sop: List[Dict]):
        """通用Agent执行入口（集成PEV闭环、边界控制）"""
        sop_steps = self.plan(base_sop)  # Plan：生成初始SOP+预案
        current_step = 0
        
        while current_step < len(sop_steps):
            step = sop_steps[current_step]
            step_name = step["name"]
            print(f"开始执行步骤{current_step+1} [{step_name}]")
            
            # Execute：执行步骤
            execute_success, execute_result = self.execute(step)
            if not execute_success:
                error_msg = execute_result
                verify_result = (False, {"error_msg": error_msg, "verify_type": "execute"})
            else:
                # Verify：双重校验
                verify_success, verify_result = self.verify(step, execute_result, current_step)
                if verify_success:
                    # 校验通过，生成快照，执行下一步
                    self._save_snapshot(current_step, execute_result, step["params"])
                    print(f"步骤{current_step+1} 执行、校验均通过")
                    current_step += 1
                    continue
                else:
                    error_msg = verify_result["error_msg"]
            
            # 异常处理：边界控制+策略执行
            boundary_result = self.control_boundary(current_step, error_msg)
            if boundary_result == "terminate":
                send_alert(f"步骤{current_step+1} 修正次数超出上限，终止任务")
                break
            elif boundary_result == "abandon":
                send_alert(f"步骤{current_step+1} 修正成本过高，建议重新执行")
                break
            
            # 边界内，执行修正/重规划
            self.correction_record[current_step] = self.correction_record.get(current_step, 0) + 1
            strategy = self.reflection_module.decide_strategy(error_msg, current_step, self._load_snapshots())
            if strategy["strategy"] == "local_correction":
                # 局部修正
                corrected_step = correct_error(strategy["prev_snapshot"]["result"], error_msg)
                sop_steps[current_step]["params"] = corrected_step["params"]
            elif strategy["strategy"] == "branch_replan":
                # 分支重规划（复用预案）
                if "工具异常" in error_msg:
                    sop_steps[current_step] = step["branch_plans"]["tool_error"](step)
                elif "规则异常" in error_msg:
                    sop_steps[current_step] = step["branch_plans"]["rule_error"](step)
            
            print(f"步骤{current_step+1} 修正完成，重新执行")
    
    def _save_snapshot(self, step_index: int, result: any, params: Dict):
        """通用快照保存逻辑"""
        snapshot_key = f"pev_snapshot_step_{step_index}"
        snapshot_data = {"result": result, "params": params, "cost": self.resource_cost}
        self.snapshot_cache.set(snapshot_key, json.dumps(snapshot_data))
    
    def _load_snapshots(self) -> Dict:
        """通用快照加载逻辑"""
        snapshot_data = {}
        keys = self.snapshot_cache.keys()
        for key in keys:
            step_index = int(key.decode().split("_")[-1])
            snapshot_data[f"step_{step_index}"] = json.loads(self.snapshot_cache.get(key))
        return snapshot_data

# 通用执行示例（可直接复用）
if __name__ == "__main__":
    # 定义通用SOP（适配任意场景）
    base_sop = [
        {
            "name": "数据/指令抓取",
            "func": data_extract,  # 通用数据抓取函数（可替换）
            "params": {"source": "interface", "url": "https://example.com/api/data"}
        },
        {
            "name": "数据处理/转换",
            "func": data_process,  # 通用数据处理函数（可替换）
            "params": {"format": "json"}
        },
        # 可继续添加其他步骤...
    ]
    # 初始化PEV Agent，执行任务
    pev_agent = PEVAgent(EXCEPTION_CLASSIFY, CORRECTION_BOUNDARY)
    pev_agent.run(base_sop)
    

异常类型

具体场景示例

处理建议

判断依据

临时可恢复异常

接口超时、网络波动、平台/工具临时维护、卡顿

原地重试（最多3次，间隔30-60秒）

异常不影响核心前提（如规则、工具、数据），且可在短时间内恢复

可补偿异常

敏感词命中、参数错误、格式错误、文本不通顺

局部修正（定向修改内容/参数）+ 断点续跑

异常仅影响当前步骤，不改变任务核心目标，修正成本低于重新执行前序步骤

前提变更异常

规则迭代、工具下线、核心参数变更、依赖故障

分支重规划（重执行异常节点及后续关联步骤）

异常导致任务前提条件变更，但核心目标未消失，局部修正无法解决

无法解决异常

工具永久下线、权限不足、核心数据缺失、严重违规

终止任务 + 推送人工介入提醒

异常导致任务核心目标无法实现，或修正成本远超重新启动任务