vee
289 lines
12 KiB
Python
289 lines
12 KiB
Python
"""Measure text semantics and optional real/synthetic WAV speech regressions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import statistics
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from cognitive_agent.infrastructure.speech.sensevoice import SpeechTranscriber
|
|
from cognitive_agent.infrastructure.speech.understanding import (
|
|
UnderstandingTranscriptProcessor,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
TEXT_FIXTURE = REPOSITORY_ROOT / "tests" / "fixtures" / "speech_text_regression.json"
|
|
DEFAULT_AUDIO_DIR = REPOSITORY_ROOT / "artifacts" / "tmp" / "speech-regression-audio"
|
|
NUMBER = re.compile(r"-?\d+(?:\.\d+)?")
|
|
CAPABILITY_ALIGNMENT_TEXTS = (
|
|
"把遮阳帘打开",
|
|
"打开前大灯",
|
|
"打开主驾座椅通风",
|
|
"把主驾座椅调整为零重力模式",
|
|
"打开一半车窗",
|
|
"打开氛围灯并调整为红色",
|
|
"打开主驾座椅加热",
|
|
"打开副驾座椅通风",
|
|
"打开遮阳帘，然后把副驾座椅向后调整50度",
|
|
"打开四分之一车窗",
|
|
"打开副驾座椅通风和加热，然后将副驾座椅靠背调整到最大位置",
|
|
"打开主驾座椅通风和加热，并且把空调调整为24度",
|
|
"打开副驾座椅加热以及通风",
|
|
"关闭空调和打开氛围灯",
|
|
"打开前排和后排空调",
|
|
"打开主驾和副驾车窗",
|
|
)
|
|
|
|
AUDIO_EXPECTATIONS: dict[str, dict[str, Any]] = {
|
|
"01_filler_hvac.wav": {
|
|
"raw": "呃，那个那个那个打开空调到24度。",
|
|
"clean": "打开空调并设置为24度。",
|
|
"capabilities": ["climate.set_power", "climate.set_temperature"],
|
|
},
|
|
"02_seat_refine.wav": {
|
|
"raw": "座椅调整到呃20呃，靠背调整到24度。",
|
|
"clean": "将座椅靠背调整到24度。",
|
|
"capabilities": ["seat.set_backrest_recline"],
|
|
},
|
|
"03_domain_repair.wav": {
|
|
"raw": "打开空调，不对，把座椅靠背向后调24度。",
|
|
"clean": "将座椅靠背向后调整24度。",
|
|
"capabilities": ["seat.set_backrest_recline"],
|
|
},
|
|
"04_multi.wav": {
|
|
"raw": "空调调到24度，然后把座椅靠背调到20度。",
|
|
"clean": "将空调设置为24度，然后将座椅靠背调整到20度。",
|
|
"capabilities": ["climate.set_temperature", "seat.set_backrest_recline"],
|
|
},
|
|
"05_number_repair.wav": {
|
|
"raw": "空调调到24，不对，25度。",
|
|
"clean": "将空调设置为25度。",
|
|
"capabilities": ["climate.set_temperature"],
|
|
},
|
|
"06_direction_repair.wav": {
|
|
"raw": "把座椅往前，不对，往后一点。",
|
|
"clean": "将座椅向后调整一点。",
|
|
"capabilities": ["seat.set_driver_axis_native"],
|
|
},
|
|
"07_negation.wav": {"raw": "不要打开空调。", "clean": "不要打开空调。", "capabilities": []},
|
|
"08_chat.wav": {"raw": "今天天气不错。", "clean": "", "capabilities": []},
|
|
"09_fast.wav": {
|
|
"raw": "打开空调，然后把座椅靠背向后调24度。",
|
|
"clean": "打开空调，然后将座椅靠背向后调整24度。",
|
|
"capabilities": ["climate.set_power", "seat.set_backrest_recline"],
|
|
},
|
|
"10_slow.wav": {
|
|
"raw": "打开空调，然后把座椅靠背向后调24度。",
|
|
"clean": "打开空调，然后将座椅靠背向后调整24度。",
|
|
"capabilities": ["climate.set_power", "seat.set_backrest_recline"],
|
|
},
|
|
}
|
|
|
|
|
|
def ratio(passed: int, total: int) -> float:
|
|
return round(passed / total, 4) if total else 0.0
|
|
|
|
|
|
def percentile(values: list[float], quantile: float) -> float:
|
|
if not values:
|
|
return 0.0
|
|
ordered = sorted(values)
|
|
index = round((len(ordered) - 1) * quantile)
|
|
return round(ordered[index], 3)
|
|
|
|
|
|
def numeric_parameters(commands: list[dict[str, Any]]) -> list[float]:
|
|
return sorted(
|
|
float(value)
|
|
for command in commands
|
|
for name, value in command.get("parameters", {}).items()
|
|
if name not in set(command.get("qualifiers", {}).get("implicit_parameters", ()))
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
)
|
|
|
|
|
|
def expected_numbers(clean: str) -> list[float]:
|
|
return sorted(float(value) for value in NUMBER.findall(clean))
|
|
|
|
|
|
def score_cases(cases: list[dict[str, Any]]) -> dict[str, Any]:
|
|
def domains(capabilities: list[str | None]) -> set[str]:
|
|
return {
|
|
capability.split(".", 1)[0]
|
|
for capability in capabilities
|
|
if capability
|
|
}
|
|
|
|
total = len(cases)
|
|
def parameter_numbers(item: dict[str, Any]) -> list[float]:
|
|
return [
|
|
float(value)
|
|
for value in item.get("parameter_numbers", expected_numbers(item["clean"]))
|
|
]
|
|
|
|
semantic = sum(item["actual_clean"] == item["clean"] for item in cases)
|
|
intent = sum(item["actual_capabilities"] == item["capabilities"] for item in cases)
|
|
device = sum(
|
|
domains(item["actual_capabilities"]) == domains(item["capabilities"])
|
|
for item in cases
|
|
)
|
|
parameter = sum(item["actual_numbers"] == parameter_numbers(item) for item in cases)
|
|
corrections = [item for item in cases if item.get("cancelled", 0)]
|
|
correction_success = sum(
|
|
item["actual_capabilities"] == item["capabilities"]
|
|
and item["actual_cancelled"] >= item.get("cancelled", 0)
|
|
for item in corrections
|
|
)
|
|
no_command = [item for item in cases if not item["capabilities"]]
|
|
false_activation = sum(bool(item["actual_capabilities"]) for item in no_command)
|
|
wrong_device = sum(
|
|
domains(item["actual_capabilities"]) != domains(item["capabilities"])
|
|
for item in cases
|
|
)
|
|
wrong_number = sum(item["actual_numbers"] != parameter_numbers(item) for item in cases)
|
|
semantic_times = [item["semantic_ms"] for item in cases]
|
|
resolution_times = [item["command_resolution_ms"] for item in cases]
|
|
return {
|
|
"case_count": total,
|
|
"semantic_clean_accuracy": ratio(semantic, total),
|
|
"intent_exact_match": ratio(intent, total),
|
|
"device_accuracy": ratio(device, total),
|
|
"action_accuracy": ratio(intent, total),
|
|
"parameter_exact_match": ratio(parameter, total),
|
|
"correction_success_rate": ratio(correction_success, len(corrections)),
|
|
"false_command_activation_rate": ratio(false_activation, len(no_command)),
|
|
"severe_wrong_device_count": wrong_device,
|
|
"severe_wrong_number_count": wrong_number,
|
|
"semantic_processing_p50_ms": percentile(semantic_times, 0.50),
|
|
"semantic_processing_p95_ms": percentile(semantic_times, 0.95),
|
|
"command_resolution_p50_ms": percentile(resolution_times, 0.50),
|
|
"command_resolution_p95_ms": percentile(resolution_times, 0.95),
|
|
}
|
|
|
|
|
|
def run_text() -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
fixtures = json.loads(TEXT_FIXTURE.read_text(encoding="utf-8"))
|
|
cases: list[dict[str, Any]] = []
|
|
for fixture in fixtures:
|
|
started = time.perf_counter()
|
|
result = processor.process(fixture["raw"])
|
|
semantic_ms = (time.perf_counter() - started) * 1000
|
|
commands = [command.to_dict() for command in result.commands]
|
|
cases.append(
|
|
{
|
|
**fixture,
|
|
"actual_clean": result.clean_text,
|
|
"actual_capabilities": [command["capability"] for command in commands],
|
|
"actual_numbers": numeric_parameters(commands),
|
|
"actual_cancelled": len(result.cancelled_fragments),
|
|
"semantic_ms": semantic_ms,
|
|
"command_resolution_ms": result.command_resolution_ms,
|
|
}
|
|
)
|
|
return score_cases(cases), cases
|
|
|
|
|
|
def run_audio(audio_dir: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
transcriber = SpeechTranscriber(REPOSITORY_ROOT)
|
|
cases: list[dict[str, Any]] = []
|
|
for name, expected in AUDIO_EXPECTATIONS.items():
|
|
path = audio_dir / name
|
|
if not path.is_file():
|
|
continue
|
|
wall_started = time.perf_counter()
|
|
result = transcriber.transcribe_wav(
|
|
path.read_bytes(), utterance_id=f"benchmark-{path.stem}"
|
|
)
|
|
wall_ms = (time.perf_counter() - wall_started) * 1000
|
|
commands = list(result["commands"])
|
|
cases.append(
|
|
{
|
|
"name": name,
|
|
**expected,
|
|
"cancelled": 1 if "repair" in name or "refine" in name else 0,
|
|
"actual_raw": result["raw_text"],
|
|
"actual_clean": result["clean_text"],
|
|
"actual_capabilities": [command["capability"] for command in commands],
|
|
"actual_numbers": numeric_parameters(commands),
|
|
"actual_cancelled": len(result["cancelled_fragments"]),
|
|
"semantic_ms": float(result["timings"]["semantic_ms"]),
|
|
"command_resolution_ms": float(result["timings"]["command_resolution_ms"]),
|
|
"asr_ms": float(result["timings"]["asr_ms"]),
|
|
"end_to_end_ms": float(result["timings"]["total_ms"]),
|
|
"wall_ms": wall_ms,
|
|
"duplicate_callbacks": int(result["assembly"]["duplicate_callbacks"]),
|
|
}
|
|
)
|
|
metrics = score_cases(cases)
|
|
raw_exact = sum(item["actual_raw"] == item["raw"] for item in cases)
|
|
metrics.update(
|
|
{
|
|
"raw_transcription_accuracy": ratio(raw_exact, len(cases)),
|
|
"duplicate_transcript_rate": ratio(
|
|
sum(item["duplicate_callbacks"] > 0 for item in cases), len(cases)
|
|
),
|
|
"asr_inference_p50_ms": percentile([item["asr_ms"] for item in cases], 0.50),
|
|
"asr_inference_p95_ms": percentile([item["asr_ms"] for item in cases], 0.95),
|
|
"end_to_end_p50_ms": percentile([item["end_to_end_ms"] for item in cases], 0.50),
|
|
"end_to_end_p95_ms": percentile([item["end_to_end_ms"] for item in cases], 0.95),
|
|
"cold_start_end_to_end_ms": round(cases[0]["end_to_end_ms"], 3) if cases else 0.0,
|
|
"steady_state_end_to_end_p50_ms": percentile(
|
|
[item["end_to_end_ms"] for item in cases[1:]], 0.50
|
|
),
|
|
"steady_state_end_to_end_p95_ms": percentile(
|
|
[item["end_to_end_ms"] for item in cases[1:]], 0.95
|
|
),
|
|
"wall_mean_ms": round(statistics.fmean(item["wall_ms"] for item in cases), 3)
|
|
if cases
|
|
else 0.0,
|
|
}
|
|
)
|
|
return metrics, cases
|
|
|
|
|
|
def run_capability_resolution(*, iterations: int = 100) -> dict[str, float | int]:
|
|
processor = UnderstandingTranscriptProcessor(REPOSITORY_ROOT)
|
|
semantic_times: list[float] = []
|
|
resolution_times: list[float] = []
|
|
for _ in range(iterations):
|
|
for text in CAPABILITY_ALIGNMENT_TEXTS:
|
|
result = processor.process(text)
|
|
semantic_times.append(result.processing_ms)
|
|
resolution_times.append(result.command_resolution_ms)
|
|
return {
|
|
"sample_count": len(semantic_times),
|
|
"semantic_p50_ms": percentile(semantic_times, 0.50),
|
|
"semantic_p95_ms": percentile(semantic_times, 0.95),
|
|
"command_resolution_p50_ms": percentile(resolution_times, 0.50),
|
|
"command_resolution_p95_ms": percentile(resolution_times, 0.95),
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--audio-dir", type=Path, default=DEFAULT_AUDIO_DIR)
|
|
parser.add_argument("--details", action="store_true")
|
|
args = parser.parse_args()
|
|
text_metrics, text_cases = run_text()
|
|
payload: dict[str, Any] = {
|
|
"text": text_metrics,
|
|
"capability_alignment": run_capability_resolution(),
|
|
}
|
|
if args.audio_dir.is_dir():
|
|
audio_metrics, audio_cases = run_audio(args.audio_dir)
|
|
payload["audio"] = audio_metrics
|
|
if args.details:
|
|
payload["audio_cases"] = audio_cases
|
|
if args.details:
|
|
payload["text_cases"] = text_cases
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|