"""Live full-duplex demo: browser mic (16 kHz PCM over a WebSocket) -> VAD/turn-taking -> streaming ASR (partials) -> mix-matched LLM reply + validator -> Flash-DPO TTS streamed back sentence by sentence; barge-in stops playback at once. ASR_URL=http://127.0.0.1:8002/asr LLM_BASE=http://127.0.0.1:8001/v1 python server.py (port 7860)""" import asyncio, json, os, time, threading, re import numpy as np, httpx, uvicorn from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from turn import TurnDetector, FR, SR, FMS from respond import Responder from tts_engine import TTS from offline import BC, trim HERE = os.path.dirname(os.path.abspath(__file__)) ASR_URL = os.environ.get("ASR_URL", "http://127.0.0.1:8002/asr") app = FastAPI() TTS_ = TTS(os.environ.get("AGENT_VOICE", "anchor_male")) LOCK = threading.Lock() BCW = {k: trim(TTS_.synth(v, seed=3), TTS_.sr) for k, v in BC.items()} acli = httpx.AsyncClient(timeout=30) import mix as _M, logging; _M.estimate('warm up', asr=True); logging.getLogger('httpx').setLevel(logging.WARNING) def pcm16(w): return (np.clip(w, -1, 1) * 32767).astype(np.int16).tobytes() async def asr(y): r = await acli.post(ASR_URL, content=np.ascontiguousarray(y, np.float32).tobytes()) return r.json() def synth_locked(c, seed): with LOCK: t = time.perf_counter(); w = TTS_.synth(c, seed=seed); return w, time.perf_counter() - t @app.get("/") def index(): return HTMLResponse(open(f"{HERE}/static/index.html", encoding="utf-8").read()) @app.websocket("/ws") async def ws(w: WebSocket): await w.accept() loop = asyncio.get_running_loop() td = TurnDetector(eot_ms=int(os.environ.get("EOT_MS", 600))); resp = Responder(); hist = [] st = dict(play_until=0.0, task=None, gen=0, last_partial=0.0, partial_busy=False, lang="eng") pending = bytearray() from tts_engine import chunk_tagged async def partial(y): st["partial_busy"] = True try: r = await asr(y); await w.send_json({"type": "partial", "text": r["text"], "asr_ms": r["ms"]}) except Exception: pass st["partial_busy"] = False gate = {} # gen -> asyncio.Event set at end-of-turn (speculative responses start at spec_ms of silence) async def speak(audio, speech_end_wall, gen): """ASR -> mix/LLM/validator -> first TTS chunk are computed speculatively (from spec_ms of silence); nothing is sent until end-of-turn opens the gate, and the whole response is dropped if the user resumes (gen changes).""" r = await asr(audio); text = r["text"].strip() if len(text.split()) == 0 or st["gen"] != gen: return rep = await loop.run_in_executor(None, resp.reply, text, list(hist)) chunks = chunk_tagged(rep["reply"]) if st["gen"] != gen or not chunks: return wav0, ds0 = await loop.run_in_executor(None, synth_locked, chunks[0], 0) if gen in gate: await gate[gen].wait() if st["gen"] != gen: return st["committed"] = gen; st["lang"] = rep["target"]["matrix"] await w.send_json({"type": "final", "text": text, "asr_ms": r["ms"]}) await w.send_json({"type": "reply", "text": rep["reply"], "target": rep["target"], "reply_mix": rep["reply_mix"], "how": rep["how"], "llm_s": rep["t_llm"]}) played = [] for i, c in enumerate(chunks): if i == 0: wav, ds = wav0, ds0 else: wav, ds = await loop.run_in_executor(None, synth_locked, c, i) if st["gen"] != gen: break await w.send_bytes(pcm16(wav)) now = time.time(); st["play_until"] = max(st["play_until"], now) + len(wav) / TTS_.sr; played.append(c) if i == 0: await w.send_json({"type": "metrics", "latency_s": round(now - speech_end_wall, 3), "asr_ms": r["ms"], "llm_s": rep["t_llm"], "tts_first_s": round(ds, 3), "eot_ms": round(td.eot_f * FMS * 1000)}) hist.append((text, " ".join(played) + ("" if len(played) == len(chunks) else " [interrupted]"))) if st.get("committed") == gen: st["committed"] = None try: while True: msg = await w.receive() if msg.get("type") == "websocket.disconnect": break if msg.get("text"): d = json.loads(msg["text"]) if d.get("type") == "reset": hist.clear(); resp.style = None; await w.send_json({"type": "info", "text": "conversation reset"}) continue if not msg.get("bytes"): continue pending += msg["bytes"] while len(pending) >= FR * 2: fr = np.frombuffer(bytes(pending[:FR * 2]), np.int16).astype(np.float32) / 32768; del pending[:FR * 2] busy = time.time() < st["play_until"] or st.get("committed") is not None for name, info in td.feed(fr, agent_busy=busy): if name == "start": await w.send_json({"type": "vad", "speech": True}) elif name == "barge": st["gen"] += 1; st["play_until"] = 0.0; st["committed"] = None await w.send_json({"type": "stop", "reason": "barge-in"}) elif name == "backchannel": await w.send_json({"type": "backchannel"}); await w.send_bytes(pcm16(0.8 * BCW.get(st["lang"], BCW["eng"]))) elif name == "spec": st["gen"] += 1; g = st["gen"]; gate[g] = asyncio.Event() speech_end_wall = time.time() - (td.t - info["speech_end"]) st["task"] = asyncio.create_task(speak(info["audio"], speech_end_wall, g)) elif name == "resume": if st["gen"] in gate: gate[st["gen"]].set() # release the waiting speculative task; it sees the new gen and exits st["gen"] += 1 # user kept talking: drop the speculative response elif name == "eot": await w.send_json({"type": "vad", "speech": False, "speech_s": round(info["speech_s"], 2)}) if st["gen"] in gate: gate[st["gen"]].set() else: st["gen"] += 1 speech_end_wall = time.time() - (td.t - info["speech_end"]) st["task"] = asyncio.create_task(speak(info["audio"], speech_end_wall, st["gen"])) if td.in_speech and td.t - st["last_partial"] >= 0.64 and not st["partial_busy"]: st["last_partial"] = td.t; asyncio.create_task(partial(td.partial_audio())) except WebSocketDisconnect: pass if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), log_level="warning")