智能体构建：打破停机瓶颈：智能体Skill热更新、灰度发布与回滚机制全流程指南.142

# 主入口：Skill 热更新 + 灰度发布 + 回滚全流程演示
from core.skill_registry import SkillRegistry
from core.dynamic_loader import DynamicLoader
from core.gray_router import GrayRouter
from core.monitor import SkillMonitor
from core.rollback import RollbackManager

if __name__ == "__main__":
    print("=" * 60)
    print("🔥 企业级智能体 Skill 热更新与灰度发布系统启动")
    print("=" * 60)

    # ====================== 0. 初始版本调用测试 ======================
    print("\n📌 步骤0：初始版本调用测试（更新前）")
    DynamicLoader.load_skill("bill_query", "./skills/bill_query_v1_0_0.py")
    import sys
    if "bill_query_v1_0_0" in sys.modules:
        old_skill = sys.modules["bill_query_v1_0_0"]
        print("[初始版本 v1.0.0] 调用结果:", old_skill.execute("user_test"))

    # ====================== 1. 注册新版本 Skill（灰度模式）======================
    print("\n📌 步骤1：注册新版本 Skill（灰度发布中）")
    SkillRegistry.register_skill(
        skill_name="bill_query",
        version="1.0.1",
        path="./skills/bill_query_v1_0_1.py",
        gray_users=["user001", "user002", "admin001"]
    )

    # ====================== 2. 热加载新版本 Skill ======================
    print("\n📌 步骤2：动态热加载 Skill（不停服）")
    DynamicLoader.load_skill("bill_query", "./skills/bill_query_v1_0_1.py")

    # ====================== 3. 灰度路由测试 ======================
    print("\n📌 步骤3：灰度路由分发测试")
    # 灰度用户 → 新版本
    print("[灰度用户] user001 →", GrayRouter.route_skill("bill_query", "user001"))
    # 普通用户 → 旧版本
    print("[普通用户] user003 →", GrayRouter.route_skill("bill_query", "user003"))

    # ====================== 4. 监控指标采集 ======================
    print("\n📌 步骤4：实时监控 Skill 运行指标")
    SkillMonitor.collect_metrics("bill_query")

    # ====================== 5. 灰度通过 → 全量发布 ======================
    print("\n" + "="*50)
    print("✅ 灰度验证无异常，执行全量发布")
    print("="*50)
    SkillRegistry.update_skill_status("bill_query", "online")

    # ====================== 6. 模拟异常 → 一键回滚 ======================
    print("\n" + "="*50)
    print("⚠️  模拟新版本异常，触发自动回滚")
    print("="*50)
    RollbackManager.rollback_skill("bill_query", old_version="1_0_0")

    # ====================== 7. 回滚后验证 ======================
    print("\n📌 步骤7：回滚后验证 - 调用不同版本Skill")
    
    # 重新加载旧版本
    DynamicLoader.load_skill("bill_query", "./skills/bill_query_v1_0_0.py")
    
    # 动态导入并执行
    import sys
    if "bill_query_v1_0_0" in sys.modules:
        old_skill = sys.modules["bill_query_v1_0_0"]
        print("[回滚后-旧版本 v1.0.0] 调用结果:", old_skill.execute("user_test"))
    
    if "bill_query_v1_0_1" in sys.modules:
        new_skill = sys.modules["bill_query_v1_0_1"]
        print("[新版本 v1.0.1] 调用结果:", new_skill.execute("user_test"))
    
    # 验证当前路由
    print("\n📌 回滚后路由验证:")
    print("[任意用户] user001 →", GrayRouter.route_skill("bill_query", "user001"))

    print("\n" + "="*60)
    print("🎯 全流程执行完成：热更新 → 灰度 → 全量 → 回滚 → 验证")
    print("="*60)

# Redis 统一配置
import redis

def get_redis_client():
    """获取 Redis 连接客户端（单例模式）"""
    try:
        client = redis.Redis(
            host="localhost",
            port=6379,
            db=0,
            decode_responses=True,
            socket_connect_timeout=3
        )
        # 测试连接
        client.ping()
        return client
    except Exception as e:
        raise Exception(f"Redis 连接失败，请启动 Redis 服务：{str(e)}")

# 全局 Redis 客户端
redis_client = get_redis_client()

import json
from config.redis_config import redis_client

class SkillRegistry:
    """技能注册与元数据管理中心"""

    @staticmethod
    def register_skill(skill_name: str, version: str, path: str, gray_users: list = None):
        """注册/更新技能信息"""
        skill_data = {
            "version": version,
            "path": path,
            "gray_users": gray_users or [],
            "status": "gray",  # gray=灰度, online=全量, offline=下线
            "update_time": __import__('time').time()
        }
        redis_client.set(f"skill:{skill_name}", json.dumps(skill_data))

    @staticmethod
    def get_skill_info(skill_name: str):
        """获取技能详情"""
        data = redis_client.get(f"skill:{skill_name}")
        return json.loads(data) if data else None

    @staticmethod
    def update_skill_status(skill_name: str, status: str):
        """更新技能发布状态"""
        info = SkillRegistry.get_skill_info(skill_name)
        if info:
            info["status"] = status
            redis_client.set(f"skill:{skill_name}", json.dumps(info))
            print(f"🔄 技能【{skill_name}】状态已更新：{status}")
            # 输出当前版本调用测试
            import sys
            import os
            current_path = info.get("path", "")
            module_name = os.path.basename(current_path).replace(".py", "")
            if module_name in sys.modules:
                skill_module = sys.modules[module_name]
                if hasattr(skill_module, "execute"):
                    print(f"📌 当前版本调用测试: {skill_module.execute('test_user')}")

