Registry 模块¶

核心文件¶

src/by_framework/core/extensions/registry.py - PluginRegistry
src/by_framework/client/registry.py - WorkerRegistry

PluginRegistry¶

管理插件注册和发现。

主要方法¶

class PluginRegistry:
    def register_bundle(self, plugin: Plugin) -> None:
        """注册插件"""

    def get_plugin(self, plugin_id: str) -> Optional[Plugin]:
        """获取指定ID的插件"""

    def list_plugins(self) -> List[Plugin]:
        """列出所有已注册插件"""

    def build_agent_configs(self) -> Dict[str, AgentConfig]:
        """构建所有Agent配置"""

    def clear(self) -> None:
        """清空所有注册"""

WorkerRegistry¶

管理 Worker 的在线状态和心跳。

主要方法¶

class WorkerRegistry:
    def __init__(self, redis_client: Redis) -> None:
        self._redis = redis_client

    async def register_worker(
        self,
        worker_id: str,
        agent_types: List[str],
    ) -> None:
        """注册 Worker"""

    async def unregister_worker(self, worker_id: str) -> None:
        """注销 Worker"""

    async def heartbeat(self, worker_id: str) -> None:
        """发送心跳"""

    async def get_online_workers(
        self,
        agent_type: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """获取在线 Workers"""

    async def is_worker_online(self, worker_id: str) -> bool:
        """检查 Worker 是否在线"""

Redis Key 模式¶

Key	描述
`worker:active:{worker_id}`	Worker 心跳注册
`worker:members:{agent_type}`	Agent type 成员集合