小智终端 5种 场景模式 OK

wangzhibo

2026-09-11 659c470eba542383987aad227b7f918100f131fa

 xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/registry.py

@@ -1,4 +1,10 @@
"""Load the Phase 1 App / Task registry from JSON."""
"""Load the Phase 1 App / Task registry from JSON.

Registry lives on xiaozhi-server only (bundled apps.json, optional data/apps.json).
Task voice_enter / voice_exit / enter_prompt live here so new Apps do not need a
firmware flash. The device only keeps a small offline MultiNet set (mode switch,
repeater, exit repeater, return to chat) plus NVS pending / last-repeater state.
"""

import json
import os
@@ -11,8 +17,14 @@
logger = setup_logging()

_REGISTRY: Optional[Dict[str, Any]] = None
_VOICE_INDEX: Optional[list] = None

_BUNDLED_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "apps.json")
_MENU_PRIO = 0
_EXIT_PRIO = 1
_CHAT_ENTER_PRIO = 2
_ENTER_PRIO = 3
_RESERVED_PRIO = 4


def _load_registry() -> Dict[str, Any]:
@@ -54,6 +66,93 @@
    return _REGISTRY


def reset_registry_cache() -> None:
    """Tests may swap the registry; drop derived indexes."""
    global _VOICE_INDEX
    _VOICE_INDEX = None


def iter_tasks():
    apps = get_registry().get("apps") or {}
    for app_id, app in apps.items():
        if not isinstance(app, dict):
            continue
        for task_id, task in (app.get("tasks") or {}).items():
            if isinstance(task, dict):
                yield app_id, app, task_id, task


def _phrase_list(raw) -> list:
    if raw is None:
        return []
    if isinstance(raw, str):
        text = raw.strip()
        return [text] if text else []
    phrases = []
    for item in raw:
        text = str(item or "").strip()
        if text:
            phrases.append(text)
    return phrases


def _build_voice_index() -> list:
    """Longer phrase first, then menu/exit/chat-enter/enter/reserved."""
    entries = []
    registry = get_registry()
    for phrase in _phrase_list(registry.get("voice_menu_phrases")):
        entries.append((phrase, ("__menu__", ""), _MENU_PRIO))
    for app_id, app, task_id, task in iter_tasks():
        default_task = app.get("default_task") or task_id
        for phrase in _phrase_list(task.get("voice_exit")):
            entries.append((phrase, ("chat", "conversation"), _EXIT_PRIO))
        for phrase in _phrase_list(task.get("voice_enter")):
            if app_id == "chat":
                entries.append((phrase, ("chat", default_task or task_id), _CHAT_ENTER_PRIO))
            else:
                entries.append((phrase, (app_id, task_id), _ENTER_PRIO))
    for phrase in _phrase_list(registry.get("reserved_voice_enter")):
        entries.append((phrase, ("__reserved__", ""), _RESERVED_PRIO))
    entries.sort(key=lambda item: (-len(item[0]), item[2]))
    return entries


def _voice_index() -> list:
    global _VOICE_INDEX
    if _VOICE_INDEX is None:
        _VOICE_INDEX = _build_voice_index()
    return _VOICE_INDEX


def match_registry_voice(filtered: str) -> Optional[Tuple[str, str]]:
    """Map stripped ASR text to (app_id, task_id) or ('__menu__'|'__reserved__', '')."""
    if not filtered:
        return None
    for phrase, target, _prio in _voice_index():
        if phrase and phrase in filtered:
            return target
    return None


def voice_menu_text() -> str:
    labels = []
    chat_hint = "我们来聊天吧"
    for app_id, _app, task_id, task in iter_tasks():
        enters = _phrase_list(task.get("voice_enter"))
        if app_id == "chat" and task_id == "conversation" and enters:
            chat_hint = enters[0]
        name = str(task.get("menu_name") or "").strip()
        if name:
            labels.append(name)
    if not labels:
        return f"想聊天？也可以说{chat_hint}。"
    if len(labels) == 1:
        body = labels[0]
    else:
        body = "、".join(labels[:-1]) + f"，还是{labels[-1]}"
    return f"想{body}？也可以说{chat_hint}。"


def default_app_id() -> str:
    return get_registry().get("default_app_id") or "chat"