执行目标
diff --git a/config/native_can_relative_controls.yaml b/config/native_can_relative_controls.yaml
new file mode 100644
index 0000000..a577e14
--- /dev/null
+++ b/config/native_can_relative_controls.yaml
@@ -0,0 +1,32 @@
+apiVersion: cognitive-agent/v1alpha1
+kind: NativeCanRelativeControlPolicy
+metadata:
+ id: denza-n7-relative-controls
+ version: 1.0.0
+spec:
+ controls:
+ climate_temperature:
+ default_step: 1
+ minimum: 18
+ maximum: 32
+ unit: celsius
+ climate_fan:
+ default_step: 1
+ minimum: 1
+ maximum: 7
+ unit: level
+ ambient_brightness:
+ default_step: 1
+ minimum: 1
+ maximum: 5
+ unit: level
+ shade_position:
+ default_step: 10
+ minimum: 0
+ maximum: 100
+ unit: percentage_point
+ window_position:
+ default_step: 10
+ minimum: 0
+ maximum: 100
+ unit: percentage_point
diff --git a/contracts/schemas/intent-ir.schema.json b/contracts/schemas/intent-ir.schema.json
index a92c5df..cab41e7 100755
--- a/contracts/schemas/intent-ir.schema.json
+++ b/contracts/schemas/intent-ir.schema.json
@@ -12,7 +12,7 @@
"spec": {
"type": "object",
"additionalProperties": false,
- "required": ["utterance", "language", "mode", "detected_domains", "requests", "confidence", "requires_confirmation"],
+ "required": ["utterance", "language", "mode", "detected_domains", "requests", "actions", "constraints", "ambiguities", "unresolved_spans", "coverage", "confidence", "requires_confirmation", "execution_eligible"],
"properties": {
"utterance": {"type": "string", "minLength": 1, "maxLength": 4000},
"language": {"type": "string", "pattern": "^[a-z]{2}(?:-[A-Z]{2})?$"},
@@ -39,6 +39,45 @@
}
}
},
+ "actions": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["id", "domain", "capability", "target", "operation", "parameters", "source_span", "confidence", "provenance", "risk_level", "confirmation_policy"],
+ "properties": {
+ "id": {"type": "string", "pattern": "^[a-z][a-z0-9_-]*$"},
+ "domain": {"type": "string", "pattern": "^[a-z][a-z0-9_]*$"},
+ "capability": {"type": "string", "minLength": 1},
+ "target": {"$ref": "common.schema.json#/$defs/target"},
+ "operation": {"type": "string", "minLength": 1},
+ "parameters": {"type": "object"},
+ "source_span": {"$ref": "#/$defs/sourceSpan"},
+ "confidence": {"type": "object", "additionalProperties": {"type": "number", "minimum": 0, "maximum": 1}},
+ "provenance": {"type": "object", "required": ["parser", "rule_ids"], "properties": {"parser": {"type": "string"}, "rule_ids": {"type": "array", "items": {"type": "string"}}}, "additionalProperties": true},
+ "risk_level": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
+ "confirmation_policy": {"type": "string", "enum": ["never", "on_ambiguity", "always"]},
+ "resolution": {"type": "object"}
+ }
+ }
+ },
+ "constraints": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "additionalProperties": true,
+ "required": ["id", "type", "source_span"],
+ "properties": {
+ "id": {"type": "string"},
+ "type": {"type": "string"},
+ "domain": {"type": "string"},
+ "capability": {"type": "string"},
+ "target": {"$ref": "common.schema.json#/$defs/target"},
+ "parameters": {"type": "object"},
+ "source_span": {"$ref": "#/$defs/sourceSpan"}
+ }
+ }
+ },
"ambiguities": {
"type": "array",
"items": {
@@ -48,13 +87,51 @@
"properties": {
"field": {"type": "string", "minLength": 1},
"question": {"type": "string", "minLength": 1},
- "options": {"type": "array", "items": {"type": ["string", "number", "boolean"]}}
+ "options": {"type": "array", "items": {"type": ["string", "number", "boolean"]}},
+ "reason": {"type": "string"},
+ "risk_level": {"type": "string"}
}
},
"default": []
},
+ "unresolved_spans": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["text", "start", "end", "reason"],
+ "properties": {
+ "text": {"type": "string"},
+ "start": {"type": "integer", "minimum": 0},
+ "end": {"type": "integer", "minimum": 0},
+ "reason": {"type": "string"}
+ }
+ }
+ },
+ "coverage": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["complete", "covered_spans"],
+ "properties": {
+ "complete": {"type": "boolean"},
+ "covered_spans": {"type": "array", "items": {"$ref": "#/$defs/sourceSpan"}}
+ }
+ },
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
- "requires_confirmation": {"type": "boolean"}
+ "requires_confirmation": {"type": "boolean"},
+ "execution_eligible": {"type": "boolean"}
+ }
+ }
+ },
+ "$defs": {
+ "sourceSpan": {
+ "type": "object",
+ "additionalProperties": false,
+ "required": ["text", "start", "end"],
+ "properties": {
+ "text": {"type": "string"},
+ "start": {"type": "integer", "minimum": 0},
+ "end": {"type": "integer", "minimum": 0}
}
}
}
diff --git a/domains/seat/src/seat_domain/generator.py b/domains/seat/src/seat_domain/generator.py
index d52ca63..b13320c 100755
--- a/domains/seat/src/seat_domain/generator.py
+++ b/domains/seat/src/seat_domain/generator.py
@@ -8,6 +8,7 @@ from typing import Any
TARGETS = {
"passenger": {"device": "seat", "zone": "front_right", "instance": "passenger"},
+ "driver": {"device": "seat", "zone": "front_left", "instance": "driver"},
}
CAPABILITIES = {
"ventilation": "seat.set_ventilation",
@@ -29,8 +30,145 @@ def _now() -> str:
return datetime.now(UTC).isoformat().replace("+00:00", "Z")
+def _source_span(utterance: str, source: Any = None) -> dict[str, Any]:
+ text = source if isinstance(source, str) and source else utterance
+ start = utterance.find(text)
+ if start < 0:
+ text = utterance
+ start = 0
+ return {"text": text, "start": start, "end": start + len(text)}
+
+
+def _semantic_target(domain: str, target: str) -> dict[str, str]:
+ if domain == "seat" and target in TARGETS:
+ return dict(TARGETS[target])
+ return {"device": domain, "zone": target}
+
+
+def _canonical_semantic_actions(
+ parsed: dict[str, Any],
+ utterance: str,
+ provider: str,
+ capability_metadata: dict[str, dict[str, str]],
+) -> list[dict[str, Any]]:
+ actions: list[dict[str, Any]] = []
+
+ def append(
+ *,
+ domain: str,
+ capability: str,
+ target: str,
+ operation: str,
+ parameters: dict[str, Any],
+ source: dict[str, Any],
+ ) -> None:
+ policy = capability_metadata.get(capability, {})
+ confidence = source.get("confidence")
+ if not isinstance(confidence, dict):
+ confidence = {"overall": float(parsed.get("confidence", 0))}
+ provenance = source.get("provenance")
+ if not isinstance(provenance, dict):
+ provenance = {"parser": provider, "rule_ids": []}
+ semantic: dict[str, Any] = {
+ "id": f"action_{len(actions) + 1}",
+ "domain": domain,
+ "capability": capability,
+ "target": _semantic_target(domain, target),
+ "operation": operation,
+ "parameters": parameters,
+ "source_span": _source_span(utterance, source.get("source_span")),
+ "confidence": confidence,
+ "provenance": provenance,
+ "risk_level": str(policy.get("risk_level", "high")),
+ "confirmation_policy": str(
+ policy.get("confirmation_policy", "on_ambiguity")
+ ),
+ }
+ resolution = source.get("relative_resolution", source.get("motion_intent"))
+ if isinstance(resolution, dict):
+ semantic["resolution"] = dict(resolution)
+ actions.append(semantic)
+
+ for action in parsed.get("actions", []):
+ target = str(action["target"])
+ if action.get("action_type") == "position":
+ parameters = {"position_pct": action["position_pct"]}
+ if action.get("angle_deg") is not None:
+ parameters["angle_deg"] = action["angle_deg"]
+ append(
+ domain="seat",
+ capability=MOTION_CAPABILITIES[str(action["axis"])],
+ target=target,
+ operation="set",
+ parameters=parameters,
+ source=action,
+ )
+ elif action.get("action_type") == "dynamic_mode":
+ append(
+ domain="seat",
+ capability=DYNAMIC_MODE_CAPABILITY,
+ target=target,
+ operation="activate",
+ parameters={"mode_id": action["mode_id"]},
+ source=action,
+ )
+ elif action.get("action_type") == "dynamic_mode_stop":
+ append(
+ domain="seat",
+ capability=DYNAMIC_MODE_STOP_CAPABILITY,
+ target=target,
+ operation="deactivate",
+ parameters={},
+ source=action,
+ )
+ else:
+ parameters = {"state": action["state"]}
+ if action.get("requested_level") is not None:
+ parameters["level"] = action["requested_level"]
+ append(
+ domain="seat",
+ capability=CAPABILITIES[str(action["function"])],
+ target=target,
+ operation="set",
+ parameters=parameters,
+ source=action,
+ )
+
+ for action in [
+ *parsed.get("domain_actions", []),
+ *parsed.get("native_can_actions", []),
+ ]:
+ append(
+ domain=str(action["domain"]),
+ capability=str(action["capability"]),
+ target=str(action["target"]),
+ operation=str(action.get("operation", "set")),
+ parameters=dict(action.get("parameters", {})),
+ source=action,
+ )
+
+ unique: list[dict[str, Any]] = []
+ signatures: set[str] = set()
+ for action in actions:
+ signature = repr(
+ (
+ action["domain"], action["capability"], action["target"],
+ action["operation"], action["parameters"],
+ )
+ )
+ if signature not in signatures:
+ signatures.add(signature)
+ action["id"] = f"action_{len(unique) + 1}"
+ unique.append(action)
+ return unique
+
+
def build_intent_ir(
- parsed: dict[str, Any], utterance: str, request_id: str, provider: str
+ parsed: dict[str, Any],
+ utterance: str,
+ request_id: str,
+ provider: str,
+ capability_metadata: dict[str, dict[str, str]] | None = None,
) -> dict[str, Any]:
requests = []
for action in parsed["actions"]:
@@ -154,7 +292,57 @@ def build_intent_ir(
}
)
- detected_domains = parsed.get("detected_domains", ["seat"])
+ semantic_actions = _canonical_semantic_actions(
+ parsed, utterance, provider, capability_metadata or {}
+ )
+ constraints = []
+ for index, raw in enumerate(parsed.get("semantic_constraints", []), start=1):
+ constraint = dict(raw)
+ constraint["id"] = str(constraint.get("id") or f"constraint_{index}")
+ if not isinstance(constraint.get("source_span"), dict):
+ constraint["source_span"] = _source_span(
+ utterance, constraint.get("source_span")
+ )
+ if isinstance(constraint.get("target"), str):
+ constraint["target"] = _semantic_target(
+ str(constraint.get("domain", "seat")), constraint["target"]
+ )
+ constraints.append(constraint)
+
+ unresolved_spans = []
+ for raw in parsed.get("unresolved_spans", []):
+ item = {"text": raw} if isinstance(raw, str) else dict(raw)
+ unresolved_spans.append(
+ {
+ **_source_span(utterance, item.get("text")),
+ "reason": str(
+ item.get("reason", "not_mapped_to_registered_capability")
+ ),
+ }
+ )
+
+ covered_spans = []
+ covered_signatures: set[tuple[int, int]] = set()
+ for item in [*semantic_actions, *constraints]:
+ span = item["source_span"]
+ signature = (int(span["start"]), int(span["end"]))
+ if signature not in covered_signatures:
+ covered_signatures.add(signature)
+ covered_spans.append(dict(span))
+
+ detected_domains = list(parsed.get("detected_domains", ["seat"]))
+ for action in semantic_actions:
+ if action["domain"] not in detected_domains:
+ detected_domains.append(action["domain"])
+ ambiguities = list(parsed.get("ambiguities", []))
+ requires_confirmation = bool(
+ ambiguities
+ or unresolved_spans
+ or any(
+ action["confirmation_policy"] == "always"
+ for action in semantic_actions
+ )
+ )
return {
"apiVersion": "cognitive-agent/v1alpha1",
@@ -173,9 +361,17 @@ def build_intent_ir(
"mode": "cross_domain" if len(detected_domains) > 1 else "single_domain",
"detected_domains": detected_domains,
"requests": requests,
- "ambiguities": parsed["ambiguities"],
+ "actions": semantic_actions,
+ "constraints": constraints,
+ "ambiguities": ambiguities,
+ "unresolved_spans": unresolved_spans,
+ "coverage": {
+ "complete": not unresolved_spans,
+ "covered_spans": covered_spans,
+ },
"confidence": parsed["confidence"],
- "requires_confirmation": bool(parsed["ambiguities"]),
+ "requires_confirmation": requires_confirmation,
+ "execution_eligible": bool(semantic_actions) and not requires_confirmation,
},
}
diff --git a/src/cognitive_agent/application/native_can_intent.py b/src/cognitive_agent/application/native_can_intent.py
index 9ddbe14..ee23407 100644
--- a/src/cognitive_agent/application/native_can_intent.py
+++ b/src/cognitive_agent/application/native_can_intent.py
@@ -26,6 +26,11 @@ WINDOW_TARGETS = {
"右后": "rear_right",
"后右": "rear_right",
}
+WINDOW_GROUP_MEMBERS = {
+ "all": ["front_left", "front_right", "rear_left", "rear_right"],
+ "front": ["front_left", "front_right"],
+ "rear": ["rear_left", "rear_right"],
+}
OUTLET_TARGETS = {
"主驾左": "driver_left",
"主驾右": "driver_right",
@@ -543,24 +548,55 @@ def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
break
if "车窗" in compact:
- explicit_target = next(
- (value for label, value in WINDOW_TARGETS.items() if label in compact),
+ explicit_targets = []
+ for label, target_name in WINDOW_TARGETS.items():
+ if target_name in WINDOW_GROUP_MEMBERS or label not in compact:
+ continue
+ if target_name not in explicit_targets:
+ explicit_targets.append(target_name)
+ broad_target = next(
+ (
+ target_name
+ for label, target_name in WINDOW_TARGETS.items()
+ if target_name in WINDOW_GROUP_MEMBERS and label in compact
+ ),
None,
)
- target = explicit_target or "front_left"
+ exclusions = {
+ target_name
+ for label, target_name in WINDOW_TARGETS.items()
+ if target_name not in WINDOW_GROUP_MEMBERS
+ and re.search(
+ rf"{re.escape(label)}(?:车窗)?(?:除外|除去|不包括|不含)",
+ compact,
+ )
+ }
+ if broad_target and exclusions:
+ targets = [
+ target_name
+ for target_name in WINDOW_GROUP_MEMBERS[broad_target]
+ if target_name not in exclusions
+ ]
+ elif explicit_targets:
+ targets = [target_name for target_name in explicit_targets if target_name not in exclusions]
+ else:
+ targets = [broad_target or "front_left"]
position = re.search(
r"(?:位置|开度|调到|设为|设置为|开启到|打开到)(\d{1,3})%",
compact,
)
half_position = bool(re.search(r"(?:开启|打开|开到|开度)?(?:一半|半开)", compact))
position_pct = int(position[1]) if position else 50 if half_position else None
- if position_pct is not None and target not in {"all", "front", "rear"} and position_pct <= 100:
+ if position_pct is not None and all(target not in WINDOW_GROUP_MEMBERS for target in targets) and position_pct <= 100:
reason = (
"用户指定了单个车窗的绝对开度。"
- if explicit_target
+ if explicit_targets or broad_target
else "用户未指定车窗目标，按驾驶台默认选择主驾左前窗。"
)
- actions.append(_action("window", "window.set_position", target, {"position_pct": position_pct}, reason))
+ actions.extend(
+ _action("window", "window.set_position", target, {"position_pct": position_pct}, reason)
+ for target in targets
+ )
elif position_pct is not None:
# Absolute position feedback is verified for one window only. Do
# not turn a half-open request for a broad scope into full motion.
@@ -575,10 +611,13 @@ def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
if motion:
reason = (
"用户指定了车窗目标和动作。"
- if explicit_target
+ if explicit_targets or broad_target
else "用户未指定车窗目标，按驾驶台默认选择主驾左前窗。"
)
- actions.append(_action("window", "window.set_motion", target, {"motion": motion}, reason))
+ actions.extend(
+ _action("window", "window.set_motion", target, {"motion": motion}, reason)
+ for target in targets
+ )
vehicle_lock_request = bool(
re.search(
@@ -786,9 +825,40 @@ def _conflicts_with_driver_native(
return False
-def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str, Any]:
+def merge_native_can_intent(
+ parsed: dict[str, Any],
+ utterance: str,
+ *,
+ relative_actions: list[dict[str, Any]] | None = None,
+ relative_ambiguities: list[dict[str, Any]] | None = None,
+ relative_claimed_controls: set[tuple[str, str]] | None = None,
+) -> dict[str, Any]:
"""Add deterministic vehicle-CAN actions without duplicating model requests."""
+ relative_actions = list(relative_actions or [])
+ relative_ambiguities = list(relative_ambiguities or [])
+ relative_claimed_controls = set(relative_claimed_controls or set())
+
+ def is_claimed(candidate: dict[str, Any]) -> bool:
+ capability = str(candidate.get("capability", ""))
+ target = str(candidate.get("target", candidate.get("target_scope", "")))
+ return (
+ (capability, target) in relative_claimed_controls
+ or (capability, "*") in relative_claimed_controls
+ )
+
+ if relative_claimed_controls:
+ parsed = dict(parsed)
+ for key in (
+ "domain_actions",
+ "native_can_actions",
+ "deferred_requests",
+ "model_requests",
+ ):
+ parsed[key] = [
+ item for item in parsed.get(key, []) if not is_claimed(item)
+ ]
+
clauses = [part for part in re.split(r"[，,。；;！!？?]+", utterance) if part.strip()]
exact_clause_actions = [
parse_exact_native_can_command(clause) for clause in clauses
@@ -797,7 +867,12 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
action for clause_actions in exact_clause_actions for action in clause_actions
]
fully_exact = bool(clauses) and all(exact_clause_actions)
- actions = exact_actions or _parse_chinese_commands(utterance)
+ actions = [
+ item
+ for item in (exact_actions or _parse_chinese_commands(utterance))
+ if not is_claimed(item)
+ ]
+ actions.extend(relative_actions)
implicit_default_capabilities = {
str(action.get("capability", ""))
for action in actions
@@ -833,7 +908,7 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
parsed,
unresolved=not has_default_driver_seat_action,
)
- if not actions:
+ if not actions and not relative_ambiguities:
return parsed
result = dict(parsed)
@@ -946,6 +1021,11 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
"parameters": dict(item["parameters"]),
"reason": item["reason"],
"execution_backend": "native_can",
+ **(
+ {"relative_resolution": dict(item["relative_resolution"])}
+ if item.get("relative_resolution")
+ else {}
+ ),
}
for item in unique_actions
],
@@ -963,10 +1043,42 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
for name, value in item["parameters"].items()
],
"reason": item["reason"],
+ **(
+ {"relative_resolution": dict(item["relative_resolution"])}
+ if item.get("relative_resolution")
+ else {}
+ ),
}
for item in unique_actions
],
]
+ constraints = list(result.get("semantic_constraints", []))
+ for match in re.finditer(
+ r"(?P左前|前左|右前|前右|左后|后左|右后|后右)(?:车窗)?"
+ r"(?:除外|除去|不包括|不含)",
+ utterance,
+ ):
+ constraints.append(
+ {
+ "type": "exclude_target",
+ "domain": "window",
+ "target": WINDOW_TARGETS[match.group("target")],
+ "source_span": {
+ "text": match.group(0),
+ "start": match.start(),
+ "end": match.end(),
+ },
+ }
+ )
+ result["semantic_constraints"] = constraints
+ result["ambiguities"] = list(result.get("ambiguities", []))
+ for ambiguity in relative_ambiguities:
+ signature = (ambiguity.get("field"), ambiguity.get("reason"))
+ if not any(
+ (item.get("field"), item.get("reason")) == signature
+ for item in result["ambiguities"]
+ ):
+ result["ambiguities"].append(ambiguity)
domains = list(result.get("detected_domains", []))
for item in unique_actions:
if item["domain"] not in domains:
@@ -986,6 +1098,12 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
item
for item in result.get("ambiguities", [])
if item.get("field") != "capability"
+ and not (
+ unique_actions
+ and not _looks_like_seat_request(utterance)
+ and str(item.get("field")) in {"target", "target_scope"}
+ and "当前台架仅支持副驾座椅" in str(item.get("question", ""))
+ )
and not (
resolved_driver_seat
and (
@@ -1006,6 +1124,11 @@ def merge_native_can_intent(parsed: dict[str, Any], utterance: str) -> dict[str,
result["decision_summary"] = [
*existing_summaries,
*[item["reason"] for item in unique_actions],
+ *[
+ f"相对指令未执行：{ambiguity['question']}"
+ for ambiguity in relative_ambiguities
+ ],
]
- result["confidence"] = max(float(result.get("confidence", 0)), 0.98)
+ if unique_actions:
+ result["confidence"] = max(float(result.get("confidence", 0)), 0.98)
return result
diff --git a/src/cognitive_agent/application/native_can_relative.py b/src/cognitive_agent/application/native_can_relative.py
new file mode 100644
index 0000000..b6e56bf
--- /dev/null
+++ b/src/cognitive_agent/application/native_can_relative.py
@@ -0,0 +1,424 @@
+"""Resolve relative vehicle controls against the current synchronized state."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+CLAUSE_SEPARATOR = re.compile(r"[，,；;。！？!?\n]+")
+NUMBER_TOKEN = r"(?:\d{1,3}|[一二两三四五六七八九十]+)"
+GENERIC_INCREASE = r"增加|提高|升高|上调|调高|增大|加大|调大|再高|更高"
+GENERIC_DECREASE = r"减少|降低|下降|下调|调低|减小|调小|再低|更低"
+AMBIENT_INCREASE = rf"{GENERIC_INCREASE}|调亮|变亮|更亮|再亮|亮一点"
+AMBIENT_DECREASE = rf"{GENERIC_DECREASE}|调暗|变暗|更暗|再暗|暗一点"
+OPENING_INCREASE = (
+ r"(?:增加|提高|增大|调大)(?:开度|开启程度)|"
+ r"(?:开度|开启程度)(?:增加|提高|增大|调大)|"
+ r"开大|多开|再打开|再开启|再开"
+)
+OPENING_DECREASE = (
+ r"(?:减少|降低|减小|调小)(?:开度|开启程度)|"
+ r"(?:开度|开启程度)(?:减少|降低|减小|调小)|"
+ r"关小|少开|再关闭|再关上|再关"
+)
+WINDOW_INCREASE = rf"{OPENING_INCREASE}|再降下|再下降|多降"
+WINDOW_DECREASE = rf"{OPENING_DECREASE}|再升起|再上升|多升"
+
+CHINESE_DIGITS = {
+ "一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5,
+ "六": 6, "七": 7, "八": 8, "九": 9, "十": 10,
+}
+WINDOW_TARGET_LABELS = {
+ "左前": ("front_left",), "前左": ("front_left",),
+ "右前": ("front_right",), "前右": ("front_right",),
+ "左后": ("rear_left",), "后左": ("rear_left",),
+ "右后": ("rear_right",), "后右": ("rear_right",),
+ "前排": ("front_left", "front_right"),
+ "后排": ("rear_left", "rear_right"),
+ "全部": ("front_left", "front_right", "rear_left", "rear_right"),
+ "所有": ("front_left", "front_right", "rear_left", "rear_right"),
+ "四个": ("front_left", "front_right", "rear_left", "rear_right"),
+ "全车": ("front_left", "front_right", "rear_left", "rear_right"),
+}
+
+
+@dataclass(frozen=True)
+class RelativePolicy:
+ default_step: int
+ minimum: int
+ maximum: int
+ unit: str
+
+
+@dataclass
+class RelativeResolution:
+ actions: list[dict[str, Any]] = field(default_factory=list)
+ ambiguities: list[dict[str, Any]] = field(default_factory=list)
+ claimed_controls: set[tuple[str, str]] = field(default_factory=set)
+
+
+class NativeCanRelativeResolver:
+ """Convert relative language into bounded absolute native-CAN actions."""
+
+ def __init__(self, repository_root: Path) -> None:
+ document = yaml.safe_load(
+ (repository_root / "config" / "native_can_relative_controls.yaml").read_text(
+ encoding="utf-8"
+ )
+ ) or {}
+ if document.get("kind") != "NativeCanRelativeControlPolicy":
+ raise ValueError("unsupported native CAN relative-control policy")
+ raw_controls = document.get("spec", {}).get("controls", {})
+ self.policies = {
+ name: self._policy(name, raw_controls.get(name))
+ for name in (
+ "climate_temperature", "climate_fan", "ambient_brightness",
+ "shade_position", "window_position",
+ )
+ }
+
+ @staticmethod
+ def _policy(name: str, raw: Any) -> RelativePolicy:
+ if not isinstance(raw, dict):
+ raise TypeError(f"relative-control policy {name} must be a mapping")
+ values = {key: raw.get(key) for key in ("default_step", "minimum", "maximum")}
+ if any(isinstance(value, bool) or not isinstance(value, int) for value in values.values()):
+ raise TypeError(f"relative-control policy {name} values must be integers")
+ if values["default_step"] <= 0 or values["minimum"] >= values["maximum"]:
+ raise ValueError(f"relative-control policy {name} has invalid range")
+ unit = raw.get("unit")
+ if not isinstance(unit, str) or not unit:
+ raise TypeError(f"relative-control policy {name} unit must be a string")
+ return RelativePolicy(unit=unit, **values)
+
+ @staticmethod
+ def _clauses(utterance: str) -> list[str]:
+ return [
+ re.sub(r"\s+", "", clause)
+ for clause in CLAUSE_SEPARATOR.split(utterance)
+ if clause.strip()
+ ]
+
+ def matches(self, utterance: str) -> bool:
+ clauses = self._clauses(utterance)
+ for index, clause in enumerate(clauses):
+ if (
+ ("温度" in clause and self._operation(clause, GENERIC_INCREASE, GENERIC_DECREASE))
+ or (re.search(r"风量|风速", clause) and self._operation(clause, GENERIC_INCREASE, GENERIC_DECREASE))
+ or (self._ambient_clause(clauses, index) and self._operation(clause, AMBIENT_INCREASE, AMBIENT_DECREASE))
+ or (re.search(r"遮阳帘|天幕帘", clause) and self._operation(clause, OPENING_INCREASE, OPENING_DECREASE))
+ or ("车窗" in clause and self._operation(clause, WINDOW_INCREASE, WINDOW_DECREASE))
+ ):
+ return True
+ return False
+
+ def resolve(self, utterance: str, context: dict[str, Any] | None) -> RelativeResolution:
+ resolution = RelativeResolution()
+ domain_state = (
+ context.get("domain_devices", {}) if isinstance(context, dict) else {}
+ )
+ clauses = self._clauses(utterance)
+ for index, clause in enumerate(clauses):
+ if "温度" in clause:
+ self._resolve_temperature(clause, domain_state, resolution)
+ if re.search(r"风量|风速", clause):
+ self._resolve_fan(clause, domain_state, resolution)
+ if self._ambient_clause(clauses, index):
+ self._resolve_ambient(clause, domain_state, resolution)
+ if re.search(r"遮阳帘|天幕帘", clause):
+ self._resolve_shade(clause, domain_state, resolution)
+ if "车窗" in clause:
+ window_clause = clause
+ if index + 1 < len(clauses) and re.search(r"除外|除去|不包括|不含", clauses[index + 1]):
+ window_clause += clauses[index + 1]
+ self._resolve_windows(window_clause, domain_state, resolution)
+ resolution.actions = self._deduplicate_actions(resolution.actions)
+ return resolution
+
+ @staticmethod
+ def _ambient_clause(clauses: list[str], index: int) -> bool:
+ clause = clauses[index]
+ if "氛围灯" in clause:
+ return not re.search(r"颜色|色号|色温", clause) and bool(
+ re.search(r"亮度|强度|调亮|变亮|更亮|再亮|调暗|变暗|更暗|再暗", clause)
+ )
+ return index > 0 and "氛围灯" in clauses[index - 1] and bool(re.match(r"(?:亮度|强度)", clause))
+
+ @staticmethod
+ def _operation(clause: str, increase_pattern: str, decrease_pattern: str) -> tuple[str, re.Match[str]] | None:
+ if re.search(rf"(?:{increase_pattern}|{decrease_pattern})(?:到|至|为){NUMBER_TOKEN}", clause):
+ return None
+ increase = re.search(increase_pattern, clause)
+ decrease = re.search(decrease_pattern, clause)
+ if bool(increase) == bool(decrease):
+ return None
+ return ("increase", increase) if increase else ("decrease", decrease) # type: ignore[arg-type]
+
+ @staticmethod
+ def _number(text: str) -> int | None:
+ if text.isdigit():
+ return int(text)
+ if text in CHINESE_DIGITS:
+ return CHINESE_DIGITS[text]
+ if "十" not in text:
+ return None
+ tens, _, ones = text.partition("十")
+ return CHINESE_DIGITS.get(tens, 1) * 10 + CHINESE_DIGITS.get(ones, 0)
+
+ @classmethod
+ def _delta(cls, clause: str, match: re.Match[str], policy: RelativePolicy) -> int:
+ tail = clause[match.end():]
+ if re.match(r"(?:了)?一点(?:点)?", tail):
+ return policy.default_step
+ explicit = re.match(rf"(?:了)?(?P{NUMBER_TOKEN})(?:档|级|度|%|个百分点)?", tail)
+ return cls._number(explicit.group("number")) if explicit else policy.default_step
+
+ @staticmethod
+ def _claim(resolution: RelativeResolution, capability: str, *targets: str) -> None:
+ resolution.claimed_controls.update((capability, target) for target in targets)
+
+ @staticmethod
+ def _ambiguity(resolution: RelativeResolution, *, field_name: str, question: str, reason: str) -> None:
+ resolution.ambiguities.append({"field": field_name, "question": question, "reason": reason})
+
+ @staticmethod
+ def _feedback_is_physical(value: Any) -> bool:
+ return isinstance(value, dict) and value.get("feedback_source") == "physical_can_ch1"
+
+ @classmethod
+ def _value(
+ cls,
+ resolution: RelativeResolution,
+ value: Any,
+ *,
+ field_name: str,
+ label: str,
+ require_physical: bool,
+ ) -> int | None:
+ if isinstance(value, dict):
+ raw = value.get("value", value.get("position_pct", value.get("temperature_c")))
+ physical = cls._feedback_is_physical(value)
+ else:
+ raw = value
+ physical = False
+ if isinstance(raw, bool) or not isinstance(raw, (int, float)):
+ cls._ambiguity(
+ resolution,
+ field_name=field_name,
+ question=f"缺少{label}的有效状态，无法计算相对目标。",
+ reason="native_can_relative_state_unknown",
+ )
+ return None
+ if require_physical and not physical:
+ cls._ambiguity(
+ resolution,
+ field_name=field_name,
+ question=f"{label}尚无 CH1 实车反馈，真实执行不能依据虚拟状态计算相对目标。",
+ reason="native_can_relative_state_not_physical",
+ )
+ return None
+ return int(raw)
+
+ @staticmethod
+ def _target(current: int, delta: int, direction: str, policy: RelativePolicy) -> tuple[int, bool]:
+ requested = current + delta if direction == "increase" else current - delta
+ target = min(policy.maximum, max(policy.minimum, requested))
+ return target, target != requested
+
+ @classmethod
+ def _at_limit(
+ cls,
+ resolution: RelativeResolution,
+ *,
+ current: int,
+ direction: str,
+ policy: RelativePolicy,
+ field_name: str,
+ label: str,
+ ) -> bool:
+ at_limit = (direction == "increase" and current >= policy.maximum) or (
+ direction == "decrease" and current <= policy.minimum
+ )
+ if at_limit:
+ cls._ambiguity(
+ resolution,
+ field_name=field_name,
+ question=f"{label}当前已在{'上限' if direction == 'increase' else '下限'} {current}，不会重复发送控制命令。",
+ reason="native_can_relative_target_at_limit",
+ )
+ return at_limit
+
+ @staticmethod
+ def _action(
+ *,
+ domain: str,
+ capability: str,
+ target: str,
+ parameters: dict[str, Any],
+ current: int,
+ delta: int,
+ direction: str,
+ label: str,
+ clamped: bool,
+ state_source: str,
+ ) -> dict[str, Any]:
+ verb = "提高" if direction == "increase" else "降低"
+ target_value = next(iter(parameters.values()))
+ return {
+ "domain": domain,
+ "capability": capability,
+ "target": target,
+ "operation": "set",
+ "parameters": parameters,
+ "reason": f"用户要求{label}{verb}{delta}；基于当前状态 {current} 计算绝对目标 {target_value}{'，已按允许范围收敛' if clamped else ''}。",
+ "relative_resolution": {
+ "direction": direction,
+ "current": current,
+ "delta": delta,
+ "target": target_value,
+ "clamped": clamped,
+ "state_source": state_source,
+ },
+ }
+
+ @staticmethod
+ def _require_physical(domain_state: dict[str, Any]) -> bool:
+ return domain_state.get("vehicle_status", {}).get("feedback_source") == "physical_can_ch1"
+
+ def _resolve_temperature(self, clause: str, state: dict[str, Any], resolution: RelativeResolution) -> None:
+ operation = self._operation(clause, GENERIC_INCREASE, GENERIC_DECREASE)
+ if operation is None:
+ return
+ direction, match = operation
+ target = "passenger" if re.search(r"副驾|乘客位", clause) else "driver"
+ self._claim(resolution, "climate.set_temperature", target)
+ zone = state.get("climate", {}).get("zones", {}).get(target, {})
+ current = self._value(
+ resolution,
+ {"value": zone.get("temperature_c"), "feedback_source": state.get("climate", {}).get("feedback_source")},
+ field_name=f"climate.{target}_temperature_c",
+ label=f"{'副驾' if target == 'passenger' else '主驾'}温度",
+ require_physical=self._require_physical(state),
+ )
+ if current is None:
+ return
+ policy = self.policies["climate_temperature"]
+ if self._at_limit(resolution, current=current, direction=direction, policy=policy, field_name=f"climate.{target}_temperature_c", label="空调温度"):
+ return
+ delta = self._delta(clause, match, policy)
+ target_value, clamped = self._target(current, delta, direction, policy)
+ resolution.actions.append(self._action(domain="climate", capability="climate.set_temperature", target=target, parameters={"temperature_c": target_value}, current=current, delta=delta, direction=direction, label="空调温度", clamped=clamped, state_source=state.get("climate", {}).get("feedback_source", "conversation_state")))
+
+ def _resolve_fan(self, clause: str, state: dict[str, Any], resolution: RelativeResolution) -> None:
+ operation = self._operation(clause, GENERIC_INCREASE, GENERIC_DECREASE)
+ if operation is None:
+ return
+ direction, match = operation
+ self._claim(resolution, "climate.set_fan_level", "vehicle")
+ climate = state.get("climate", {})
+ current = self._value(resolution, {"value": climate.get("fan_level"), "feedback_source": climate.get("feedback_source")}, field_name="climate.fan_level", label="空调风量", require_physical=self._require_physical(state))
+ if current is None:
+ return
+ policy = self.policies["climate_fan"]
+ if self._at_limit(resolution, current=current, direction=direction, policy=policy, field_name="climate.fan_level", label="空调风量"):
+ return
+ delta = self._delta(clause, match, policy)
+ target_value, clamped = self._target(current, delta, direction, policy)
+ resolution.actions.append(self._action(domain="climate", capability="climate.set_fan_level", target="vehicle", parameters={"level": target_value}, current=current, delta=delta, direction=direction, label="空调风量", clamped=clamped, state_source=climate.get("feedback_source", "conversation_state")))
+
+ def _resolve_ambient(self, clause: str, state: dict[str, Any], resolution: RelativeResolution) -> None:
+ operation = self._operation(clause, AMBIENT_INCREASE, AMBIENT_DECREASE)
+ if operation is None:
+ return
+ direction, match = operation
+ capability = "ambient_light.set_brightness_level"
+ self._claim(resolution, capability, "vehicle")
+ ambient = state.get("ambient_light", {})
+ level = ambient.get("front_brightness_level")
+ if level is None:
+ brightness_pct = ambient.get("brightness_pct")
+ level = round(float(brightness_pct) * 5 / 100) if isinstance(brightness_pct, (int, float)) else None
+ current = self._value(resolution, {"value": level, "feedback_source": ambient.get("feedback_source")}, field_name="ambient_light.brightness_level", label="氛围灯亮度", require_physical=self._require_physical(state))
+ if current is None:
+ return
+ policy = self.policies["ambient_brightness"]
+ if self._at_limit(resolution, current=current, direction=direction, policy=policy, field_name="ambient_light.brightness_level", label="氛围灯亮度"):
+ return
+ delta = self._delta(clause, match, policy)
+ target_value, clamped = self._target(current, delta, direction, policy)
+ resolution.actions.append(self._action(domain="ambient_light", capability=capability, target="vehicle", parameters={"brightness_level": target_value}, current=current, delta=delta, direction=direction, label="氛围灯亮度", clamped=clamped, state_source=ambient.get("feedback_source", "conversation_state")))
+
+ def _resolve_shade(self, clause: str, state: dict[str, Any], resolution: RelativeResolution) -> None:
+ operation = self._operation(clause, OPENING_INCREASE, OPENING_DECREASE)
+ if operation is None:
+ return
+ direction, match = operation
+ self._claim(resolution, "roof.set_shade_position", "vehicle")
+ roof = state.get("roof", {})
+ current = self._value(resolution, {"value": roof.get("shade_position_pct"), "feedback_source": roof.get("feedback_source")}, field_name="roof.shade_position_pct", label="遮阳帘开度", require_physical=self._require_physical(state))
+ if current is None:
+ return
+ policy = self.policies["shade_position"]
+ if self._at_limit(resolution, current=current, direction=direction, policy=policy, field_name="roof.shade_position_pct", label="遮阳帘开度"):
+ return
+ delta = self._delta(clause, match, policy)
+ target_value, clamped = self._target(current, delta, direction, policy)
+ resolution.actions.append(self._action(domain="roof", capability="roof.set_shade_position", target="vehicle", parameters={"position_pct": target_value}, current=current, delta=delta, direction=direction, label="遮阳帘开度", clamped=clamped, state_source=roof.get("feedback_source", "conversation_state")))
+
+ @staticmethod
+ def _window_targets(clause: str) -> tuple[str, ...]:
+ individual = []
+ for label, targets in tuple(WINDOW_TARGET_LABELS.items())[:8]:
+ if label in clause and targets[0] not in individual:
+ individual.append(targets[0])
+ if individual:
+ return tuple(individual)
+ for label in ("全部", "所有", "四个", "全车", "前排", "后排"):
+ if label in clause:
+ return WINDOW_TARGET_LABELS[label]
+ return ()
+
+ def _resolve_windows(self, clause: str, state: dict[str, Any], resolution: RelativeResolution) -> None:
+ operation = self._operation(clause, WINDOW_INCREASE, WINDOW_DECREASE)
+ if operation is None:
+ return
+ direction, match = operation
+ targets = self._window_targets(clause)
+ if re.search(r"除外|除去|不包括|不含", clause):
+ self._ambiguity(resolution, field_name="window.target", question="相对调整多个车窗时暂不支持排除目标，请明确列出需要调整的车窗。", reason="native_can_relative_exclusion_unsupported")
+ return
+ if not targets:
+ self._ambiguity(resolution, field_name="window.target", question="相对调整车窗开度需要明确单窗、前排、后排或全部车窗。", reason="native_can_relative_target_unknown")
+ return
+ policy = self.policies["window_position"]
+ delta = self._delta(clause, match, policy)
+ windows = state.get("window", {})
+ positions = windows.get("positions", {})
+ resolved: list[dict[str, Any]] = []
+ for target in targets:
+ self._claim(resolution, "window.set_position", target)
+ position = positions.get(target, {})
+ current = self._value(resolution, position, field_name=f"window.{target}_position_pct", label=f"{target}车窗开度", require_physical=self._require_physical(state))
+ if current is None:
+ return
+ if self._at_limit(resolution, current=current, direction=direction, policy=policy, field_name=f"window.{target}_position_pct", label=f"{target}车窗开度"):
+ return
+ target_value, clamped = self._target(current, delta, direction, policy)
+ resolved.append(self._action(domain="window", capability="window.set_position", target=target, parameters={"position_pct": target_value}, current=current, delta=delta, direction=direction, label=f"{target}车窗开度", clamped=clamped, state_source=position.get("feedback_source", windows.get("feedback_source", "conversation_state"))))
+ resolution.actions.extend(resolved)
+
+ @staticmethod
+ def _deduplicate_actions(actions: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ unique: list[dict[str, Any]] = []
+ seen: set[tuple[str, str, str]] = set()
+ for action in actions:
+ key = (action["capability"], action["target"], repr(action["parameters"]))
+ if key not in seen:
+ seen.add(key)
+ unique.append(action)
+ return unique
diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py
index 0a950f1..899beeb 100755
--- a/src/cognitive_agent/application/seat_service.py
+++ b/src/cognitive_agent/application/seat_service.py
@@ -53,6 +53,12 @@ from cognitive_agent.application.model_plan_adapter import ( # noqa: E402
SEAT_MOTION_CAPABILITIES,
)
from cognitive_agent.application.native_can_intent import merge_native_can_intent # noqa: E402
+from cognitive_agent.application.native_can_relative import ( # noqa: E402
+ NativeCanRelativeResolver,
+)
+from cognitive_agent.application.semantic_ir import ( # noqa: E402
+ apply_language_safety_guards,
+)
from cognitive_agent.application.native_can_feedback import ( # noqa: E402
VehicleSafetyPreflightError,
authoritative_physical_ch1_feedback_state,
@@ -75,6 +81,7 @@ from cognitive_agent.application.zcan_seat_executor import ( # noqa: E402
from cognitive_agent.infrastructure.llm.openai_responses import ( # noqa: E402
OpenAIIntentError,
OpenAIIntentInterpreter,
+ REASONING_EFFORTS,
)
from cognitive_agent.infrastructure.can.hardware import ( # noqa: E402
HardwareSafetyError,
@@ -202,6 +209,7 @@ class SeatDemoService:
self.hardware_settings = SeatHardwareSettings.from_env(repository_root)
self.vehicle_can_settings = VehicleCanSettings.from_env(repository_root)
self.native_can_runtime = NativeCanRuntime(repository_root)
+ self.native_can_relative_resolver = NativeCanRelativeResolver(repository_root)
self.hardware_execution = PHYSICAL_BENCH_COORDINATOR
self.hardware_access = PHYSICAL_BENCH_ACCESS
# Shared by the short-lived hardware executor instances. A process must
@@ -217,6 +225,16 @@ class SeatDemoService:
repository_root, {"seat", *self.domain_runtimes}
)
self.capability_context = self.capability_registry.snapshot()
+ self.semantic_capability_metadata = {
+ capability["name"]: {
+ "risk_level": capability.get("risk_level", "high"),
+ "confirmation_policy": capability.get(
+ "confirmation_policy", "on_ambiguity"
+ ),
+ }
+ for domain in self.capability_context["domains"]
+ for capability in domain["capabilities"]
+ }
self.model_plan_adapter = ModelPlanAdapter(self.capability_context)
self.vehicle_profiles = VehicleProfileRegistry(repository_root / "config" / "vehicles")
self.mode_profiles = ModeProfileRegistry(repository_root / "config" / "modes")
@@ -316,6 +334,9 @@ class SeatDemoService:
},
"openai_available": self.openai.available,
"openai_model": self.openai.model,
+ "openai_models": self.openai.model_catalog,
+ "openai_reasoning_effort_default": self.openai.reasoning_effort,
+ "openai_reasoning_efforts": list(REASONING_EFFORTS),
"openai_responses_endpoint": self.openai.responses_endpoint,
"conversation_memory": True,
"default_vehicle_id": default_vehicle.id,
@@ -633,6 +654,7 @@ class SeatDemoService:
*,
hardware_client_id: str | None = None,
hardware_access_bypass: bool = True,
+ reasoning_effort: str | None = None,
) -> dict[str, Any]:
# Creating a reusable scene is a configuration workflow. Synthesize
# and validate it in dry-run even when the console currently targets
@@ -649,6 +671,7 @@ class SeatDemoService:
approval_token,
vehicle_id,
simulation_fault,
+ reasoning_effort,
)
if hardware_access_bypass or self.driver_only_ch1_execution_active:
with self.hardware_execution.reserve():
@@ -660,6 +683,7 @@ class SeatDemoService:
approval_token,
vehicle_id,
simulation_fault,
+ reasoning_effort,
)
authorized_client_id = self.hardware_access.ensure_authorized(hardware_client_id)
with self.hardware_execution.reserve():
@@ -672,6 +696,7 @@ class SeatDemoService:
approval_token,
vehicle_id,
simulation_fault,
+ reasoning_effort,
)
def _process_once(
@@ -683,6 +708,7 @@ class SeatDemoService:
approval_token: str | None = None,
vehicle_id: str | None = None,
simulation_fault: str = "normal",
+ reasoning_effort: str | None = None,
) -> dict[str, Any]:
utterance = utterance.strip()
if not utterance:
@@ -691,6 +717,11 @@ class SeatDemoService:
raise ValueError("utterance must be at most 1000 characters")
if provider not in {"auto", "openai", "local"}:
raise ValueError("provider must be auto, openai, or local")
+ effective_reasoning_effort = (
+ self.openai.reasoning_effort
+ if reasoning_effort is None
+ else self.openai.validate_reasoning_effort(reasoning_effort)
+ )
if execution_mode not in {"dry_run", "simulated_hardware", "hardware"}:
raise ValueError(
"execution_mode must be dry_run, simulated_hardware, or hardware"
@@ -755,6 +786,8 @@ class SeatDemoService:
"conversation_turn": context["turn_count"],
"status": "processing",
"provider_requested": provider,
+ "reasoning_effort_requested": effective_reasoning_effort,
+ "reasoning_effort_used": None,
"execution_mode": execution_mode,
"hardware_execution_allowed": (
hardware_request and not control_plane_request
@@ -828,6 +861,7 @@ class SeatDemoService:
provider,
context,
vehicle_profile,
+ reasoning_effort=reasoning_effort,
)
if self._is_dynamic_mode_stop_request(utterance):
parsed = self._apply_dynamic_mode_stop_request(parsed, utterance)
@@ -835,7 +869,25 @@ class SeatDemoService:
parsed = self._apply_mode_profiles(
parsed, matched_modes, context, vehicle_profile
)
- parsed = merge_native_can_intent(parsed, utterance)
+ relative_resolution = (
+ self.native_can_relative_resolver.resolve(utterance, context)
+ if self.native_can_relative_resolver.matches(utterance)
+ else None
+ )
+ parsed = merge_native_can_intent(
+ parsed,
+ utterance,
+ relative_actions=(
+ relative_resolution.actions if relative_resolution else None
+ ),
+ relative_ambiguities=(
+ relative_resolution.ambiguities if relative_resolution else None
+ ),
+ relative_claimed_controls=(
+ relative_resolution.claimed_controls if relative_resolution else None
+ ),
+ )
+ parsed = apply_language_safety_guards(parsed, utterance)
parsed = self._partition_native_actions(
parsed,
promote_supported=(hardware_request and self.vehicle_can_settings.enabled),
@@ -891,10 +943,43 @@ class SeatDemoService:
if provider_notice:
parsed["decision_summary"].append(provider_notice)
result["provider_used"] = provider_used
+ if provider_used.startswith("openai:"):
+ result["reasoning_effort_used"] = effective_reasoning_effort
result["decision_summary"] = parsed["decision_summary"]
result["deferred_requests"] = parsed.get("deferred_requests", [])
result["model_requests"] = parsed.get("model_requests", [])
+ if (
+ parsed.get("semantic_constraints")
+ and not parsed.get("actions")
+ and not parsed.get("domain_actions")
+ and not parsed.get("native_can_actions")
+ and not parsed.get("ambiguities")
+ and not parsed.get("unresolved_spans")
+ ):
+ intent_ir = build_intent_ir(
+ parsed,
+ utterance,
+ request_id,
+ provider_used,
+ self.semantic_capability_metadata,
+ )
+ result["intent_ir"] = intent_ir
+ result["validation"] = self.contracts.validate(intent_ir)
+ result["status"] = "no_action"
+ self._trace(
+ traces,
+ request_id,
+ "safety_check",
+ "succeeded",
+ "检测到禁止或保持约束；未生成、编译或发送任何 CAN 动作。",
+ time.perf_counter(),
+ {"constraint_count": len(intent_ir["spec"]["constraints"]), "execution_eligible": False},
+ domain="orchestration",
+ )
+ self._persist(result)
+ return result
+
context_started = time.perf_counter()
self._trace(
traces,
@@ -960,7 +1045,13 @@ class SeatDemoService:
vehicle_profile=vehicle_profile,
)
- intent_ir = build_intent_ir(parsed, utterance, request_id, provider_used)
+ intent_ir = build_intent_ir(
+ parsed,
+ utterance,
+ request_id,
+ provider_used,
+ self.semantic_capability_metadata,
+ )
result["intent_ir"] = intent_ir
generation_started = time.perf_counter()
function_config = build_function_config(parsed, request_id, provider_used)
@@ -2791,7 +2882,13 @@ class SeatDemoService:
"""Run isolated adapters under the selected physical-bus policy."""
vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
generation_started = time.perf_counter()
- intent_ir = build_intent_ir(parsed, utterance, request_id, provider_used)
+ intent_ir = build_intent_ir(
+ parsed,
+ utterance,
+ request_id,
+ provider_used,
+ self.semantic_capability_metadata,
+ )
intent_actions = [
*parsed.get("domain_actions", []),
*parsed.get("native_can_actions", []),
@@ -2823,6 +2920,47 @@ class SeatDemoService:
)
result["intent_ir"] = intent_ir
+ if intent_ir["spec"]["requires_confirmation"]:
+ validation = self.contracts.validate(intent_ir)
+ validation.extend(
+ {
+ "severity": "error",
+ "code": "AMBIGUOUS_INTENT",
+ "path": item.get("field", "utterance"),
+ "message": item.get("question", "控制目标需要确认。"),
+ }
+ for item in intent_ir["spec"]["ambiguities"]
+ )
+ validation.extend(
+ {
+ "severity": "error",
+ "code": "UNRESOLVED_UTTERANCE_SPAN",
+ "path": "utterance",
+ "message": (
+ f"未解析原文“{item['text']}”：{item['reason']}。"
+ ),
+ }
+ for item in intent_ir["spec"]["unresolved_spans"]
+ )
+ result["validation"] = validation
+ result["status"] = "needs_confirmation"
+ self._trace(
+ traces,
+ request_id,
+ "safety_check",
+ "failed",
+ "语义意图存在歧义或未覆盖片段；未进入 CAN 编译与执行。",
+ time.perf_counter(),
+ {
+ "ambiguity_count": len(intent_ir["spec"]["ambiguities"]),
+ "unresolved_span_count": len(intent_ir["spec"]["unresolved_spans"]),
+ "execution_eligible": False,
+ },
+ domain="orchestration",
+ )
+ self._persist(result)
+ return result
+
configs: dict[str, dict[str, Any]] = {}
if parsed.get("actions"):
seat_config = build_function_config(parsed, request_id, provider_used)
@@ -3531,6 +3669,8 @@ class SeatDemoService:
provider: str,
context: dict[str, Any],
vehicle_profile: VehicleProfile | None = None,
+ *,
+ reasoning_effort: str | None = None,
) -> tuple[dict[str, Any], str, str | None]:
vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
if provider == "local":
@@ -3539,7 +3679,15 @@ class SeatDemoService:
if not self.openai.available:
raise ValueError("GPT 模式需要设置 OPENAI_API_KEY")
try:
- plan = self.openai.interpret(utterance, context, self.capability_context)
+ interpret_options = {}
+ if reasoning_effort is not None:
+ interpret_options["reasoning_effort"] = reasoning_effort
+ plan = self.openai.interpret(
+ utterance,
+ context,
+ self.capability_context,
+ **interpret_options,
+ )
return (
self.model_plan_adapter.adapt(
plan,
@@ -3558,7 +3706,15 @@ class SeatDemoService:
)
if self.openai.available:
try:
- plan = self.openai.interpret(utterance, context, self.capability_context)
+ interpret_options = {}
+ if reasoning_effort is not None:
+ interpret_options["reasoning_effort"] = reasoning_effort
+ plan = self.openai.interpret(
+ utterance,
+ context,
+ self.capability_context,
+ **interpret_options,
+ )
return (
self.model_plan_adapter.adapt(
plan,
diff --git a/src/cognitive_agent/application/semantic_ir.py b/src/cognitive_agent/application/semantic_ir.py
new file mode 100644
index 0000000..cda6ec9
--- /dev/null
+++ b/src/cognitive_agent/application/semantic_ir.py
@@ -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打开|开启|启动|关闭|关掉|解锁|开锁|上锁|落锁|锁上|进入|"
+ r"切换到|调用|恢复|保存|升高|降低|后仰|前倾|前移|后移)"
+)
+PRESERVE_SEAT = re.compile(r"(?:不要|别|不想)(?:再)?动(?P主驾|副驾)")
+CONTROL_VERB = re.compile(
+ r"打开|开启|启动|关闭|关掉|解锁|开锁|上锁|落锁|锁上|进入|切换到|"
+ r"调用|恢复|保存|升高|降低|后仰|前倾|前移|后移|调到|调为|设置为"
+)
+FUZZY_UNMAPPED_CONTROL = re.compile(
+ r"(?P(?:座椅|靠背|腿托|空调|车窗|风口|灯光?)"
+ 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
diff --git a/src/cognitive_agent/infrastructure/llm/openai_responses.py b/src/cognitive_agent/infrastructure/llm/openai_responses.py
index a6123a0..0e54b48 100755
--- a/src/cognitive_agent/infrastructure/llm/openai_responses.py
+++ b/src/cognitive_agent/infrastructure/llm/openai_responses.py
@@ -10,6 +10,9 @@ import urllib.request
from typing import Any
+REASONING_EFFORTS = ("low", "medium", "high")
+
+
def _env_text(name: str, default: str) -> str:
"""Return a non-blank environment value or its documented default."""
@@ -221,7 +224,9 @@ class OpenAIIntentInterpreter:
).rstrip("/")
self.timeout_seconds = float(os.getenv("OPENAI_TIMEOUT_SECONDS", "30"))
self.max_retries = max(0, int(os.getenv("OPENAI_MAX_RETRIES", "2")))
- self.reasoning_effort = _env_text("OPENAI_REASONING_EFFORT", "medium")
+ self.reasoning_effort = self.validate_reasoning_effort(
+ _env_text("OPENAI_REASONING_EFFORT", "medium")
+ )
self.streaming = os.getenv("OPENAI_STREAM", "false").strip().lower() in {
"1",
"true",
@@ -237,11 +242,31 @@ class OpenAIIntentInterpreter:
def responses_endpoint(self) -> str:
return f"{self.base_url}/responses"
+ @property
+ def model_catalog(self) -> list[dict[str, Any]]:
+ return [
+ {
+ "id": self.model,
+ "label": self.model,
+ "provider": "openai",
+ "reasoning_efforts": list(REASONING_EFFORTS),
+ }
+ ]
+
+ @staticmethod
+ def validate_reasoning_effort(value: str) -> str:
+ normalized = str(value).strip().lower()
+ if normalized not in REASONING_EFFORTS:
+ raise ValueError("reasoning_effort must be low, medium, or high")
+ return normalized
+
def interpret(
self,
utterance: str,
context: dict[str, Any] | None = None,
capability_context: dict[str, Any] | None = None,
+ *,
+ reasoning_effort: str | None = None,
) -> dict[str, Any]:
if not self.api_key:
raise OpenAIIntentError("OPENAI_API_KEY is not configured")
@@ -256,10 +281,15 @@ class OpenAIIntentInterpreter:
}
],
}
+ effective_reasoning_effort = (
+ self.reasoning_effort
+ if reasoning_effort is None
+ else self.validate_reasoning_effort(reasoning_effort)
+ )
payload = {
"model": self.model,
"store": False,
- "reasoning": {"effort": self.reasoning_effort},
+ "reasoning": {"effort": effective_reasoning_effort},
"input": [
{
"role": "system",
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index 59998c0..ab86ba7 100755
--- a/src/cognitive_agent/server.py
+++ b/src/cognitive_agent/server.py
@@ -313,6 +313,11 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
else None
),
hardware_access_bypass=self._is_unproxied_loopback_request(),
+ reasoning_effort=(
+ str(payload["reasoning_effort"])
+ if payload.get("reasoning_effort") is not None
+ else None
+ ),
)
self._json(result, HTTPStatus.CREATED)
except HardwareAccessRequiredError as exc:
diff --git a/tests/unit/test_can_test_branch_fusion.py b/tests/unit/test_can_test_branch_fusion.py
new file mode 100644
index 0000000..57410a6
--- /dev/null
+++ b/tests/unit/test_can_test_branch_fusion.py
@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from cognitive_agent.application.native_can_relative import NativeCanRelativeResolver
+from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
+
+
+class CanTestBranchFusionTests(unittest.TestCase):
+ def setUp(self) -> None:
+ self.service = SeatDemoService(REPOSITORY_ROOT)
+ self.temporary_directory = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary_directory.cleanup)
+ self.service.artifact_root = Path(self.temporary_directory.name)
+
+ def test_negated_vehicle_control_never_compiles_or_executes(self) -> None:
+ result = self.service.process(
+ "不要打开左前车窗",
+ provider="local",
+ conversation_id="conv-negative-window",
+ )
+
+ self.assertEqual(result["status"], "no_action")
+ spec = result["intent_ir"]["spec"]
+ self.assertEqual(spec["actions"], [])
+ self.assertEqual(spec["constraints"][0]["capability"], "window.set_motion")
+ self.assertFalse(spec["execution_eligible"])
+ self.assertEqual(result["function_configs"], {})
+
+ def test_multiple_windows_and_exclusion_are_deterministic(self) -> None:
+ multiple = self.service.process(
+ "打开左前和右前车窗",
+ provider="local",
+ conversation_id="conv-two-windows",
+ )
+ self.assertEqual(multiple["status"], "succeeded")
+ self.assertEqual(
+ [item["target"]["zone"] for item in multiple["intent_ir"]["spec"]["actions"]],
+ ["front_left", "front_right"],
+ )
+
+ excluded = self.service.process(
+ "打开全部车窗，左前除外",
+ provider="local",
+ conversation_id="conv-excluded-window",
+ )
+ self.assertEqual(excluded["status"], "succeeded")
+ self.assertEqual(
+ [item["target"]["zone"] for item in excluded["intent_ir"]["spec"]["actions"]],
+ ["front_right", "rear_left", "rear_right"],
+ )
+ self.assertEqual(
+ excluded["intent_ir"]["spec"]["constraints"][0]["type"],
+ "exclude_target",
+ )
+
+ def test_unresolved_mixed_control_fails_closed_before_compile(self) -> None:
+ result = self.service.process(
+ "打开左前车窗，再把座椅调舒服点",
+ provider="local",
+ conversation_id="conv-unresolved-mixed",
+ )
+
+ self.assertEqual(result["status"], "needs_confirmation")
+ self.assertFalse(result["intent_ir"]["spec"]["execution_eligible"])
+ self.assertEqual(result["function_configs"], {})
+ self.assertIn(
+ "UNRESOLVED_UTTERANCE_SPAN",
+ {item["code"] for item in result["validation"]},
+ )
+
+ def test_relative_fan_uses_current_state_and_records_resolution(self) -> None:
+ result = self.service.process(
+ "把风量提高两档",
+ provider="local",
+ conversation_id="conv-relative-fan",
+ )
+
+ self.assertEqual(result["status"], "succeeded")
+ action = result["intent_ir"]["spec"]["actions"][0]
+ self.assertEqual(action["capability"], "climate.set_fan_level")
+ self.assertEqual(action["parameters"], {"level": 2})
+ self.assertEqual(action["resolution"]["current"], 0)
+ self.assertEqual(action["resolution"]["delta"], 2)
+
+ def test_relative_vehicle_control_requires_physical_feedback_in_hardware_context(self) -> None:
+ resolver = NativeCanRelativeResolver(REPOSITORY_ROOT)
+ context = self.service.conversations.snapshot("conv-relative-physical")
+ context["domain_devices"]["vehicle_status"]["feedback_source"] = "physical_can_ch1"
+
+ result = resolver.resolve("把风量提高一档", context)
+
+ self.assertEqual(result.actions, [])
+ self.assertEqual(
+ result.ambiguities[0]["reason"],
+ "native_can_relative_state_not_physical",
+ )
+
+ def test_reasoning_effort_reaches_openai_payload(self) -> None:
+ self.service.openai.api_key = "test-key"
+ plan = {
+ "detected_domains": ["window"],
+ "requests": [],
+ "ambiguities": [],
+ "decision_summary": ["no action"],
+ "confidence": 0.9,
+ }
+ with patch.object(self.service.openai, "interpret", return_value=plan) as interpret:
+ self.service.process(
+ "你好",
+ provider="openai",
+ conversation_id="conv-reasoning-effort",
+ reasoning_effort="high",
+ )
+
+ self.assertEqual(interpret.call_args.kwargs["reasoning_effort"], "high")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/unit/test_frontend_reasoning_layout_contract.py b/tests/unit/test_frontend_reasoning_layout_contract.py
new file mode 100644
index 0000000..c513259
--- /dev/null
+++ b/tests/unit/test_frontend_reasoning_layout_contract.py
@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from pathlib import Path
+import unittest
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+CONSOLE_SOURCE = REPOSITORY_ROOT / "apps" / "web-console" / "console-v2.js"
+CONSOLE_STYLES = REPOSITORY_ROOT / "apps" / "web-console" / "console-v2.css"
+
+
+class FrontendReasoningLayoutContractTests(unittest.TestCase):
+ def test_reasoning_control_is_removed_from_fixed_height_request_form(self) -> None:
+ source = CONSOLE_SOURCE.read_text(encoding="utf-8")
+ self.assertIn('const reasoningField = $("#reasoningField", originalForm);', source)
+ self.assertIn('reasoningField.classList.add("v2-reasoning-field");', source)
+ self.assertIn("composer.append(trigger);", source)
+ self.assertIn("document.body.append(reasoningField);", source)
+ self.assertIn('trigger.setAttribute("aria-haspopup", "listbox");', source)
+
+ def test_v2_reasoning_control_has_compact_bounded_layout(self) -> None:
+ styles = CONSOLE_STYLES.read_text(encoding="utf-8")
+ self.assertIn(".console-v2 .v2-reasoning-field .reasoning-segmented", styles)
+ self.assertIn(".console-v2 .v2-reasoning-trigger", styles)
+ self.assertIn("position: fixed;", styles)
+ self.assertIn("min-height: 42px;", styles)
+ self.assertIn("white-space: nowrap;", styles)
+
+
+if __name__ == "__main__":
+ unittest.main()