1

main/xiaozhi-server/core/handle/intentHandler.py

@@ -17,6 +17,41 @@ from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
 TAG = __name__
 
 
+STRICT_HAZARD_MARK_PHRASES = {
+    "标记急转弯": "sharp_turn",
+    "标记起转弯": "sharp_turn",
+    "标记急弯": "sharp_turn",
+    "标记左转弯": "left_turn",
+    "标记右转弯": "right_turn",
+    "标记障碍": "obstacle",
+    "标记颠簸": "bump",
+    "标记大坑": "bump",
+    "标记坑": "bump",
+    "标记跳台": "jump",
+    "标记陡坡": "slope",
+    "标记涉水": "water",
+    "标记水坑": "water",
+}
+
+NAV_ALERT_ECHO_PHRASES = {
+    "前方左转弯",
+    "前方右转弯",
+    "准备急转弯",
+    "前方颠簸",
+    "前方水坑",
+    "前方障碍",
+    "前方跳台",
+    "前方陡坡",
+    "前方风险点",
+    "注意左转弯",
+    "注意右转弯",
+    "注意急转弯",
+    "注意颠簸",
+    "注意水坑",
+    "注意障碍",
+}
+
+
 async def handle_user_intent(conn: "ConnectionHandler", text):
     # 预处理输入文本，处理可能的JSON格式
     try:
@@ -37,6 +72,10 @@ async def handle_user_intent(conn: "ConnectionHandler", text):
     if await checkWakeupWords(conn, filtered_text):
         return True
 
+    if should_ignore_nav_alert_echo(filtered_text):
+        conn.logger.bind(tag=TAG).info(f"忽略导航播报回采文本: {filtered_text}")
+        return True
+
     direct_function_call = match_direct_nav_command(filtered_text)
     if direct_function_call is not None:
         function_name, function_args = direct_function_call
@@ -133,27 +172,20 @@ def match_direct_nav_command(filtered_text: str):
         return "nav_stop_run", {}
 
     if text.startswith("标记"):
-        if any(word in text for word in ("急转弯", "起转弯", "急弯")):
-            return "nav_mark_hazard", {"hazard_type": "sharp_turn"}
-        if "左转弯" in text:
-            return "nav_mark_hazard", {"hazard_type": "left_turn"}
-        if "右转弯" in text:
-            return "nav_mark_hazard", {"hazard_type": "right_turn"}
-        if "障碍" in text:
-            return "nav_mark_hazard", {"hazard_type": "obstacle"}
-        if any(word in text for word in ("颠簸", "大坑", "坑")):
-            return "nav_mark_hazard", {"hazard_type": "bump"}
-        if "跳台" in text:
-            return "nav_mark_hazard", {"hazard_type": "jump"}
-        if "陡坡" in text:
-            return "nav_mark_hazard", {"hazard_type": "slope"}
-        if any(word in text for word in ("涉水", "水坑")):
-            return "nav_mark_hazard", {"hazard_type": "water"}
-        return "nav_mark_hazard", {"hazard_type": "other"}
+        if text in STRICT_HAZARD_MARK_PHRASES:
+            return "nav_mark_hazard", {"hazard_type": STRICT_HAZARD_MARK_PHRASES[text]}
+        return "nav_mark_hazard_reject", {}
 
     return None
 
 
+def should_ignore_nav_alert_echo(filtered_text: str) -> bool:
+    text = (filtered_text or "").strip()
+    if not text:
+        return False
+    return text in NAV_ALERT_ECHO_PHRASES
+
+
 async def dispatch_function_call(
     conn: "ConnectionHandler",
     original_text: str,

main/xiaozhi-server/plugins_func/functions/nav_copilot.py

@@ -155,6 +155,24 @@ async def nav_mark_hazard(conn, hazard_type: str, note: str | None = None):
     )
 
 
+nav_mark_hazard_reject_desc = {
+    "type": "function",
+    "function": {
+        "name": "nav_mark_hazard_reject",
+        "description": "当用户进入风险点录制口令但没有说出有效风险类型时，提示其保持安静并等待重新录入。",
+        "parameters": {"type": "object", "properties": {}, "required": []},
+    },
+}
+
+
+@register_function("nav_mark_hazard_reject", nav_mark_hazard_reject_desc, ToolType.SYSTEM_CTL)
+async def nav_mark_hazard_reject(conn):
+    return ActionResponse(
+        action=Action.RESPONSE,
+        response="请保持安静，等待录入风险点。",
+    )
+
+
 nav_start_run_desc = {
     "type": "function",
     "function": {