【2026 最新】Gander 語音 Agent 怎麼做？新手必學 5 個可插話、可取消的 Mock 實戰技巧

你正在請語音助理整理一份報告，它才說到一半，你突然補一句：「等等，改成只列三點。」真正困難的不是把 ASR（把語音轉成文字）、LLM（負責理解與推理的大模型）、TTS（把文字念成語音）接成一條線，而是讓 Gander 語音 Agent 類型的系統同時做到：先停止舊語音、接住新意圖、處理背景任務，最後還不能把晚到的舊答案念出來。

這篇專為第一次碰語音 Agent runtime（控制事件與狀態的執行層）的讀者寫。我們不先下載大模型，也不租三張 GPU；只用 Python 標準庫、假 ASR、假 TTS 與一個延遲 worker（由 asyncio Task 排程的背景 coroutine），做出能插話、取消、逾時、去重與攔截舊結果的最小版本。你會看到完整程式、事件 trace（依時間排列的流水帳）、四種故障注入，以及升級官方 Gander stack（完整系統組合）前該補的評測。

先說清楚範圍：這是受 Gander 公開架構啟發的教學 Mock，不是 Gander 模型或 runtime 的重製，也不產生真實音訊。本文核對的是 2026 年 9 月 11 日可見的官方 repo、模型卡與論文 v2；所有 benchmark 都視為作者報告，不能當成 AlphaLab 的獨立重現。

先說結論：Gander 語音 Agent 要先學狀態，不是先學模型

全雙工語音 Agent ＝ 快路徑（先接話）＋慢路徑（把事做完）＋雙重交付閘門（插話時立刻撤權，final 後再綁定新任務）。

• 插話不等於取消：停止正在播放的聲音，和取消背景 Brain 任務，是兩個不同動作。
• 取消不等於停止：worker 可能忽略取消、剛好完成，甚至已經呼叫外部工具；交付前仍要同時驗證 generation 與 delivery epoch。
• Mock 不驗模型好不好：它只讓事件順序與狀態不變量先變得可觀察、可測。

把 runtime 想成餐廳的出餐檯：前場先回「收到」，後場慢慢做菜；客人一開口改單，出餐檯先立刻關閉舊取餐號，等新單確認後才發新號碼。前者是 delivery epoch（交付許可版本），後者是 generation（任務世代）；兩者都吻合，餐點才可送出。

Gander 是什麼？小腦、Runtime、大腦各管一件事

Gander 官方 repo與技術報告 v2把完整系統拆成三層。Front Cerebellum（前端小腦）持續看原始音訊與影像，負責聽、說、打斷與工具動作；Agent Orchestration Runtime 綁定可信的完成語句、任務狀態、權限與交付；Back Brain（後端大腦）則非同步處理長時間推理、工具、檔案與工作流。若你已讀過 AI Agent Harness，可以把 Runtime 理解成更重視即時對話與任務世代的執行層。

官方的 streaming Thinker–Talker 使用一秒因果單元，Thinker 先判斷互動狀態，再讓分離的 Talker 串流合成語音。一秒是模型切分時間的架構值，不是端到端延遲保證。固定版本模型卡列出的核心互動控制是 listen、speak、interrupt，工具呼叫則走結構化任務動作；不要把它簡化成「四個等價按鈕」。

Gander 語音 Agent 的 interrupt，為什麼不是 cancel？

在目前公開實作裡，模型的 interrupt 與瀏覽器的播放控制處理「先別再說」；背景任務的取消則由 task_resolve(cancel) 走另一條生命週期。官方 runtime 也以 generation 丟棄晚到的舊音訊或 worker event。白話說：嘴巴停下、廚房停工、舊餐不出門，是三道不同閘門。

拆成 5 個零件：Gander 語音 Agent Mock 的最小設計

① 「可信 final」：partial 只能預覽，final 才能派工

