⚡

排队系统分析器Skill queuing-analyzer

排队系统分析器是一个基于排队论的AI分析工具，用于对各类等待线系统进行数学建模和性能评估。它支持M/M/1、M/M/c、M/G/1等多种经典排队模型，能够计算系统利用率、平均等待时间、队列长度等关键性能指标，并应用于呼叫中心人员配置、服务能力规划、系统优化等场景。关键词：排队论，系统分析，性能评估，数学模型，服务优化，呼叫中心，容量规划，仿真建模。

预测建模 1 次安装 2 次浏览 更新于 2/25/2026

name: queuing-analyzer description: 排队系统分析技能，用于对等待线系统进行解析评估。 allowed-tools: Bash(*) Read Write Edit Glob Grep WebFetch metadata: author: babysitter-sdk version: “1.0.0” category: simulation backlog-id: SK-IE-007

排队分析器

您是 排队分析器 - 一个使用排队论对等待线系统进行解析评估的专门技能。

概述

此技能支持AI驱动的排队分析，包括：

  • M/M/1、M/M/c、M/G/1 模型计算
  • 稳态性能指标计算（Lq, Wq, L, W）
  • 服务器利用率分析
  • 概率计算（等待时间 > 阈值）
  • 呼叫中心人员配置的埃尔朗C公式
  • 有限总体模型
  • 排队网络分析

前提条件

  • Python 3.8+ 及 scipy, numpy
  • 理解排队论符号

能力

1. M/M/1 队列分析

def mm1_queue(arrival_rate, service_rate):
    """
    M/M/1 队列性能指标
    - 泊松到达，指数服务，单服务器
    """
    lambda_ = arrival_rate
    mu = service_rate

    # 利用率
    rho = lambda_ / mu

    if rho >= 1:
        return {"error": "系统不稳定 (rho >= 1)"}

    # 性能指标
    L = rho / (1 - rho)  # 系统中期望数量
    Lq = rho**2 / (1 - rho)  # 队列中期望数量
    W = 1 / (mu - lambda_)  # 系统中期望时间
    Wq = rho / (mu - lambda_)  # 队列中期望时间

    # 概率
    P0 = 1 - rho  # 系统空闲概率
    Pn = lambda n: (1 - rho) * rho**n  # 系统中有n个的概率

    return {
        "model": "M/M/1",
        "arrival_rate": lambda_,
        "service_rate": mu,
        "utilization": rho,
        "L": L,
        "Lq": Lq,
        "W": W,
        "Wq": Wq,
        "P0": P0,
        "P_wait": rho,
        "stable": rho < 1
    }

2. M/M/c 队列分析

from scipy.special import factorial
import numpy as np

def mmc_queue(arrival_rate, service_rate, num_servers):
    """
    M/M/c 队列性能指标
    - 多个并行服务器
    """
    lambda_ = arrival_rate
    mu = service_rate
    c = num_servers

    rho = lambda_ / (c * mu)

    if rho >= 1:
        return {"error": "系统不稳定 (rho >= 1)"}

    # 计算 P0
    sum_term = sum((c * rho)**n / factorial(n) for n in range(c))
    last_term = (c * rho)**c / (factorial(c) * (1 - rho))
    P0 = 1 / (sum_term + last_term)

    # 埃尔朗C公式（等待概率）
    C = ((c * rho)**c / factorial(c)) * (1 / (1 - rho)) * P0

    # 性能指标
    Lq = C * rho / (1 - rho)
    L = Lq + lambda_ / mu
    Wq = Lq / lambda_
    W = Wq + 1 / mu

    return {
        "model": "M/M/c",
        "arrival_rate": lambda_,
        "service_rate": mu,
        "servers": c,
        "utilization": rho,
        "L": L,
        "Lq": Lq,
        "W": W,
        "Wq": Wq,
        "P0": P0,
        "P_wait": C,
        "stable": rho < 1
    }

3. M/G/1 队列（Pollaczek-Khinchin）

def mg1_queue(arrival_rate, service_mean, service_variance):
    """
    使用Pollaczek-Khinchin公式的M/G/1队列
    - 一般服务时间分布
    """
    lambda_ = arrival_rate
    Es = service_mean
    Var_s = service_variance

    # 服务时间的二阶矩
    Es2 = Var_s + Es**2

    rho = lambda_ * Es

    if rho >= 1:
        return {"error": "系统不稳定 (rho >= 1)"}

    # Pollaczek-Khinchin公式
    Lq = (lambda_**2 * Es2) / (2 * (1 - rho))
    L = Lq + rho
    Wq = Lq / lambda_
    W = Wq + Es

    return {
        "model": "M/G/1",
        "arrival_rate": lambda_,
        "service_mean": Es,
        "service_variance": Var_s,
        "utilization": rho,
        "L": L,
        "Lq": Lq,
        "W": W,
        "Wq": Wq,
        "stable": rho < 1
    }

4. 呼叫中心人员配置的埃尔朗C公式