import importlib
import sys
import os

class DynamicLoader:
    """技能热加载核心模块"""

    @staticmethod
    def load_skill(skill_name: str, skill_path: str):
        """动态加载/重载技能模块"""
        try:
            # 添加技能目录到系统路径
            base_dir = os.path.dirname(skill_path)
            if base_dir not in sys.path:
                sys.path.append(base_dir)

            # 获取模块名
            module_file = os.path.basename(skill_path)
            module_name = module_file.replace(".py", "")

            # 加载/重载（热更新关键）
            if module_name in sys.modules:
                importlib.reload(sys.modules[module_name])
            else:
                importlib.import_module(module_name)

            print(f"✅ 技能【{skill_name}】热加载成功：{skill_path}")
            return True

        except Exception as e:
            print(f"❌ 技能【{skill_name}】加载失败：{str(e)}")
            return False

    @staticmethod
    def unload_skill(skill_name: str):
        """从内存卸载技能"""
        print(f"🗑️  技能【{skill_name}】已从运行内存卸载")

from core.skill_registry import SkillRegistry

class GrayRouter:
    """请求路由：灰度用户/普通用户分发"""

    @staticmethod
    def is_gray_user(user_id: str, gray_list: list) -> bool:
        """判断是否为灰度用户"""
        return user_id in gray_list

    @staticmethod
    def route_skill(skill_name: str, user_id: str) -> str:
        """路由到对应版本的技能路径"""
        skill_info = SkillRegistry.get_skill_info(skill_name)
        if not skill_info:
            raise Exception(f"技能【{skill_name}】未注册")

        # 灰度模式：仅灰度用户使用新版本
        if skill_info["status"] == "gray":
            if GrayRouter.is_gray_user(user_id, skill_info["gray_users"]):
                return f"路由到【新版本 {skill_info['version']}】"
            else:
                return "路由到【稳定旧版本 v1.0.0】"

        # 全量模式：所有用户使用新版本
        elif skill_info["status"] == "online":
            return f"路由到【全量版本 {skill_info['version']}】"

        else:
            return "⚠️  技能已下线"

import time
import psutil
from datetime import datetime

class SkillMonitor:
    """实时监控技能运行质量指标"""

    @staticmethod
    def collect_metrics(skill_name: str):
        """采集并输出监控指标（企业版可接入 Prometheus）"""
        metrics = {
            "skill": skill_name,
            "success_rate": "100%",
            "response_time": "0.28s",
            "error_rate": "0%",
            "cpu_usage": f"{psutil.cpu_percent(interval=0.1)}%",
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }

        # 打印监控日志
        print(f"📊 监控指标 {metrics['time']}：")
        for k, v in metrics.items():
            if k != "time" and k != "skill":
                print(f"   {k}: {v}")

        # 写入日志文件
        with open("./logs/skill_metrics.log", "a", encoding="utf-8") as f:
            f.write(str(metrics) + "\n")

        return metrics

from core.dynamic_loader import DynamicLoader
from core.skill_registry import SkillRegistry
import json
from config.redis_config import redis_client

class RollbackManager:
    """技能版本一键回滚"""

    @staticmethod
    def rollback_skill(skill_name: str, old_version: str = "1.0.0"):
        """回滚到指定稳定版本"""
        print(f"⏪ 开始回滚技能【{skill_name}】到版本 {old_version}")

        # 构建旧版本路径
        old_path = f"./skills/{skill_name}_v{old_version}.py"

        # 热加载旧版本
        success = DynamicLoader.load_skill(skill_name, old_path)

        if success:
            # 更新注册信息：版本号和路径都恢复为旧版本
            skill_info = SkillRegistry.get_skill_info(skill_name)
            if skill_info:
                skill_info["version"] = old_version.replace("_", ".")
                skill_info["path"] = old_path
                skill_info["status"] = "online"
                redis_client.set(f"skill:{skill_name}", json.dumps(skill_info))
            
            print(f"✅ 回滚完成：【{skill_name}】已恢复至稳定版本 {old_version}")

# 账单查询技能 v1.0.0（稳定版）
def execute(user_id):
    return f"【v1.0.0】用户 {user_id} 当前账户余额：10000 元"

# 账单查询技能 v1.0.1（新版本）
def execute(user_id):
    return f"【v1.0.1】用户 {user_id} 余额：10000 元 | 本月账单已自动分期"

{'skill': 'bill_query', 'success_rate': '100%', 'response_time': '0.28s', 'error_rate': '0%', 'cpu_usage': '40.4%', 'time': '2026-04-07 20:33:13'}
{'skill': 'bill_query', 'success_rate': '100%', 'response_time': '0.28s', 'error_rate': '0%', 'cpu_usage': '7.6%', 'time': '2026-04-07 20:39:10'}
{'skill': 'bill_query', 'success_rate': '100%', 'response_time': '0.28s', 'error_rate': '0%', 'cpu_usage': '3.4%', 'time': '2026-04-07 20:42:18'}
......