依托Harness工程管控+Skills技能封装，完整实现AI产品稳定研发与规模化落地应用22.9

"""
Harness + Skills 大模型工程化落地完整示例
核心能力复刻本文全部理论：权限驾驭、参数驾驭、频次驾驭、结果驾驭、重试驾驭、状态驾驭
适配大模型AI产品标准化研发、可复用、可迭代落地
"""
import time
import random
from typing import Dict, Any, Optional, List

# ===================== 全局状态记忆模块（状态驾驭）=====================
# 实现短期会话记忆 + 任务断点状态存储
class SessionMemory:
    def __init__(self):
        self.session_cache: Dict[str, Dict[str, Any]] = {}

    def save_progress(self, session_id: str, key: str, value: Any) -> None:
        """保存任务进度与上下文，支持断点续跑"""
        if session_id not in self.session_cache:
            self.session_cache[session_id] = {}
        self.session_cache[session_id][key] = value

    def get_progress(self, session_id: str, key: str) -> Optional[Any]:
        """读取历史任务状态"""
        return self.session_cache.get(session_id, {}).get(key, None)

# ===================== Harness 核心管控模块（六层驾驭）=====================
class HarnessEngine:
    # 频次驾驭：单会话最大调用次数限制
    MAX_CALL_TIMES = 10
    # 重试驾驭：最大重试次数
    MAX_RETRY_TIMES = 3

    def __init__(self):
        self.memory = SessionMemory()
        self.call_count: Dict[str, int] = {}
        # 高危操作权限白名单
        self.safe_permission_list = ["user_basic", "user_admin"]

    # 1. 权限驾驭：分级权限校验
    def permission_control(self, session_id: str, permission: str, is_high_risk: bool = False) -> Dict[str, Any]:
        risk_label = "【高危操作】" if is_high_risk else "【普通操作】"
        print(f"  >> [权限驾驭]{risk_label} 校验权限级别：{permission}，白名单：{self.safe_permission_list}")
        if is_high_risk and permission not in self.safe_permission_list:
            print(f"  >> [权限驾驭] ✗ 权限不足，高危操作需管理员权限")
            return {"status": False, "msg": "权限不足，高危操作需管理员权限", "data": None}
        print(f"  >> [权限驾驭] ✓ 权限校验通过")
        return {"status": True, "msg": "权限校验通过", "data": None}

    # 2. 频次驾驭：防止无限循环调用
    def frequency_control(self, session_id: str) -> Dict[str, Any]:
        count = self.call_count.get(session_id, 0) + 1
        print(f"  >> [频次驾驭] 当前会话第 {count} 次调用，上限 {self.MAX_CALL_TIMES} 次")
        if count > self.MAX_CALL_TIMES:
            print(f"  >> [频次驾驭] ✗ 调用频次超限，终止任务")
            return {"status": False, "msg": "调用频次超限，终止任务", "data": None}
        self.call_count[session_id] = count
        print(f"  >> [频次驾驭] ✓ 频次校验通过")
        return {"status": True, "msg": "频次校验通过", "data": None}

    # 3. 参数驾驭：参数合法性、完整性校验
    def params_verify(self, params: Dict[str, Any], need_keys: List[str]) -> Dict[str, Any]:
        print(f"  >> [参数驾驭] 必需参数清单：{need_keys}，传入参数：{list(params.keys())}")
        for key in need_keys:
            if key not in params or not params[key]:
                print(f"  >> [参数驾驭] ✗ 参数缺失：{key}")
                return {"status": False, "msg": f"参数缺失：{key}", "data": None}
        print(f"  >> [参数驾驭] ✓ 参数完整性校验通过")
        return {"status": True, "msg": "参数校验通过", "data": params}

    # 4. 重试驾驭：阶梯式失败重试
    def retry_exec(self, func, *args, **kwargs) -> Dict[str, Any]:
        print(f"  >> [重试驾驭] 开始执行核心逻辑，最大重试次数：{self.MAX_RETRY_TIMES}")
        retry_num = 0
        while retry_num < self.MAX_RETRY_TIMES:
            try:
                result = func(*args, **kwargs)
                if retry_num > 0:
                    print(f"  >> [重试驾驭] ✓ 第 {retry_num} 次重试成功")
                else:
                    print(f"  >> [重试驾驭] ✓ 首次执行成功")
                return result
            except Exception as e:
                retry_num += 1
                wait_sec = 0.5 * retry_num
                print(f"  >> [重试驾驭] ✗ 第 {retry_num} 次执行失败：{e}，等待 {wait_sec:.1f}s 后重试")
                time.sleep(wait_sec)
                if retry_num >= self.MAX_RETRY_TIMES:
                    print(f"  >> [重试驾驭] ✗ 已达最大重试次数 {self.MAX_RETRY_TIMES}，放弃重试")
                    return {"status": False, "msg": f"任务重试{self.MAX_RETRY_TIMES}次失败：{str(e)}", "data": None}

    # 5. 结果驾驭：结构化清洗、标准化输出
    def result_clean(self, raw_result: str) -> Dict[str, Any]:
        print(f"  >> [结果驾驭] 原始输出长度：{len(raw_result)} 字符，开始结构化清洗")
        # 简单清洗：去除空格、换行，适配结构化场景
        clean_data = raw_result.strip().replace("\n", "").replace(" ", "")
        if not clean_data:
            print(f"  >> [结果驾驭] ✗ 模型输出为空，结果无效")
            return {"status": False, "msg": "模型输出为空，结果无效", "data": None}
        print(f"  >> [结果驾驭] ✓ 清洗完成，输出长度：{len(clean_data)} 字符")
        return {"status": True, "msg": "结果清洗完成", "data": clean_data}

    # 6. 全链路日志观测
    def log_record(self, session_id: str, skill_name: str, res: Dict[str, Any]) -> None:
        log_info = {
            "session_id": session_id,
            "skill": skill_name,
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "result": res
        }
        print(f"  >> [全链路日志] {log_info}")