痛點：串流 ASR 會重送 partial（尚未確認的轉錄假設）、改寫文字，若每次 partial 都啟動 Brain，同一個請求可能執行多次。解法：以 utterance_id 去重，只讓 runtime 已信任的第一次 final（該段不再修訂的轉錄結果，不保證文法上是完整句）配到新的 turn 與 generation（遞增的任務世代編號）。本 Mock 把 on_final() 當成信任邊界，沒有自行驗證 ASR；連送兩次相同 final，trace 只能出現一次 turn_finalized，同 ID 不同文字則記為 final_conflict。

② 「快路徑」：先回應，不等背景工作完成

痛點：長任務可能跑數十秒，若 TTS 等 worker，對話像斷線。解法：final 到達後立刻排一段簡短 acknowledgement，例如「收到，我先處理」，並把第一個假 TTS chunk 的時間記成 first_audio。它只測 runtime 排程，不代表真實模型或網路延遲。

③ 「慢路徑」：worker 有 timeout，也有明確終態

痛點：背景工作若永遠卡住，前端只剩一句「正在處理」。解法：用 asyncio.wait() 設回應期限；期限到就先撤銷交付、要求本機 task 取消並另行收尾。這不是遠端工作已停止的證明，更不會回滾副作用。只有 epoch 與 generation 都仍有效的任務能播逾時訊息；過期 timeout 只記 stale_timeout_suppressed。

④ 「雙層取消」：先停 TTS，再通知 worker

痛點：使用者一開口，Agent 的音訊 queue 還在播放；舊結果也可能在下一個 final 抵達前完成。解法：speech_start 先令 delivery_epoch += 1 並取消 speaker，當下就讓舊輸出失去交付權；新 final 到達後，再對舊 run 設定合作式 asyncio.Event。這個 Event 是本文自訂協定，不是 Task.cancel()；Python 官方文件提醒，Task.cancel() 也只是取消請求，coroutine 仍可能攔截或延後處理。

⑤ 「世代閘門」：取消失敗，舊結果仍不能交付

痛點：取消與完成常在同一個競態窗口發生；即使外部 API 沒提供可用的中止介面，舊答案也不能回來。解法：每次 final 令 generation += 1；worker、timeout 與每個 TTS chunk 交付前，都要確認自己的 generation 和 speech-start 時保存的 delivery epoch 仍是目前值。任一不符就拒收。這也堵住「使用者已開口，但新 final 還沒到」的空窗；可再搭配 Agent Runtime Controls 理解撤權設計。

動手做：可插話、可取消的 Python Mock

你只需要 Python 3.9 以上，不需安裝套件。先打開終端機（macOS 的 Terminal、Linux shell 或 Windows 的 WSL），輸入 python3 --version；看到 3.9 或更高版本後，建立一個空資料夾，把下面內容存成 mock_voice_agent.py。為了讓機制看得見，假 TTS 每 0.01 秒吐一段字，假 Brain 則用 sleep 模擬長任務；這些時間都是測試 fixture（固定的測試條件），不是效能成績。

import asyncio
from dataclasses import dataclass
from time import monotonic


class WorkerCancelled(Exception):
    """Cooperative worker stop, separate from asyncio task cancellation."""


@dataclass
class Run:
    generation: int
    delivery_epoch: int
    cancel: asyncio.Event
    task: asyncio.Task


