"""Load the Phase 1 App / Task registry from JSON.""" import json import os from typing import Any, Dict, Optional, Tuple from config.config_loader import get_project_dir from config.logger import setup_logging TAG = __name__ logger = setup_logging() _REGISTRY: Optional[Dict[str, Any]] = None _BUNDLED_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "apps.json") def _load_registry() -> Dict[str, Any]: candidates = [ os.path.join(get_project_dir(), "data", "apps.json"), _BUNDLED_PATH, ] for path in candidates: if not os.path.isfile(path): continue try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict) and data.get("apps"): logger.bind(tag=TAG).info(f"已加载 App 注册表: {path}") return data except Exception as e: logger.bind(tag=TAG).error(f"加载 App 注册表失败 {path}: {e}") logger.bind(tag=TAG).warning("未找到 apps.json,使用内置最小注册表") return { "default_app_id": "chat", "apps": { "chat": { "name": "聊天", "mode": "CHAT", "default_task": "conversation", "tool_policy": "inherit_agent", "system_prompt_mode": "agent_proactive", "tasks": {"conversation": {"interaction_mode": "CHAT", "skills": []}}, } }, } def get_registry() -> Dict[str, Any]: global _REGISTRY if _REGISTRY is None: _REGISTRY = _load_registry() return _REGISTRY def default_app_id() -> str: return get_registry().get("default_app_id") or "chat" def get_app(app_id: str) -> Optional[Dict[str, Any]]: if not app_id: return None return get_registry().get("apps", {}).get(app_id) def get_task(app_id: str, task_id: str) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]: app = get_app(app_id) if not app: return None, None tasks = app.get("tasks") or {} if not task_id: task_id = app.get("default_task") or "" task = tasks.get(task_id) if not task: return app, None return app, task