Class LoadBalancing

LoadBalancing.py:125–653

class LoadBalancing:
    def __init__(self) -> None:
        """初始化LoadBalancing实例，获取健康检查数据和配置"""
        # 这里要维护一个包含滑动窗口逻辑的, 每个供应商的每个模型的表现情况队列.
        # 然后先按照损坏概率去降序排序.
        # 再按照执行时间去降序排序.
        self.healthy = Harness_localAPI.check_healthy()
        self.source_price = source_price
        self.source_ranking = source_ranking
        self.source_mapping = source_mapping
        # 添加日志记录器
        self.logger = logging.getLogger(__name__)
        # 防止日志向上传播，避免重复打印
        self.logger.propagate = False


    def _check_valid_model(self, source_name, model_name):
        """检查模型是否在源上有效
        
        Args:
            source_name (str): 源名称
            model_name (str): 模型名称
            
        Returns:
            bool: 如果模型有效返回True，否则返回False
        """
        # 处理多模态模型的情况
        base_model_name = model_name[:-3] if model_name.endswith("_mm") else model_name
        
        # 检查映射是否存在且非None
        if (source_name in self.source_mapping and 
            base_model_name in self.source_mapping[source_name] and 
            self.source_mapping[source_name][base_model_name] is not None):
            return True
        return False

    def get_combinedRanking(self, ranking_1, ranking_2, weight):
        """融合两个不同指标的排序
        
        将两个不同指标的排序结果按给定权重融合为一个新的排序结果。
        
        Args:
            ranking_1 (dict): 第一个指标的排序结果
            ranking_2 (dict): 第二个指标的排序结果
            weight (float): ranking_2的权重，范围[0,1]
            
        Returns:
            dict: 融合后的排序结果
        """
        ranking_3 = {}
        
        len_1 = len(ranking_1)
        len_2 = len(ranking_2)
        
        # 创建副本避免修改原始数据
        tmp_ranking_1 = ranking_1.copy()
        tmp_ranking_2 = ranking_2.copy()