# ===================== 标准化Skills技能封装（单一职责、可复用、可迭代）=====================
class DataAnalysisSkill:
    """
    数据分析标准化Skill
    符合本文拆解原则：单一职责、通用场景、粒度适中、低耦合
    能力：接收业务文本数据，完成统计分析与结论生成
    """
    # 技能固定入参规范
    NEED_PARAMS = ["data_content", "analyse_target"]

    def __init__(self, harness: HarnessEngine):
        self.harness = harness
        self.skill_name = "business_data_analysis"

    def exec(self, session_id: str, permission: str, params: Dict[str, Any]) -> Dict[str, Any]:
        print(f"\n{'─'*50}")
        print(f"【Skill执行】{self.skill_name} | 会话：{session_id}")
        print(f"{'─'*50}")

        # 步骤1：频次校验
        print(f"[步骤1/7] 频次校验（Harness·频次驾驭层）")
        freq_check = self.harness.frequency_control(session_id)
        if not freq_check["status"]:
            print(f"  ✗ 频次校验未通过，任务中止\n")
            return freq_check

        # 步骤2：参数校验
        print(f"[步骤2/7] 参数校验（Harness·参数驾驭层）")
        param_check = self.harness.params_verify(params, self.NEED_PARAMS)
        if not param_check["status"]:
            print(f"  ✗ 参数校验未通过，任务中止\n")
            return param_check

        # 步骤3：权限校验（本技能含数据分析导出，判定为高危操作）
        print(f"[步骤3/7] 权限校验（Harness·权限驾驭层）")
        perm_check = self.harness.permission_control(session_id, permission, is_high_risk=True)
        if not perm_check["status"]:
            print(f"  ✗ 权限校验未通过，任务中止\n")
            return perm_check

        # 步骤4：重试机制执行核心技能逻辑
        print(f"[步骤4/7] 核心逻辑执行（Harness·重试驾驭层）")
        def core_logic():
            # 模拟大模型分析推理逻辑
            data = params["data_content"]
            target = params["analyse_target"]
            # 模拟模型输出，统一返回Dict以匹配retry_exec的预期结构
            raw_output = f"基于数据【{data}】，针对目标【{target}】完成分析：数据合规，核心指标正常，业务运行稳定"
            return {"status": True, "msg": "分析完成", "data": raw_output}

        raw_res = self.harness.retry_exec(core_logic)
        if not raw_res["status"]:
            print(f"  ✗ 核心逻辑执行失败，任务中止\n")
            return raw_res

        # 步骤5：结果结构化清洗
        print(f"[步骤5/7] 结果清洗（Harness·结果驾驭层）")
        final_res = self.harness.result_clean(raw_res["data"])

        # 步骤6：状态留存，支持断点续跑
        print(f"[步骤6/7] 状态留存（Harness·状态驾驭层）")
        self.harness.memory.save_progress(session_id, self.skill_name, final_res)
        print(f"  >> [状态驾驭] ✓ 任务进度已保存，会话 {session_id} 支持断点续跑")

        # 步骤7：全链路日志记录
        print(f"[步骤7/7] 全链路日志记录")
        self.harness.log_record(session_id, self.skill_name, final_res)

        print(f"\n{'─'*50}")
        print(f"【Skill执行完毕】{self.skill_name} → 状态：{final_res['status']}")
        print(f"{'─'*50}\n")
        return final_res

