"""DuplexJev on ZeroGPU: ten voice-agent decisions from raw speech in one forward pass.""" import os import time import gradio as gr import numpy as np import spaces from duplexjev import Question, default_table from duplexjev.table import LABELS MODEL_ID = os.environ.get("DUPLEXJEV_MODEL", "adventists-ai/DuplexJev-32B-Para") N_PERM = int(os.environ.get("DUPLEXJEV_NPERM", "1")) # 2 = average over option orders (same answers in our tests, 1.5x slower) GPU_SIZE = os.environ.get("DUPLEXJEV_GPU", "xlarge") # 32B in bf16 needs the full 96 GB card MOCK = os.environ.get("DUPLEXJEV_MOCK") == "1" # local UI test without a GPU if MOCK: class _Mock: def decide(self, audio, qs): rng = np.random.default_rng(len(qs)) out = {} for q in qs: p = rng.dirichlet(np.ones(len(q.options))) out[q.id] = {"answer": q.options[int(p.argmax())], "confidence": float(p.max()), "probs": dict(zip(q.options, map(float, p)))} return out DECIDER = _Mock() else: from duplexjev import Decider DECIDER = Decider.from_pretrained(MODEL_ID, device="cuda") # placed on cuda at import time, as ZeroGPU expects SR = 16000 # Chinese option / label -> English gloss, so the page reads in English whatever the speech language from duplexjev.table import _EN, _ZH # same ids and option order, except gender ZH2EN = {} for (_i, _q, _eo), (_j, _zq, _zo) in zip(_EN, _ZH): if _i == "gender": ZH2EN.update({"男性": "male", "女性": "female"}) else: ZH2EN.update(dict(zip(_zo, _eo))) ZH2EN.update({"开口回应": "reply", "不出声，继续等（这不是对助手说的）": "stay silent (not addressed to the assistant)"}) LABEL2EN = {LABELS["zh"][k]: LABELS["en"][k] for k in LABELS["zh"]} LABEL2EN["是不是对助手说"] = "addressed to the assistant?" def _en(x, table): return f"{table[x]} · {x}" if x in table else x def _to16k(audio): """Gradio gives (sr, int16 array); resample with a polyphase filter (plain interpolation aliases and makes speech sound harsher, which skews emotion and speaking-style answers).""" from math import gcd from scipy.signal import resample_poly sr, x = audio x = np.asarray(x) if x.dtype.kind == "i": x = x.astype(np.float32) / float(np.iinfo(x.dtype).max + 1) x = x.astype(np.float32) if x.ndim == 2: x = x.mean(axis=1) if sr != SR: g = gcd(int(sr), SR) x = resample_poly(x, SR // g, int(sr) // g).astype(np.float32) return x[: SR * 20] # at most 20 s def table_text(lang): rows = [] for q in default_table(lang): rows.append(f"{LABELS[lang].get(q.id, q.id)} | {q.text} | " + "; ".join(q.options)) if lang == "zh": rows.append("是不是对助手说 | 助手已经讲完，正在等用户。听到这段声音，助手该怎么做？ | 开口回应; 不出声，继续等（这不是对助手说的）") return "\n".join(rows) def parse_table(text, lang): qs = [] for i, line in enumerate(l for l in text.splitlines() if l.strip()): parts = [p.strip() for p in line.split("|")] if len(parts) < 3: raise gr.Error(f"Line {i + 1}: use name | question | option 1; option 2; …") opts = [o.strip() for o in parts[2].split(";") if o.strip()] if len(opts) < 2: raise gr.Error(f"Line {i + 1}: give at least two options, separated by ';'") qs.append(Question(parts[0] or f"q{i + 1}", parts[1], opts, lang=lang)) if not qs: raise gr.Error("Add at least one question.") if len(qs) > 16: raise gr.Error("At most 16 questions per clip in this demo.") return qs @spaces.GPU(duration=6, size=GPU_SIZE) # a call takes ~0.3 s; a short reservation leaves visitors more free runs def _run(x, qs, n_perm): t = time.perf_counter() res = DECIDER.decide(x, qs) if MOCK else DECIDER.decide(x, qs, n_perm=n_perm) return res, (time.perf_counter() - t) * 1000 def decide(audio, lang, text, n_perm=None): if audio is None: raise gr.Error("Record or upload a clip first (or pick an example).") x = _to16k(audio) if len(x) < SR * 0.2: raise gr.Error("The clip is too short.") qs = parse_table(text, lang) res, ms = _run(x, qs, int(n_perm or N_PERM)) rows = [] for q in qs: r = res[q.id] tr = (lambda o: ZH2EN.get(o, o)) if lang == "zh" else (lambda o: o) others = " · ".join(f"{tr(o)} {p:.0%}" for o, p in sorted(r["probs"].items(), key=lambda kv: -kv[1]) if o != r["answer"]) name = _en(q.id, LABEL2EN) if lang == "zh" else q.id ans = _en(r["answer"], ZH2EN) if lang == "zh" else r["answer"] rows.append([name, ans, f"{r['confidence']:.0%}", others]) head = f"### {len(qs)} decisions · 1 forward pass · 0 decode steps · **{ms:.0f} ms** on the GPU" if lang == "zh": head += f"\nChinese speech: the questions use the Chinese wording the model was trained with; answers shown in English · 中文。{len(qs)} 个判断，GPU 上 {ms:.0f} 毫秒。" return head, rows def set_lang(lang): return table_text(lang) EX_EN = [ ["examples/en_barge_in.wav", "en"], ["examples/en_unfinished.wav", "en"], ["examples/en_backchannel.wav", "en"], ["examples/laughter.wav", "en"], ] EX_ZH = [ ["examples/zh_wait.wav", "zh"], ["examples/zh_side_talk.wav", "zh"], ["examples/zh_happy.wav", "zh"], ["examples/zh_complete.wav", "zh"], ] INTRO = """ # 🎧 DuplexJev — stop transcribing to decide **The decision layer for voice agents.** Give it a clip and the questions your agent needs answered — *has the user finished? should I stop talking? are they angry?* Every answer comes from **one forward pass**: no ASR, no decoding. 10 decisions in 92 ms, 21× faster than ASR → LLM ([paper](https://arxiv.org/abs/2610.02638)). ⭐ [GitHub](https://github.com/adventists-ai/duplexjev) · 🤗 [Models](https://huggingface.co/adventists-ai) · 🔌 Pipecat: `pip install pipecat-duplexjev` · LiveKit: `pip install livekit-plugins-duplexjev` · 👁️ **New:** voice + picture together → [DuplexJev-Vision demo](https://huggingface.co/spaces/BitKnight/DuplexJev-Vision) · 🏢 Commercial licence & hosted API: **jiejin@adventists.ai** ⏱️ The first run after the Space has been idle takes a few seconds while Hugging Face attaches a GPU; after that each run takes about 0.3 s on the GPU. Free GPU time per visitor is limited by Hugging Face — [log in](https://huggingface.co/login) for more runs. 首次运行需要几秒分配 GPU，之后每次约 0.3 秒；登录 Hugging Face 可获得更多免费次数。 """ FOOT = """ Model: DuplexJev-32B-Para (weights CC BY-NC 4.0), one full ZeroGPU card. Audio is processed for this request only and not stored. Edit the questions freely: one per line, `name | question | option 1; option 2; …`. GPU time shown excludes the queue and network. 音频只用于本次判断，不保存。 """ with gr.Blocks(title="DuplexJev — speech to decisions", theme=gr.themes.Soft(primary_hue="emerald")) as demo: gr.Markdown(INTRO) with gr.Row(): with gr.Column(scale=5): audio = gr.Audio(sources=["microphone", "upload"], type="numpy", label="Speech clip (≤ 20 s)") lang = gr.Radio([("English speech", "en"), ("Chinese speech · 中文语音", "zh")], value="en", label="Speech language (picks the question wording) / 语音语言") with gr.Accordion("Questions (editable) / 判断题（可编辑）", open=False): table = gr.Textbox(value=table_text("en"), lines=11, show_label=False) btn = gr.Button("Decide ⚡", variant="primary") nperm = gr.Number(value=N_PERM, visible=False) gr.Examples(EX_EN, inputs=[audio, lang], label="🇬🇧 English speech examples", example_labels=["angry barge-in: “Wait, wait — stop…”", "stops mid-request: “…and, um —”", "“mm-hmm, yeah” backchannel", "laughter"]) gr.Examples(EX_ZH, inputs=[audio, lang], label="🇨🇳 Chinese speech examples · 中文语音示例 (answers in English · 中文)", example_labels=["“Wait, let me think” · 等一下我想想", "back-seat chat, not to the assistant · 后排闲聊", "happy thanks · 开心道谢", "navigation request · 导航请求"]) with gr.Column(scale=6): head = gr.Markdown("### Results appear here") out = gr.Dataframe(headers=["decision", "answer", "confidence", "other options"], wrap=True, interactive=False, max_height=1000, column_widths=["18%", "34%", "13%", "35%"]) gr.Markdown(FOOT) lang.change(set_lang, lang, table) btn.click(decide, [audio, lang, table, nperm], [head, out], api_name="decide") if __name__ == "__main__": demo.queue(max_size=40).launch()