class MockVoiceAgent:
    def __init__(self):
        self.generation = 0
        self.delivery_epoch = 0
        self.seen_finals = {}
        self.runs = []
        self.speaker = None
        self.cleanup_tasks = []
        self.trace = []
        self.first_audio_ms = {}
        self.final_received_at = {}

    def log(self, event, **fields):
        row = {"event": event, **fields}
        self.trace.append(row)
        print(row)

    def is_current(self, generation, delivery_epoch):
        return (
            generation == self.generation
            and delivery_epoch == self.delivery_epoch
        )

    async def on_partial(self, utterance_id, text):
        self.log("partial_buffered", utterance=utterance_id, text=text)

    async def on_speech_start(self):
        # Revoke delivery immediately; do not wait for ASR finalization.
        self.delivery_epoch += 1
        self.log("delivery_revoked", delivery_epoch=self.delivery_epoch)
        if self.speaker and not self.speaker.done():
            self.speaker.cancel()
            self.log("tts_cancel_requested", reason="barge_in")

    async def on_final(self, utterance_id, text, *, delay=0.20,
                       timeout=0.50, honor_cancel=True, fail=False):
        if utterance_id in self.seen_finals:
            old_text = self.seen_finals[utterance_id]
            event = (
                "duplicate_final_ignored"
                if old_text == text
                else "final_conflict"
            )
            self.log(event, utterance=utterance_id)
            return

        # This method is the trusted ASR-final boundary in the mock.
        self.seen_finals[utterance_id] = text
        self.generation += 1
        generation = self.generation
        delivery_epoch = self.delivery_epoch
        self.final_received_at[generation] = monotonic()
        self.log(
            "turn_finalized",
            generation=generation,
            delivery_epoch=delivery_epoch,
            text=text,
        )

        for run in self.runs:
            if not run.task.done():
                run.cancel.set()
                self.log(
                    "worker_cancel_requested",
                    old=run.generation,
                    new=generation,
                )

        self._speak(
            "收到，我先處理。", generation, delivery_epoch, "fast_ack"
        )
        cancel = asyncio.Event()
        task = asyncio.create_task(
            self._finish(
                text,
                generation,
                delivery_epoch,
                cancel,
                delay,
                timeout,
                honor_cancel,
                fail,
            )
        )
        self.runs.append(Run(generation, delivery_epoch, cancel, task))

    def _speak(self, text, generation, delivery_epoch, purpose):
        if not self.is_current(generation, delivery_epoch):
            self.log(
                "tts_stale_suppressed",
                generation=generation,
                purpose=purpose,
            )
            return
        if self.speaker and not self.speaker.done():
            self.speaker.cancel()
        self.speaker = asyncio.create_task(
            self._fake_tts(text, generation, delivery_epoch, purpose)
        )

    async def _fake_tts(self, text, generation, delivery_epoch, purpose):
        try:
            for chunk in text.split("，"):
                await asyncio.sleep(0.01)
                if not self.is_current(generation, delivery_epoch):
                    self.log(
                        "tts_stale_suppressed",
                        generation=generation,
                        purpose=purpose,
                    )
                    return
                if generation not in self.first_audio_ms:
                    latency_ms = (
                        monotonic() - self.final_received_at[generation]
                    ) * 1000
                    self.first_audio_ms[generation] = latency_ms
                    self.log(
                        "first_audio",
                        generation=generation,
                        latency_ms=round(latency_ms, 2),
                    )
                self.log(
                    "tts_chunk",
                    generation=generation,
                    purpose=purpose,
                    chunk=chunk,
                )
        except asyncio.CancelledError:
            self.log(
                "tts_interrupted", generation=generation, purpose=purpose
            )

    async def _fake_brain(self, text, cancel, delay, honor_cancel, fail):
        elapsed = 0.0
        while elapsed < delay:
            await asyncio.sleep(0.01)
            elapsed += 0.01
            if honor_cancel and cancel.is_set():
                raise WorkerCancelled
        if fail:
            raise RuntimeError("injected worker failure")
        return f"完成：{text}"

    async def _drain(self, task, generation):
        try:
            await task
        except (asyncio.CancelledError, WorkerCancelled):
            self.log("worker_cleanup_finished", generation=generation)
        except Exception as error:
            self.log(
                "worker_cleanup_failed",
                generation=generation,
                error=type(error).__name__,
            )

    async def _finish(self, text, generation, delivery_epoch, cancel,
                      delay, timeout, honor_cancel, fail):
        brain_task = asyncio.create_task(
            self._fake_brain(text, cancel, delay, honor_cancel, fail)
        )
        try:
            done, _ = await asyncio.wait({brain_task}, timeout=timeout)
        except asyncio.CancelledError:
            brain_task.cancel()
            await self._drain(brain_task, generation)
            self.log("supervisor_cancelled", generation=generation)
            raise

        if not done:
            # This is a response deadline, not proof that remote work stopped.
            cancel.set()
            brain_task.cancel()
            cleanup = asyncio.create_task(
                self._drain(brain_task, generation)
            )
            self.cleanup_tasks.append(cleanup)
            if self.is_current(generation, delivery_epoch):
                self.log("worker_timeout", generation=generation)
                self._speak(
                    "背景任務逾時。",
                    generation,
                    delivery_epoch,
                    "timeout",
                )
            else:
                self.log(
                    "stale_timeout_suppressed", generation=generation
                )
            return

        try:
            result = brain_task.result()
        except WorkerCancelled:
            self.log("worker_cancelled", generation=generation)
            return
        except asyncio.CancelledError:
            self.log("worker_task_cancelled", generation=generation)
            raise
        except Exception as error:
            self.log(
                "worker_failed",
                generation=generation,
                error=type(error).__name__,
            )
            if self.is_current(generation, delivery_epoch):
                self._speak(
                    "背景任務失敗。",
                    generation,
                    delivery_epoch,
                    "failure",
                )
            return

        if not self.is_current(generation, delivery_epoch):
            self.log("stale_result_suppressed", generation=generation)
            return

        self.log("result_accepted", generation=generation, result=result)
        self._speak(
            result, generation, delivery_epoch, "worker_result"
        )

    async def settle(self):
        await asyncio.gather(*(run.task for run in self.runs))
        if self.cleanup_tasks:
            await asyncio.gather(*self.cleanup_tasks)
        if self.speaker:
            await self.speaker


