大模型服务熔断限流计费联动：异常流量风控拦截与超限自动降配架构实践.165

import time
from collections import deque
import threading
import matplotlib.pyplot as plt
import numpy as np
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class LlmFuseLimitController:
    def __init__(self):
        # ===================== 熔断配置 =====================
        # 熔断三种状态：关闭、开启、半开
        self.fuse_state = "CLOSE"
        # 滑动窗口：保存最近100条请求成功状态
        self.request_window = deque(maxlen=100)
        # 失败率触发阈值
        self.fail_rate_threshold = 0.5
        # 熔断静默等待时间 秒
        self.fuse_sleep_second = 10
        # 熔断开启时间记录
        self.fuse_open_timestamp = 0
        # 半开状态允许探测请求数量
        self.half_open_test_count = 5
        self.half_open_pass_count = 0

        # ===================== 令牌桶限流配置 =====================
        # 每秒生成令牌数量 代表全局QPS
        self.token_generate_rate = 10
        # 令牌桶最大容量 允许突发流量
        self.token_bucket_max = 20
        self.token_bucket_current = self.token_bucket_max
        # 上一次补充令牌时间
        self.last_refill_time = time.time()

        # 线程锁 保证并发安全
        self.lock = threading.Lock()

    # 令牌桶补充与拿取令牌
    def try_get_token(self):
        with self.lock:
            now = time.time()
            # 计算距离上次补令牌的时间间隔
            time_delta = now - self.last_refill_time
            # 按速率补充令牌
            add_token = time_delta * self.token_generate_rate
            self.token_bucket_current = min(self.token_bucket_current + add_token, self.token_bucket_max)
            self.last_refill_time = now

            # 有令牌则消耗一个，放行请求
            if self.token_bucket_current >= 1:
                self.token_bucket_current -= 1
                return True
            # 无令牌 限流拦截
            return False

    # 记录单次请求执行结果
    def record_request_result(self, is_success: bool):
        with self.lock:
            self.request_window.append(is_success)
            # 半开状态统计探测成功数
            if self.fuse_state == "HALF_OPEN" and is_success:
                self.half_open_pass_count += 1

    # 熔断状态检查与状态机流转
    def check_fuse_status(self):
        with self.lock:
            now = time.time()

            # 状态1：熔断开启
            if self.fuse_state == "OPEN":
                # 判断是否达到静默时间，进入半开
                if now - self.fuse_open_timestamp >= self.fuse_sleep_second:
                    self.fuse_state = "HALF_OPEN"
                    self.half_open_pass_count = 0
                    print("【熔断】静默结束，进入半开探测状态")
                return False

            # 状态2：半开探测
            if self.fuse_state == "HALF_OPEN":
                # 探测请求全部成功，恢复正常关闭
                if self.half_open_pass_count >= self.half_open_test_count:
                    self.fuse_state = "CLOSE"
                    print("【熔断】探测通过，服务恢复关闭状态")
                    return True
                # 探测过程中暂时放行少量请求
                return True

            # 状态3：正常关闭 计算失败率判断是否触发熔断
            if len(self.request_window) >= 20:
                fail_num = sum(1 for res in self.request_window if not res)
                fail_rate = fail_num / len(self.request_window)
                if fail_rate >= self.fail_rate_threshold:
                    self.fuse_state = "OPEN"
                    self.fuse_open_timestamp = now
                    print(f"【熔断】失败率{fail_rate:.2f}超标，触发熔断开启")
                    return False
            return True

