File size: 7,023 Bytes
ac573f1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """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")
|