feat: integrate safer vehicle CAN reasoning
This commit is contained in:
vee
@@ -0,0 +1,177 @@
|
||||
"""Fail-closed language guards applied before semantic IntentIR generation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from copy import deepcopy
|
||||
from typing import Any
|
||||
|
||||
|
||||
CLAUSE_SEPARATOR = re.compile(r"[，,；;。！？!?\n]+")
|
||||
NEGATED_OPERATION = re.compile(
|
||||
r"(?:不要|别|不想|无需|不必|禁止)"
|
||||
r"(?:再|去|把|将|给我|帮我|立即|马上){0,2}"
|
||||
r"(?P<verb>打开|开启|启动|关闭|关掉|解锁|开锁|上锁|落锁|锁上|进入|"
|
||||
r"切换到|调用|恢复|保存|升高|降低|后仰|前倾|前移|后移)"
|
||||
)
|
||||
PRESERVE_SEAT = re.compile(r"(?:不要|别|不想)(?:再)?动(?P<target>主驾|副驾)")
|
||||
CONTROL_VERB = re.compile(
|
||||
r"打开|开启|启动|关闭|关掉|解锁|开锁|上锁|落锁|锁上|进入|切换到|"
|
||||
r"调用|恢复|保存|升高|降低|后仰|前倾|前移|后移|调到|调为|设置为"
|
||||
)
|
||||
FUZZY_UNMAPPED_CONTROL = re.compile(
|
||||
r"(?P<text>(?:座椅|靠背|腿托|空调|车窗|风口|灯光?)"
|
||||
r"(?:调|弄|搞)(?:得|到)?(?:舒服|合适|好)(?:一点|点|些)?)"
|
||||
)
|
||||
|
||||
|
||||
def _span(utterance: str, match: re.Match[str]) -> dict[str, Any]:
|
||||
return {"text": match.group(0), "start": match.start(), "end": match.end()}
|
||||
|
||||
|
||||
def _target_from_text(text: str, anchor: int = 0) -> tuple[str | None, str | None]:
|
||||
candidates = [
|
||||
(abs(match.start() - anchor), match.start(), domain, target)
|
||||
for pattern, domain, target in (
|
||||
(r"左前", "window", "front_left"),
|
||||
(r"右前", "window", "front_right"),
|
||||
(r"左后", "window", "rear_left"),
|
||||
(r"右后", "window", "rear_right"),
|
||||
(r"主驾|驾驶位|司机位", "seat", "driver"),
|
||||
(r"副驾|副驾驶|乘客位", "seat", "passenger"),
|
||||
(r"后排", "door", "rear"),
|
||||
)
|
||||
for match in re.finditer(pattern, text)
|
||||
]
|
||||
if candidates:
|
||||
_, _, domain, target = min(candidates)
|
||||
return domain, target
|
||||
if "车窗" in text:
|
||||
return "window", "all" if re.search(r"全部|所有|四个|全车", text) else None
|
||||
if re.search(r"遮阳帘|天幕帘", text):
|
||||
return "roof", "vehicle"
|
||||
if "空调" in text:
|
||||
return "climate", "vehicle"
|
||||
if "氛围灯" in text:
|
||||
return "ambient_light", "vehicle"
|
||||
return None, None
|
||||
|
||||
|
||||
def _constraint(utterance: str, match: re.Match[str]) -> dict[str, Any]:
|
||||
clause = next(
|
||||
(item for item in CLAUSE_SEPARATOR.split(utterance) if match.group(0) in item),
|
||||
utterance,
|
||||
)
|
||||
domain, target = _target_from_text(clause, match.start())
|
||||
verb = match.group("verb")
|
||||
result: dict[str, Any] = {
|
||||
"type": "prohibit_action",
|
||||
"source_span": _span(utterance, match),
|
||||
}
|
||||
if domain:
|
||||
result["domain"] = domain
|
||||
if target:
|
||||
result["target"] = target
|
||||
parameters: dict[str, Any] = {}
|
||||
if domain == "window":
|
||||
result["capability"] = "window.set_motion"
|
||||
parameters["motion"] = "open" if verb in {"打开", "开启", "启动"} else "close"
|
||||
elif domain == "roof":
|
||||
result["capability"] = "roof.set_shade_motion"
|
||||
parameters["motion"] = "open" if verb in {"打开", "开启", "启动"} else "close"
|
||||
elif domain == "climate":
|
||||
result["capability"] = "climate.set_power"
|
||||
parameters["state"] = "on" if verb in {"打开", "开启", "启动"} else "off"
|
||||
elif domain == "ambient_light":
|
||||
result["capability"] = "ambient_light.set_power"
|
||||
parameters["state"] = "on" if verb in {"打开", "开启", "启动"} else "off"
|
||||
elif domain == "seat":
|
||||
if "通风" in clause:
|
||||
result["capability"] = "seat.set_ventilation"
|
||||
elif "加热" in clause:
|
||||
result["capability"] = "seat.set_heating"
|
||||
elif "靠背" in clause:
|
||||
result["capability"] = "seat.set_backrest_recline"
|
||||
if result.get("capability") in {"seat.set_ventilation", "seat.set_heating"}:
|
||||
parameters["state"] = "on" if verb in {"打开", "开启", "启动"} else "off"
|
||||
if parameters:
|
||||
result["parameters"] = parameters
|
||||
return result
|
||||
|
||||
|
||||
def _remove_seat_target(parsed: dict[str, Any], target: str) -> None:
|
||||
parsed["actions"] = [
|
||||
action for action in parsed.get("actions", []) if action.get("target") != target
|
||||
]
|
||||
for key in ("domain_actions", "native_can_actions", "deferred_requests", "model_requests"):
|
||||
parsed[key] = [
|
||||
request
|
||||
for request in parsed.get(key, [])
|
||||
if not (
|
||||
request.get("domain") == "seat"
|
||||
and request.get("target", request.get("target_scope")) == target
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def apply_language_safety_guards(parsed: dict[str, Any], utterance: str) -> dict[str, Any]:
|
||||
"""Preserve negation and unresolved control language without producing CAN."""
|
||||
|
||||
result = deepcopy(parsed)
|
||||
constraints = list(result.get("semantic_constraints", []))
|
||||
unresolved = list(result.get("unresolved_spans", []))
|
||||
|
||||
for match in PRESERVE_SEAT.finditer(utterance):
|
||||
target = "driver" if match.group("target") == "主驾" else "passenger"
|
||||
constraints.append(
|
||||
{
|
||||
"type": "prohibit_change",
|
||||
"domain": "seat",
|
||||
"target": target,
|
||||
"source_span": _span(utterance, match),
|
||||
}
|
||||
)
|
||||
_remove_seat_target(result, target)
|
||||
|
||||
negated = list(NEGATED_OPERATION.finditer(utterance))
|
||||
if negated:
|
||||
constraints.extend(_constraint(utterance, item) for item in negated)
|
||||
for key in (
|
||||
"actions", "domain_actions", "native_can_actions", "deferred_requests",
|
||||
"model_requests", "preset_actions", "preset_events",
|
||||
):
|
||||
result[key] = []
|
||||
result["ambiguities"] = []
|
||||
if len(CONTROL_VERB.findall(utterance)) > len(negated):
|
||||
unresolved.append(
|
||||
{
|
||||
"text": utterance,
|
||||
"reason": "mixed_negated_and_positive_operations_require_atomic_confirmation",
|
||||
}
|
||||
)
|
||||
result.setdefault("ambiguities", []).append(
|
||||
{
|
||||
"field": "utterance",
|
||||
"question": "同一句中同时包含否定和执行动作，请拆分或确认完整目标。",
|
||||
"reason": "mixed_negated_and_positive_operations",
|
||||
}
|
||||
)
|
||||
result["decision_summary"] = [
|
||||
*result.get("decision_summary", []),
|
||||
"检测到否定控制约束；未把“不要/不想”错误转换为相反动作。",
|
||||
]
|
||||
result["confidence"] = min(float(result.get("confidence", 0)), 0.8)
|
||||
|
||||
if not result.get("actions") and not result.get("domain_actions"):
|
||||
for match in FUZZY_UNMAPPED_CONTROL.finditer(utterance):
|
||||
if not any(item.get("text") == match.group("text") for item in unresolved):
|
||||
unresolved.append(
|
||||
{
|
||||
"text": match.group("text"),
|
||||
"reason": "fuzzy_control_goal_not_resolved_to_absolute_target",
|
||||
}
|
||||
)
|
||||
|
||||
result["semantic_constraints"] = constraints
|
||||
result["unresolved_spans"] = unresolved
|
||||
return result
|
||||
Reference in New Issue
Block a user