大模型应用：避免大模型服务雪崩：深入解析AI场景下熔断机制设计与应用实践.160

import time
import random
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from enum import Enum
from collections import deque

# 定义熔断器三种运行状态
class CircuitState(Enum):
    CLOSED = "闭合状态-正常通行"
    OPEN = "打开状态-熔断拦截"
    HALF_OPEN = "半打开状态-试探探测"

class LLMCircuitBreaker:
    """
    大模型专属熔断器实现
    适配LLM长耗时、高异常特性，支持滑动窗口、三态流转、降级联动
    """
    def __init__(
        self,
        window_time: int = 10,        # 滑动统计窗口时长(秒)
        fail_threshold: float = 0.5,  # 触发熔断的失败率阈值
        cool_down_time: int = 30,     # 熔断冷却恢复时长(秒)
        probe_count: int = 1          # 半开启状态单次试探请求数量
    ):
        # 初始化默认状态
        self.state = CircuitState.CLOSED
        # 核心配置参数
        self.window_time = window_time
        self.fail_threshold = fail_threshold
        self.cool_down_time = cool_down_time
        self.probe_count = probe_count
        self.probe_req_num = 0

        # 滑动窗口：存储元组(时间戳, 是否请求失败)
        self.record_queue = deque()
        # 熔断开启的起始时间戳
        self.open_start_time = 0

    def _clear_expired_record(self):
        """清理滑动窗口内过期的历史请求记录"""
        current_time = time.time()
        # 剔除超出窗口时间的老旧数据
        while self.record_queue and current_time - self.record_queue[0][0] > self.window_time:
            self.record_queue.popleft()

    def _get_fail_rate(self) -> float:
        """计算当前滑动窗口内的请求失败率"""
        self._clear_expired_record()
        total_request = len(self.record_queue)
        if total_request == 0:
            return 0.0
        # 统计异常请求数量
        fail_count = sum(1 for _, is_fail in self.record_queue if is_fail)
        return fail_count / total_request

    def _record_result(self, is_fail: bool):
        """单条请求结果写入滑动窗口"""
        self._clear_expired_record()
        self.record_queue.append((time.time(), is_fail))

    def _state_transfer(self):
        """核心：熔断器三态自动流转逻辑"""
        current_time = time.time()

        # 1. 闭合状态：检测失败率，达到阈值则熔断
        if self.state == CircuitState.CLOSED:
            fail_rate = self._get_fail_rate()
            if fail_rate >= self.fail_threshold:
                self.state = CircuitState.OPEN
                self.open_start_time = current_time
                print(f"\n【系统预警】大模型服务异常累积")
                print(f"当前窗口失败率：{fail_rate:.2f} | 触发熔断阈值：{self.fail_threshold}")
                print(f"熔断器状态变更：{self.state.value}")

        # 2. 打开状态：冷却时间结束，切换至半开启探测
        elif self.state == CircuitState.OPEN:
            if current_time - self.open_start_time >= self.cool_down_time:
                self.state = CircuitState.HALF_OPEN
                self.probe_req_num = 0
                print(f"\n【冷却完成】熔断保护周期结束")
                print(f"熔断器状态变更：{self.state.value}，开始试探性检测服务")

    def allow_request(self) -> bool:
        """判断当前请求是否允许调用大模型"""
        self._state_transfer()
        if self.state == CircuitState.CLOSED:
            return True
        elif self.state == CircuitState.OPEN:
            return False
        # 半开启状态：限制试探请求数量
        else:
            if self.probe_req_num < self.probe_count:
                self.probe_req_num += 1
                return True
            return False

    def report_result(self, is_fail: bool):
        """上报请求执行结果，驱动状态二次变更"""
        self._record_result(is_fail)
        if self.state == CircuitState.HALF_OPEN:
            if is_fail:
                # 试探请求失败，服务未恢复，重回熔断
                self.state = CircuitState.OPEN
                self.open_start_time = time.time()
                print(f"\n【探测失败】大模型服务仍存在异常")
                print(f"重新进入熔断保护：{self.state.value}")
            else:
                # 试探请求成功，服务恢复，切换正常模式
                self.state = CircuitState.CLOSED
                self.record_queue.clear()
                print(f"\n【探测成功】大模型服务恢复正常运行")
                print(f"熔断器解除限制：{self.state.value}，全量流量恢复通行")

# 模拟真实大模型推理服务
def mock_llm_invoke() -> tuple[bool, str]:
    """
    模拟LLM真实运行状态：
    70% 正常推理、20% 推理超时、10% 服务内部报错
    模拟长耗时推理，贴合真实业务场景
    """
    random_seed = random.random()
    # 模拟大模型推理耗时
    time.sleep(random.uniform(1.0, 3.0))
    if random_seed < 0.7:
        return True, "AI推理结果：随着人工智能技术迭代，大模型已广泛应用于各行业数字化升级，为业务赋能增效。"
    elif random_seed < 0.9:
        return False, "异常：大模型推理任务超时，GPU队列拥堵，响应链路中断"
    else:
        return False, "异常：模型服务内部错误，参数解析失败，推理进程异常退出"

# 大模型专属降级兜底方案
def llm_fallback_response() -> str:
    """AI服务不可用时的标准化降级回复"""
    return "当前AI推理服务临时繁忙，为保障基础使用，已为您提供兜底回答。您可以简化问题稍后重试，感谢理解。"

