Files
Cognitive_Agent/src/cognitive_agent/application/semantic_ir.py
T

597 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Fail-closed language guards applied before semantic IntentIR generation."""
from __future__ import annotations
import json
import re
from collections.abc import Mapping
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"(?:调|弄|搞)(?:得|到)?(?:舒服|合适|好)(?:一点|点|些)?)"
)
OUTLET_MODE_DISABLE = re.compile(
r"(?:关闭|关掉|停止|取消)(?:一下)?(?:空调)?(?:的)?"
r"(?P<feature>吹面|吹脸|吹脚|除霜)(?:出风口|出风|模式)?|"
r"(?P<feature_after>吹面|吹脸|吹脚|除霜)(?:出风口|出风|模式)?"
r"(?:关闭|关掉|停止|取消)"
)
POSITIVE_OUTLET_MODE = re.compile(
r"(?:打开|开启|启用|切换到|切换为|改为|设为|设置为|调整为|调为)"
r"(?:空调)?(?:的)?(?:吹面|吹脸|吹脚|除霜)"
)
DRIVER_SEAT_THERMAL_MIDDLE = re.compile(
r"(?:通风|加热)[^,,;;。!?!?\n]{0,16}"
r"(?:中等(?:档|级)?|中档|中级|中间档|适中|(?:调到|调为|调|设为|设置为|调整到|调整为)中)|"
r"(?:中等(?:档|级)?|中档|中级|中间档|适中)"
r"[^,,;;。!?!?\n]{0,16}(?:通风|加热)"
)
META_NON_EXECUTION = re.compile(
r"复述|引用|翻译|改写|总结|解释|分析|造句|"
r"故事|剧本|小说|作文|角色|主角|假设|假如|如果|"
r"示例|举例|这句话|以下内容"
)
_EXECUTION_LIST_FIELDS = (
"actions",
"domain_actions",
"native_can_actions",
"deferred_requests",
"model_requests",
"preset_actions",
"preset_events",
)
_SEAT_CONTEXT = re.compile(
r"座椅|座位|靠背|椅背|坐盆|坐垫|腿托|脚托|"
r"座椅通风|座椅加热|按摩|座椅记忆|零重力|木马|摇篮"
)
_SEAT_DRIVER_SCOPE = re.compile(
r"\bdriver\b|\bfront[-_ ]?left\b|主驾|主驾驶|驾驶位|驾驶员|"
r"驾驶席|驾驶座(?:椅)?|司机位|司机座(?:椅)?|左前"
)
_SEAT_PASSENGER_SCOPE = re.compile(
r"\bpassenger\b|\bfront[-_ ]?right\b|副驾|副驾驶|乘客|乘员|右前"
)
_SEAT_UNSUPPORTED_SCOPE = re.compile(
r"前排|后排|后座|二排|第二排|中排|第三排|三排|"
r"左后|右后|前舱|后舱|左侧|右侧|左边|右边|"
r"所有座椅|全部座椅|全车座椅"
)
def resolve_explicit_seat_scopes(
utterance: str,
*,
assume_seat_context: bool = False,
) -> frozenset[str] | None:
"""Resolve scopes only from clauses that actually describe a seat action."""
scopes: set[str] = set()
relevant_clause_seen = False
clauses = [item for item in CLAUSE_SEPARATOR.split(utterance) if item.strip()]
for clause in clauses or [utterance]:
if not assume_seat_context and not _SEAT_CONTEXT.search(clause):
continue
relevant_clause_seen = True
if _SEAT_UNSUPPORTED_SCOPE.search(clause):
scopes.add("unsupported")
continue
if _SEAT_DRIVER_SCOPE.search(clause):
scopes.add("driver")
if _SEAT_PASSENGER_SCOPE.search(clause):
scopes.add("passenger")
if not relevant_clause_seen or not scopes:
return None
return frozenset(scopes)
def _filter_explicit_seat_scope(
result: dict[str, Any],
requested_scopes: frozenset[str],
) -> bool:
"""Remove seat operations outside the user's explicit scope."""
allowed: set[str] = set()
if "driver" in requested_scopes:
allowed.update(("driver", "front_left"))
if "passenger" in requested_scopes:
allowed.update(("passenger", "front_right"))
if "unsupported" in requested_scopes:
allowed.clear()
removed = False
for key in _EXECUTION_LIST_FIELDS:
kept: list[Any] = []
for item in result.get(key, []):
if not isinstance(item, dict):
kept.append(item)
continue
capability = str(item.get("capability", ""))
is_seat = item.get("domain") == "seat" or capability.startswith("seat.")
if not is_seat:
kept.append(item)
continue
target = str(item.get("target", item.get("target_scope", "")))
if target in allowed:
kept.append(item)
else:
removed = True
result[key] = kept
return removed
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 _remove_capability(parsed: dict[str, Any], capability: str) -> None:
for key in (
"actions",
"domain_actions",
"native_can_actions",
"deferred_requests",
"model_requests",
):
parsed[key] = [
request
for request in parsed.get(key, [])
if request.get("capability") != capability
]
def _parameter_mapping(item: Mapping[str, Any]) -> dict[str, Any]:
parameters: object = item.get("parameters", {})
if isinstance(parameters, list):
return {
str(parameter.get("name")): parameter.get("value")
for parameter in parameters
if isinstance(parameter, dict) and parameter.get("name")
}
if isinstance(parameters, Mapping):
normalized = {str(name): value for name, value in parameters.items()}
if normalized:
return normalized
# The passenger-seat domain predates the capability-shaped actions used by
# the other domains. Normalize its legacy action fields here so the shared
# conflict gate runs before FunctionConfig compilation and execution.
legacy: dict[str, Any] = {}
for source, target in (
("state", "state"),
("requested_level", "level"),
("position_pct", "position_pct"),
("angle_deg", "angle_deg"),
):
if item.get(source) is not None:
legacy[target] = item.get(source)
return legacy
def _control_capability(item: Mapping[str, Any]) -> str:
capability = str(item.get("capability", ""))
if capability:
return capability
function = str(item.get("function", ""))
if function in {"ventilation", "heating"}:
return f"seat.set_{function}"
if item.get("action_type") == "position":
return {
"longitudinal": "seat.set_longitudinal",
"cushion_tilt": "seat.set_cushion_tilt",
"backrest_recline": "seat.set_backrest_recline",
"legrest_height": "seat.set_legrest_height",
}.get(str(item.get("axis", "")), "")
return ""
def _control_conflict_key(
item: Mapping[str, Any],
parameters: Mapping[str, Any],
) -> tuple[str, str, str, str]:
target: object = item.get("target", item.get("target_scope", ""))
if isinstance(target, Mapping):
target = target.get("zone", target.get("device", ""))
capability = _control_capability(item)
domain = str(item.get("domain", "")) or (
capability.split(".", 1)[0] if "." in capability else ""
)
selector = ""
if capability == "seat.set_driver_axis_native":
selector = f"axis:{parameters.get('axis', '')}"
elif capability == "seat.set_driver_massage_native":
selector = f"action:{parameters.get('action', '')}"
return domain, capability, str(target), selector
def _conflicting_control_keys(
result: dict[str, Any],
) -> set[tuple[str, str, str, str]]:
"""Find contradictory values, while allowing complementary subcommands.
Some capabilities intentionally emit more than one operation for the same
target: massage mode plus strength, or ambient colour plus brightness.
Those parameter sets are complementary. A conflict exists only when the
same setting parameter receives multiple values. ``action`` and ``axis``
select a subcommand and therefore are not themselves setting values.
"""
values: dict[tuple[str, str, str, str], dict[str, set[str]]] = {}
# ``model_requests`` is audit metadata and can still contain a built-in
# mode's superseded defaults. The executable action lists below have
# already applied explicit-user-value precedence, so conflict decisions
# must be based on those lists only.
for field in ("actions", "domain_actions", "native_can_actions"):
for item in result.get(field, []):
if not isinstance(item, dict) or not _control_capability(item):
continue
parameters = _parameter_mapping(item)
if not parameters:
continue
key = _control_conflict_key(item, parameters)
settings = values.setdefault(key, {})
for name, value in parameters.items():
if str(name) in {"action", "axis"}:
continue
settings.setdefault(str(name), set()).add(
json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
)
return {
key
for key, settings in values.items()
if any(len(candidates) > 1 for candidates in settings.values())
}
def _remove_conflicting_controls(
result: dict[str, Any],
conflicts: set[tuple[str, str, str, str]],
) -> None:
for field in _EXECUTION_LIST_FIELDS:
kept: list[Any] = []
for item in result.get(field, []):
if not isinstance(item, dict) or not _control_capability(item):
kept.append(item)
continue
key = _control_conflict_key(item, _parameter_mapping(item))
if key not in conflicts:
kept.append(item)
result[field] = kept
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", []))
meta_match = META_NON_EXECUTION.search(utterance)
if meta_match is not None:
for key in _EXECUTION_LIST_FIELDS:
result[key] = []
constraints.append(
{
"type": "prohibit_meta_language_execution",
"source_span": _span(utterance, meta_match),
}
)
unresolved.append(
{
"text": utterance,
"reason": "meta_language_is_not_vehicle_authorization",
}
)
result["ambiguities"] = []
result["decision_summary"] = [
*result.get("decision_summary", []),
"检测到复述、翻译或创作类元语言;未将其中的控制词当作车辆授权。",
]
result["confidence"] = min(float(result.get("confidence", 0)), 0.8)
explicit_seat_scopes = resolve_explicit_seat_scopes(utterance)
if explicit_seat_scopes is not None:
removed = _filter_explicit_seat_scope(result, explicit_seat_scopes)
if removed:
constraints.append(
{
"type": "restrict_explicit_seat_scope",
"domain": "seat",
"target": ",".join(sorted(explicit_seat_scopes)),
}
)
result["decision_summary"] = [
*result.get("decision_summary", []),
"已移除与用户显式座位不一致的座椅动作,未扩大控制范围。",
]
if "unsupported" in explicit_seat_scopes:
unresolved.append(
{
"text": utterance,
"reason": "explicit_seat_scope_is_not_safely_callable",
}
)
result["ambiguities"] = [
*result.get("ambiguities", []),
{
"field": "target_scope",
"question": "当前请求的座位范围不能在不扩大目标的前提下执行。",
"reason": "explicit_seat_scope_is_not_safely_callable",
},
]
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 _EXECUTION_LIST_FIELDS:
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)
disabled_outlets = list(OUTLET_MODE_DISABLE.finditer(utterance))
if disabled_outlets and POSITIVE_OUTLET_MODE.search(utterance) is None:
_remove_capability(result, "climate.set_outlet_mode")
for match in disabled_outlets:
feature = match.group("feature") or match.group("feature_after")
constraints.append(
{
"type": "prohibit_action",
"domain": "climate",
"capability": "climate.set_outlet_mode",
"parameters": {"disabled_feature": feature},
"source_span": _span(utterance, match),
}
)
unresolved.append(
{
"text": match.group(0),
"reason": "outlet_feature_disable_requires_absolute_mode",
}
)
result["ambiguities"] = [
*result.get("ambiguities", []),
{
"field": "climate.outlet_mode",
"question": (
"当前CAN白名单只能设置完整出风模式;请明确改为吹脚、除霜等目标模式。"
),
"reason": "outlet_feature_disable_requires_absolute_mode",
},
]
result["decision_summary"] = [
*result.get("decision_summary", []),
"检测到关闭单一出风方向的请求;未将其错误转换为开启该出风模式。",
]
result["confidence"] = min(float(result.get("confidence", 0)), 0.8)
thermal_middle = DRIVER_SEAT_THERMAL_MIDDLE.search(utterance)
if thermal_middle is not None and not re.search(
r"副驾|副驾驶|乘客|右前", utterance
):
feature = "通风" if "通风" in thermal_middle.group(0) else "加热"
native_capability = (
"seat.set_driver_ventilation_native"
if feature == "通风"
else "seat.set_driver_heating_native"
)
canonical_capability = (
"seat.set_ventilation"
if feature == "通风"
else "seat.set_heating"
)
_remove_capability(result, native_capability)
_remove_capability(result, canonical_capability)
constraint = {
"type": "unsupported_discrete_level",
"domain": "seat",
"capability": native_capability,
"parameters": {"requested_level": "medium"},
"source_span": _span(utterance, thermal_middle),
}
if not any(
item.get("type") == constraint["type"]
and item.get("capability") == native_capability
and item.get("parameters") == constraint["parameters"]
for item in constraints
):
constraints.append(constraint)
unresolved.append(
{
"text": thermal_middle.group(0),
"reason": "driver_seat_thermal_middle_level_unsupported",
}
)
result["ambiguities"] = [
*result.get("ambiguities", []),
{
"field": (
"seat.driver.ventilation.level"
if feature == "通风"
else "seat.driver.heating.level"
),
"question": (
f"主驾座椅{feature}实车白名单只有低档和高档,"
"请明确选择低档或高档。"
),
"reason": "driver_seat_thermal_middle_level_unsupported",
},
]
result["decision_summary"] = [
*result.get("decision_summary", []),
f"识别到主驾座椅{feature}中档请求;实车协议不支持中档,未降级为其他档位。",
]
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",
}
)
conflicts = _conflicting_control_keys(result)
if conflicts:
_remove_conflicting_controls(result, conflicts)
for domain, capability, target, selector in sorted(conflicts):
constraint = {
"type": "reject_conflicting_absolute_values",
"domain": domain,
"capability": capability,
"target": target,
}
if selector:
constraint["selector"] = selector
constraints.append(constraint)
unresolved.append(
{
"text": utterance,
"reason": "conflicting_values_for_same_capability_target",
}
)
result["ambiguities"] = [
*result.get("ambiguities", []),
{
"field": "parameters",
"question": "同一设备目标包含互相冲突的设定值,已停止相关操作,请明确唯一值。",
"reason": "conflicting_values_for_same_capability_target",
},
]
result["decision_summary"] = [
*result.get("decision_summary", []),
"检测到同一能力与目标的冲突值,相关动作均未下发。",
]
result["semantic_constraints"] = constraints
result["unresolved_spans"] = unresolved
return result