# 测试模拟：熔断 + 限流 双机制验证
if __name__ == "__main__":
    print("="*60)
    print("测试目的：验证熔断器三状态流转 + 令牌桶限流机制")
    print("  - 请求0-49：正常期，全部成功")
    print("  - 请求50-99：故障期，模拟50%连续失败，触发熔断")
    print("  - 请求100+：恢复期，熔断静默10秒后半开探测")
    print("="*60)

    controller = LlmFuseLimitController()

    # 记录数据用于可视化
    normal_count = limit_count = fuse_count = fail_count = 0
    normal_set = set()
    fuse_state_history = []
    token_history = []

    for req_id in range(150):
        # 记录当前状态
        fuse_state_history.append(controller.fuse_state)
        token_history.append(controller.token_bucket_current)

        # 第一步：限流校验（令牌桶算法，控制QPS）
        if not controller.try_get_token():
            # 令牌耗尽，请求被限流，不进入后续处理
            limit_count += 1
            print(f"[-限流] 请求{req_id:3d} >> 令牌桶空，限流拦截（不计费）")
            time.sleep(0.08)
            continue

        # 第二步：熔断校验（状态机：CLOSE/OPEN/HALF_OPEN）
        if not controller.check_fuse_status():
            # 熔断开启或探测未通过，快速失败
            fuse_count += 1
            print(f"[熔断] 请求{req_id:3d} >> 熔断器开启，快速失败（不计费）")
            time.sleep(0.08)
            continue

        # 模拟业务请求成功/失败规律（50-99号请求模拟故障）
        if 50 <= req_id < 100:
            success = False  # 模拟下游服务故障
        else:
            success = True   # 正常响应

        # 记录结果用于熔断器统计失败率
        controller.record_request_result(success)

        if success:
            normal_count += 1
            normal_set.add(req_id)
        else:
            fail_count += 1

        status_text = "推理成功，正常计费" if success else "推理失败（下游故障），不计费"
        state_tag = "[√正常]" if success else "[×故障]"
        print(f"{state_tag} 请求{req_id:3d} >> 通过限流+熔断校验，{status_text}")
        time.sleep(0.05)

    print("="*60)
    print("测试完成：熔断器保护下游免被压垮，限流控制整体QPS")
    print("="*60)

    # ========== 可视化：熔断限流过程分析 ==========
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    fig.suptitle('大模型服务熔断限流机制运行分析', fontsize=14, fontweight='bold')

    # 图1: 请求状态分布饼图
    ax1 = axes[0, 0]
    labels = ['正常请求', '限流拦截', '熔断拦截', '故障失败']
    sizes = [normal_count, limit_count, fuse_count, fail_count]
    colors = ['#52c41a', '#faad14', '#f5222d', '#ff7a45']
    explode = (0.02, 0.02, 0.05, 0.02)
    wedges, texts, autotexts = ax1.pie(sizes, explode=explode, labels=labels, colors=colors,
                                        autopct='%1.1f%%', startangle=90, pctdistance=0.75)
    ax1.set_title('请求处理状态分布', fontsize=11, fontweight='bold')
    # 添加中心圆形成环形图
    centre_circle = plt.Circle((0, 0), 0.50, fc='white')
    ax1.add_patch(centre_circle)
    ax1.text(0, 0, f'总请求\n150', ha='center', va='center', fontsize=12, fontweight='bold')

    # 图2: 熔断器状态流转时序
    ax2 = axes[0, 1]
    time_points = np.arange(len(fuse_state_history))
    state_map = {'CLOSE': 2, 'OPEN': 0, 'HALF_OPEN': 1}
    state_values = [state_map[s] for s in fuse_state_history]
    colors_map = {'CLOSE': '#52c41a', 'OPEN': '#f5222d', 'HALF_OPEN': '#faad14'}
    scatter_colors = [colors_map[s] for s in fuse_state_history]

    ax2.scatter(time_points, state_values, c=scatter_colors, s=20, alpha=0.7)
    ax2.fill_between(time_points, -0.3, state_values, alpha=0.3, color='#1890ff')
    ax2.set_yticks([0, 1, 2])
    ax2.set_yticklabels(['OPEN\n(熔断)', 'HALF_OPEN\n(半开)', 'CLOSE\n(正常)'])
    ax2.set_xlabel('请求序号', fontsize=10)
    ax2.set_title('熔断器状态流转时序', fontsize=11, fontweight='bold')
    ax2.axvline(x=50, color='red', linestyle='--', alpha=0.5, label='故障开始')
    ax2.axvline(x=100, color='green', linestyle='--', alpha=0.5, label='故障结束')
    ax2.legend(loc='lower right', fontsize=8)
    ax2.set_ylim(-0.5, 2.5)
    ax2.grid(True, alpha=0.3)

    # 图3: 令牌桶水位变化
    ax3 = axes[1, 0]
    ax3.plot(token_history, color='#1890ff', linewidth=1.5, label='令牌桶水位')
    ax3.axhline(y=controller.token_bucket_max, color='green', linestyle='--', alpha=0.7, label='最大容量')
    ax3.fill_between(range(len(token_history)), 0, token_history, alpha=0.3, color='#1890ff')
    ax3.set_xlabel('请求序号', fontsize=10)
    ax3.set_ylabel('令牌数量', fontsize=10)
    ax3.set_title('令牌桶水位动态变化', fontsize=11, fontweight='bold')
    ax3.legend(loc='upper right', fontsize=9)
    ax3.grid(True, alpha=0.3)
    ax3.set_ylim(0, controller.token_bucket_max + 2)

    # 图4: 各阶段请求统计柱状图
    ax4 = axes[1, 1]
    categories = ['正常期\n(0-49)', '故障期\n(50-99)', '恢复期\n(100-149)']
    normal_data = [sum(1 for i in range(0, 50) if i in normal_set),
                   sum(1 for i in range(50, 100) if i in normal_set),
                   sum(1 for i in range(100, 150) if i in normal_set)]
    blocked_data = [50 - normal_data[0], 50 - normal_data[1], 50 - normal_data[2]]

    x = np.arange(len(categories))
    width = 0.35
    bars1 = ax4.bar(x - width/2, normal_data, width, label='通过请求', color='#52c41a', alpha=0.8)
    bars2 = ax4.bar(x + width/2, blocked_data, width, label='拦截/失败', color='#f5222d', alpha=0.8)

    ax4.set_ylabel('请求数量', fontsize=10)
    ax4.set_title('三阶段请求处理对比', fontsize=11, fontweight='bold')
    ax4.set_xticks(x)
    ax4.set_xticklabels(categories)
    ax4.legend(fontsize=9)
    ax4.grid(True, alpha=0.3, axis='y')

    # 在柱子上添加数值
    for bar in bars1:
        height = bar.get_height()
        ax4.annotate(f'{int(height)}',
                    xy=(bar.get_x() + bar.get_width() / 2, height),
                    xytext=(0, 3), textcoords="offset points",
                    ha='center', va='bottom', fontsize=9)
    for bar in bars2:
        height = bar.get_height()
        ax4.annotate(f'{int(height)}',
                    xy=(bar.get_x() + bar.get_width() / 2, height),
                    xytext=(0, 3), textcoords="offset points",
                    ha='center', va='bottom', fontsize=9)

    plt.tight_layout()
    plt.savefig('fuse_limit_analysis.png', dpi=150, bbox_inches='tight', facecolor='white')
    print("\n[可视化] 熔断限流分析图已保存: fuse_limit_analysis.png")
    plt.show()