def plot_circuit_breaker_trend(results: list, states: list):
    """
    绘制熔断器请求过程趋势图
    results: 每条记录为 (请求序号, 状态, 说明)
             状态: 'success' / 'fail' / 'blocked'
    states:  每个请求对应的熔断器状态记录
    """
    plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
    plt.rcParams['axes.unicode_minus'] = False

    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 10), sharex=True)
    fig.suptitle('大模型熔断器请求过程趋势图', fontsize=16, weight='bold', color='#1a1a2e')

    req_nums = [r[0] for r in results]
    req_labels = [f"Req {r[0]}" for r in results]

    # ---------- 上图：请求结果分布 ----------
    status_colors = {'success': '#2ecc71', 'fail': '#e74c3c', 'blocked': '#95a5a6'}
    result_map = {'success': 1, 'fail': 2, 'blocked': 0}
    y_vals = [result_map[r[1]] for r in results]
    color_list = [status_colors[r[1]] for r in results]

    for i, (x, y) in enumerate(zip(req_nums, y_vals)):
        ax1.scatter(x, y, color=color_list[i], s=120, zorder=3,
                     edgecolor='white', linewidth=1.5)
        # 标注说明
        ax1.annotate(results[i][2], xy=(x, y), xytext=(0, 8),
                     textcoords='offset points', fontsize=8,
                     ha='center', color=color_list[i])

    ax1.set_yticks([0, 1, 2])
    ax1.set_yticklabels(['被拦截', '成功', '失败'], fontsize=11)
    ax1.set_ylabel('请求结果', fontsize=12, color='#333')
    ax1.grid(alpha=0.3, linestyle='--')
    ax1.set_xlim(0.5, max(req_nums) + 0.5)
    ax1.set_ylim(-0.5, 2.5)
    ax1.set_facecolor('#fafafa')

    # 图例
    legend_patches = [
        mpatches.Patch(color='#2ecc71', label='成功通行'),
        mpatches.Patch(color='#e74c3c', label='放行但失败'),
        mpatches.Patch(color='#95a5a6', label='熔断拦截'),
    ]
    ax1.legend(handles=legend_patches, loc='upper left', fontsize=10)

    # ---------- 下图：熔断器状态变化 ----------
    state_color_map = {
        'CLOSED': '#2ecc71',
        'OPEN': '#e74c3c',
        'HALF_OPEN': '#f39c12'
    }
    state_y = {'CLOSED': 1, 'HALF_OPEN': 2, 'OPEN': 0}

    # 分段绘制状态背景色块
    seg_start_idx = 0
    prev_state = states[0]
    for i, st in enumerate(states):
        if st != prev_state:
            # 绘制前一段状态
            ax2.axhspan(state_y[prev_state] - 0.3, state_y[prev_state] + 0.3,
                        xmin=seg_start_idx / len(states),
                        xmax=i / len(states),
                        color=state_color_map[prev_state], alpha=0.25)
            seg_start_idx = i
            prev_state = st
    # 绘制最后一段状态
    ax2.axhspan(state_y[prev_state] - 0.3, state_y[prev_state] + 0.3,
                xmin=seg_start_idx / len(states),
                xmax=1.0,
                color=state_color_map[prev_state], alpha=0.25)

    # 绘制状态散点及标签
    for i, st in enumerate(states):
        y_pos = state_y[st]
        ax2.scatter(i + 1, y_pos, color=state_color_map[st], s=80, zorder=3,
                     edgecolor='white', linewidth=1.5)
        ax2.text(i + 1, y_pos + 0.35, st, fontsize=8, ha='center', color=state_color_map[st])

    ax2.set_yticks([0, 1, 2])
    ax2.set_yticklabels(['熔断打开', '正常闭合', '半打开'], fontsize=11)
    ax2.set_ylabel('熔断器状态', fontsize=12, color='#333')
    ax2.set_xlabel('请求序号', fontsize=12, color='#333')
    ax2.set_xticks(req_nums)
    ax2.set_xticklabels(req_labels, rotation=45, fontsize=9)
    ax2.grid(alpha=0.3, linestyle='--')
    ax2.set_xlim(0.5, max(req_nums) + 0.5)
    ax2.set_ylim(-0.5, 2.5)
    ax2.set_facecolor('#fafafa')

    plt.tight_layout(rect=[0, 0, 1, 0.96])
    plt.savefig('160.熔断器请求过程趋势图.png', facecolor='white', dpi=150, bbox_inches='tight')
    print('\n[已保存] 160.熔断器请求过程趋势图.png')
    plt.show()

# 主程序测试逻辑
if __name__ == "__main__":
    # 初始化大模型熔断器，适配AI场景参数
    breaker = LLMCircuitBreaker(
        window_time=8,
        fail_threshold=0.4,
        cool_down_time=15,
        probe_count=1
    )

    print("===== 大模型熔断机制模拟测试启动 =====")
    print("模拟连续20次AI问答请求，自动触发熔断、冷却、探测、恢复全流程\n")

    # 数据记录列表
    results = []   # (请求序号, 状态, 说明)
    states = []    # 每个请求对应的熔断器状态名称

    # 模拟连续业务请求
    for req_index in range(20):
        print(f"—————— 第 {req_index + 1} 次 AI 请求 ——————")

        # 熔断校验（内部会自动触发状态流转）
        is_allow = breaker.allow_request()
        states.append(breaker.state.name)

        if is_allow:
            # 放行：调用大模型
            success, result = mock_llm_invoke()
            status_label = 'success' if success else 'fail'
            desc = '成功' if success else '失败'
            print(f"模型调用结果：{result}")
            # 上报结果
            breaker.report_result(not success)
            results.append((req_index + 1, status_label, desc))
        else:
            # 拦截：执行降级
            fallback_msg = llm_fallback_response()
            print(f"请求已被熔断拦截 | 降级回复：{fallback_msg}")
            results.append((req_index + 1, 'blocked', '拦截'))
        # 模拟用户请求间隔（测试时可缩短）
        time.sleep(0.3)

    # 绘制趋势图
    print("\n===== 正在生成趋势图 =====")
    plot_circuit_breaker_trend(results, states)