async def demo():
    agent = MockVoiceAgent()

    # Revised partials never start work.
    await agent.on_partial("u0", "幫我整")
    await agent.on_partial("u0", "幫我整理")
    assert agent.generation == 0 and not agent.runs

    # An old result finishes inside the speech-start -> delayed-final gap.
    await agent.on_final(
        "u1", "幫我整理這份報告", delay=0.04, honor_cancel=False
    )
    await asyncio.sleep(0.015)
    await agent.on_speech_start()
    await asyncio.sleep(0.05)

    # Instant result replaces the ack; dedupe and conflict stay terminal.
    await agent.on_final("u2", "改成只列三點", delay=0.0)
    await agent.on_final("u2", "改成只列三點")
    await agent.on_final("u2", "改成只列五點")
    await asyncio.sleep(0.04)

    # A superseded timeout must not speak over the new turn.
    await agent.on_final(
        "u3", "慢速舊任務", delay=0.20, timeout=0.04,
        honor_cancel=False,
    )
    await asyncio.sleep(0.015)
    await agent.on_speech_start()
    await asyncio.sleep(0.005)
    await agent.on_final("u4", "新的短任務", delay=0.02)
    await asyncio.sleep(0.06)

    # Current timeout and worker failure both reach explicit terminal states.
    await agent.on_final(
        "u5", "會逾時的任務", delay=0.20, timeout=0.04
    )
    await asyncio.sleep(0.06)
    await agent.on_final("u6", "會失敗的任務", delay=0.01, fail=True)
    await agent.settle()

    by_event = {}
    for row in agent.trace:
        by_event.setdefault(row["event"], []).append(row)

    assert [row["generation"] for row in by_event["turn_finalized"]] == [
        1, 2, 3, 4, 5, 6
    ]
    assert [row["generation"] for row in by_event["result_accepted"]] == [
        2, 4
    ]
    assert by_event["stale_result_suppressed"][0]["generation"] == 1
    assert by_event["stale_timeout_suppressed"][0]["generation"] == 3
    assert by_event["worker_timeout"][0]["generation"] == 5
    assert by_event["worker_failed"][0]["generation"] == 6
    assert len(by_event["duplicate_final_ignored"]) == 1
    assert len(by_event["final_conflict"]) == 1
    assert 2 in agent.first_audio_ms
    assert any(
        row.get("generation") == 2
        and row.get("purpose") == "worker_result"
        for row in by_event["tts_chunk"]
    )
    assert any(
        row.get("old") == 3 for row in by_event["worker_cancel_requested"]
    )
    print("PASS: epoch fence, generation fence, dedupe, timeout, failure")


