Language hints support for RT melia-1 model

[rakeshv247]

Language hints support for RT melia-1 model

[giorgosHadji]

LGTM, have you tested fully?

[rakeshv247]

Manual RT testing of language_hints / language_hints_strict

Tested against a local RT container serving melia-1. Audio: 160s, 16 kHz mono, containing Spanish, Portuguese, French, Dutch and English. One session per case.

case	language	language_hints	strict	LanguageInfo	word language tags	transcript
no-hints	multi	—	—	en, es, pt, fr, nl	es 232, pt 168, en 80, fr 70, nl 67	all 5 languages transcribed
strict-true	multi	es,fr	true	es, fr	es 463, fr 126	non-empty
strict-omitted	multi	es,fr	—	es, fr	es 463, fr 126	non-empty, identical to strict-true
strict-false	multi	es,fr	false	none	none	(empty)
override-base-language	en	es,pt,fr	true	pt, es, fr	es 259, pt 227, fr 139	non-empty, no en words
hint-absent-from-audio	multi	de	true	de	de 606	non-empty, every word tagged de
unsupported-hint	multi	xyz	—	rejected	—	lang(s) ['xyz'] in language_hints is not a supported language
multi-as-hint	multi	multi	—	rejected	—	lang(s) ['multi'] in language_hints is not a supported language
empty-hints	multi	[]	—	rejected	—	schema validation error (minItems: 1)
strict-without-hints	multi	—	true	rejected	—	schema validation error

Test script

Needs an RT endpoint serving a next-gen model and a 16-bit mono PCM WAV with several languages in it.

python test_rt_language_hints.py --url ws://<host>:<port> --audio <multilingual.wav>
python test_rt_language_hints.py --url ... --audio ... --cases strict-false,no-hints
python test_rt_language_hints.py --url ... --audio ... --list

test_rt_language_hints.py

#!/usr/bin/env python3
"""Manual test matrix for RT `language_hints` / `language_hints_strict` (PR #137).

Streams a WAV file to an RT endpoint once per case and reports, for each, the
languages the server announced, the per-word language tags, and the transcript.
Per-word tags are the only way to observe hint enforcement -- LanguageInfo just
announces languages, it does not tag words.

Usage:
    python test_rt_language_hints.py --url ws://<host>:<port> --audio <multilingual.wav>
    python test_rt_language_hints.py --url ... --audio ... --cases strict-false,no-hints
    python test_rt_language_hints.py --url ... --audio ... --list

Requires an RT endpoint serving a next-gen model (language hints are not
supported by enhanced/standard) and a 16-bit mono PCM WAV. For the multilingual
cases to mean anything the audio needs several languages in it; the reported
runs used a 160s sample containing Spanish, Portuguese, French, Dutch and
English, and the hint cases below name languages from that set.
"""

from __future__ import annotations

import argparse
import asyncio
import io
import wave
from dataclasses import dataclass
from dataclasses import field
from typing import Optional

from speechmatics.rt import AsyncClient
from speechmatics.rt import AudioEncoding
from speechmatics.rt import AudioFormat
from speechmatics.rt import ServerMessageType
from speechmatics.rt import TranscriptionConfig
from speechmatics.rt import TranscriptionError


@dataclass
class Case:
    name: str
    language: str
    hints: Optional[list[str]] = None
    strict: Optional[bool] = None
    expect: str = ""


# One case per documented behaviour of the feature, plus the rejection paths.
CASES = [
    Case("no-hints", "multi", expect="all languages in the audio are detected"),
    Case("strict-true", "multi", ["es", "fr"], True, expect="only the hinted languages"),
    Case("strict-omitted", "multi", ["es", "fr"], None, expect="same as strict-true"),
    Case("strict-false", "multi", ["es", "fr"], False, expect="bias-only regime, not implemented yet"),
    Case("override-base-language", "en", ["es", "pt", "fr"], True, expect="hints win over language=en"),
    Case("hint-absent-from-audio", "multi", ["de"], True, expect="restricts rather than biases"),
    Case("unsupported-hint", "multi", ["xyz"], None, expect="rejected: not a supported language"),
    Case("multi-as-hint", "multi", ["multi"], None, expect="rejected: not a supported language"),
    Case("empty-hints", "multi", [], None, expect="rejected: schema minItems 1"),
    Case("strict-without-hints", "multi", None, True, expect="rejected: schema requires language_hints"),
]


@dataclass
class Result:
    case: Case
    announced: list[str] = field(default_factory=list)
    word_langs: dict[str, int] = field(default_factory=dict)
    finals: list[str] = field(default_factory=list)
    error: Optional[str] = None

    @property
    def transcript(self) -> str:
        return " ".join(self.finals).strip()


