| """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 = {} |
|
|
| 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() |
| st["gen"] += 1 |
| 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") |
|
|