def erlang_c_staffing(arrival_rate, service_rate, target_service_level,
                      target_wait_time):
    """
    确定达到服务水平目标所需的最少服务器数量
    """
    lambda_ = arrival_rate
    mu = service_rate

    # 稳定所需的最少服务器
    min_servers = int(np.ceil(lambda_ / mu))

    for c in range(min_servers, min_servers + 100):
        result = mmc_queue(lambda_, mu, c)

        if result.get('error'):
            continue

        # 服务水平：P(等待 <= 目标)
        # SL = 1 - C * exp(-(c*mu - lambda) * target_wait)
        C = result['P_wait']
        exp_term = np.exp(-(c * mu - lambda_) * target_wait_time)
        service_level = 1 - C * exp_term

        if service_level >= target_service_level:
            return {
                "recommended_servers": c,
                "achieved_service_level": service_level,
                "target_service_level": target_service_level,
                "P_wait": C,
                "utilization": result['utilization'],
                "avg_wait": result['Wq']
            }

    return {"error": "无法达到目标服务水平"}

5. 有限总体（M/M/c/K/K）

def finite_population_queue(arrival_rate, service_rate, num_servers,
                           population_size):
    """
    有限总体队列（机器维修模型）
    """
    lambda_ = arrival_rate  # 每个客户的到达率
    mu = service_rate
    c = num_servers
    K = population_size

    # 使用递归的状态概率
    P = np.zeros(K + 1)
    P[0] = 1  # 临时值

    for n in range(1, K + 1):
        if n <= c:
            P[n] = P[n-1] * (K - n + 1) * lambda_ / (n * mu)
        else:
            P[n] = P[n-1] * (K - n + 1) * lambda_ / (c * mu)

    # 归一化
    P = P / P.sum()

    # 性能指标
    L = sum(n * P[n] for n in range(K + 1))
    Lq = sum((n - c) * P[n] for n in range(c + 1, K + 1))

    # 有效到达率
    lambda_eff = sum((K - n) * lambda_ * P[n] for n in range(K))

    W = L / lambda_eff if lambda_eff > 0 else 0
    Wq = Lq / lambda_eff if lambda_eff > 0 else 0

    return {
        "model": "M/M/c/K/K",
        "servers": c,
        "population": K,
        "L": L,
        "Lq": Lq,
        "W": W,
        "Wq": Wq,
        "effective_arrival_rate": lambda_eff,
        "state_probabilities": P.tolist()
    }

6. 排队网络（Jackson网络）

def jackson_network(arrival_rates, service_rates, routing_matrix):
    """
    开放Jackson网络分析
    arrival_rates: 每个节点的外部到达率
    service_rates: 每个节点的服务率
    routing_matrix: 从i路由到j的概率
    """
    n_nodes = len(service_rates)

    # 求解有效到达率
    # lambda_i = gamma_i + sum_j(lambda_j * r_ji)
    R = np.array(routing_matrix)
    gamma = np.array(arrival_rates)

    # lambda = gamma + lambda * R => lambda = gamma * (I - R)^-1
    I = np.eye(n_nodes)
    lambdas = np.linalg.solve((I - R.T), gamma)

    # 将每个队列分析为M/M/1
    results = []
    for i in range(n_nodes):
        result = mm1_queue(lambdas[i], service_rates[i])
        result['node'] = i
        result['effective_arrival_rate'] = lambdas[i]
        results.append(result)

    # 网络总计
    L_total = sum(r['L'] for r in results if 'L' in r)

    return {
        "model": "Jackson_Network",
        "effective_arrival_rates": lambdas.tolist(),
        "node_results": results,
        "total_L": L_total
    }

流程集成

此技能与以下流程集成：

  • queuing-system-analysis.js
  • capacity-planning-analysis.js
  • discrete-event-simulation-modeling.js

输出格式

{
  "model": "M/M/c",
  "parameters": {
    "arrival_rate": 10,
    "service_rate": 4,
    "servers": 3
  },
  "performance_measures": {
    "utilization": 0.833,
    "L": 6.01,
    "Lq": 3.51,
    "W": 0.601,
    "Wq": 0.351
  },
  "probabilities": {
    "P0": 0.045,
    "P_wait": 0.702
  },
  "service_level": {
    "P_wait_less_5min": 0.82
  },
  "recommendations": [
    "高利用率 - 考虑增加服务器"
  ]
}

工具/库

库 描述 使用场景
scipy 科学计算 核心计算
queueing Python包 队列分析
queuecomputer ® R包 高级模型
Custom 手写代码 特定需求

最佳实践

  1. 验证稳定性 - 在计算前检查 rho < 1
  2. 验证假设 - 测试泊松到达
  3. 考虑有限总体 - 当适用时
  4. 与仿真比较 - 对于复杂系统
  5. 敏感性分析 - 测试参数变化
  6. 记录符号 - 使用Kendall符号

约束

  • 明确报告所有假设
  • 标记不稳定系统
  • 记录解析模型的局限性
  • 对于复杂情况推荐仿真