def load_wav(path: str) -> tuple[bytes, AudioFormat]:
    with wave.open(path, "rb") as wav:
        if wav.getsampwidth() != 2 or wav.getnchannels() != 1:
            raise SystemExit(f"{path}: need 16-bit mono PCM, got {wav.getsampwidth() * 8}-bit / {wav.getnchannels()}ch")
        frames = wav.readframes(wav.getnframes())
        rate = wav.getframerate()
    print(f"[audio] {path}: {rate} Hz mono s16le, {len(frames) / (2 * rate):.1f}s\n")
    return frames, AudioFormat(encoding=AudioEncoding.PCM_S16LE, sample_rate=rate)


async def run_case(case: Case, frames: bytes, audio_format: AudioFormat, args: argparse.Namespace) -> Result:
    result = Result(case=case)
    config = TranscriptionConfig(
        language=case.language,
        model=args.model,
        language_hints=case.hints,
        language_hints_strict=case.strict,
    )
    print(f"--- {case.name}: language={case.language!r} hints={case.hints!r} strict={case.strict!r}")

    try:
        async with AsyncClient(url=args.url, api_key=args.api_key) as client:

            @client.on(ServerMessageType.LANGUAGE_INFO)
            def on_language_info(msg: dict) -> None:
                result.announced.append(msg.get("language", "?"))

            @client.on(ServerMessageType.ADD_TRANSCRIPT)
            def on_final(msg: dict) -> None:
                for res in msg.get("results", []):
                    for alt in res.get("alternatives", []):
                        lang = alt.get("language")
                        if lang:
                            result.word_langs[lang] = result.word_langs.get(lang, 0) + 1
                text = msg["metadata"]["transcript"].strip()
                if text:
                    result.finals.append(text)

            await client.transcribe(
                io.BytesIO(frames),
                transcription_config=config,
                audio_format=audio_format,
                timeout=args.timeout,
            )
    except TranscriptionError as exc:
        result.error = str(exc).split("\n")[0]

    if result.error:
        print(f"    error: {result.error[:120]}")
    else:
        tags = ", ".join(f"{k} {v}" for k, v in sorted(result.word_langs.items(), key=lambda kv: -kv[1]))
        print(f"    announced: {result.announced or 'none'} | word tags: {tags or 'none'}")
        print(f"    transcript: {result.transcript[:100] or '(empty)'}")
    return result


def print_table(results: list[Result]) -> None:
    print("\n| case | language | hints | strict | LanguageInfo | word language tags | transcript |")
    print("|---|---|---|---|---|---|---|")
    for r in results:
        hints = ",".join(r.case.hints) if r.case.hints else ("[]" if r.case.hints == [] else "—")
        strict = {None: "—", True: "true", False: "false"}[r.case.strict]
        if r.error:
            reason = r.error[:60].replace("|", "/")
            print(f"| {r.case.name} | {r.case.language} | {hints} | {strict} | rejected | — | `{reason}` |")
            continue
        tags = ", ".join(f"{k} {v}" for k, v in sorted(r.word_langs.items(), key=lambda kv: -kv[1])) or "none"
        announced = ", ".join(r.announced) or "none"
        excerpt = (r.transcript[:40] + "…") if r.transcript else "**(empty)**"
        print(f"| {r.case.name} | {r.case.language} | {hints} | {strict} | {announced} | {tags} | {excerpt} |")


async def main() -> None:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--url", required=False, help="RT websocket URL, e.g. ws://host:9000")
    p.add_argument("--audio", required=False, help="16-bit mono PCM WAV")
    p.add_argument("--api-key", default="dev", help="ignored by a local container, required by SaaS")
    p.add_argument("--model", default="melia-1")
    p.add_argument("--cases", default=None, help="comma-separated case names (default: all)")
    p.add_argument("--list", action="store_true", help="list case names and exit")
    p.add_argument("--timeout", type=float, default=300.0)
    args = p.parse_args()

    if args.list:
        for c in CASES:
            print(f"{c.name:24s} {c.expect}")
        return
    if not args.url or not args.audio:
        raise SystemExit("--url and --audio are required")

    wanted = args.cases.split(",") if args.cases else None
    cases = [c for c in CASES if wanted is None or c.name in wanted]
    if not cases:
        raise SystemExit(f"no cases matched {args.cases!r}; use --list")

    frames, audio_format = load_wav(args.audio)
    results = [await run_case(c, frames, audio_format, args) for c in cases]
    print_table(results)


if __name__ == "__main__":
    asyncio.run(main())

[giorgosHadji]

LGTM ship it