asyncio.run(demo())

執行：

python3 -m py_compile mock_voice_agent.py
PYTHONASYNCIODEBUG=1 python3 mock_voice_agent.py

不要只看最後的 PASS。trace 會驗證：generation 1 在 speech-start 到 delayed-final 的空窗完成，仍被 stale_result_suppressed；generation 2 的即時結果取代 acknowledgement，卻仍正確記到 first_audio；generation 3 的過期 timeout 不搶麥克風；generation 5 與 6 分別進入 timeout 和 failure 終態。這條因果鏈比單看「沒有 exception」更有用。若要把 loop、tool 與觀測性再拆深一層，可接著做 30 行 Agent Harness 與 Agent Observability。

故障注入：別只測正常對話

一個語音 Agent demo 最容易只演「使用者說完、Agent 回完」；真正會破壞狀態的，通常是重送、半句、自我修正、逾時和取消競態。每個案例先寫不變量，再注入失敗：

1. 重複 final：同 ID、同文字只忽略一次；同 ID、不同文字必須記錄 conflict，不能默默吞掉。
2. 修訂 partial：畫面可以更新，但 generation 與 Brain task 數量都不變。
3. worker timeout：只有雙重閘門仍有效的任務收到逾時回應；舊任務只留 trace。
4. 取消競態：故意讓舊 worker 忽略合作式 cancel，並讓它在 delayed-final 空窗完成，結果仍不得交付。

首回應時間也要先固定定義。本文 Mock 使用「收到 ASR final → 第一個假 TTS chunk」；真實產品還應分開量 speech_onset → interruption_detected → playback_cleared → last_audible_sample。想建立更完整的語音測試集，可沿用 Voice Agent Eval 的方法，把噪音、長停頓、backchannel、回音、第三人說話與 mid-sentence correction 都做成可重播案例。

實務上至少有兩只時鐘：對話時鐘關心「使用者開始說話後，最後一個可聽樣本何時消失」；任務時鐘關心「取消請求後，worker 何時進入 terminal state」。前者慢會讓人覺得 Agent 搶話，後者失控會浪費算力或留下副作用。兩者不能合成一個平均延遲，否則你不知道問題在播放器、模型、網路還是工具。

何時該把 Gander 語音 Agent Mock 升級成真模型？

官方 release profile目前要求 Linux、Conda、相容 CUDA 12.4 的 NVIDIA 環境與三張實體 GPU：GPU 0 跑 Thinker、GPU 1 跑 Talker、GPU 2 跑 managed ASR；還要下載 MiniCPM-o 4.5、Gander Thinker／Talker、faster-whisper-large-v3。在這份固定 commit 的 requirements 與 serve config 中，硬體欄位只列三張 physical GPU，沒有列 GPU 型號或最低 VRAM；因此本文不從模型檔案大小自行推成硬體保證。

Back Brain 的 provider 介面設計為可替換，但目前 repo 內附的Codex provider與 quick start 使用已登入的 codex app-server --stdio。範例的 model: null 表示 Gander 不在 thread/start 請求覆寫模型，實際模型由那套 Codex 安裝的設定與預設值決定；論文 benchmark 指定的模型只屬那次評估配置，不應搬成日常部署預設。需要工具呼叫時，也可先讀 Code-Implemented Tool Calls，把參數驗證與副作用邊界補齊。

升級的判斷不是「Mock 的 sleep 很快」，而是你開始需要回答 Mock 無法回答的問題：真實聲學下會不會誤判發話權？Talker 要多久停聲？視覺線索是否真的改善指代？三張 GPU 能否滿足你的併發？上圖的資料取自作者論文的 Full-Duplex-Bench v3 結果；多項成績啟用了不同元件與 judge，不能跨表直接比較。最值得帶走的是評測缺口，而不是「最好」的標語。可用 AI Evals 的資料集、grader、regression 三層，把自己的語言、麥克風與工具環境納入。

