优化语音输入/普通文本输入

2026-09-03 13:36:40 +08:00
parent 009d8ca73e
commit 32d179143f
26 changed files with 2318 additions and 53 deletions
@@ -15,6 +15,44 @@ from cognitive_agent.infrastructure.speech.understanding import (
 REPOSITORY_ROOT = Path(__file__).resolve().parents[2]


+def _canonical_actions(result: dict[str, object]) -> list[dict[str, object]]:
+    intent_ir = result.get("intent_ir")
+    assert isinstance(intent_ir, dict)
+    spec = intent_ir.get("spec")
+    assert isinstance(spec, dict)
+    actions = spec.get("actions")
+    assert isinstance(actions, list)
+    return [
+        {
+            "domain": action.get("domain"),
+            "capability": action.get("capability"),
+            "target": action.get("target"),
+            "operation": action.get("operation"),
+            "parameters": action.get("parameters"),
+        }
+        for action in actions
+        if isinstance(action, dict)
+    ]
+
+
+def _sorted_canonical_actions(result: dict[str, object]) -> list[dict[str, object]]:
+    return sorted(
+        _canonical_actions(result),
+        key=lambda action: (
+            str(action["domain"]),
+            str(action["capability"]),
+            repr(action["target"]),
+            repr(action["parameters"]),
+        ),
+    )
+
+
+def _mode_service(directory: str) -> SeatDemoService:
+    service = SeatDemoService(REPOSITORY_ROOT)
+    service.artifact_root = Path(directory) / "artifacts"
+    return service
+
+
@pytest.mark.parametrize(
    ("utterance", "capability", "target", "parameters"),
    (
@@ -194,3 +232,157 @@ def test_every_registered_native_runtime_capability_keeps_native_bridge_binding(
            assert [
                action["capability"] for action in parsed["native_can_actions"]
            ] == [capability]
+
+
+def test_text_and_speech_share_implicit_mode_target_and_start_command() -> None:
+    processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
+    understanding = processor.process("将座椅调整为木马模式。").to_dict()
+    understanding["utterance_id"] = "mode-start-equivalence"
+
+    with tempfile.TemporaryDirectory() as directory, patch.dict(
+        os.environ,
+        {
+            "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false",
+            "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"),
+        },
+        clear=False,
+    ):
+        service = _mode_service(directory)
+        text_result = service.process(
+            "将座椅调整为木马模式。",
+            provider="local",
+            conversation_id="conv-text-mode-start",
+        )
+        speech_result = service.process(
+            understanding["clean_text"],
+            provider="local",
+            conversation_id="conv-speech-mode-start",
+            _speech_understanding=understanding,
+        )
+
+    assert understanding["execution_eligible"] is True
+    assert text_result["status"] == speech_result["status"] == "succeeded"
+    assert _canonical_actions(text_result) == _canonical_actions(speech_result)
+    assert _canonical_actions(text_result) == [
+        {
+            "domain": "seat",
+            "capability": "seat.start_dynamic_mode",
+            "target": {
+                "device": "seat",
+                "zone": "front_right",
+                "instance": "passenger",
+            },
+            "operation": "activate",
+            "parameters": {"mode_id": 1},
+        }
+    ]
+
+
+def test_text_and_speech_share_dynamic_mode_stop_command() -> None:
+    processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
+    understanding = processor.process("关闭木马模式").to_dict()
+    understanding["utterance_id"] = "mode-stop-equivalence"
+
+    with tempfile.TemporaryDirectory() as directory, patch.dict(
+        os.environ,
+        {
+            "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false",
+            "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"),
+        },
+        clear=False,
+    ):
+        service = _mode_service(directory)
+        text_result = service.process(
+            "关闭木马模式",
+            provider="local",
+            conversation_id="conv-text-mode-stop",
+        )
+        speech_result = service.process(
+            understanding["clean_text"],
+            provider="local",
+            conversation_id="conv-speech-mode-stop",
+            _speech_understanding=understanding,
+        )
+
+    assert understanding["clean_text"] == "关闭木马模式。"
+    assert understanding["execution_eligible"] is True
+    assert text_result["status"] == speech_result["status"] == "succeeded"
+    assert _canonical_actions(text_result) == _canonical_actions(speech_result)
+    assert _canonical_actions(text_result)[0]["capability"] == "seat.stop_dynamic_mode"
+    assert _canonical_actions(text_result)[0]["parameters"] == {}
+
+
+def test_explicit_incompatible_mode_target_fails_closed_for_both_inputs() -> None:
+    processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
+    understanding = processor.process("将主驾座椅调整为木马模式。").to_dict()
+
+    with tempfile.TemporaryDirectory() as directory, patch.dict(
+        os.environ,
+        {
+            "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false",
+            "COGNITIVE_AGENT_MODE_PROFILE_ROOT": str(Path(directory) / "modes"),
+        },
+        clear=False,
+    ):
+        service = _mode_service(directory)
+        text_result = service.process(
+            "将主驾座椅调整为木马模式。",
+            provider="local",
+            conversation_id="conv-text-mode-rejected",
+        )
+
+    assert understanding["semantic_status"] == "recognized_unsupported"
+    assert understanding["execution_eligible"] is False
+    assert text_result["status"] == "needs_confirmation"
+    assert _canonical_actions(text_result) == []
+
+
+def test_text_and_speech_share_colloquial_multidomain_commands() -> None:
+    utterance = "先打开主驾座椅通风，再打开副驾座椅通风，最后把空调调成二十四度"
+    processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
+    understanding = processor.process(utterance).to_dict()
+    understanding["utterance_id"] = "multidomain-equivalence"
+
+    with tempfile.TemporaryDirectory() as directory, patch.dict(
+        os.environ,
+        {
+            "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
+            "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false",
+        },
+        clear=False,
+    ):
+        service = _mode_service(directory)
+        text_result = service.process(
+            utterance,
+            provider="local",
+            conversation_id="conv-text-multidomain",
+        )
+        speech_result = service.process(
+            understanding["clean_text"],
+            provider="local",
+            conversation_id="conv-speech-multidomain",
+            _speech_understanding=understanding,
+        )
+
+    assert understanding["clean_text"] == (
+        "打开主驾座椅通风，然后打开副驾座椅通风，然后将空调设置为24度。"
+    )
+    assert understanding["execution_eligible"] is True
+    assert text_result["status"] == speech_result["status"] == "succeeded"
+    assert _sorted_canonical_actions(text_result) == _sorted_canonical_actions(
+        speech_result
+    )
+    assert {
+        action["capability"] for action in _canonical_actions(text_result)
+    } == {
+        "seat.set_driver_ventilation_native",
+        "seat.set_ventilation",
+        "climate.set_temperature",
+    }