Agent Registry & Discovery — 动态 Agent 能力注册与路由

Skill-Agent-Registry-Discovery · 10-MAS

causalexperimentforecastingmulti_agentpricing供应链与补货MAS与智能体工程定价与利润风控与合规WF-A 智能补货WF-D 选品扫描WF-F 动态定价

实现难度⭐⭐⭐☆☆

业务优先级⭐⭐⭐⭐⭐

业务视角

适用角色运营负责人 / CTO · 产品经理 · CEO

适用平台Amazon PPC + 库存 + 定价多 Agent 协作 · TikTok 内容运营流水线

什么情况下用运营任务太碎，选品/定价/广告/客服同时跑，人手严重不足；重复性运营动作需要 7×24 响应但没有足够人力

成功是什么样的多个 AI Agent 协作自动完成跨系统运营任务，运营团队人效提升 3-5 倍，7×24 无人值守运营

业务痛点

运营人手不够任务太多价格变化没有及时响应重复性工作占据太多时间想做 7×24 监控但没人盯

1. 解决的问题

静态工具注册（配置文件写死 Agent 列表）无法应对 MAS 三大动态性：① Agent 数量动态扩缩；② 能力随版本演化；③ 健康状态实时变化（宕机/过载/SLO 降级）。

2. 核心算法逻辑

3. 业务应用场景

场景一：WF-A 补货 Agent 灰度升级

新版补货 Agent v2（支持多货币汇率预测）上线时自动向 Registry 注册，声明新能力 `["replenishment", "fx_prediction"]`。Registry 广播变更通知，Orchestrator 灰度路由 10% 流量至 v2，每 5 分钟检查 SLO：v2 连续 3 次 fitness > v1 且 HEALTHY，自动升权重至 100%，v1 优雅下线（deregister）。

场景二：WF-D 选品 Agent 池（多能力调度）

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

ROI：MAS 从静态配置升级为动态服务网格，支持热更新和蓝绿发布，Agent 版本迭代零停机；Fitness 路由减少低质量决策暴露率
难度：⭐⭐⭐☆☆ | 优先级：⭐⭐⭐⭐⭐

7. 代码模板

代码块数量：1 · 路径：paper2skills-code/mas/agent_registry_discovery

"""
Agent Registry & Discovery — 动态注册与路由
来源：Agent Registry 2025-2026 + MCP/A2A 协议扩展
"""
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Set


class HealthStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    DOWN = "down"


class RoutingStrategy(Enum):
    FITNESS_FIRST = "fitness_first"
    LATENCY_FIRST = "latency_first"
    ROUND_ROBIN = "round_robin"


@dataclass
class AgentCapability:
    skill_names: List[str]
    domains: List[str]
    slo_target: float               # 如 0.999
    version: str
    fitness: float = 1.0            # 0.0-1.0，综合延迟+成功率
    p95_latency_ms: float = 100.0

    def matches(self, required_skills: List[str], domain: Optional[str] = None) -> float:
        if not required_skills:
            return 0.0
        required_set: Set[str] = set(required_skills)
        own_set: Set[str] = set(self.skill_names)
        intersection = required_set & own_set
        union = required_set | own_set
        jaccard = len(intersection) / len(union) if union else 0.0
        domain_bonus = 0.1 if domain and domain in self.domains else 0.0
        return min(1.0, jaccard + domain_bonus)


@dataclass
class AgentRegistration:
    agent_id: str
    endpoint: str
    capabilities: AgentCapability
    health_status: HealthStatus = HealthStatus.HEALTHY
    last_heartbeat: float = field(default_factory=time.time)
    registered_at: float = field(default_factory=time.time)
    _consecutive_failures: int = 0

    def update_heartbeat(self) -> None:
        self.last_heartbeat = time.time()
        self._consecutive_failures = 0
        self.health_status = HealthStatus.HEALTHY

    def mark_failure(self) -> None:
        self._consecutive_failures += 1

8. 论文来源

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