# ===================== 工程调度入口（Planner 全局调度）=====================
class AIAssistantPlanner:
    def __init__(self):
        # 初始化工程管控引擎
        self.harness = HarnessEngine()
        # 注册所有标准化Skills
        self.skill_map = {
            "data_analysis": DataAnalysisSkill(self.harness)
        }

    def run_task(self, session_id: str, skill_key: str, permission: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """统一任务调度入口"""
        if skill_key not in self.skill_map:
            return {"status": False, "msg": "不存在该技能，任务调度失败", "data": None}
        skill = self.skill_map[skill_key]
        return skill.exec(session_id, permission, params)

# ===================== 测试运行（完整落地演示）=====================
if __name__ == "__main__":
    print("=" * 60)
    print("  Harness + Skills 大模型工程化 — 完整链路演示")
    print("=" * 60)
    print("  六层驾驭：权限 → 频次 → 参数 → 重试 → 结果 → 状态")
    print("=" * 60)

    # 初始化AI工程调度器
    print("\n【初始化】创建 HarnessEngine 全局管控引擎 + SessionMemory 状态记忆模块")
    ai_engine = AIAssistantPlanner()
    print("【初始化】注册 Skills: data_analysis（数据分析技能）")
    print("【初始化】AI工程调度器就绪")

    # 模拟用户会话与参数
    test_session = "session_20260730_001"
    test_permission = "user_admin"
    test_params = {
        "data_content": "2026年7月业务数据：订单量2300，转化率18%，退款率2%",
        "analyse_target": "月度业务数据健康度分析"
    }

    print(f"\n【任务输入】会话ID：{test_session}")
    print(f"【任务输入】权限等级：{test_permission}")
    print(f"【任务输入】数据内容：{test_params['data_content']}")
    print(f"【任务输入】分析目标：{test_params['analyse_target']}")

    # 执行完整Harness+Skills任务链路
    result = ai_engine.run_task(
        session_id=test_session,
        skill_key="data_analysis",
        permission=test_permission,
        params=test_params
    )

    print("【最终业务输出结果】")
    print(f"  status : {result['status']}")
    print(f"  msg    : {result['msg']}")
    print(f"  data   : {result['data']}")
    print("\n" + "=" * 60)
    print("  全链路执行完毕 — 六层驾驭全部通过")
    print("=" * 60)