minghanchen
389 lines
14 KiB
Python
389 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cognitive_agent.application.seat_service import SeatDemoService
|
|
from cognitive_agent.infrastructure.speech.understanding import (
|
|
UnderstandingTranscriptProcessor,
|
|
)
|
|
|
|
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"),
|
|
(
|
|
(
|
|
"打开一半车窗",
|
|
"window.set_position",
|
|
{"device": "window", "zone": "front_left"},
|
|
{"position_pct": 50},
|
|
),
|
|
(
|
|
"把副驾腿托向上调整到最大位置",
|
|
"seat.set_legrest_height",
|
|
{"device": "seat", "zone": "front_right", "instance": "passenger"},
|
|
{"position_pct": 100},
|
|
),
|
|
(
|
|
"把遮阳帘打开",
|
|
"roof.set_shade_motion",
|
|
{"device": "roof", "zone": "vehicle"},
|
|
{"motion": "open"},
|
|
),
|
|
(
|
|
"遮阳帘打开一半",
|
|
"roof.set_shade_motion",
|
|
{"device": "roof", "zone": "vehicle"},
|
|
{"motion": "half"},
|
|
),
|
|
(
|
|
"把遮阳帘打开到四分之一",
|
|
"roof.set_shade_position",
|
|
{"device": "roof", "zone": "vehicle"},
|
|
{"position_pct": 25},
|
|
),
|
|
),
|
|
)
|
|
def test_server_owned_speech_semantics_bypass_second_language_parse(
|
|
utterance: str,
|
|
capability: str,
|
|
target: dict[str, str],
|
|
parameters: dict[str, int],
|
|
) -> None:
|
|
processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
understanding = processor.process(utterance).to_dict()
|
|
understanding["utterance_id"] = "bridge-regression"
|
|
|
|
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 = SeatDemoService(REPOSITORY_ROOT)
|
|
service.artifact_root = Path(directory) / "artifacts"
|
|
with patch.object(
|
|
service,
|
|
"_interpret",
|
|
side_effect=AssertionError("cleanText must not be parsed a second time"),
|
|
):
|
|
result = service.process(
|
|
understanding["clean_text"],
|
|
provider="local",
|
|
execution_mode="dry_run",
|
|
_speech_understanding=understanding,
|
|
)
|
|
|
|
assert result["status"] == "succeeded"
|
|
assert result["provider_used"] == "speech_semantic"
|
|
assert result["speech_understanding"]["raw_text"] == utterance
|
|
actions = result["intent_ir"]["spec"]["actions"]
|
|
assert len(actions) == 1
|
|
assert actions[0]["capability"] == capability
|
|
assert actions[0]["target"] == target
|
|
assert actions[0]["parameters"] == parameters
|
|
assert actions[0]["provenance"]["parser"] == "speech_semantic"
|
|
|
|
|
|
def test_non_executable_speech_cannot_enter_orchestration_bridge() -> None:
|
|
processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
understanding = processor.process("打开前大灯").to_dict()
|
|
understanding["utterance_id"] = "bridge-unsupported"
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
|
|
},
|
|
clear=False,
|
|
):
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
context = service._context_for_vehicle(
|
|
service.conversations.snapshot("conv-speech-bridge"),
|
|
service.vehicle_profiles.get(None),
|
|
)
|
|
with pytest.raises(ValueError, match="not execution eligible"):
|
|
service._adapt_speech_understanding(
|
|
understanding,
|
|
context,
|
|
service.vehicle_profiles.get(None),
|
|
)
|
|
|
|
|
|
def test_every_registered_native_runtime_capability_keeps_native_bridge_binding() -> None:
|
|
def valid_value(spec: dict[str, object]) -> object:
|
|
allowed = spec.get("allowed_values")
|
|
if isinstance(allowed, list) and allowed:
|
|
return allowed[0]
|
|
value_type = str(spec.get("type", "string"))
|
|
if value_type in {"integer", "number"}:
|
|
return spec.get("minimum", 0)
|
|
if value_type == "boolean":
|
|
return True
|
|
if value_type == "object":
|
|
return {}
|
|
if value_type == "array":
|
|
return []
|
|
return "test"
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
|
|
},
|
|
clear=False,
|
|
):
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
context = service._context_for_vehicle(
|
|
service.conversations.snapshot("conv-native-speech-matrix"),
|
|
service.vehicle_profiles.get(None),
|
|
)
|
|
registered = {
|
|
definition.name: definition
|
|
for definition in service.capability_registry.definitions()
|
|
if definition.name in service.native_can_runtime.supported_capabilities
|
|
}
|
|
assert registered
|
|
for capability, definition in registered.items():
|
|
parameters = {
|
|
str(spec["name"]): valid_value(spec)
|
|
for spec in definition.parameters
|
|
if spec.get("required")
|
|
}
|
|
target = definition.target_scopes[0]
|
|
understanding = {
|
|
"raw_text": f"native bridge matrix: {capability}",
|
|
"execution_eligible": True,
|
|
"commands": [
|
|
{
|
|
"status": "active",
|
|
"negated": False,
|
|
"resolution_status": "executable",
|
|
"validation_issues": [],
|
|
"confidence": 1.0,
|
|
"source_text": capability,
|
|
"executor_payload": {
|
|
"domain": definition.domain,
|
|
"capability": capability,
|
|
"target": target,
|
|
"operation": "set",
|
|
"parameters": parameters,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
parsed = service._adapt_speech_understanding(
|
|
understanding,
|
|
context,
|
|
service.vehicle_profiles.get(None),
|
|
)
|
|
assert parsed["deferred_requests"] == [], capability
|
|
assert parsed["actions"] == [], capability
|
|
assert parsed["domain_actions"] == [], capability
|
|
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",
|
|
}
|