尧图网站开发 
1. 为什么我要用 Claude Code 搭一个 Whisper 实时转写服务语音转文字这件事看起来简单真做起来坑不少。我最早是用现成的云服务 API按分钟计费短音频还行一旦要做会议实时字幕、客服质检这种长时流式场景成本和延迟都很难接受。后来转向本地 Whisper识别质量确实够用但怎么把「文件转写」升级成「边说边出字」的实时链路中间要处理音频分片、采样率对齐、WebSocket 回传、断线重连一堆细节。这篇就聚焦一个目标用 Claude Code 从零搭一个 Whisper 实时流式语音转文字服务技术栈是 FastAPI WebSocket音频分片推送、识别结果实时回传。适合谁有 Python 基础、想跑通实时 ASR Demo 的开发者或者正在做会议记录、直播字幕、语音助手原型的同学。我会给出可复制的 config.toml 骨架含统一 Key/API 通道配置、服务启动命令和端到端验证动作你照着敲就能跑起来。核心检索词先摆出来Claude Code 负责帮你生成和补全工程代码Whisper 负责识别FastAPI 提供 HTTP 与 WebSocket 接口WebSocket 负责实时音频流。整条链路是「浏览器/客户端采集音频 → 分片推送 → 服务端缓冲 → Whisper 识别 → 结果回传」。2. 前置准备TaoToken 统一 Key 与 API 通道配置在写业务代码之前先把模型调用的通道配好。我习惯把模型访问统一走一个入口这样切换模型、管理 Key 都方便。TaoToken 提供统一的 API 通道官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。你需要先去控制台创建一个 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完在 API Keys 页面可以查看和管理页面是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你后面要做长期编码或者 Agent 类任务可以了解下 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。这里要说明一点Whisper 本身是本地模型识别不消耗远程额度但我们在工程里会预留一个「文本后处理」环节比如识别完做标点恢复、术语纠错这一步可以走统一 API 通道。所以 config.toml 里我会把两套配置都放进去本地 Whisper 参数 远程通道参数互不干扰。先建项目目录初始化依赖mkdir whisper-stream-asr cd whisper-stream-asr python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn[standard] websockets openai-whisper numpy python-multipartmacOS 上还需要 FFmpeg 处理音频格式brew install ffmpegWindows 用户去 FFmpeg 官网下压缩包把 bin 目录加进 PATH 即可。装完用ffmpeg -version验证一下。3. 可复制配置config.toml 骨架与加载代码Claude Code 在这步很好用你把需求描述清楚它能直接生成结构合理的配置加载模块。我实测下来让它「按 toml dataclass 的方式生成配置加载支持环境变量覆盖」出来的代码基本不用大改。先写 config.toml# config.toml - Whisper 实时流式服务配置 [server] host 0.0.0.0 port 8003 temp_dir ./data/temp output_dir ./data/output [whisper] # 模型大小: tiny / base / small / medium / large model_size base device cpu language # 留空表示自动检测 task transcribe # transcribe 或 translate fp16 false temperature 0.0 [audio] sample_rate 16000 channels 1 max_file_size 52428800 # 50MB split_threshold 300 # 超过 300 秒自动分片 split_duration 60 # 每片 60 秒 [stream] buffer_duration 3.0 # 缓冲 3 秒触发一次识别 sample_rate 16000 overlap_duration 0.5 # 分片重叠避免丢字 [taotoken] # 统一 API 通道用于文本后处理等远程调用 base_url https://taotoken.net/api api_key # 建议用环境变量 TAOTOKEN_API_KEY 覆盖 model claude-sonnet-4-5 timeout 30然后是加载代码用 dataclass 映射支持环境变量覆盖敏感字段# config_loader.py import os import tomllib from dataclasses import dataclass, field from pathlib import Path from typing import Optional dataclass class ServerConfig: host: str 0.0.0.0 port: int 8003 temp_dir: str ./data/temp output_dir: str ./data/output dataclass class WhisperConfig: model_size: str base device: str cpu language: str task: str transcribe fp16: bool False temperature: float 0.0 dataclass class AudioConfig: sample_rate: int 16000 channels: int 1 max_file_size: int 50 * 1024 * 1024 split_threshold: int 300 split_duration: int 60 dataclass class StreamConfig: buffer_duration: float 3.0 sample_rate: int 16000 overlap_duration: float 0.5 dataclass class TaoTokenConfig: base_url: str https://taotoken.net/api api_key: str model: str claude-sonnet-4-5 timeout: int 30 dataclass class AppConfig: server: ServerConfig field(default_factoryServerConfig) whisper: WhisperConfig field(default_factoryWhisperConfig) audio: AudioConfig field(default_factoryAudioConfig) stream: StreamConfig field(default_factoryStreamConfig) taotoken: TaoTokenConfig field(default_factoryTaoTokenConfig) def load_config(path: str config.toml) - AppConfig: raw {} p Path(path) if p.exists(): with open(p, rb) as f: raw tomllib.load(f) cfg AppConfig( serverServerConfig(**raw.get(server, {})), whisperWhisperConfig(**raw.get(whisper, {})), audioAudioConfig(**raw.get(audio, {})), streamStreamConfig(**raw.get(stream, {})), taotokenTaoTokenConfig(**raw.get(taotoken, {})), ) # 环境变量优先避免把 Key 写进文件 env_key os.getenv(TAOTOKEN_API_KEY) if env_key: cfg.taotoken.api_key env_key return cfg config load_config()这里有个细节tomllib是 Python 3.11 起内置的如果你用 3.10 及以下换成tomli并pip install tomli即可。配置加载完跑一句验证python -c from config_loader import config; print(config.whisper.model_size, config.taotoken.base_url)输出base https://taotoken.net/api就说明配置通了。4. 核心链路音频分片、Whisper 识别与 WebSocket 回传配置就绪后进入主体。整条实时链路拆成三块音频预处理、Whisper 引擎封装、WebSocket 服务。4.1 音频预处理统一采样率与分片实时流最容易翻车的地方是采样率不一致。浏览器采集通常是 48kHzWhisper 要 16kHz 单声道。我写了个预处理模块负责把任意输入统一成 float32 的 16kHz 单声道数组# audio_utils.py import numpy as np class AudioProcessor: def __init__(self, sample_rate: int 16000): self.sample_rate sample_rate def pcm16_to_float32(self, data: bytes) - np.ndarray: 把 16bit PCM 字节流转成 [-1, 1] 的 float32 数组 arr np.frombuffer(data, dtypenp.int16).astype(np.float32) return arr / 32768.0 def resample(self, audio: np.ndarray, src_rate: int, dst_rate: int) - np.ndarray: 线性插值重采样够用且无额外依赖 if src_rate dst_rate: return audio duration len(audio) / src_rate dst_len int(duration * dst_rate) src_idx np.linspace(0, len(audio) - 1, dst_len) return np.interp(src_idx, np.arange(len(audio)), audio).astype(np.float32) def split_chunks(self, audio: np.ndarray, chunk_sec: int 60): 长音频分片返回 (片段, 起始秒, 结束秒) chunk_len chunk_sec * self.sample_rate chunks [] for start in range(0, len(audio), chunk_len): end min(start chunk_len, len(audio)) chunks.append((audio[start:end], start / self.sample_rate, end / self.sample_rate)) return chunks4.2 Whisper 引擎封装把模型加载和转写包一层避免每次请求都重新加载模型# whisper_engine.py import numpy as np import whisper from config_loader import config class WhisperEngine: def __init__(self): print(f加载 Whisper 模型: {config.whisper.model_size} ({config.whisper.device})) self.model whisper.load_model( config.whisper.model_size, deviceconfig.whisper.device ) def transcribe(self, audio: np.ndarray, language: str ) - dict: options { task: config.whisper.task, temperature: config.whisper.temperature, fp16: config.whisper.fp16, } if language: options[language] language result self.model.transcribe(audio, **options) segments [ { id: i, start: round(s[start], 2), end: round(s[end], 2), text: s[text].strip(), } for i, s in enumerate(result.get(segments, [])) ] return { text: result[text].strip(), language: result.get(language, unknown), segments: segments, }4.3 WebSocket 实时服务这是整篇的核心。客户端把音频按小块推过来服务端累积到 3 秒就触发一次识别把结果回传同时保留 0.5 秒重叠避免边界丢字# app.py import numpy as np from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from audio_utils import AudioProcessor from config_loader import config from whisper_engine import WhisperEngine app FastAPI(titleWhisper 实时流式语音转文字服务) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*] ) processor AudioProcessor(sample_rateconfig.stream.sample_rate) engine WhisperEngine() app.get(/health) async def health(): return {status: ok, model: config.whisper.model_size} app.websocket(/ws/stream) async def ws_stream(ws: WebSocket): await ws.accept() buffer np.array([], dtypenp.float32) threshold int(config.stream.buffer_duration * config.stream.sample_rate) overlap int(config.stream.overlap_duration * config.stream.sample_rate) try: while True: data await ws.receive_bytes() chunk processor.pcm16_to_float32(data) buffer np.concatenate([buffer, chunk]) if len(buffer) threshold: result engine.transcribe(buffer, languageconfig.whisper.language) await ws.send_json({ type: partial, text: result[text], language: result[language], timestamp: round(len(buffer) / config.stream.sample_rate, 2), is_final: False, }) # 保留重叠部分避免分片边界丢字 buffer buffer[-overlap:] if overlap 0 else np.array([], dtypenp.float32) except WebSocketDisconnect: if len(buffer) 0: result engine.transcribe(buffer, languageconfig.whisper.language) print(f[最终片段] {result[text]})启动命令uvicorn app:app --host 0.0.0.0 --port 8003 --reload看到加载 Whisper 模型: base (cpu)和Uvicorn running on http://0.0.0.0:8003就说明服务起来了。5. 端到端验证从音频推送到识别回传服务起来后写个客户端脚本模拟实时推送。这里用 Python 的 websockets 库把一段 WAV 按 100ms 一块推过去# client_test.py import asyncio import wave import websockets async def send_audio(path: str): with wave.open(path, rb) as wf: assert wf.getframerate() 16000, 请先用 ffmpeg 转成 16kHz 单声道 assert wf.getnchannels() 1 frames wf.readframes(wf.getnframes()) chunk_size 1600 * 2 # 100ms 的 16bit 数据 async with websockets.connect(ws://localhost:8003/ws/stream) as ws: for i in range(0, len(frames), chunk_size): await ws.send(frames[i:i chunk_size]) await asyncio.sleep(0.1) # 模拟实时 try: msg await asyncio.wait_for(ws.recv(), timeout0.01) print(识别回传:, msg) except asyncio.TimeoutError: pass asyncio.run(send_audio(sample_16k.wav))准备测试音频用 FFmpeg 转成标准格式ffmpeg -i input.mp3 -ar 16000 -ac 1 -sample_fmt s16 sample_16k.wav运行客户端你会看到类似输出{type: partial, text: 大家好欢迎来到语音识别演示, language: zh, timestamp: 3.0, is_final: false} {type: partial, text: 今天我们讲实时流式转写, language: zh, timestamp: 6.0, is_final: false}每 3 秒回传一次识别结果说明整条链路通了。如果要做文本后处理比如加标点、纠术语可以在engine.transcribe之后调一次统一 API 通道把识别文本发过去做润色Key 从环境变量读export TAOTOKEN_API_KEY你的Key6. 本篇常见错排查跑不通的时候八成是下面几个问题。报错AssertionError: 请先用 ffmpeg 转成 16kHz 单声道客户端音频格式不对。Whisper 对采样率敏感48kHz 直接喂进去识别会乱。用上面那条 ffmpeg 命令转一遍或者在前端采集时就用AudioContext指定 16kHz。WebSocket 连不上报 403 或直接断开检查 uvicorn 启动参数里有没有--host 0.0.0.0以及防火墙是否放行 8003 端口。如果是浏览器端连接注意ws://和wss://要和页面协议匹配HTTPS 页面不能连ws://。识别结果重复或丢字这是分片重叠参数没调好。overlap_duration太小会丢字太大会重复。我实测 0.5 秒比较稳。如果还是重复可以在回传前做一次文本去重比较上一段结尾和新段开头。模型加载慢或内存爆base模型约 140MBlarge要 3GB 以上。CPU 环境建议先用base或small。如果显存够把device改成cuda、fp16改成true速度能快好几倍。tomllib导入失败Python 版本低于 3.11。要么升级 Python要么pip install tomli并把import tomllib改成import tomli as tomllib。API 通道调用返回 401Key 没配或环境变量没生效。确认TAOTOKEN_API_KEY已经 export或者直接在 config.toml 里填不推荐提交到仓库。接入细节可以看文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。7. 继续往下走把 Demo 变成能用的服务跑通 Demo 只是第一步。真要上线还有几件事值得做把识别结果落库做历史查询加一个前端页面用MediaRecorder采集音频给 WebSocket 加心跳和断线重连以及把 Whisper 换成faster-whisper进一步压延迟。如果你后面要做更复杂的编码任务比如给这个服务加鉴权、加限流、接数据库可以试试 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 配合 Claude Code 做长期迭代会顺手很多。想直接体验模型对话能力可以走 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 。最后留个实用技巧调试实时链路时先把buffer_duration调成 1 秒这样回传快、容易观察等逻辑稳定了再调回 3 秒降低识别频率。另外 Whisper 的initial_prompt参数很好用把你的领域术语塞进去识别准确率能明显提升比如做医疗转写就填常见药名。 
1. 为什么我要用 Claude Code 搭一个 Whisper 实时转写服务语音转文字这件事看起来简单真做起来坑不少。我最早是用现成的云服务 API按分钟计费短音频还行一旦要做会议实时字幕、客服质检这种长时流式场景成本和延迟都很难接受。后来转向本地 Whisper识别质量确实够用但怎么把「文件转写」升级成「边说边出字」的实时链路中间要处理音频分片、采样率对齐、WebSocket 回传、断线重连一堆细节。这篇就聚焦一个目标用 Claude Code 从零搭一个 Whisper 实时流式语音转文字服务技术栈是 FastAPI WebSocket音频分片推送、识别结果实时回传。适合谁有 Python 基础、想跑通实时 ASR Demo 的开发者或者正在做会议记录、直播字幕、语音助手原型的同学。我会给出可复制的 config.toml 骨架含统一 Key/API 通道配置、服务启动命令和端到端验证动作你照着敲就能跑起来。核心检索词先摆出来Claude Code 负责帮你生成和补全工程代码Whisper 负责识别FastAPI 提供 HTTP 与 WebSocket 接口WebSocket 负责实时音频流。整条链路是「浏览器/客户端采集音频 → 分片推送 → 服务端缓冲 → Whisper 识别 → 结果回传」。2. 前置准备TaoToken 统一 Key 与 API 通道配置在写业务代码之前先把模型调用的通道配好。我习惯把模型访问统一走一个入口这样切换模型、管理 Key 都方便。TaoToken 提供统一的 API 通道官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。你需要先去控制台创建一个 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完在 API Keys 页面可以查看和管理页面是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你后面要做长期编码或者 Agent 类任务可以了解下 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。这里要说明一点Whisper 本身是本地模型识别不消耗远程额度但我们在工程里会预留一个「文本后处理」环节比如识别完做标点恢复、术语纠错这一步可以走统一 API 通道。所以 config.toml 里我会把两套配置都放进去本地 Whisper 参数 远程通道参数互不干扰。先建项目目录初始化依赖mkdir whisper-stream-asr cd whisper-stream-asr python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn[standard] websockets openai-whisper numpy python-multipartmacOS 上还需要 FFmpeg 处理音频格式brew install ffmpegWindows 用户去 FFmpeg 官网下压缩包把 bin 目录加进 PATH 即可。装完用ffmpeg -version验证一下。3. 可复制配置config.toml 骨架与加载代码Claude Code 在这步很好用你把需求描述清楚它能直接生成结构合理的配置加载模块。我实测下来让它「按 toml dataclass 的方式生成配置加载支持环境变量覆盖」出来的代码基本不用大改。先写 config.toml# config.toml - Whisper 实时流式服务配置 [server] host 0.0.0.0 port 8003 temp_dir ./data/temp output_dir ./data/output [whisper] # 模型大小: tiny / base / small / medium / large model_size base device cpu language # 留空表示自动检测 task transcribe # transcribe 或 translate fp16 false temperature 0.0 [audio] sample_rate 16000 channels 1 max_file_size 52428800 # 50MB split_threshold 300 # 超过 300 秒自动分片 split_duration 60 # 每片 60 秒 [stream] buffer_duration 3.0 # 缓冲 3 秒触发一次识别 sample_rate 16000 overlap_duration 0.5 # 分片重叠避免丢字 [taotoken] # 统一 API 通道用于文本后处理等远程调用 base_url https://taotoken.net/api api_key # 建议用环境变量 TAOTOKEN_API_KEY 覆盖 model claude-sonnet-4-5 timeout 30然后是加载代码用 dataclass 映射支持环境变量覆盖敏感字段# config_loader.py import os import tomllib from dataclasses import dataclass, field from pathlib import Path from typing import Optional dataclass class ServerConfig: host: str 0.0.0.0 port: int 8003 temp_dir: str ./data/temp output_dir: str ./data/output dataclass class WhisperConfig: model_size: str base device: str cpu language: str task: str transcribe fp16: bool False temperature: float 0.0 dataclass class AudioConfig: sample_rate: int 16000 channels: int 1 max_file_size: int 50 * 1024 * 1024 split_threshold: int 300 split_duration: int 60 dataclass class StreamConfig: buffer_duration: float 3.0 sample_rate: int 16000 overlap_duration: float 0.5 dataclass class TaoTokenConfig: base_url: str https://taotoken.net/api api_key: str model: str claude-sonnet-4-5 timeout: int 30 dataclass class AppConfig: server: ServerConfig field(default_factoryServerConfig) whisper: WhisperConfig field(default_factoryWhisperConfig) audio: AudioConfig field(default_factoryAudioConfig) stream: StreamConfig field(default_factoryStreamConfig) taotoken: TaoTokenConfig field(default_factoryTaoTokenConfig) def load_config(path: str config.toml) - AppConfig: raw {} p Path(path) if p.exists(): with open(p, rb) as f: raw tomllib.load(f) cfg AppConfig( serverServerConfig(**raw.get(server, {})), whisperWhisperConfig(**raw.get(whisper, {})), audioAudioConfig(**raw.get(audio, {})), streamStreamConfig(**raw.get(stream, {})), taotokenTaoTokenConfig(**raw.get(taotoken, {})), ) # 环境变量优先避免把 Key 写进文件 env_key os.getenv(TAOTOKEN_API_KEY) if env_key: cfg.taotoken.api_key env_key return cfg config load_config()这里有个细节tomllib是 Python 3.11 起内置的如果你用 3.10 及以下换成tomli并pip install tomli即可。配置加载完跑一句验证python -c from config_loader import config; print(config.whisper.model_size, config.taotoken.base_url)输出base https://taotoken.net/api就说明配置通了。4. 核心链路音频分片、Whisper 识别与 WebSocket 回传配置就绪后进入主体。整条实时链路拆成三块音频预处理、Whisper 引擎封装、WebSocket 服务。4.1 音频预处理统一采样率与分片实时流最容易翻车的地方是采样率不一致。浏览器采集通常是 48kHzWhisper 要 16kHz 单声道。我写了个预处理模块负责把任意输入统一成 float32 的 16kHz 单声道数组# audio_utils.py import numpy as np class AudioProcessor: def __init__(self, sample_rate: int 16000): self.sample_rate sample_rate def pcm16_to_float32(self, data: bytes) - np.ndarray: 把 16bit PCM 字节流转成 [-1, 1] 的 float32 数组 arr np.frombuffer(data, dtypenp.int16).astype(np.float32) return arr / 32768.0 def resample(self, audio: np.ndarray, src_rate: int, dst_rate: int) - np.ndarray: 线性插值重采样够用且无额外依赖 if src_rate dst_rate: return audio duration len(audio) / src_rate dst_len int(duration * dst_rate) src_idx np.linspace(0, len(audio) - 1, dst_len) return np.interp(src_idx, np.arange(len(audio)), audio).astype(np.float32) def split_chunks(self, audio: np.ndarray, chunk_sec: int 60): 长音频分片返回 (片段, 起始秒, 结束秒) chunk_len chunk_sec * self.sample_rate chunks [] for start in range(0, len(audio), chunk_len): end min(start chunk_len, len(audio)) chunks.append((audio[start:end], start / self.sample_rate, end / self.sample_rate)) return chunks4.2 Whisper 引擎封装把模型加载和转写包一层避免每次请求都重新加载模型# whisper_engine.py import numpy as np import whisper from config_loader import config class WhisperEngine: def __init__(self): print(f加载 Whisper 模型: {config.whisper.model_size} ({config.whisper.device})) self.model whisper.load_model( config.whisper.model_size, deviceconfig.whisper.device ) def transcribe(self, audio: np.ndarray, language: str ) - dict: options { task: config.whisper.task, temperature: config.whisper.temperature, fp16: config.whisper.fp16, } if language: options[language] language result self.model.transcribe(audio, **options) segments [ { id: i, start: round(s[start], 2), end: round(s[end], 2), text: s[text].strip(), } for i, s in enumerate(result.get(segments, [])) ] return { text: result[text].strip(), language: result.get(language, unknown), segments: segments, }4.3 WebSocket 实时服务这是整篇的核心。客户端把音频按小块推过来服务端累积到 3 秒就触发一次识别把结果回传同时保留 0.5 秒重叠避免边界丢字# app.py import numpy as np from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from audio_utils import AudioProcessor from config_loader import config from whisper_engine import WhisperEngine app FastAPI(titleWhisper 实时流式语音转文字服务) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*] ) processor AudioProcessor(sample_rateconfig.stream.sample_rate) engine WhisperEngine() app.get(/health) async def health(): return {status: ok, model: config.whisper.model_size} app.websocket(/ws/stream) async def ws_stream(ws: WebSocket): await ws.accept() buffer np.array([], dtypenp.float32) threshold int(config.stream.buffer_duration * config.stream.sample_rate) overlap int(config.stream.overlap_duration * config.stream.sample_rate) try: while True: data await ws.receive_bytes() chunk processor.pcm16_to_float32(data) buffer np.concatenate([buffer, chunk]) if len(buffer) threshold: result engine.transcribe(buffer, languageconfig.whisper.language) await ws.send_json({ type: partial, text: result[text], language: result[language], timestamp: round(len(buffer) / config.stream.sample_rate, 2), is_final: False, }) # 保留重叠部分避免分片边界丢字 buffer buffer[-overlap:] if overlap 0 else np.array([], dtypenp.float32) except WebSocketDisconnect: if len(buffer) 0: result engine.transcribe(buffer, languageconfig.whisper.language) print(f[最终片段] {result[text]})启动命令uvicorn app:app --host 0.0.0.0 --port 8003 --reload看到加载 Whisper 模型: base (cpu)和Uvicorn running on http://0.0.0.0:8003就说明服务起来了。5. 端到端验证从音频推送到识别回传服务起来后写个客户端脚本模拟实时推送。这里用 Python 的 websockets 库把一段 WAV 按 100ms 一块推过去# client_test.py import asyncio import wave import websockets async def send_audio(path: str): with wave.open(path, rb) as wf: assert wf.getframerate() 16000, 请先用 ffmpeg 转成 16kHz 单声道 assert wf.getnchannels() 1 frames wf.readframes(wf.getnframes()) chunk_size 1600 * 2 # 100ms 的 16bit 数据 async with websockets.connect(ws://localhost:8003/ws/stream) as ws: for i in range(0, len(frames), chunk_size): await ws.send(frames[i:i chunk_size]) await asyncio.sleep(0.1) # 模拟实时 try: msg await asyncio.wait_for(ws.recv(), timeout0.01) print(识别回传:, msg) except asyncio.TimeoutError: pass asyncio.run(send_audio(sample_16k.wav))准备测试音频用 FFmpeg 转成标准格式ffmpeg -i input.mp3 -ar 16000 -ac 1 -sample_fmt s16 sample_16k.wav运行客户端你会看到类似输出{type: partial, text: 大家好欢迎来到语音识别演示, language: zh, timestamp: 3.0, is_final: false} {type: partial, text: 今天我们讲实时流式转写, language: zh, timestamp: 6.0, is_final: false}每 3 秒回传一次识别结果说明整条链路通了。如果要做文本后处理比如加标点、纠术语可以在engine.transcribe之后调一次统一 API 通道把识别文本发过去做润色Key 从环境变量读export TAOTOKEN_API_KEY你的Key6. 本篇常见错排查跑不通的时候八成是下面几个问题。报错AssertionError: 请先用 ffmpeg 转成 16kHz 单声道客户端音频格式不对。Whisper 对采样率敏感48kHz 直接喂进去识别会乱。用上面那条 ffmpeg 命令转一遍或者在前端采集时就用AudioContext指定 16kHz。WebSocket 连不上报 403 或直接断开检查 uvicorn 启动参数里有没有--host 0.0.0.0以及防火墙是否放行 8003 端口。如果是浏览器端连接注意ws://和wss://要和页面协议匹配HTTPS 页面不能连ws://。识别结果重复或丢字这是分片重叠参数没调好。overlap_duration太小会丢字太大会重复。我实测 0.5 秒比较稳。如果还是重复可以在回传前做一次文本去重比较上一段结尾和新段开头。模型加载慢或内存爆base模型约 140MBlarge要 3GB 以上。CPU 环境建议先用base或small。如果显存够把device改成cuda、fp16改成true速度能快好几倍。tomllib导入失败Python 版本低于 3.11。要么升级 Python要么pip install tomli并把import tomllib改成import tomli as tomllib。API 通道调用返回 401Key 没配或环境变量没生效。确认TAOTOKEN_API_KEY已经 export或者直接在 config.toml 里填不推荐提交到仓库。接入细节可以看文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。7. 继续往下走把 Demo 变成能用的服务跑通 Demo 只是第一步。真要上线还有几件事值得做把识别结果落库做历史查询加一个前端页面用MediaRecorder采集音频给 WebSocket 加心跳和断线重连以及把 Whisper 换成faster-whisper进一步压延迟。如果你后面要做更复杂的编码任务比如给这个服务加鉴权、加限流、接数据库可以试试 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 配合 Claude Code 做长期迭代会顺手很多。想直接体验模型对话能力可以走 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 。最后留个实用技巧调试实时链路时先把buffer_duration调成 1 秒这样回传快、容易观察等逻辑稳定了再调回 3 秒降低识别频率。另外 Whisper 的initial_prompt参数很好用把你的领域术语塞进去识别准确率能明显提升比如做医疗转写就填常见药名。