from __future__ import annotations import re from typing import Any from app.schemas.intent import IntentDefinition class ResponsePolicy: _DEFAULT_INTENT_HINTS = { "cabin_nav_to": "导航", "cabin_nav_cancel": "结束导航", "cabin_set_ac": "调空调", "cabin_ac_on": "打开空调", "cabin_ac_off": "关闭空调", "cabin_fan_up": "调大风量", "cabin_fan_down": "调小风量", "cabin_defog_front_on": "打开前挡除雾", "cabin_defog_rear_on": "打开后挡除雾", "cabin_window_open": "打开车窗", "cabin_window_close": "关闭车窗", "cabin_sunroof_open": "打开天窗", "cabin_sunroof_close": "关闭天窗", "cabin_trunk_open": "打开后备箱", "cabin_trunk_close": "关闭后备箱", "cabin_lock_doors": "锁车门", "cabin_unlock_doors": "解锁车门", "cabin_play_music": "播放音乐", "cabin_pause_music": "暂停音乐", "cabin_next_track": "下一首", "cabin_previous_track": "上一首", "cabin_volume_up": "调大音量", "cabin_volume_down": "调小音量", "cabin_volume_mute": "静音", "cabin_lights_on": "打开车灯", "cabin_lights_off": "关闭车灯", "cabin_reading_light_on": "打开阅读灯", "cabin_reading_light_off": "关闭阅读灯", "cabin_seat_heat_on": "打开座椅加热", "cabin_seat_heat_off": "关闭座椅加热", "cabin_seat_vent_on": "打开座椅通风", "cabin_seat_vent_off": "关闭座椅通风", "cabin_mirror_fold": "折叠后视镜", "cabin_mirror_unfold": "展开后视镜", "cabin_wiper_on": "打开雨刷", "cabin_wiper_off": "关闭雨刷", "cabin_screen_brightness_up": "调亮屏幕", "cabin_screen_brightness_down": "调暗屏幕", "cabin_answer_call": "接听电话", "cabin_hang_up_call": "挂断电话", "cs_query_order": "查订单", "cs_query_logistics": "查物流", "cs_cancel_order": "取消订单", "cs_transfer_human": "转人工", } def __init__( self, templates: dict[str, str] | None = None, intent_hints: dict[str, str] | None = None, ) -> None: self._templates = templates or {} self._intent_hints = {**self._DEFAULT_INTENT_HINTS, **(intent_hints or {})} def ask_for_slot(self, intent: IntentDefinition, slot_name: str, default_template: str) -> str: if slot_name == "order_id": if intent.intent_id == "cs_cancel_order": return self._template("ask_cancel_order_id", "请告诉我订单号。") return self._template("ask_order_id", "请提供订单号。") if slot_name == "destination": return self._template("ask_destination", "请告诉我要去哪里。") if slot_name == "temperature": return self._template("ask_temperature", "请告诉我要设置多少度。") if slot_name == "media_query": return self._template("ask_media_query", "想听什么风格或者具体的歌名？") return default_template.strip() or "请补充一个关键信息。" def workflow_result(self, intent: IntentDefinition, plugin_result: dict[str, Any]) -> str: if not plugin_result.get("success", True): return self._template("workflow_failed", "这次没处理成功，请稍后再试。") message = str(plugin_result.get("message") or "").strip() if not message: return self.ack(intent) if len(message) > 42: return message[:39].rstrip("，。；; ") + "..." return message def workflow_summary(self, messages: list[str]) -> str: cleaned = [item.strip() for item in messages if item and item.strip()] if not cleaned: return self._template("workflow_summary_empty", "好的，已经处理完成。") if len(cleaned) == 1: return cleaned[0] natural_clauses: list[str] = [] previous_subject: str | None = None for index, item in enumerate(cleaned[:3]): clause, subject = self._vehicle_style_clause(item, index=index, previous_subject=previous_subject) natural_clauses.append(clause) previous_subject = subject or previous_subject summary = f"好，{'，'.join(natural_clauses)}。" if len(cleaned) > 3: summary = summary.rstrip("。") + "，其余步骤也已完成。" if len(summary) > 70: return summary[:67].rstrip("，。；; ") + "..." return summary def ask_for_confirmation(self, intent: IntentDefinition, detail: str | None = None) -> str: if intent.intent_id == "cs_cancel_order": if detail: return f"即将取消订单，{detail}。请回复“确认”或“取消”。" return "即将取消订单。请回复“确认”或“取消”。" if detail: return f"{detail}。请回复“确认”或“取消”。" return "请确认是否继续执行。回复“确认”或“取消”。" def confirm_retry(self) -> str: return self._template("confirm_retry", "我需要一个明确确认。请回复“确认”继续，或回复“取消”终止。") def confirm_cancelled(self) -> str: return self._template("confirm_cancelled", "好的，已取消这一步。") def step_skipped(self, intent: IntentDefinition, reason: str | None = None) -> str: if intent.intent_id == "cs_cancel_order": base = "订单取消步骤未执行。" else: base = "这一步已跳过。" if reason: return f"{base}{reason}" return base def ack(self, intent: IntentDefinition | None = None) -> str: if intent is None: return self._template("ack_default", "收到，马上处理。") if intent.domain == "cabin": return self._template("ack_cabin", "好的，马上处理。") return self._template("ack_service", "收到，我来处理。") def reject(self) -> str: return self._template("reject", "这个我暂时做不了，但我可以帮你查询、控制或转人工。") def short_social(self, social_kind: str) -> str: if social_kind == "greeting": return self._template("short_social_greeting", "你好，我在。") if social_kind == "thanks": return self._template("short_social_thanks", "不客气。") if social_kind == "goodbye": return self._template("short_social_goodbye", "好的，有需要再叫我。") if social_kind == "capability": return self._template( "short_social_capability", "我可以帮你查订单、查物流、取消订单、导航、调空调、播放音乐或转人工。", ) return self._template("short_social_default", "我在。") def open_social_fallback(self) -> str: return self._template("open_social_fallback", "可以和你聊两句，你也可以继续告诉我想处理什么。") def with_pending_hint(self, text: str, pending_hint: str | None = None) -> str: base = text.strip() or self.open_social_fallback() hint = (pending_hint or "").strip() if not hint: return base return f"{base} {hint}" def pending_task_hint(self, status: str, pending_slots: list[str], current_intent: str | None = None) -> str | None: if status == "waiting_confirmation": return self._template("pending_confirmation_hint", "当前这一步还在等你确认，回复“确认”或“取消”即可。") if status == "waiting_slot" and pending_slots: if pending_slots[0] == "order_id": return self._template("pending_slot_order_id", "当前还缺订单号，你继续告诉我订单号就行。") if pending_slots[0] == "temperature": return self._template("pending_slot_temperature", "当前还缺温度，你继续告诉我要设置多少度就行。") if pending_slots[0] == "destination": return self._template("pending_slot_destination", "当前还缺目的地，你继续告诉我要去哪里就行。") if pending_slots[0] == "media_query": return self._template("pending_slot_media_query", "当前还缺歌名或风格，你直接说歌名、歌手或风格就行。") return self._template("pending_slot_default", "当前还缺一个关键信息，你继续补充就行。") if status == "running" and current_intent: return self._template("pending_running", "当前任务还在继续，你也可以直接继续下一个指令。") return None def task_stopped(self) -> str: return self._template("task_stopped", "好的，已停止当前任务。") def clarify(self, candidate_intents: list[str]) -> str: options = [ self._intent_hints.get(intent_id, intent_id) for intent_id in candidate_intents if intent_id ] deduped: list[str] = [] for item in options: if item not in deduped: deduped.append(item) if not deduped: return "我理解得还不够确定，你是想查询、控制，还是转人工？" if len(deduped) == 1: return f"请确认一下，你是想{deduped[0]}吗？" if len(deduped) == 2: return f"请确认一下，你是想{deduped[0]}还是{deduped[1]}？" return f"请确认一下，你是想{deduped[0]}、{deduped[1]}，还是{deduped[2]}？" def fallback(self) -> str: return self._template("fallback", "我还没完全听懂，你可以换个简短说法，或告诉我是查询、控制还是转人工。") def _template(self, key: str, default: str) -> str: value = str(self._templates.get(key, default)).strip() return value or default def _naturalize_workflow_message(self, text: str) -> str: normalized = text.strip().rstrip("。；; ") normalized = re.sub(r"^好的[，,\s]*", "", normalized) normalized = re.sub(r"^收到[，,\s]*", "", normalized) if normalized.startswith("已将"): normalized = normalized[2:] elif normalized.startswith("已经将"): normalized = normalized[3:] elif normalized.startswith("已经"): normalized = normalized[2:] elif normalized.startswith("已"): normalized = normalized[1:] normalized = normalized.strip("，, ") if not normalized: return "已经处理好了" if normalized.endswith("了"): return normalized return f"{normalized}了" def _vehicle_style_clause( self, text: str, *, index: int, previous_subject: str | None = None, ) -> tuple[str, str | None]: normalized = self._naturalize_workflow_message(text) match = re.match(r"^(打开|关闭)(.+)了$", normalized) if match: action, subject = match.groups() subject = subject.strip() if action == "打开": if previous_subject and previous_subject == subject: return "也打开了", subject if index > 0: return f"{subject}也打开了", subject return f"{subject}已经打开了", subject if previous_subject and previous_subject == subject: return "也帮你关上了", subject if index > 0: return f"{subject}也帮你关上了", subject return f"{subject}已经关上了", subject match = re.match(r"^(锁定|解锁)(.+)了$", normalized) if match: action, subject = match.groups() subject = subject.strip() action_text = "锁好了" if action == "锁定" else "解锁了" if previous_subject and previous_subject == subject: return f"也{action_text}", subject if index > 0: return f"{subject}也{action_text}", subject return f"{subject}已经{action_text}", subject match = re.match(r"^(.+)调到\s*(.+)度了$", normalized) if match: subject, value = match.groups() subject = subject.strip() value = value.strip() if previous_subject and previous_subject == subject: return f"也调到 {value} 度了", subject if index > 0: return f"{subject}也调到 {value} 度了", subject return f"{subject}调到 {value} 度了", subject match = re.match(r"^(调大|调小)(.+)了$", normalized) if match: action, subject = match.groups() subject = subject.strip() if previous_subject and previous_subject == subject: return f"也{action}了", subject if index > 0: return f"{subject}也{action}了", subject return f"{subject}已经{action}了", subject if normalized.startswith("正在播放 "): target = normalized[len("正在播放 ") :].strip() if index > 0: return f"也开始播放 {target} 了", "播放" return f"开始播放 {target} 了", "播放" if normalized.startswith("订单 ") and normalized.endswith(" 已取消"): order_text = normalized[:-4].strip() return f"{order_text}已经取消了", "订单" if normalized.startswith("订单 ") and "当前" in normalized: return normalized, "订单" return normalized, None