app.py

"""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"\n<sub>Chinese speech: the questions use the Chinese wording the model was trained with; answers shown in English · 中文。{len(qs)} 个判断，GPU 上 {ms:.0f} 毫秒。</sub>"
    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**

<sub>⏱️ 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 可获得更多免费次数。</sub>
"""

FOOT = """
<sub>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. 音频只用于本次判断，不保存。</sub>
"""

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()