这套东西做出来是什么样
先描述成品，方便你判断要不要往下做。
打开程序，对着麦克风说话，同时把摄像头对着桌面上的设备面板。你说："这个面板现在的读数我看不懂，帮我判断一下要不要停机。"助手没有马上接话，而是安静了大约两秒——这两秒里它在做多步推理，还在看你摄像头画面里的数字。然后它开口，用语音一步步告诉你判断依据，同时屏幕上滚动着同步字幕。说到一半你打断它："等等，如果按你说的做要多久？"音频立刻停住，它听到了你的插话，重新思考，再回答。
这就是"先深度思考、再实时应答"：听和说是实时的，想是慢的，两者被编排在同一条会话里。
核心做法有两块：
1. 实时通道：双向流式会话，负责麦克风采集、摄像头帧、语音合成回放、打断检测。
2. 思考通道：扩展思考（Extended Thinking）负责慢推理。它可以在会话内部直接开启，也可以通过工具调用接一个"慢一点但更聪明"的模型。
两种接法都会讲，你可以按当前模型能力选。
前置条件清单
在动手前把这些准备好：
- 一个支持实时双向流式接口的模型访问权限。本文统一用 Live API 的说法，模型 ID、可用区域、是否支持思考参数，以官方文档当前版本为准。
- API Key 或等效凭证，并且有权限调用实时接口。
- 较新的 Python 3（能正常使用
asyncio、websockets、sounddevice），以及pip可用的网络环境。 - 麦克风 + 扬声器，强烈建议用耳机。外放会引发回环，助手会听见自己说话，然后反复自我打断。
- 可选摄像头，用来看画面。
- 一个后端进程。Key 不要放到浏览器里，绕一层自己的服务端做 WebSocket 代理。
安装依赖：
```bash
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install websockets sounddevice numpy python-dotenv
```
步骤 1：搭一个能跑通的最小骨架
目录结构先定下来，后面每一步都往这里加文件：
```text
realtime-assistant/
├── .env
├── main.py # 主循环：连接、收发、编排
├── audio_io.py # 麦克风采集与扬声器播放
├── thinking.py # 思考预算路由 + 深度思考调用
└── tools.py # 工具声明与执行
```
.env 里放三个东西，注意加进 .gitignore：
```bash
LIVE_WS_URL=wss://<按官方文档填写的实时接口地址>
LIVE_MODEL=<当前可用的实时模型 ID>
LIVE_API_KEY=<你的 key>
THINK_MODEL=<当前可用的、支持扩展思考的模型 ID>
```
接口地址一定从官方文档复制，不同版本路径会变，硬编码容易踩坑。
步骤 2：发起会话，完成握手
实时接口是长连接 WebSocket。连上以后第一件事是发一条 setup，把模型、输出模态、系统指令、工具、上下文管理策略一次性声明好。
```python
main.py
import asyncio, json, os, base64
import websockets
from dotenv import load_dotenv
load_dotenv()
WS_URL = os.environ["LIVE_WS_URL"]
MODEL = os.environ["LIVE_MODEL"]
KEY = os.environ["LIVE_API_KEY"]
SYSTEM_PROMPT = """你是一个现场协助助手。
规则：
1. 简单问题直接答，不要拖。
2. 需要多步推理、对比、计算的问题，先调用 deep_think 工具再回答。
3. 回答用口语，短句，一次说一到两个要点。
4. 用户插话时立刻停止，先听完整句。"""
SETUP = {
"setup": {
"model": f"models/{MODEL}",
"generationConfig": {
"responseModalities": ["AUDIO"],
"speechConfig": {
"voiceConfig": {
"prebuiltVoiceConfig": {"voiceName": "Aoede"}
}
},
关键：把思考预算放进 generationConfig，字段位置以官方文档为准
"thinkingConfig": {
"thinkingBudget": 2048,
"includeThoughts": True,
},
},
"systemInstruction": {"parts": [{"text": SYSTEM_PROMPT}]},
"contextWindowCompression": {"slidingWindow": {}},
"sessionResumption": {},
}
}
```
连接并确认握手：
```python
async def connect():
url = f"{WS_URL}?key={KEY}"
ws = await websockets.connect(url, max_size=None, ping_interval=20)
await ws.send(json.dumps(SETUP))
raw = await ws.recv()
print("握手返回：", raw[:400])
return ws
```
如果返回里出现 setup 完成之类的字段，说明配置被接受了。如果模型不支持某个字段，接口通常会直接报错——这时先把 thinkingConfig 删掉，跑通基础会话，再进入下一步。
步骤 3：开启扩展思考的两种接法
接法 A：会话内直接开思考
就是上一步 generationConfig.thinkingConfig 的写法。适合"用户问题不复杂，但需要一点推理"的场景。特点是延迟增加可控，一般在几百毫秒到数秒之间。
要点：
thinkingBudget是思考 token 的上限，具体取值语义（关闭、自动、固定值）以官方文档为准。includeThoughts如果被支持，返回内容里可能带思考摘要片段，可以拿去在界面上显示"正在思考"。- 并非所有实时模型都接受这个字段。不确定就先试，报错就切接法 B。
接法 B：双通道——实时会话负责听和说，深度思考走工具调用
这是更稳的架构，也是"先深思考再实时应答"最通用的实现：
- 快通道（实时会话）：低延迟，负责听、说、打断。
- 慢通道（支持扩展思考的模型）：收到工具调用后做多步推理，把结论回注给会话。
声明工具：
```python
tools.py
DEEP_THINK_TOOL = {
"functionDeclarations": [
{
"name": "deep_think",
"description": "把需要多步推理、对比计算或权衡取舍的问题交给深度思考模型。"
"简单寒暄和单步事实问答不要调用。",
"parameters": {
"type": "OBJECT",
"properties": {
"question": {"type": "STRING", "description": "需要深入思考的问题原文"},
"budget_hint": {
"type": "STRING",
"enum": ["low", "high"],
"description": "问题难度提示",
},
},
"required": ["question"],
},
}
]
}
```
把这个结构塞进 SETUP["setup"]["tools"]。执行工具时用一个独立的客户端去调思考模型，不要让主循环阻塞在等待上：
```python
tools.py
import asyncio, os
async def run_deep_think(question: str, budget: int) -> str:
伪代码：换成你所用 SDK 的同步或异步调用
client = make_think_client(os.environ["THINK_MODEL"])
resp = await asyncio.to_thread(
client.generate,
contents=question,
thinking_budget=budget, # 参数名以官方文档为准
)
return extract_text(resp)
```
回注结果：
```python
async def send_tool_result(ws, call_id, name, payload):
await ws.send(json.dumps({
"toolResponse": {
"functionResponses": [{
"id": call_id,
"name": name,
"response": {"result": payload},
}]
}
}))
```
一个体验细节：调用工具要花时间，用户会以为程序卡了。让模型在调用前先说一句"我看一下"，或者由客户端在发送工具请求时播放一段短提示音。这在实时对话里比省那半秒更重要。
步骤 4：管理思考预算
预算的本质是"用多少延迟和成本换多少推理深度"。三件事必须做：
第一，按难度路由，不要一个值走天下。
```python
thinking.py
HARD_MARKERS = ("为什么", "对比", "权衡", "推演", "证明", "计算",
"方案", "设计", "排查", "debug", "根因")
def pick_budget(text: str, low: int = 0, high: int = 4096) -> int:
text = text.strip()
if len(text) < 12 and not any(m in text for m in HARD_MARKERS):
return low # 短问题、寒暄：不思考或极少思考
if any(m in text for m in HARD_MARKERS) or len(text) > 60:
return high # 复杂问题：给足预算
return (low + high) // 2
```
第二，给每个档位测延迟。 录 20 段真实提问，分别跑低档和高档，记录从"用户说完"到"听到第一个音节"的时间。你会得到一张表：低档可能不到一秒，高档可能好几秒。这张表决定了你的路由阈值该定在哪。
第三，把思考量记进日志。 每次调用记录：原始问题、选中的预算、实际消耗的思考 token、端到端延迟。跑一周你就能看出哪些问题被浪费了预算，哪些被低估了。
一个反直觉的结论：高档位不是默认选项。用户能容忍的沉默通常在 1～3 秒之间，超过就会开始怀疑断线。宁可让助手先说"我想一下"，再慢慢答。
步骤 5：采集与推流音视频
音频输入一般是 16 位单声道 PCM，采样率按官方文档要求（常见是 16 kHz 输入、24 kHz 输出，具体以文档为准）。采样率对不上，听到的就是快放或慢放的怪声。
```python
audio_io.py
import queue, asyncio
import sounddevice as sd
IN_RATE = 16000
OUT_RATE = 24000
CHUNK_MS = 20
FRAME_BYTES = IN_RATE * CHUNK_MS // 1000 * 2 # int16 = 2 字节
class Mic:
def __init__(self):
self.q = queue.Queue()
self.stream = sd.RawInputStream(
samplerate=IN_RATE, channels=1, dtype="int16",
blocksize=FRAME_BYTES // 2,
callback=lambda indata, *_: self.q.put(bytes(indata)),
)
def start(self):
self.stream.start()
async def chunks(self):
while True:
yield await asyncio.to_thread(self.q.get)
```
推流：
```python
import base64, json
async def pump_audio(ws, mic):
async for pcm in mic.chunks():
await ws.send(json.dumps({
"realtimeInput": {
"audio": {
"data": base64.b64encode(pcm).decode(),
"mimeType": f"audio/pcm;rate={IN_RATE}",
}
}
}))
```
摄像头帧同样走 realtimeInput，换成 video 字段和 image/jpeg。每帧都发会浪费带宽，实际用 1 秒 1 帧到 2 秒 1 帧就够了，因为你要的是"看见现场"，不是"看视频"。
```python
async def pump_video(ws, camera, interval=1.0):
while True:
jpeg = camera.grab_jpeg() # 返回 bytes
await ws.send(json.dumps({
"realtimeInput": {
"video": {
"data": base64.b64encode(jpeg).decode(),
"mimeType": "image/jpeg",
}
}
}))
await asyncio.sleep(interval)
```
步骤 6：接收流式输出，边收边放
服务端返回的音频是一小块一小块来的。用一个带清空能力的播放队列，才能配合打断。
```python
audio_io.py
class Speaker:
def __init__(self):
self.stream = sd.RawOutputStream(
samplerate=OUT_RATE, channels=1, dtype="int16")
self.stream.start()
def write(self, pcm: bytes):
self.stream.write(pcm)
def flush(self):
关键：丢弃还没播出去的数据，实现"说停就停"
self.stream.abort()
self.stream.start()
```
主接收循环，同时处理音频、字幕、思考标记和打断信号：
```python
async def recv_loop(ws, speaker, on_tool_call):
async for raw in ws:
msg = json.loads(raw)
sc = msg.get("serverContent")
if sc:
if sc.get("interrupted"):
speaker.flush()
print("[已打断]")
turn = sc.get("modelTurn")
if turn:
for part in turn.get("parts", []):
inline = part.get("inlineData")
if inline and inline.get("data"):
speaker.write(base64.b64decode(inline["data"]))
if part.get("thought"):
print("[思考中] ...")
elif part.get("text"):
print("字幕:", part["text"], flush=True)
if sc.get("turnComplete"):
print("[本轮结束]")
tc = msg.get("toolCall")
if tc:
for fc in tc.get("functionCalls", []):
await on_tool_call(ws, fc)
```
thought 字段是否返回、返回什么，以官方文档为准。有就在界面上展示，没有就靠工具调用前的提示音兜底。
步骤 7：打断处理
打断是实时助手最难调的部分。分成三层：
第一层，本地清空。 收到 interrupted 立刻 speaker.flush()，同时把待播队列丢掉。播放延迟要压到几十毫秒级，否则用户说完一整句，旧回答还在念。
第二层，让服务端知道你在说话。 默认由服务端做语音活动检测。如果要自己控制，可以在检测到用户开口和收声时发送活动开始 / 活动结束事件：
```python
async def mark_activity(ws, started: bool):
await ws.send(json.dumps({
"realtimeInput": {
"activityStart" if started else "activityEnd": {}
}
}))
```
字段名与是否支持手动模式以官方文档为准。
第三层，防回声。 这是新手最容易忽略的。扬声器外放时，麦克风会收到助手自己的声音，服务端判定"用户说话了"，于是自我打断，陷入循环。解决顺序：
1. 用耳机，一步到位。
2. 用系统级回声消除，或开启音频输入设备的回音消除选项。
3. 助手说话期间调高本地语音活动检测阈值，只在明显高于背景电平时才算插话。
打断后不要关连接。 会话要继续，只是新的一轮从你的插话开始。有些实现会在打断后出现"两个回答叠在一起"，多数原因是播放队列没有真正清干净。
步骤 8：组装主循环并加长连接保护
```python
async def main():
mic = Mic(); mic.start()
speaker = Speaker()
camera = open_camera() # 你的实现
ws = await connect()
await ws.send(json.dumps({"setup": {**SETUP["setup"],
"tools": [DEEP_THINK_TOOL]}}))
重新发送带工具的 setup（按你的客户端 API 组织）
async def on_tool_call(ws, fc):
args = fc.get("args", {})
q = args.get("question", "")
budget = 4096 if args.get("budget_hint") == "high" else 1024
try:
result = await run_deep_think(q, budget)
except Exception as e:
result = f"深度思考调用失败：{e}"
await send_tool_result(ws, fc.get("id"), fc.get("name"), result)
await asyncio.gather(
pump_audio(ws, mic),
pump_video(ws, camera),
recv_loop(ws, speaker, on_tool_call),
)
if __name__ == "__main__":
asyncio.run(main())
```
长连接的三个保护动作：
- 上下文压缩：
contextWindowCompression让会话在长时间运行后自动滑窗，避免被长度上限截断。 - 会话恢复：开启
sessionResumption后，服务端会周期性下发恢复句柄。把它存下来，重连时在 setup 里带上，可以延续对话。 - 提前离场通知：连接到期前服务端一般会发一个通知事件。收到后别等它断，主动保存句柄并新建连接。
如果中间加了 Nginx 之类的反向代理，记得关掉代理缓冲、放宽读写超时，否则音频会被攒成一大块才转发，延迟直接翻倍：
```nginx
location /ws/ {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_buffering off;
proxy_read_timeout 3600s;
proxy_send_timeout 3600s;
}
```
常见坑与排错
听到的声音又快又尖 / 又慢又闷。 采样率不匹配。检查三处：麦克风设备采样率、发送时 mimeType 里声明的采样率、播放设备的采样率。三个必须一致。
助手一直打断自己。 回环。先用耳机验证，如果耳机下正常，就是外放问题；如果耳机下仍出现，检查是否把输出音频错误地也推给了输入。
提问后沉默很久才开口。 先看思考预算档位是不是选高了，再看是不是每轮都在走工具调用。给工具调用加一条规则："寒暄和单步问答不要调用"。
回答卡在半句。 播放队列的消费速度跟不上写入速度，或者 interrupted 之后队列没清空。给播放器加一个队列上限，超过就丢弃最旧的数据。
连了十几分钟就断。 上下文触顶或连接到期。开启上下文压缩与会话恢复，并处理提前离场通知。
浏览器里能跑，部署到服务器就不行。 十有八九是代理缓冲或超时。按上面的 Nginx 配置改。
调试时无法复现。 别用麦克风，改用一段固定音频文件按真实节奏回放，同时保存服务端原始事件流到日志文件。这样每次运行输入完全一致，问题才可定位。
成本悄悄涨上去。 思考预算是按 token 计费的，高档位调用频繁会明显拉高开销。把每次调用的预算和实际消耗打日志，做个每日统计。
下一步建议
跑通最小闭环之后，按这个顺序加东西：
1. 加一个"思考可见"界面。把思考片段、工具调用、字幕分三栏展示。用户看到"它在想什么"，对延迟的容忍度会明显提高。
2. 做一份评估集。录 30 段真实提问，标注"该不该调用深度思考"和"可接受的延迟上限"，每次改动路由阈值就跑一遍。
3. 接入检索。把内部文档、设备手册塞进检索层，让深度思考带上事实依据，而不是凭空推理。
4. 加记忆。把用户偏好、上次结论存成结构化记录，会话开始时注进系统指令。
5. 做降级路径。思考模型超时或限流时，让实时通道用一句"这个我需要再确认，先给你一个初步判断"接住，而不是直接报错。
最后提醒一句：模型名称、参数名、字段位置、计费方式都会随版本变化，动手前把官方文档当前版本对应页面开在浏览器里，本文代码里的字段名以那一页为准。