vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完善 N7 智能座舱交互与控制链路

Browse Source
This commit is contained in:
vee
2026-09-07 15:34:59 +08:00
parent 32d179143f
commit c96970f068
100 changed files with 30088 additions and 1174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
domains/seat/src/seat_domain/__init__.py
+12 -5
View File
@@ -9,23 +9,30 @@ from .executor import (
from .generator import build_function_config, build_function_preset, build_intent_ir
from .hardware_compiler import T2GSeatCompiler
from .hardware_executor import HardwareExecutionError, HardwareSeatExecutor
from .intent import parse_local_intent, resolve_contextual_intent
from .intent import (
SeatEndpointResolution,
parse_local_intent,
resolve_contextual_intent,
resolve_seat_axis_endpoint,
)
from .matrix import SeatCanMatrix
from .validator import validate_semantics
__all__ = [
"SeatCanMatrix",
"SeatCompiler",
"T2GSeatCompiler",
"SIMULATED_HARDWARE_FAULTS",
"HardwareExecutionError",
"HardwareSeatExecutor",
"SIMULATED_HARDWARE_FAULTS",
"SeatCanMatrix",
"SeatCompiler",
"SeatEndpointResolution",
"SimulatedHardwareSeatExecutor",
"T2GSeatCompiler",
"VirtualSeatExecutor",
"build_function_config",
"build_function_preset",
"build_intent_ir",
"parse_local_intent",
"resolve_contextual_intent",
"resolve_seat_axis_endpoint",
"validate_semantics",
]
domains/seat/src/seat_domain/intent.py
+294 -55
View File
@@ -6,12 +6,13 @@ import math
import re
import unicodedata
from copy import deepcopy
from dataclasses import dataclass
from typing import Any
TARGET_PATTERN = re.compile(
r"主副驾|前排右侧|右前座椅|右前座|副驾驶座椅|副驾驶座|副驾驶|副驾位|"
r"乘客座椅|乘客座|乘客位|主驾驶座椅|主驾驶座|主驾驶|驾驶位|驾驶座|"
r"乘客座椅|乘客座|乘客位|乘员座椅|乘员座|乘员位|乘员|"
r"主驾驶座椅|主驾驶座|主驾驶|驾驶位|驾驶员|驾驶座|"
r"司机位|司机座|主驾|前排|副驾"
)
FEATURE_PATTERN = re.compile(r"通风|加热")
@@ -90,7 +91,7 @@ SEAT_CONTROL_PATTERN = re.compile(
)
PASSENGER_ONLY_QUESTION = "当前所有座椅能力仅控制副驾，不支持主驾座椅目标。"
UNSUPPORTED_SEAT_SCOPE_PATTERN = re.compile(
r"后排|第二排|二排|第三排|三排|左前座椅|左前座|左前|"
r"后排|后座|中排|第二排|二排|第三排|三排|左前座椅|左前座|左前|"
r"右后座椅|右后座|右后|左后座椅|左后座|左后|"
r"所有座椅|全部座椅|全车座椅"
)
@@ -107,6 +108,7 @@ TARGETS = {
"主驾驶座椅": ["driver"],
"驾驶座": ["driver"],
"司机座": ["driver"],
"驾驶员": ["driver"],
"副驾": ["passenger"],
"副驾驶": ["passenger"],
"副驾驶座": ["passenger"],
@@ -115,6 +117,10 @@ TARGETS = {
"乘客位": ["passenger"],
"乘客座": ["passenger"],
"乘客座椅": ["passenger"],
"乘员": ["passenger"],
"乘员位": ["passenger"],
"乘员座": ["passenger"],
"乘员座椅": ["passenger"],
"右前座": ["passenger"],
"右前座椅": ["passenger"],
"前排右侧": ["passenger"],
@@ -162,8 +168,41 @@ MOTION_AXIS_ALIASES = {
"坐垫",
),
"backrest_recline": ("靠背", "椅背"),
"legrest_height": ("腿部支撑", "小腿托", "腿托", "脚托"),
"legrest_height": (
"腿部支撑高度",
"小腿托高度",
"腿托高度",
"脚托高度",
"腿部支撑",
"小腿托",
"腿托",
"脚托",
),
}
def _axis_alias_present(axis: str, alias: str, text: str) -> bool:
"""Return whether an axis alias is unambiguous in the supplied text.
``高度`` by itself is a supported shorthand for the passenger seat-height
axis, but it also occurs inside explicit leg-rest phrases such as
``腿托高度``. Treating that suffix as a second cushion alias makes the
parser choose the cushion axis before it reaches the more-specific leg-rest
axis. The explicit component name must win.
"""
if axis == "cushion_tilt" and alias == "高度":
legrest_aliases = MOTION_AXIS_ALIASES["legrest_height"]
if any(f"{legrest_alias}高度" in text for legrest_alias in legrest_aliases):
return False
return alias in text
def _axis_present(axis: str, text: str) -> bool:
return any(
_axis_alias_present(axis, alias, text)
for alias in MOTION_AXIS_ALIASES[axis]
)
MOTION_ANGLE_RANGES = {
"backrest_recline": (95, 163),
"legrest_height": (95, 167),
@@ -183,8 +222,8 @@ RELATIVE_DIRECTION_SIGNS = {
"decrease": {"向上", "往上", "抬高", "抬起", "升高", "上"},
},
"backrest_recline": {
"increase": {"向后", "往后", "朝后", "后仰", "后"},
"decrease": {"向前", "往前", "朝前", "前倾", "前"},
"increase": {"向后", "往后", "朝后", "后移", "后仰", "后倾", "后"},
"decrease": {"向前", "往前", "朝前", "前移", "前倾", "前"},
},
"legrest_height": {
"increase": {"向上", "往上", "抬高", "抬起", "升高", "上"},
@@ -217,6 +256,85 @@ MOTION_ENDPOINT_LABELS = {
"legrest_height": {"increase": "最上方", "decrease": "最下方"},
}
# Seat motion endpoints all share the same language shape: a physical axis,
# an optional direction, a motion verb, and a terminal position. Keeping the
# endpoint vocabulary in one table prevents one axis (historically the
# backrest) from accepting natural phrases that another axis (the leg rest or
# seat height) silently rejects.
MOTION_PHYSICAL_ENDPOINTS: dict[str, dict[int, tuple[str, ...]]] = {
"longitudinal": {
0: ("最前面", "最前方", "最前位置", "最前"),
100: ("最后面", "最后方", "最后位置", "最后"),
},
"cushion_tilt": {
0: ("最上方", "最高位置", "最高"),
100: ("最下方", "最低位置", "最低"),
},
"backrest_recline": {
0: ("调直", "立直", "竖直", "最前方", "最前位置", "最前"),
100: (
"完全放倒",
"放倒到最后",
"最后方",
"最后位置",
"最后",
),
},
"legrest_height": {
0: ("收起", "收回", "最下方", "最低位置", "最低"),
100: ("完全抬起", "最上方", "最高位置", "最高"),
},
}
MOTION_DEFAULT_EXTREMA: dict[str, dict[str, int | None]] = {
# "最大位置" has no physical meaning for a bidirectional rail unless the
# user also gives a direction. The remaining axes have stable product
# semantics for maximum/minimum posture.
"longitudinal": {"最大": None, "最小": None},
"cushion_tilt": {"最大": 0, "最小": 100},
"backrest_recline": {"最大": 100, "最小": 0},
"legrest_height": {"最大": 100, "最小": 0},
}
MOTION_LEGACY_DIRECTIONAL_ENDPOINTS: dict[str, dict[int, tuple[str, ...]]] = {
"longitudinal": {
0: ("向前移到底", "推到最前"),
100: ("向后移到底", "移到最后"),
},
"cushion_tilt": {
0: ("抬到头", "抬到底"),
100: ("降到头", "降到底"),
},
"backrest_recline": {
0: ("前倾到底",),
100: ("放倒到底", "后仰到底", "后移到底"),
},
"legrest_height": {
0: ("降到头", "降到底"),
100: ("升到头", "升到底", "抬到底"),
},
}
MOTION_ENDPOINT_DIRECTION_PATTERN = (
r"向前|往前|朝前|前移|前倾|向后|往后|朝后|后移|后仰|后倾|"
r"向上|往上|抬高|抬起|升高|向下|往下|降低|降下|前|后|上|下"
)
MOTION_ENDPOINT_FILLER_PATTERN = (
r"(?:再|继续|接着|然后|移动|移|转动|旋转|转|调节|调整|调|"
r"设置|设|倾斜|抬|升|降|推|拉)*"
)
MOTION_GENERIC_TERMINAL_PATTERN = (
r"(?:最大|最小|极限)(?:位置|角度|行程)?|(?:到|至)(?:底|头|尽头|末端)"
)
@dataclass(frozen=True)
class SeatEndpointResolution:
"""Result of resolving one absolute endpoint phrase for one seat axis."""
matched: bool
position_pct: int | None = None
question: str | None = None
direction: str | None = None
endpoint: str | None = None
def _chinese_integer(token: str) -> int | None:
if token.isdigit():
@@ -286,13 +404,137 @@ def _relative_direction(axis: str, token: str) -> str | None:
return None
def _endpoint_position(axis: str, endpoint: str) -> int | None:
for position_pct, tokens in MOTION_PHYSICAL_ENDPOINTS[axis].items():
if endpoint in tokens:
return position_pct
for position_pct, tokens in MOTION_LEGACY_DIRECTIONAL_ENDPOINTS[axis].items():
if endpoint in tokens:
return position_pct
return None
def resolve_seat_axis_endpoint(
text: str,
axis: str,
*,
aliases: tuple[str, ...] | None = None,
) -> SeatEndpointResolution:
"""Resolve compositional absolute endpoint language for one seat axis.
Directional terminal phrases use the physical polarity already shared by
relative and timed motion. Named physical endpoints are checked against
an explicit direction so contradictory requests fail closed instead of
silently choosing one half of the sentence.
"""
if axis not in MOTION_AXIS_ALIASES:
raise ValueError(f"unknown seat motion axis: {axis}")
normalized = _normalize_text(text)
axis_aliases = aliases or MOTION_AXIS_ALIASES[axis]
alias_pattern = "(?:" + "|".join(
re.escape(alias) for alias in sorted(set(axis_aliases), key=len, reverse=True)
) + ")"
named_endpoints = {
token
for tokens in MOTION_PHYSICAL_ENDPOINTS[axis].values()
for token in tokens
}
named_endpoints.update(
token
for tokens in MOTION_LEGACY_DIRECTIONAL_ENDPOINTS[axis].values()
for token in tokens
)
endpoint_pattern = "(?:" + "|".join(
[
*(re.escape(token) for token in sorted(named_endpoints, key=len, reverse=True)),
MOTION_GENERIC_TERMINAL_PATTERN,
]
) + ")"
directional = re.search(
rf"{alias_pattern}(?:位置|角度)?{MOTION_ENDPOINT_FILLER_PATTERN}"
rf"(?P<direction>{MOTION_ENDPOINT_DIRECTION_PATTERN})"
rf"{MOTION_ENDPOINT_FILLER_PATTERN}(?:到|至|为|成)?"
rf"(?P<endpoint>{endpoint_pattern})(?![+-]?\d)",
normalized,
)
if directional is not None:
direction = directional.group("direction")
endpoint = directional.group("endpoint")
operation = _relative_direction(axis, direction)
if operation is None:
allowed = "或".join(MOTION_ENDPOINT_LABELS[axis].values())
return SeatEndpointResolution(
matched=True,
question=(
f"{MOTION_AXIS_LABELS[axis]}不支持“{direction}”方向；"
f"请改用与该部件一致的方向，或直接指定{allowed}。"
),
direction=direction,
endpoint=endpoint,
)
direction_position = 100 if operation == "increase" else 0
named_position = _endpoint_position(axis, endpoint)
if named_position is not None and named_position != direction_position:
return SeatEndpointResolution(
matched=True,
question=(
f"{MOTION_AXIS_LABELS[axis]}的方向“{direction}”与端点"
f"“{endpoint}”相冲突；请只保留一个一致的方向和目标位置。"
),
direction=direction,
endpoint=endpoint,
)
return SeatEndpointResolution(
matched=True,
position_pct=direction_position,
direction=direction,
endpoint=endpoint,
)
direct = re.search(
rf"{alias_pattern}(?:位置|角度)?{MOTION_ENDPOINT_FILLER_PATTERN}"
rf"(?:到|至|为|成)?(?P<endpoint>{endpoint_pattern})(?![+-]?\d)",
normalized,
)
if direct is None:
return SeatEndpointResolution(matched=False)
endpoint = direct.group("endpoint")
position_pct = _endpoint_position(axis, endpoint)
if position_pct is not None:
return SeatEndpointResolution(
matched=True,
position_pct=position_pct,
endpoint=endpoint,
)
generic = re.match(r"(?P<extent>最大|最小)", endpoint)
if generic is not None:
position_pct = MOTION_DEFAULT_EXTREMA[axis][generic.group("extent")]
if position_pct is not None:
return SeatEndpointResolution(
matched=True,
position_pct=position_pct,
endpoint=endpoint,
)
return SeatEndpointResolution(
matched=True,
question=(
f"{MOTION_AXIS_LABELS[axis]}的“{endpoint}”缺少明确方向；"
f"请说明移动到{MOTION_ENDPOINT_LABELS[axis]['decrease']}还是"
f"{MOTION_ENDPOINT_LABELS[axis]['increase']}，也可以直接指定0到100%的位置。"
),
endpoint=endpoint,
)
def _directional_motion_axis(clause: str, direction: str) -> str | None:
"""Resolve an omitted passenger-seat axis from the movement direction."""
explicit_axes = [
axis
for axis, aliases in MOTION_AXIS_ALIASES.items()
if any(alias in clause for alias in aliases)
for axis in MOTION_AXIS_ALIASES
if _axis_present(axis, clause)
]
if len(explicit_axes) == 1:
return explicit_axes[0]
@@ -944,8 +1186,8 @@ def parse_local_intent(
mentioned_axes = [
axis
for axis, aliases in MOTION_AXIS_ALIASES.items()
if any(alias in text for alias in aliases)
for axis in MOTION_AXIS_ALIASES
if _axis_present(axis, text)
]
if ZERO_POSITION_PATTERN.search(text) and (mentioned_axes or "四轴" in text):
reset_axes = list(MOTION_AXIS_ALIASES) if "四轴" in text else mentioned_axes
@@ -1027,8 +1269,8 @@ def parse_local_intent(
clause_axes = [
axis
for axis, aliases in MOTION_AXIS_ALIASES.items()
if any(alias in clause for alias in aliases)
for axis in MOTION_AXIS_ALIASES
if _axis_present(axis, clause)
]
percentages = list(
re.finditer(r"(?P<position>[+-]?\d{1,3}(?:\.\d+)?)%", clause)
@@ -1118,6 +1360,8 @@ def parse_local_intent(
}
for axis, aliases in timed_aliases.items():
for alias in aliases:
if not _axis_alias_present(axis, alias, clause):
continue
match = re.search(
TIMED_MOTION_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
clause,
@@ -1154,6 +1398,8 @@ def parse_local_intent(
if axis in timed_axes or axis in directional_absolute_axes:
continue
for alias in aliases:
if not _axis_alias_present(axis, alias, clause):
continue
angle_match = re.search(
RELATIVE_ANGLE_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
clause,
@@ -1232,54 +1478,45 @@ def parse_local_intent(
motion_actions.append(action)
endpoint_actions: dict[str, int] = {}
endpoint_axes: set[str] = set()
has_specific_non_longitudinal_axis = any(
any(alias in clause for alias in MOTION_AXIS_ALIASES[axis])
_axis_present(axis, clause)
for axis in ("cushion_tilt", "backrest_recline", "legrest_height")
)
has_longitudinal_axis = any(
alias in clause for alias in MOTION_AXIS_ALIASES["longitudinal"]
) or (
has_longitudinal_axis = _axis_present("longitudinal", clause) or (
not has_specific_non_longitudinal_axis
and any(alias in clause for alias in ("座椅", "座位"))
)
if has_longitudinal_axis:
if re.search(r"最前面|最前|向前移到底|推到最前", clause):
endpoint_actions["longitudinal"] = 0
elif re.search(r"最后面|最后|向后移到底|移到最后", clause):
endpoint_actions["longitudinal"] = 100
cushion_alias = r"(?:座椅高度|高度|坐垫倾角|坐盆倾角|坐盆|坐垫)"
if re.search(
cushion_alias + r"(?:抬到最上方|调到最上方|移到最上方|最高|抬到头|抬到底)",
clause,
):
endpoint_actions["cushion_tilt"] = 0
elif re.search(
cushion_alias + r"(?:降到最下方|调到最下方|移到最下方|最低|降到头|降到底)",
clause,
):
endpoint_actions["cushion_tilt"] = 100
backrest_alias = r"(?:靠背|椅背)"
if re.search(backrest_alias + r"(?:调直|立直|竖直)", clause):
endpoint_actions["backrest_recline"] = 0
elif re.search(
backrest_alias
+ r"(?:放倒到最后|放倒到底|完全放倒|后仰到底|向后移到底|最大角度)",
clause,
):
endpoint_actions["backrest_recline"] = 100
legrest_alias = r"(?:腿部支撑|小腿托|腿托|脚托)"
if re.search(
legrest_alias
+ r"(?:收起|收回|降到最下方|调到最下方|降到最低|最低|降到头|降到底)",
clause,
):
endpoint_actions["legrest_height"] = 0
elif re.search(
legrest_alias
+ r"(?:升到最上方|抬到最上方|调到最上方|升到最高|抬到最高|最高|完全抬起|升到头|升到底|抬到底)",
clause,
):
endpoint_actions["legrest_height"] = 100
endpoint_aliases = dict(MOTION_AXIS_ALIASES)
if has_longitudinal_axis and not _axis_present("longitudinal", clause):
endpoint_aliases["longitudinal"] = (
*MOTION_AXIS_ALIASES["longitudinal"],
"座椅",
"座位",
)
for axis, aliases in endpoint_aliases.items():
if axis == "longitudinal" and not has_longitudinal_axis:
continue
if axis != "longitudinal" and not _axis_present(axis, clause):
continue
resolution = resolve_seat_axis_endpoint(
clause,
axis,
aliases=aliases,
)
if not resolution.matched:
continue
endpoint_axes.add(axis)
if resolution.question:
ambiguities.append(
{
"field": f"passenger.position.{axis}",
"question": resolution.question,
}
)
continue
if resolution.position_pct is not None:
endpoint_actions[axis] = resolution.position_pct
for axis, position_pct in endpoint_actions.items():
target_zh = _nearest_preceding_target(clause, len(clause)) or "副驾"
if "driver" in TARGETS[target_zh]:
@@ -1298,13 +1535,15 @@ def parse_local_intent(
for axis, aliases in MOTION_AXIS_ALIASES.items():
if (
axis in endpoint_actions
axis in endpoint_axes
or axis in relative_axes
or axis in timed_axes
or axis in directional_absolute_axes
):
continue
for alias in aliases:
if not _axis_alias_present(axis, alias, clause):
continue
position_match = re.search(
POSITION_PATTERN_TEMPLATE.format(alias=re.escape(alias)),
clause,