class LlmBillingRiskManager:
    def __init__(self):
        # 恶意IP黑名单
        self.risk_ip_blacklist = {"192.168.1.100", "10.0.0.55", "172.16.0.88"}
        # 记录用户每小时Token消耗
        self.user_hour_token_usage = {}
        # 单用户每小时最大配额
        self.user_max_hour_token = 10000
        # 计费规则：每1000 Token 0.1元
        self.price_per_thousand_token = 0.1

    # 恶意IP、恶意行为风控校验
    def risk_check(self, user_id: str, client_ip: str) -> tuple[bool, str]:
        if client_ip in self.risk_ip_blacklist:
            return True, "命中恶意IP黑名单，请求拦截"
        # 可扩展：高频请求检测、违规内容检测
        return False, "请求风控校验通过"

    # 异常流量Token超限校验
    def token_limit_check(self, user_id: str, use_token: int) -> tuple[bool, str]:
        current_used = self.user_hour_token_usage.get(user_id, 0)
        if current_used + use_token > self.user_max_hour_token:
            return True, "单用户每小时Token配额超限，拦截请求"
        # 未超限则累加消耗
        self.user_hour_token_usage[user_id] = current_used + use_token
        return False, "Token配额校验通过"

    # 计费计算逻辑
    def calc_billing_fee(self, token_num: int, is_request_valid: bool) -> tuple[float, str]:
        # 无效请求一律不计费
        if not is_request_valid:
            return 0.0, "无效/拦截请求，免计费"
        # 有效请求按规则计费
        fee = (token_num / 1000) * self.price_per_thousand_token
        return round(fee, 4), "有效推理请求，正常计费"

