"""Unified start_app_task / switch_app / mode helpers."""
|
|
from __future__ import annotations
|
|
import uuid
|
from typing import Any, Dict, Optional, TYPE_CHECKING
|
|
from config.logger import setup_logging
|
from plugins_func.register import Action, ActionResponse
|
|
from .registry import default_app_id, get_app, get_task
|
|
if TYPE_CHECKING:
|
from core.connection import ConnectionHandler
|
|
TAG = __name__
|
logger = setup_logging()
|
|
SOURCE_PRIORITY = {
|
"mqtt_immediate": 3,
|
"schedule": 2,
|
"voice": 1,
|
"default_chat": 0,
|
}
|
|
TRANSLATOR_PROMPT = (
|
"你是一台翻译机。只输出译文，不要解释、不要拼音课、不要追问、不要闲聊。"
|
"未指定语言时：中文译成英文，英文译成中文；其它语言默认译成中文。"
|
)
|
|
CHAT_PROACTIVE_HINT = (
|
"你是用户的语音伙伴。进入对话后可以先主动短句邀请聊天，例如「我们来聊天吧」。"
|
"用户开口后按正常对话回答，不要反复催促。"
|
)
|
|
VOICE_MENU = "想聊天、听复读，还是打开翻译？也可以说我们来聊天吧。"
|
|
CHAT_PHRASES = ("我们来聊天吧", "聊会天吧", "回到聊天", "退出翻译", "结束翻译", "退出复读")
|
TRANSLATE_PHRASES = ("打开翻译", "开始翻译", "英语翻译", "翻译模式")
|
REPEATER_PHRASES = ("开始复读", "打开复读", "英语学习", "听复读")
|
|
DEVICE_PLAY_TOOL = "self_mp3_play"
|
DEVICE_STOP_TOOL = "self_mp3_stop"
|
|
|
def source_priority(source: str) -> int:
|
return SOURCE_PRIORITY.get(source or "voice", 1)
|
|
|
def _ensure_runtime_fields(conn: "ConnectionHandler") -> None:
|
if not hasattr(conn, "active_app_id"):
|
conn.active_app_id = default_app_id()
|
if not hasattr(conn, "current_task_id"):
|
conn.current_task_id = "conversation"
|
if not hasattr(conn, "interaction_mode"):
|
conn.interaction_mode = "CHAT"
|
if not hasattr(conn, "current_source"):
|
conn.current_source = "default_chat"
|
if not hasattr(conn, "tool_policy"):
|
conn.tool_policy = "inherit_agent"
|
if not hasattr(conn, "tool_allowlist"):
|
conn.tool_allowlist = []
|
if not hasattr(conn, "original_system_prompt"):
|
conn.original_system_prompt = conn.prompt
|
|
|
def filter_tools(conn: "ConnectionHandler", tools: Dict[str, Any]) -> Dict[str, Any]:
|
"""inherit_agent keeps the live agent toolset; allowlist keeps named tools only."""
|
_ensure_runtime_fields(conn)
|
if getattr(conn, "tool_policy", "inherit_agent") != "allowlist":
|
return tools
|
allow = set(getattr(conn, "tool_allowlist", None) or [])
|
if not allow:
|
return {}
|
filtered = {}
|
for name, definition in tools.items():
|
if name in allow:
|
filtered[name] = definition
|
return filtered
|
|
|
def _refresh_tools(conn: "ConnectionHandler") -> None:
|
handler = getattr(conn, "func_handler", None)
|
if handler and hasattr(handler, "upload_functions_desc"):
|
handler.upload_functions_desc()
|
|
|
def _apply_prompt(conn: "ConnectionHandler", app: Dict[str, Any]) -> None:
|
mode = app.get("system_prompt_mode") or "none"
|
if conn.original_system_prompt is None:
|
conn.original_system_prompt = conn.prompt
|
if mode == "translator":
|
conn.change_system_prompt(
|
"你现在是英语翻译机。用户说话后应调用 translate；用户说退出翻译、回到聊天时调用 switch_app，app_id=chat。"
|
"不要讲笑话、不要查天气、不要闲聊。"
|
)
|
elif mode == "agent_proactive":
|
base = conn.original_system_prompt or conn.prompt or ""
|
if CHAT_PROACTIVE_HINT not in (base or ""):
|
conn.change_system_prompt((base + "\n" + CHAT_PROACTIVE_HINT).strip())
|
else:
|
conn.change_system_prompt(base)
|
else:
|
if conn.original_system_prompt:
|
conn.change_system_prompt(conn.original_system_prompt)
|
|
|
async def _execute_named_tool(conn: "ConnectionHandler", tool_name: str, arguments: Dict[str, Any]):
|
handler = getattr(conn, "func_handler", None)
|
if not handler or not hasattr(handler, "tool_manager"):
|
return ActionResponse(action=Action.ERROR, response="工具处理器未就绪")
|
try:
|
return await handler.tool_manager.execute_tool(tool_name, arguments or {})
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"执行工具 {tool_name} 失败: {e}")
|
return ActionResponse(action=Action.ERROR, response=str(e))
|
|
|
async def _stop_current_skills(conn: "ConnectionHandler") -> None:
|
mode = getattr(conn, "interaction_mode", "CHAT")
|
if mode == "PLAY_AUDIO":
|
await _execute_named_tool(conn, DEVICE_STOP_TOOL, {})
|
conn.awaiting_voice_app_choice = False
|
|
|
def _speak(conn: "ConnectionHandler", text: str) -> None:
|
if not text:
|
return
|
from core.handle.intentHandler import speak_txt
|
|
if not getattr(conn, "sentence_id", None):
|
conn.sentence_id = uuid.uuid4().hex
|
try:
|
speak_txt(conn, text)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"主动播报失败: {e}")
|
|
|
def _greeting_text(conn: "ConnectionHandler") -> str:
|
return (conn.config or {}).get("proactive_chat_greeting") or "我们来聊天吧。"
|
|
|
async def start_app_task(
|
conn: "ConnectionHandler",
|
app_id: str,
|
task_id: str = "",
|
params: Optional[Dict[str, Any]] = None,
|
source: str = "voice",
|
execute_skills: bool = True,
|
) -> ActionResponse:
|
"""Single entry for MQTT / schedule / voice / default chat."""
|
_ensure_runtime_fields(conn)
|
params = params or {}
|
app, task = get_task(app_id, task_id)
|
if not app:
|
return ActionResponse(action=Action.ERROR, response=f"未知的 App: {app_id}")
|
if not task:
|
default_task = app.get("default_task") or ""
|
app, task = get_task(app_id, default_task)
|
if not task:
|
return ActionResponse(action=Action.ERROR, response=f"未知的 Task: {app_id}/{task_id}")
|
|
resolved_task_id = task_id or app.get("default_task") or "conversation"
|
new_p = source_priority(source)
|
cur_p = source_priority(getattr(conn, "current_source", "default_chat"))
|
same_task = (
|
getattr(conn, "active_app_id", "") == app_id
|
and getattr(conn, "current_task_id", "") == resolved_task_id
|
)
|
if new_p < cur_p and not same_task:
|
logger.bind(tag=TAG).info(
|
f"拒绝低优先级任务 {source}({new_p}) < {conn.current_source}({cur_p})"
|
)
|
return ActionResponse(
|
action=Action.NONE,
|
response="当前有更高优先级的任务，先完成后再切换",
|
)
|
|
if not same_task or new_p >= cur_p:
|
await _stop_current_skills(conn)
|
|
conn.active_app_id = app_id
|
conn.current_task_id = resolved_task_id
|
conn.interaction_mode = task.get("interaction_mode") or app.get("mode") or "CHAT"
|
conn.current_source = source
|
conn.tool_policy = app.get("tool_policy") or "inherit_agent"
|
conn.tool_allowlist = list(app.get("tools") or [])
|
_apply_prompt(conn, app)
|
_refresh_tools(conn)
|
|
logger.bind(tag=TAG).info(
|
f"start_app_task app={app_id} task={resolved_task_id} mode={conn.interaction_mode} source={source}"
|
)
|
|
if execute_skills:
|
for skill in task.get("skills") or []:
|
skill_name = (skill or {}).get("name") or ""
|
if skill_name in ("self.mp3.play", DEVICE_PLAY_TOOL):
|
play_args = {
|
"name": params.get("name") or "",
|
"repeat": int(params.get("repeat") or 1),
|
"power_off_after": bool(params.get("power_off_after") or False),
|
}
|
if params.get("speed") is not None:
|
play_args["speed"] = params.get("speed")
|
if not play_args["name"]:
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response="没有指定要复读的文件名，请先用 self.mp3.get_list 确认 files[].name",
|
)
|
result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
|
if result and result.action in (Action.ERROR, Action.NOTFOUND):
|
return ActionResponse(
|
action=Action.RESPONSE,
|
response=result.response or "这台设备没有本地复读播放能力",
|
)
|
elif skill_name == "self.power_off":
|
await _execute_named_tool(conn, "self_power_off", {})
|
|
if conn.interaction_mode == "CHAT":
|
enable = (conn.config or {}).get("enable_proactive_greeting", True)
|
woken = bool(getattr(conn, "just_woken_up", False))
|
if enable and not woken:
|
_speak(conn, _greeting_text(conn))
|
|
app_name = app.get("name") or app_id
|
return ActionResponse(
|
action=Action.NONE,
|
result=f"{app_id}/{resolved_task_id}",
|
response=f"已打开{app_name}",
|
)
|
|
|
async def switch_app(conn: "ConnectionHandler", app_id: str, source: str = "voice") -> ActionResponse:
|
app = get_app(app_id)
|
if not app:
|
return ActionResponse(action=Action.ERROR, response=f"不支持的 App: {app_id}")
|
task_id = app.get("default_task") or ""
|
return await start_app_task(conn, app_id, task_id, {}, source)
|
|
|
async def on_user_wakeup(conn: "ConnectionHandler") -> None:
|
"""User wake word returns to chat; do not treat [app_task] as wakeup."""
|
_ensure_runtime_fields(conn)
|
if getattr(conn, "interaction_mode", "CHAT") == "CHAT" and getattr(conn, "active_app_id", "chat") == "chat":
|
return
|
await start_app_task(conn, "chat", "conversation", {}, source="voice")
|
|
|
def match_voice_switch(text: str) -> Optional[tuple]:
|
"""Map a short phrase to (app_id, task_id) or ('__menu__', '')."""
|
if not text:
|
return None
|
_, filtered = _strip(text)
|
if filtered in ("切换模式", "qie huan mo shi"):
|
return ("__menu__", "")
|
for p in CHAT_PHRASES:
|
if p in filtered:
|
return ("chat", "conversation")
|
for p in TRANSLATE_PHRASES:
|
if p in filtered:
|
return ("english_translation", "live_translate")
|
for p in REPEATER_PHRASES:
|
if p in filtered:
|
return ("english_learning", "repeater")
|
if any(k in filtered for k in ("数字背诵", "成语接龙", "口算", "加减法")):
|
return ("__reserved__", "")
|
return None
|
|
|
def _strip(text: str) -> tuple:
|
from core.utils.util import remove_punctuation_and_length
|
|
return remove_punctuation_and_length(text)
|
|
|
def is_exit_translation(text: str) -> bool:
|
_, filtered = _strip(text)
|
return any(p in filtered for p in ("退出翻译", "结束翻译", "回到聊天", "我们来聊天吧", "聊会天吧"))
|
|
|
def _plain_user_text(text: str) -> str:
|
"""ASR 可能把原文包在 JSON 里，翻译只要 content。"""
|
if not text:
|
return ""
|
raw = str(text).strip()
|
if raw.startswith("{") and raw.endswith("}"):
|
try:
|
import json
|
|
data = json.loads(raw)
|
if isinstance(data, dict) and data.get("content"):
|
return str(data["content"]).strip()
|
except Exception:
|
pass
|
return raw
|
|
|
def _looks_like_tool_error(text: str) -> bool:
|
if not text:
|
return True
|
s = str(text)
|
if "Error executing tool" in s or "isError=True" in s or "isError=true" in s:
|
return True
|
if "未配置" in s and "api_key" in s:
|
return True
|
return False
|
|
|
async def run_translation(conn: "ConnectionHandler", text: str) -> str:
|
"""Translate with the connection LLM first (智控台已配的模型). MCP is fallback only."""
|
text = _plain_user_text(text)
|
if not text:
|
return ""
|
|
from plugins_func.register import all_function_registry
|
|
item = all_function_registry.get("translate")
|
if item:
|
try:
|
maybe = item.func(conn, text=text)
|
if hasattr(maybe, "__await__"):
|
maybe = await maybe
|
if isinstance(maybe, ActionResponse) and maybe.action not in (
|
Action.ERROR,
|
Action.NOTFOUND,
|
):
|
out = (maybe.response or maybe.result or "").strip()
|
if out and not _looks_like_tool_error(out):
|
return out
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"翻译插件失败: {e}")
|
|
handler = getattr(conn, "func_handler", None)
|
if handler and handler.has_tool("translate"):
|
result = await _execute_named_tool(conn, "translate", {"text": text})
|
if result and result.action not in (Action.ERROR, Action.NOTFOUND):
|
out = _extract_translate_text(result.response or result.result or "")
|
if out and not _looks_like_tool_error(out):
|
return out
|
|
llm = getattr(conn, "llm", None)
|
if llm and hasattr(llm, "response_no_stream"):
|
try:
|
return (
|
llm.response_no_stream(TRANSLATOR_PROMPT, text, temperature=0.1) or ""
|
).strip()
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"翻译 LLM 失败: {e}")
|
return ""
|
|
|
def _extract_translate_text(raw: Any) -> str:
|
if raw is None:
|
return ""
|
if isinstance(raw, str):
|
text = raw.strip()
|
if text.startswith("{") and "text" in text:
|
try:
|
import json
|
|
data = json.loads(text)
|
if isinstance(data, dict):
|
return str(data.get("text") or data.get("content") or text)
|
except Exception:
|
pass
|
return text
|
if hasattr(raw, "content"):
|
content = raw.content
|
if isinstance(content, list) and content:
|
first = content[0]
|
return getattr(first, "text", None) or str(first)
|
return str(content)
|
return str(raw).strip()
|
