300 lines
14 KiB
Python
300 lines
14 KiB
Python
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
|