# 业务测试：计费风控三阶段校验
if __name__ == "__main__":
    print("\n" + "="*60)
    print("测试目的：验证计费风控三层防护机制")
    print("  阶段1：恶意IP黑名单校验（安全层）")
    print("  阶段2：单用户Token配额校验（配额层）")
    print("  阶段3：有效请求计费核算（计费层）")
    print("="*60)

    billing_manager = LlmBillingRiskManager()
    # 模拟4条典型用例：正常/黑名单/超配额/无效请求
    request_list = [
        {"user_id":"u001","ip":"192.168.1.20","token":800,"valid":True},   # 正常计费
        {"user_id":"u001","ip":"192.168.1.100","token":1200,"valid":True}, # 命中黑名单
        {"user_id":"u002","ip":"10.0.0.33","token":11000,"valid":True},  # Token超配额
        {"user_id":"u003","ip":"172.16.0.22","token":500,"valid":False}, # 无效请求
    ]

    for i, req in enumerate(request_list, 1):
        print(f"\n[用例{i}] 用户{req['user_id']} | IP:{req['ip']} | Token:{req['token']} | 有效:{req['valid']}")

        # 1.风控校验：恶意IP黑名单拦截
        is_risk, risk_msg = billing_manager.risk_check(req["user_id"], req["ip"])
        if is_risk:
            print(f"  [风控拦截] {risk_msg} → 不计费")
            continue
        print(f"  [风控通过] {risk_msg}")

        # 2.Token超限校验：单用户每小时配额控制
        is_over, over_msg = billing_manager.token_limit_check(req["user_id"], req["token"])
        if is_over:
            print(f"  [配额拦截] {over_msg} → 不计费")
            continue
        print(f"  [配额通过] {over_msg}")

        # 3.计费核算：仅有效请求计费，无效/拦截请求免费
        fee, bill_msg = billing_manager.calc_billing_fee(req["token"], req["valid"])
        if fee > 0:
            print(f"  [计费成功] 费用:{fee}元 | {bill_msg}")
        else:
            print(f"  [计费豁免] 费用:0元 | {bill_msg}")

    print("\n" + "="*60)
    print("测试完成：三层防护确保计费准确、风险可控")
    print("="*60)

    # ========== 可视化：计费风控结果分析 ==========
    fig2, axes2 = plt.subplots(1, 2, figsize=(12, 5))
    fig2.suptitle('计费风控三层防护机制分析', fontsize=14, fontweight='bold')

    # 图1: 用例处理结果对比
    ax1 = axes2[0]
    case_names = ['用例1\n正常计费', '用例2\n黑名单拦截', '用例3\n配额超限', '用例4\n无效请求']
    case_fees = [0.08, 0, 0, 0]  # 实际费用
    case_tokens = [800, 1200, 11000, 500]  # 请求token数

    x = np.arange(len(case_names))
    width = 0.35

    bars1 = ax1.bar(x - width/2, [t/100 for t in case_tokens], width, label='请求Token(百)', color='#1890ff', alpha=0.8)
    bars2 = ax1.bar(x + width/2, [f*1000 for f in case_fees], width, label='实际费用(×0.001元)', color='#52c41a', alpha=0.8)

    ax1.set_ylabel('数值', fontsize=10)
    ax1.set_title('四用例Token消耗与计费结果', fontsize=11, fontweight='bold')
    ax1.set_xticks(x)
    ax1.set_xticklabels(case_names, fontsize=9)
    ax1.legend(fontsize=9)
    ax1.grid(True, alpha=0.3, axis='y')

    # 添加拦截标记
    for i, (fee, name) in enumerate(zip(case_fees, case_names)):
        if fee == 0:
            ax1.annotate('拦截', xy=(i, max(case_tokens)/200), ha='center', fontsize=10,
                        color='red', fontweight='bold',
                        bbox=dict(boxstyle='round,pad=0.3', facecolor='yellow', alpha=0.7))

    # 图2: 防护层级拦截分布
    ax2 = axes2[1]
    stages = ['风控层\n(IP黑名单)', '配额层\n(Token限额)', '计费层\n(有效核算)']
    intercept_count = [1, 1, 2]  # 拦截/豁免数量
    pass_count = [3, 2, 1]  # 通过数量

    x = np.arange(len(stages))
    bars1 = ax2.barh(x, pass_count, color='#52c41a', alpha=0.8, label='通过')
    bars2 = ax2.barh(x, intercept_count, left=pass_count, color='#f5222d', alpha=0.8, label='拦截/豁免')

    ax2.set_yticks(x)
    ax2.set_yticklabels(stages)
    ax2.set_xlabel('请求数量', fontsize=10)
    ax2.set_title('三层防护拦截效果', fontsize=11, fontweight='bold')
    ax2.legend(fontsize=9, loc='lower right')
    ax2.grid(True, alpha=0.3, axis='x')

    # 添加数值标签
    for i, (p, inter) in enumerate(zip(pass_count, intercept_count)):
        ax2.text(p/2, i, str(p), ha='center', va='center', fontsize=10, fontweight='bold', color='white')
        ax2.text(p + inter/2, i, str(inter), ha='center', va='center', fontsize=10, fontweight='bold', color='white')

    plt.tight_layout()
    plt.savefig('billing_risk_analysis.png', dpi=150, bbox_inches='tight', facecolor='white')
    print("\n[可视化] 计费风控分析图已保存: billing_risk_analysis.png")
    plt.show()