"""One small, real voice-agent turn on Modal, measured stage by stage (Idea, 22 Oct). modal run scripts/modal_voice_demo.py --out One GPU function on a T4 runs the whole cascade on a single synthetic support call: open speech-to-text (Whisper small.en) -> a small open LLM (Qwen2.5-1.5B-Instruct, fp16, not quantized) writes the agent's reply -> open text-to-speech (Kokoro-82M) speaks it. The caller's audio is itself made by Kokoro from a script below, so no real person's voice is used. Licences: Whisper MIT, Qwen2.5-1.5B-Instruct Apache-2.0, Kokoro-82M Apache-2.0. The turn runs twice in the same container: the first run includes the first-inference warm-up, the second is warm. Model loading and the container's own start are timed separately. All weights are baked into the image, so the GPU isn't billed while they download. This is one demo clip, not a client result: at this tiny volume the GPU cost is mostly start-up and idle time, so it says nothing about savings at scale. """ from __future__ import annotations import json import time from pathlib import Path import modal STT_MODEL = "openai/whisper-small.en" LLM_MODEL = "Qwen/Qwen2.5-1.5B-Instruct" CALLER_VOICE, AGENT_VOICE = "af_heart", "am_michael" CALLER = ( "Hi, I'm calling about my internet bill. I was charged twice this month, once on the " "third and again on the fifteenth, both for the same amount, fifty nine dollars and " "ninety nine cents. I've had the account for about four years and this has never " "happened before. I already checked my bank app, and both payments went through. I " "also tried the chat on your website yesterday, but it kept sending me back to the " "start. Can you tell me why I was charged twice, and how long it will take to get the " "extra payment refunded to my card? My account number is four four seven one two " "nine. Thanks." ) SYSTEM = ( "You are a phone support agent for an internet provider. Answer the caller in two or " "three short spoken sentences. No lists, no markdown. Don't invent account details." ) def _download() -> None: from huggingface_hub import snapshot_download from kokoro import KPipeline snapshot_download(STT_MODEL) snapshot_download(LLM_MODEL) KPipeline(lang_code="a") # fetches Kokoro-82M and its voices into the image image = ( modal.Image.debian_slim(python_version="3.11") .apt_install("espeak-ng", "ffmpeg") .pip_install( "torch==2.5.1", "transformers==4.46.3", "accelerate==1.1.1", "kokoro==0.9.4", "soundfile==0.12.1", "numpy<2", ) .run_function(_download) ) app = modal.App("illate-voice-demo", image=image) @app.function(gpu="T4", timeout=900) def run(called_at: float) -> dict: import io import subprocess import numpy as np import soundfile as sf import torch from kokoro import KPipeline from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline entered = time.time() out: dict = { "container_start_s": round(entered - called_at, 2), "gpu": subprocess.run( ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True, ).stdout.strip(), } def tts(text: str, voice: str) -> np.ndarray: parts = [ a.cpu().numpy() if hasattr(a, "cpu") else np.asarray(a) for _, _, a in tts_pipe(text, voice=voice) ] return np.concatenate(parts) t = time.time() tts_pipe = KPipeline(lang_code="a", device="cuda") out["load_tts_s"] = round(time.time() - t, 2) caller = tts(CALLER, CALLER_VOICE) # the input clip; not part of the measured turn buf = Path("/tmp/caller.wav") sf.write(buf, caller, 24000) t = time.time() stt = pipeline( "automatic-speech-recognition", STT_MODEL, torch_dtype=torch.float16, device="cuda" ) out["load_stt_s"] = round(time.time() - t, 2) t = time.time() tok = AutoTokenizer.from_pretrained(LLM_MODEL) llm = AutoModelForCausalLM.from_pretrained(LLM_MODEL, torch_dtype=torch.float16).to("cuda") out["load_llm_s"] = round(time.time() - t, 2) def turn() -> dict: r: dict = {} torch.cuda.synchronize() t0 = time.time() heard = stt(str(buf), return_timestamps=True)["text"].strip() torch.cuda.synchronize() r["stt_s"] = round(time.time() - t0, 3) msgs = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": heard}] ids = tok.apply_chat_template(msgs, add_generation_prompt=True, return_tensors="pt").to( "cuda" ) t0 = time.time() gen = llm.generate(ids, max_new_tokens=120, do_sample=False) torch.cuda.synchronize() r["llm_s"] = round(time.time() - t0, 3) reply = tok.decode(gen[0, ids.shape[1] :], skip_special_tokens=True).strip() r.update(llm_tokens_in=int(ids.shape[1]), llm_tokens_out=int(gen.shape[1] - ids.shape[1])) t0 = time.time() spoken = tts(reply, AGENT_VOICE) torch.cuda.synchronize() r["tts_s"] = round(time.time() - t0, 3) r["total_s"] = round(r["stt_s"] + r["llm_s"] + r["tts_s"], 3) r.update(transcript=heard, reply=reply, reply_chars=len(reply)) r["audio_out_s"] = round(len(spoken) / 24000, 2) r["_wav"] = spoken return r first, warm = turn(), turn() out["audio_in_s"] = round(len(caller) / 24000, 2) for k, r in (("first_run", first), ("warm_run", warm)): wav = r.pop("_wav") out[k] = r b = io.BytesIO() sf.write(b, wav, 24000, format="WAV") out["reply_wav"] = b.getvalue() b = io.BytesIO() sf.write(b, caller, 24000, format="WAV") out["caller_wav"] = b.getvalue() out["container_alive_s"] = round(time.time() - entered, 2) return out @app.local_entrypoint() def main(out: str): d = Path(out) d.mkdir(parents=True, exist_ok=True) called = time.time() res = run.remote(called) res["round_trip_s"] = round(time.time() - called, 2) res["called_at_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(called)) (d / "caller.wav").write_bytes(res.pop("caller_wav")) (d / "reply.wav").write_bytes(res.pop("reply_wav")) (d / "result.json").write_text(json.dumps(res, indent=2)) print(json.dumps(res, indent=2))