"""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
|