上線前最常踩的 5 個坑

1. 把 VAD 當語意終點：VAD（語音活動偵測）只知道「像不像有人說話」，不知道「嗯……」是停頓還是講完。正式版要另做 endpointing（判斷一句話何時真正結束）與 turn-taking（誰取得發話權）。
2. 以為取消會回滾副作用：停止 coroutine（可暫停與恢復的非同步工作）不會自動撤回已寄出的信或已送出的訂單。工具層要有 idempotency key（同一請求重送仍只生效一次的識別碼）、狀態查詢與可補償動作。
3. 只清模型、不清播放器：模型已停止，audio queue（待播放音訊佇列）仍可能繼續播。要同時截斷輸出 buffer，並把歷史對齊到使用者真正聽到的位置。
4. 只在記憶體保存 generation：重新連線或程序重啟後，舊事件可能重送。正式 runtime 應有持久 ledger（事件帳本）、單調 sequence（只增加的事件序號）、去重與 replay 規則。
5. 把作者 demo 當 SLA：一秒 unit、模型控制 token 與公開影片能說明設計，不能代替你的 p50／p95 停聲時間、誤打斷率、stale delivery rate 與權限拒絕測試。

另外，本文這份極小 Mock刻意省略 echo cancellation、backpressure、斷線重連、亂序事件、權限詢問、安全審查、成本與可觀測性；seen_finals、runs、trace 與時間戳也都沒有容量上限或過期機制。這些描述的是教學碼的缺口，不是對官方 Gander 功能的斷言；正式版本必須設定保留期限、容量界線與清理規則。

Gander 語音 Agent 常見問題 FAQ

1. 這份 Mock 是 Gander 的輕量版嗎？

不是。它沒有匯入 Gander、MiniCPM-o、Talker、ASR 或官方協定，只借用「即時前端＋非同步 Brain＋可撤銷交付權」的高階觀念來教 runtime semantics。

2. 不用 GPU 也能做全雙工語音 Agent 嗎？

能學控制流程，不能驗證真實語音品質。本例可測去重、取消、逾時與舊結果拒收；ASR 準確度、TTS 停聲、echo 與模型判斷仍要真 stack。

3. 使用者插話時，直接 cancel worker 就夠了嗎？

不夠。speech start 要先停音訊並更新 delivery epoch，新 final 再更新 generation、提出 worker 取消；所有輸出都通過雙重閘門，外部工具副作用則另做查詢與補償。

4. partial ASR 可以讓回應更快嗎？

可以預覽，不宜直接授權有副作用的任務。partial 會修訂與重送；若要 speculative work，至少隔離成可丟棄、唯讀的分支。

5. Gander 的一秒 unit 代表一秒內一定回話嗎？

不代表。那是串流模型的因果切分；端到端時間還包含擷取、排程、推理、合成、網路與播放。

6. 官方的 8% interrupt 是打斷成功率嗎？

不是。在該 benchmark 定義裡，它代表 Agent 在使用者結束前就開始說話的比例，也就是 premature start；Gander 論文表 3 並未列出使用者開口後多久停聲。

7. 現在就值得部署官方三 GPU 版本嗎？

只有在你需要驗證真實音訊／影像與模型互動時才值得。若連 stale result、timeout、duplicate final 都還沒測，先用 Mock 修好 runtime，會更快定位問題。

8. 下一步最該加哪個 production 功能？

先加持久事件 ledger。它讓 generation、sequence、取消與交付在重連後仍可追，接著再補真實 VAD／endpointing 與工具 idempotency。

給新手的 5 個重點

1. 快回應和慢任務應分開排程。
2. speech interruption、task cancel、stale-result suppression 是三件事。
3. speech start 立即撤權；可信 final 才建立新任務世代。
4. 先寫不變量，再注入 duplicate、conflict、partial、timeout、failure 與 race。
5. Mock 通過後，再把同一組 trace 指標搬到真模型與真網路。