MAS Scale Management — 大规模多智能体集群管理：万级并发、单调扩展、公司制架构

1. 解决的问题

1. 性能坍塌问题：新 Agent 加入时，系统路由还不了解其能力，导致任务分配混乱、性能下降

2. 核心算法逻辑

MAS 的规模扩展面临三个独特挑战，与普通分布式系统不同：

3. 业务应用场景

业务背景：双 11 期间，需要同时处理： - 10,000 个 SKU 的库存状态更新（Environment Service：数据库读写） - 5,000 个广告 Agent 实时竞价（Model Service：LLM 推理） - 2,000 个客服 Agent 处理退换货咨询（Agent Service：任务调度）

业务背景：团队将供应商评估 Agent 从 5 个扩展到 20 个（新增 15 个评估中国工厂的专业 Agent）。扩展后前 3 天评估质量下降，运营反馈结果不稳定。

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

未自动抽取；请查看原始 Skill 卡片。

7. 代码模板

代码块数量：6 · 路径：未检测到

import math
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from enum import Enum


class ServiceType(Enum):
    MODEL = "model"
    AGENT = "agent"
    ENVIRONMENT = "environment"


@dataclass
class AgentCapability:
    agent_id: str
    task_success_counts: Dict[str, int] = field(default_factory=dict)
    task_total_counts: Dict[str, int] = field(default_factory=dict)
    is_familiarized: bool = False
    joined_at: float = field(default_factory=time.time)

    def success_rate(self, task_type: str) -> float:
        total = self.task_total_counts.get(task_type, 0)
        if total == 0:
            return 0.0
        return self.task_success_counts.get(task_type, 0) / total

    def total_tasks(self) -> int:
        return sum(self.task_total_counts.values())

    def update(self, task_type: str, success: bool):
        self.task_total_counts[task_type] = self.task_total_counts.get(task_type, 0) + 1
        if success:
            self.task_success_counts[task_type] = self.task_success_counts.get(task_type, 0) + 1


class MonoScaleRouter:
    """
    Contextual Bandit 路由器：UCB1 策略保证扩容性能单调不退化
    """

    def __init__(self, exploration_coeff: float = 1.0):
        self.beta = exploration_coeff
        self.agents: Dict[str, AgentCapability] = {}
        self.t: int = 0

    def register(self, agent_id: str, familiarization_tasks: Optional[List[Dict]] = None):
        cap = AgentCapability(agent_id=agent_id)
        if familiarization_tasks:
            for task in familiarization_tasks:
                cap.update(task["task_type"], task["success"])
            cap.is_familiarized = True
        self.agents[agent_id] = cap

    def route(self, task_type: str) -> Optional[str]:
        if not self.agents:
            return None
        self.t += 1
        best_agent, best_score = None, -1.0

8. 论文来源

• 2601.07526
• 2601.23219
• 2604.01020