Evaluation Walkthrough

1
dataset = [
2
    {"audio": "samples/en_agent01.wav", "reference": "Thank you for calling.", "language": "en"},
3
    {"audio": "samples/es_call02.wav", "reference": "Hola, ¿en qué puedo ayudarte?", "language": "es"},
4
    # ... 50–200 entries
5
]

1
import os
2
import time
3
import wave
4
import requests
5
from jiwer import wer, cer
6
from whisper_normalizer.english import EnglishTextNormalizer
7
8
API_KEY = os.environ["SMALLEST_API_KEY"]
9
PULSE_URL = "https://api.smallest.ai/waves/v1/stt/?model=pulse"
10
11
normalizer = EnglishTextNormalizer()
12
13
def audio_duration_seconds(path: str) -> float:
14
    """Read the audio duration from a WAV header. Use librosa/soundfile for non-WAV inputs."""
15
    with wave.open(path, "rb") as wf:
16
        return wf.getnframes() / wf.getframerate()
17
18
def run_sample(sample: dict) -> dict:
19
    """Transcribe one sample and compute per-sample metrics."""
20
    with open(sample["audio"], "rb") as f:
21
        audio_bytes = f.read()
22
23
    t_start = time.perf_counter()
24
    resp = requests.post(
25
        PULSE_URL,
26
        params={
27
            "language": sample["language"],
28
            "word_timestamps": str(sample.get("word_timestamps", True)).lower(),
29
            "diarize": str(sample.get("diarize", False)).lower(),
30
        },
31
        headers={
32
            "Authorization": f"Bearer {API_KEY}",
33
            "Content-Type": "application/octet-stream",
34
        },
35
        data=audio_bytes,
36
        timeout=60,
37
    )
38
    latency_ms = (time.perf_counter() - t_start) * 1000
39
    resp.raise_for_status()
40
    data = resp.json()
41
42
    duration_s = audio_duration_seconds(sample["audio"])
43
    rtf = (latency_ms / 1000) / duration_s  # processing time ÷ audio duration; lower is better
44
45
    ref = normalizer(sample["reference"])
46
    hyp = normalizer(data["transcription"])
47
48
    return {
49
        "path": sample["audio"],
50
        "wer": wer(ref, hyp),
51
        "cer": cer(ref, hyp),
52
        "latency_ms": latency_ms,
53
        "rtf": rtf,
54
        "duration_s": duration_s,
55
        "transcription": data["transcription"],
56
        "words": data.get("words", []),
57
        "utterances": data.get("utterances", []),
58
    }

1
from whisper_normalizer.basic import BasicTextNormalizer
2
normalizer = BasicTextNormalizer()  # locale-agnostic; works for Hindi, Spanish, French, etc.

1
import pandas as pd
2
3
results = [run_sample(s) for s in dataset]
4
df = pd.DataFrame(results)
5
6
summary = {
7
    "samples": len(df),
8
    "avg_wer": df["wer"].mean(),
9
    "p95_wer": df["wer"].quantile(0.95),
10
    "avg_latency_ms": df["latency_ms"].mean(),
11
    "p95_latency_ms": df["latency_ms"].quantile(0.95),
12
    "avg_rtf": df["rtf"].mean(),
13
}
14
print(summary)

1
def breakdown(df):
2
    worst = df.sort_values("wer", ascending=False).head(5)
3
    return worst[["path", "wer", "transcription"]].to_dict(orient="records")
4
5
outliers = breakdown(df)

1
configs = [
2
    {"language": "en", "word_timestamps": True},
3
    {"language": "multi", "word_timestamps": True, "diarize": True},
4
]
5
6
def evaluate_config(config: dict, dataset: list) -> pd.DataFrame:
7
    forwardable = {"language", "word_timestamps", "diarize"}
8
    overridden = [{**s, **{k: v for k, v in config.items() if k in forwardable}} for s in dataset]
9
    return pd.DataFrame([run_sample(s) for s in overridden])
10
11
for config in configs:
12
    cfg_df = evaluate_config(config, dataset)
13
    print(config, cfg_df["wer"].mean(), cfg_df["latency_ms"].mean())