与文字转语音并播放(pyttsx3,ChatTTS))
一、语音转字符串(faster_whisper)1、安装 faster_whisper、pyaudio、zhconvpip install faster_whisper pyaudio zhconv 国内镜像 pip install faster_whisper pyaudio zhconv -i https://pypi.tuna.tsinghua.edu.cn/simple2、创建存放模型的文件夹3、修改模型下载地址为国内镜像修改下载的路径为刚新建存放模型的文件夹地址运行下列代码import os os.environ[HF_ENDPOINT] https://hf-mirror.com #国内镜像地址 os.environ[HF_HOME] D:/Pycharm/Project/ChatTTS/faster_whisper_models # 定义模型放置地址。指向 hub 的父目录不是 hub 本身(要使用/作为文件分割符不能用\\) import warnings warnings.filterwarnings(ignore, categoryUserWarning, modulezhconv) import threading import time import numpy as np import pyaudio from faster_whisper import WhisperModel from zhconv import convert # 需安装 MODEL_SIZE medium # 升级模型 SAMPLE_RATE 16000 # Whisper 要求的采样率 CHUNK_DURATION 10 # 每次录音的时长秒 model WhisperModel( model_size_or_path MODEL_SIZE, # 模型大小或本地路径。字符串如 tiny, base, small, medium, large-v3。也可是本地文件夹路径如 D:/models/faster-whisper-tiny devicecuda, # 运行设备cpu 或 cuda需要 NVIDIA GPU 及 CUDA 环境。GPU 能大幅加速。 compute_typefloat16, # 计算精度类型影响速度、内存和准确性 # int8内存占用最小速度较快精度略有损失CPU 上常用。 # float16半精度GPU 上速度最快精度损失小需 GPU 支持。 # float32单精度最准确但内存和计算开销最大一般只在 CPU 且追求极致精度时使用。 cpu_threads4, num_workers1 ) audio pyaudio.PyAudio() stream audio.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferint(SAMPLE_RATE * CHUNK_DURATION) ) print(监听中... (CtrlC 退出)) def transcribe_audio(audio_data): audio_np np.frombuffer(audio_data, dtypenp.int16).astype(np.float32) / 32768.0 segments, info model.transcribe( audio_np, languagezh, # 指定语言代码如 zh 中文en 英文若不指定则自动检测。 beam_size5, # 集束搜索宽度默认 5越大越准确但速度越慢。 best_of5, # 可选配合 beam_size 使用进一步挑选最佳结果 vad_filterTrue, # 是否启用内置的语音活动检测VAD可以过滤静音片段提升体验设为 True 可避免对静音进行转写。 vad_parametersdict(min_silence_duration_ms500), temperature 0.0, # 控制生成文本的随机性。对于追求准确性的转录任务可以使用较低的温度值如 0.0 # 以下为额外可调参数 condition_on_previous_text True, # False避免依赖前文减少错误累积 no_speech_threshold 0.6, # 提高静音检测灵敏度过滤无效片段默认0.6 compression_ratio_threshold 2.4, # 过滤异常压缩比的结果默认2.4 ) for segment in segments: simplified convert(segment.text, zh-cn) print(f[{segment.start:.2f}s - {segment.end:.2f}s] {simplified}) try: while True: audio_bytes stream.read(int(SAMPLE_RATE * CHUNK_DURATION)) threading.Thread(targettranscribe_audio, args(audio_bytes,), daemonTrue).start() time.sleep(CHUNK_DURATION - 0.1) # # 略微提前唤醒实现流式效果 except KeyboardInterrupt: print(\n退出) finally: stream.stop_stream() stream.close() audio.terminate()二、文字转语音并播放(pyttsx3,ChatTTS)一、pyttsx31、安装pip install pyttsx32、调用import pyttsx3 engine pyttsx3.init() engine.setProperty(rate, 150) # 语速 engine.setProperty(volume, 0.9) # 音量 engine.say(你好指令已收到。) engine.runAndWait()二、ChatTTS1、使用整合包2、手动部署1、安装condapycharm搞一个空的虚拟环境GPU版本1、下载布置cudacudnn2、 克隆项目-git clone https://github.com/2noise/ChatTTSChatTTS:A generative speech model for daily dialogue. - AtomGit#这个也行克隆完成后 # 1. 进入你克隆下来的项目文件夹 cd ChatTTS # 2. 安装项目及其所有依赖 (开发模式) pip install -e .3、 安装依赖torchtorchaudio针对不同GPU用不同版本 我的GPU装了cuda12.6 到时候再查就好了 南京大学源 pip install torch2.8.0cu126 torchaudio2.8.0cu126 --index-url https://mirrors.nju.edu.cn/pytorch/whl/cu126 原连接 pip install torch2.8.0cu126 torchaudio2.8.0cu126 --index-url https://download.pytorch.org/whl/cu126剩余依赖和cpu版本流程一样4、下载模型安装ModelScopepip install modelscope先去中文模型网https://www.modelscope.cn搜索ChatTTS#里面提供了多种模型下载方式下载完整模型库modelscope download --model AI-ModelScope/ChatTTS --local_dir D:/Models/chatTTS --local_dir D:/Models/chatTTS 是自定义的模型下载地址5、测试import os import ChatTTS import soundfile as sf import torch # 填写你的模型路径 KNOWN_MODEL_PATH C:/Users/chenzl/Desktop/gitee/ChatTTS/models # 例如/home/user/models/ChatTTS # # 1. 初始化 chat ChatTTS.Chat() # 2. 加载模型 —— 使用 load() 方法指定 sourcecustom 和 custom_path chat.load( sourcecustom, # 从自定义路径加载 custom_pathKNOWN_MODEL_PATH, compileFalse, # 首次加载建议关闭编译加快速度 # devicecuda 或 cpu 可自动检测不传则默认 ) # 3. 准备测试文本 texts [你好这是 ChatTTS 的本地模型测试。我的声音听起来自然吗] # 4. 生成音频 wavs chat.infer(texts) # 5. 保存为 WAV 文件 # sf.write(test_output.wav, wavs[0][0], 24000) sf.write(test_output.wav, wavs[0], 24000) print(✅ 测试完成请播放当前目录下的 test_output.wav 文件)CPU版本1、克隆项目-git clone https://github.com/2noise/ChatTTS2、安装依赖-用下列命令 在镜像源下载torchcpu版本与 torchaudiopip install torch torchaudio --index-url https://download.pytorch.org/whl/cpu -i https://pypi.tuna.tsinghua.edu.cn/simple下面是原本的requirements.txtnumpy3.0.0 numba torch2.1.0 torchaudio tqdm vector_quantize_pytorch transformers4.41.1 vocos IPython gradio pybase16384 pynini2.1.5; sys_platform linux WeTextProcessing; sys_platform linux nemo_text_processing; sys_platform linux av pydub requestsrequirements.txt改成下面的numpy3.0.0 numba #torch2.1.0 #torchaudio tqdm vector_quantize_pytorch transformers4.41.1 vocos IPython gradio pybase16384 pynini2.1.5; sys_platform linux WeTextProcessing; sys_platform linux nemo_text_processing; sys_platform linux av pydub requests输入命令下载别的依赖国内源下载 省时间pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/安装soundfilepip install soundfile根目录下测试3、下载模型约2.3GB4、运行代码三、会说话的TOMimport os import threading import time import queue import numpy as np import pyaudio import sounddevice as sd os.environ[HF_ENDPOINT] https://hf-mirror.com os.environ[HF_HOME] D:/Pycharm/Project/ChatTTS/faster_whisper_models import ChatTTS from faster_whisper import WhisperModel from zhconv import convert # 配置区 # Whisper 配置 WHISPER_MODEL_SIZE medium WHISPER_DEVICE cuda COMPUTE_TYPE float16 CPU_THREADS 4 NUM_WORKERS 1 SAMPLE_RATE 16000 # 录音帧参数小帧实时处理 FRAME_DURATION 0.05 # 50ms 一帧 FRAME_SIZE int(SAMPLE_RATE * FRAME_DURATION) # VAD 参数 ENERGY_THRESHOLD 0.015 # 能量阈值高于此值认为是语音环境噪音大可调高至 0.025 SILENCE_DURATION 0.8 # 连续 0.8 秒静音认为语音结束 SILENCE_FRAMES int(SILENCE_DURATION / FRAME_DURATION) MIN_SPEECH_DURATION 0.3 # 最少 0.3 秒才认为是有效语音 MIN_SPEECH_FRAMES int(MIN_SPEECH_DURATION / FRAME_DURATION) # ChatTTS 配置 CHATTTS_MODEL_PATH D:/Pycharm/Project/ChatTTS/chattts_models #填写本地模型的地址 # 初始化 print(加载 ChatTTS 模型...) chat ChatTTS.Chat() chat.load(sourcecustom, custom_pathCHATTTS_MODEL_PATH, compileTrue) print(✅ ChatTTS 加载完成) print(加载 Whisper 模型...) whisper_model WhisperModel( model_size_or_pathWHISPER_MODEL_SIZE, deviceWHISPER_DEVICE, compute_typeCOMPUTE_TYPE, cpu_threadsCPU_THREADS, num_workersNUM_WORKERS ) print(✅ Whisper 加载完成) # 队列与同步信号 audio_queue queue.Queue(maxsize30) text_queue queue.Queue() play_queue queue.Queue() stop_event threading.Event() playback_active threading.Event() # 播放期间为 True录音线程丢弃音频防回环 # 线程函数 # ---- 录音线程实时 VAD 切分语音停顿即送队列 ---- def record_microphone(): p pyaudio.PyAudio() stream p.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferFRAME_SIZE ) print( 录音开始语音停顿时自动切分CtrlC 停止...) speech_buffer [] # 累积语音帧 silence_count 0 # 连续静音帧计数 speech_count 0 # 当前语音段的总语音帧数 try: while not stop_event.is_set(): try: audio_bytes stream.read(FRAME_SIZE, exception_on_overflowFalse) except Exception as e: print(f录音读取错误: {e}) break audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 energy np.sqrt(np.mean(audio_np ** 2)) if len(audio_np) 0 else 0.0 if energy ENERGY_THRESHOLD: # 语音帧 silence_count 0 speech_buffer.append(audio_np) speech_count 1 else: # 静音帧 if speech_count 0: speech_buffer.append(audio_np) silence_count 1 if silence_count SILENCE_FRAMES: # 语音结束 if speech_count MIN_SPEECH_FRAMES: if playback_active.is_set(): # 正在播放扬声器输出丢弃以防止回环 print( 正在播放丢弃本段录音防回环) else: full_speech np.concatenate(speech_buffer) speech_bytes (np.clip(full_speech, -1.0, 1.0) * 32767).astype(np.int16).tobytes() try: audio_queue.put(speech_bytes, timeout1) except queue.Full: print(⚠️ 音频队列已满丢弃该段语音) else: # 语音太短丢弃 pass # 重置状态 speech_buffer [] silence_count 0 speech_count 0 # else: 纯静音不处理 except KeyboardInterrupt: pass finally: stream.stop_stream() stream.close() p.terminate() print( 录音线程结束) # ---- 转录线程 ---- def transcribe_worker(): while not stop_event.is_set(): try: audio_bytes audio_queue.get(timeout0.5) except queue.Empty: continue if audio_bytes is None: break try: audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 if len(audio_np) int(SAMPLE_RATE * 0.2): audio_queue.task_done() continue segments, _ whisper_model.transcribe( audio_np, languagezh, beam_size5, vad_filterTrue, vad_parametersdict(min_silence_duration_ms500), temperature0.0 ) full_text for seg in segments: try: simplified convert(seg.text, zh-cn) except Exception: simplified seg.text full_text simplified if full_text.strip(): print(f 识别文本: {full_text}) text_queue.put(full_text) except Exception as e: print(f❌ 转录错误: {e}) finally: audio_queue.task_done() # ---- 合成线程 ---- def synthesize_worker(): while not stop_event.is_set(): try: text text_queue.get(timeout0.5) except queue.Empty: continue if text is None: break print(f 开始合成: {text}) try: wavs chat.infer([text]) if not wavs or len(wavs) 0: print(⚠️ ChatTTS 返回空结果) text_queue.task_done() continue audio_data np.asarray(wavs[0]) # 确保一维 if audio_data.ndim 1: audio_data audio_data.flatten() # 归一化并裁剪到 [-1, 1]防止 int16 溢出导致爆音/播放中断 max_val np.max(np.abs(audio_data)) if max_val 1.0: audio_data audio_data / max_val audio_int16 (np.clip(audio_data, -1.0, 1.0) * 32767).astype(np.int16) if len(audio_int16) 0: play_queue.put(audio_int16) print( 合成完成加入播放队列) else: print(⚠️ 合成结果为空) except Exception as e: print(f❌ 合成错误: {e}) finally: text_queue.task_done() # ---- 播放线程播放前加锁防回环播放后解锁 ---- def play_worker(): while not stop_event.is_set(): try: audio_int16 play_queue.get(timeout0.5) except queue.Empty: continue if audio_int16 is None: break try: if len(audio_int16) 0: print(⚠️ 播放音频为空跳过) play_queue.task_done() continue audio_float audio_int16.astype(np.float32) / 32768.0 # 确保一维 if audio_float.ndim 1: audio_float audio_float.flatten() duration len(audio_float) / 24000.0 print(f 开始播放时长 {duration:.2f} 秒) # 上锁通知录音线程丢弃扬声器输出的回声 playback_active.set() sd.play(audio_float, 24000) sd.wait() playback_active.clear() # 播放结束后额外等待让扬声器余音消散 time.sleep(1) print(✅ 播放完成) except Exception as e: print(f❌ 播放错误: {e}) playback_active.clear() finally: play_queue.task_done() # 启动所有线程 threads [] t_rec threading.Thread(targetrecord_microphone) t_rec.start() threads.append(t_rec) t_trans threading.Thread(targettranscribe_worker) t_trans.start() threads.append(t_trans) t_synth threading.Thread(targetsynthesize_worker) t_synth.start() threads.append(t_synth) t_play threading.Thread(targetplay_worker) t_play.start() threads.append(t_play) # 主线程等待退出CtrlC try: while True: time.sleep(0.5) except KeyboardInterrupt: print(\n 正在退出...) # 清理同步信号 playback_active.clear() stop_event.set() # 发送哨兵值让阻塞在 get() 的线程退出 audio_queue.put(None) text_queue.put(None) play_queue.put(None) # 等待各线程结束最多等 5 秒 for t in threads: t.join(timeout5) print(✅ 已退出)四、实现语音控制打开应用1、下载语音转文字模型import os # 环境变量必须在导入模型库之前设置 # Hugging Face 镜像站。国内直连 huggingface.co 会超时WinError 10060 # 走镜像站可以正常下载和校验。 os.environ[HF_ENDPOINT] https://hf-mirror.com # 模型缓存根目录。注意 # 1. 指向 hub 的【父目录】不是 hub 本身 # 2. 必须用 / 作分隔符不能用 \否则会被当成转义字符 # 3. 实际模型会存到 .../语音转文字/hub/models--Systran--faster-whisper-small/ os.environ[HF_HOME] C:/Users/chenzl/Desktop/gitee/ChatTTS/语音转文字 from faster_whisper import WhisperModel # 加载 / 下载模型 # 这行执行时先去 HF_HOME/hub/ 找模型找不到就通过 HF_ENDPOINT 下载 # 下载完自动加载到内存。所以这行跑完模型文件已经落到硬盘上了。 model WhisperModel( model_size_or_pathsmall, # 模型规格或本地路径。 # tiny ~39M 最快准确率最低 # base ~74M 较快 # small ~244M 速度和准确率平衡你现在用的 # medium ~769M 更准明显更慢 # large-v3 ~1.5G 最准CPU 上很慢 # 也可传本地文件夹路径如 D:/models/faster-whisper-small devicecpu, # 运行设备 # cpu 普通 CPU 推理 # cuda 需要 NVIDIA GPU CUDA 环境速度大幅提升 compute_typefloat32, # 计算精度影响速度 / 内存 / 准确率 # int8 内存最小速度较快精度略损CPU 上常用 # float16 半精度GPU 上最快精度损失小 # float32 单精度最准确内存和计算开销最大 cpu_threads4, # CPU 推理线程数。一般设为物理核心数太多反而变慢 num_workers1, # 并发处理的工作线程数。单条音频转写保持 1 即可 ) print(下载完成)2、下载字符串向量化模型import os # 环境变量必须在导入模型库之前设置 # Hugging Face 官方域名在国内经常连不上会报 WinError 10060连接超时。 # 设置镜像站后所有对 huggingface.co 的请求都会被重定向到 hf-mirror.com。 os.environ[HF_ENDPOINT] https://hf-mirror.com # 模型缓存根目录。注意三点 # 1. 指向 hub 的【父目录】不是 hub 本身。 # 实际模型会存到 .../字符转向量打开软件/hub/models--BAAI--bge-m3/ # 2. 推荐用 / 作分隔符这里用 \\ 是因为 Python 字符串里 \ 是转义字符 # 写成 C:\Users 会把 \U 当成 Unicode 转义而出错所以双写 \\。 # 3. 这个变量必须在 import 模型库之前设置导入后再设就不生效了。 os.environ[HF_HOME] C:\\Users\\chenzl\\Desktop\\gitee\\ChatTTS\\字符转向量打开软件 from sentence_transformers import SentenceTransformer # 加载 / 下载 BGE-M3 # 这行执行时先去 HF_HOME/hub/ 找 BAAI/bge-m3 # 找不到 → 通过 HF_ENDPOINT 镜像站下载 → 存到 HF_HOME 并加载到内存 # 已存在 → 直接读本地不联网。 model SentenceTransformer( BAAI/bge-m3 # 模型标识格式为 组织名/模型名。 # BAAI 北京智源人工智能研究院 # bge-m3 多语言嵌入模型中文效果强向量维度 1024 # 也可传本地完整路径如 rD:\models\bge-m3 # 那样就直接读本地完全不碰 HF_HOME、不联网。 )3、最终实现import warnings warnings.filterwarnings(ignore, categoryUserWarning, modulezhconv) import threading import queue import subprocess import webbrowser import numpy as np import pyaudio from faster_whisper import WhisperModel from zhconv import convert from sentence_transformers import SentenceTransformer # 1. 配置 SAMPLE_RATE 16000 CHUNK_DURATION 10 SIM_THRESHOLD 0.7 WHISPER_MODEL_PATH rC:\Users\chenzl\Desktop\gitee\ChatTTS\语音转文字\hub\models--Systran--faster-whisper-small\snapshots\536b0662742c02347bc0e980a01041f333bce120 BGE_MODEL_PATH rC:\Users\chenzl\Desktop\gitee\ChatTTS\字符转向量打开软件\hub\models--BAAI--bge-m3\snapshots\5617a9f61b028005a4858fdac845db406aefb181 # 2. 加载 Whisper print(正在加载 Whisper 模型...) whisper_model WhisperModel( model_size_or_pathWHISPER_MODEL_PATH, devicecpu, compute_typefloat32, cpu_threads4, num_workers1, ) # 3. 加载 BGE-M3 print(正在加载 BGE-M3 模型...) embed_model SentenceTransformer(BGE_MODEL_PATH) # 4. 意图库 intents { open_deepseek: [打开deepseek, 启动ai助手, 我要搜索问题, 把网页打开, 打开网页], open_weixin: [打开微信, 启动微信, 我要聊天, 打开WeChat], open_bilibili: [打开哔哩哔哩, 启动B站, 我要看视频, 打开B站], unknown: [ 今天天气怎么样, 你叫什么名字, 现在几点了, 帮我查一下资料, 这个多少钱, 明天要下雨吗, 随便聊聊, 你好, 谢谢, ], } print(正在向量化意图库...) intent_vectors {} for name, phrases in intents.items(): vecs embed_model.encode(phrases, normalize_embeddingsTrue) intent_vectors[name] vecs.mean(axis0) print(意图库准备完成。) # 5. 匹配函数 def match_intent(text, thresholdSIM_THRESHOLD): q embed_model.encode([text], normalize_embeddingsTrue)[0] best, best_score None, -1 for name, vec in intent_vectors.items(): score float(np.dot(q, vec)) if score best_score: best, best_score name, score return (best, best_score) if best_score threshold else (None, best_score) # 6. 动作 def open_deepseek(): webbrowser.open(https://chat.deepseek.com/) def open_weixin(): subprocess.Popen([rD:\Weixin\Weixin.exe]) def open_bilibili(): subprocess.Popen([rD:\bilibili\哔哩哔哩.exe]) actions { open_deepseek: open_deepseek, open_weixin: open_weixin, open_bilibili: open_bilibili, } # 7. 语音转写线程 audio_queue queue.Queue() # 存放待转写的音频 text_queue queue.Queue() # 存放转写好的文字 def transcribe_worker(): 后台线程不断从 audio_queue 取音频转写后把文字放进 text_queue while True: audio_bytes audio_queue.get() if audio_bytes is None: break try: audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 segments, info whisper_model.transcribe( audio_np, languagezh, beam_size20, vad_filterTrue, vad_parametersdict(min_silence_duration_ms500), temperature0.0, ) pieces [] for segment in segments: simplified convert(segment.text, zh-cn) pieces.append(simplified) text .join(pieces).strip().replace( , ) if text: text_queue.put(text) except Exception as e: print(f[转写出错] {e}) # 8. 启动录音 audio pyaudio.PyAudio() stream audio.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferint(SAMPLE_RATE * CHUNK_DURATION), ) # 启动转写线程 transcribe_thread threading.Thread(targettranscribe_worker, daemonTrue) transcribe_thread.start() print(监听中... (CtrlC 退出)) # 9. 主循环 def record_loop(): 后台线程不断录音把音频塞进 audio_queue try: while True: audio_bytes stream.read(int(SAMPLE_RATE * CHUNK_DURATION), exception_on_overflowFalse) audio_queue.put(audio_bytes) except Exception as e: print(f[录音停止] {e}) record_thread threading.Thread(targetrecord_loop, daemonTrue) record_thread.start() try: while True: # 从 text_queue 取转写结果阻塞等待 try: text text_queue.get(timeout0.5) except queue.Empty: continue print(f\n️ 识别{text}) intent, score match_intent(text) if intent and intent ! unknown: actions[intent]() print(f✅ 执行{intent} (置信度 {score:.3f})) else: print(f❌ 没听懂 (最高分 {score:.3f})) except KeyboardInterrupt: print(\n退出) finally: audio_queue.put(None) # 通知转写线程退出 stream.stop_stream() stream.close() audio.terminate() 语音转字符串(faster_whisper)与文字转语音并播放(pyttsx3,ChatTTS)
一、语音转字符串(faster_whisper)1、安装 faster_whisper、pyaudio、zhconvpip install faster_whisper pyaudio zhconv 国内镜像 pip install faster_whisper pyaudio zhconv -i https://pypi.tuna.tsinghua.edu.cn/simple2、创建存放模型的文件夹3、修改模型下载地址为国内镜像修改下载的路径为刚新建存放模型的文件夹地址运行下列代码import os os.environ[HF_ENDPOINT] https://hf-mirror.com #国内镜像地址 os.environ[HF_HOME] D:/Pycharm/Project/ChatTTS/faster_whisper_models # 定义模型放置地址。指向 hub 的父目录不是 hub 本身(要使用/作为文件分割符不能用\\) import warnings warnings.filterwarnings(ignore, categoryUserWarning, modulezhconv) import threading import time import numpy as np import pyaudio from faster_whisper import WhisperModel from zhconv import convert # 需安装 MODEL_SIZE medium # 升级模型 SAMPLE_RATE 16000 # Whisper 要求的采样率 CHUNK_DURATION 10 # 每次录音的时长秒 model WhisperModel( model_size_or_path MODEL_SIZE, # 模型大小或本地路径。字符串如 tiny, base, small, medium, large-v3。也可是本地文件夹路径如 D:/models/faster-whisper-tiny devicecuda, # 运行设备cpu 或 cuda需要 NVIDIA GPU 及 CUDA 环境。GPU 能大幅加速。 compute_typefloat16, # 计算精度类型影响速度、内存和准确性 # int8内存占用最小速度较快精度略有损失CPU 上常用。 # float16半精度GPU 上速度最快精度损失小需 GPU 支持。 # float32单精度最准确但内存和计算开销最大一般只在 CPU 且追求极致精度时使用。 cpu_threads4, num_workers1 ) audio pyaudio.PyAudio() stream audio.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferint(SAMPLE_RATE * CHUNK_DURATION) ) print(监听中... (CtrlC 退出)) def transcribe_audio(audio_data): audio_np np.frombuffer(audio_data, dtypenp.int16).astype(np.float32) / 32768.0 segments, info model.transcribe( audio_np, languagezh, # 指定语言代码如 zh 中文en 英文若不指定则自动检测。 beam_size5, # 集束搜索宽度默认 5越大越准确但速度越慢。 best_of5, # 可选配合 beam_size 使用进一步挑选最佳结果 vad_filterTrue, # 是否启用内置的语音活动检测VAD可以过滤静音片段提升体验设为 True 可避免对静音进行转写。 vad_parametersdict(min_silence_duration_ms500), temperature 0.0, # 控制生成文本的随机性。对于追求准确性的转录任务可以使用较低的温度值如 0.0 # 以下为额外可调参数 condition_on_previous_text True, # False避免依赖前文减少错误累积 no_speech_threshold 0.6, # 提高静音检测灵敏度过滤无效片段默认0.6 compression_ratio_threshold 2.4, # 过滤异常压缩比的结果默认2.4 ) for segment in segments: simplified convert(segment.text, zh-cn) print(f[{segment.start:.2f}s - {segment.end:.2f}s] {simplified}) try: while True: audio_bytes stream.read(int(SAMPLE_RATE * CHUNK_DURATION)) threading.Thread(targettranscribe_audio, args(audio_bytes,), daemonTrue).start() time.sleep(CHUNK_DURATION - 0.1) # # 略微提前唤醒实现流式效果 except KeyboardInterrupt: print(\n退出) finally: stream.stop_stream() stream.close() audio.terminate()二、文字转语音并播放(pyttsx3,ChatTTS)一、pyttsx31、安装pip install pyttsx32、调用import pyttsx3 engine pyttsx3.init() engine.setProperty(rate, 150) # 语速 engine.setProperty(volume, 0.9) # 音量 engine.say(你好指令已收到。) engine.runAndWait()二、ChatTTS1、使用整合包2、手动部署1、安装condapycharm搞一个空的虚拟环境GPU版本1、下载布置cudacudnn2、 克隆项目-git clone https://github.com/2noise/ChatTTSChatTTS:A generative speech model for daily dialogue. - AtomGit#这个也行克隆完成后 # 1. 进入你克隆下来的项目文件夹 cd ChatTTS # 2. 安装项目及其所有依赖 (开发模式) pip install -e .3、 安装依赖torchtorchaudio针对不同GPU用不同版本 我的GPU装了cuda12.6 到时候再查就好了 南京大学源 pip install torch2.8.0cu126 torchaudio2.8.0cu126 --index-url https://mirrors.nju.edu.cn/pytorch/whl/cu126 原连接 pip install torch2.8.0cu126 torchaudio2.8.0cu126 --index-url https://download.pytorch.org/whl/cu126剩余依赖和cpu版本流程一样4、下载模型安装ModelScopepip install modelscope先去中文模型网https://www.modelscope.cn搜索ChatTTS#里面提供了多种模型下载方式下载完整模型库modelscope download --model AI-ModelScope/ChatTTS --local_dir D:/Models/chatTTS --local_dir D:/Models/chatTTS 是自定义的模型下载地址5、测试import os import ChatTTS import soundfile as sf import torch # 填写你的模型路径 KNOWN_MODEL_PATH C:/Users/chenzl/Desktop/gitee/ChatTTS/models # 例如/home/user/models/ChatTTS # # 1. 初始化 chat ChatTTS.Chat() # 2. 加载模型 —— 使用 load() 方法指定 sourcecustom 和 custom_path chat.load( sourcecustom, # 从自定义路径加载 custom_pathKNOWN_MODEL_PATH, compileFalse, # 首次加载建议关闭编译加快速度 # devicecuda 或 cpu 可自动检测不传则默认 ) # 3. 准备测试文本 texts [你好这是 ChatTTS 的本地模型测试。我的声音听起来自然吗] # 4. 生成音频 wavs chat.infer(texts) # 5. 保存为 WAV 文件 # sf.write(test_output.wav, wavs[0][0], 24000) sf.write(test_output.wav, wavs[0], 24000) print(✅ 测试完成请播放当前目录下的 test_output.wav 文件)CPU版本1、克隆项目-git clone https://github.com/2noise/ChatTTS2、安装依赖-用下列命令 在镜像源下载torchcpu版本与 torchaudiopip install torch torchaudio --index-url https://download.pytorch.org/whl/cpu -i https://pypi.tuna.tsinghua.edu.cn/simple下面是原本的requirements.txtnumpy3.0.0 numba torch2.1.0 torchaudio tqdm vector_quantize_pytorch transformers4.41.1 vocos IPython gradio pybase16384 pynini2.1.5; sys_platform linux WeTextProcessing; sys_platform linux nemo_text_processing; sys_platform linux av pydub requestsrequirements.txt改成下面的numpy3.0.0 numba #torch2.1.0 #torchaudio tqdm vector_quantize_pytorch transformers4.41.1 vocos IPython gradio pybase16384 pynini2.1.5; sys_platform linux WeTextProcessing; sys_platform linux nemo_text_processing; sys_platform linux av pydub requests输入命令下载别的依赖国内源下载 省时间pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/安装soundfilepip install soundfile根目录下测试3、下载模型约2.3GB4、运行代码三、会说话的TOMimport os import threading import time import queue import numpy as np import pyaudio import sounddevice as sd os.environ[HF_ENDPOINT] https://hf-mirror.com os.environ[HF_HOME] D:/Pycharm/Project/ChatTTS/faster_whisper_models import ChatTTS from faster_whisper import WhisperModel from zhconv import convert # 配置区 # Whisper 配置 WHISPER_MODEL_SIZE medium WHISPER_DEVICE cuda COMPUTE_TYPE float16 CPU_THREADS 4 NUM_WORKERS 1 SAMPLE_RATE 16000 # 录音帧参数小帧实时处理 FRAME_DURATION 0.05 # 50ms 一帧 FRAME_SIZE int(SAMPLE_RATE * FRAME_DURATION) # VAD 参数 ENERGY_THRESHOLD 0.015 # 能量阈值高于此值认为是语音环境噪音大可调高至 0.025 SILENCE_DURATION 0.8 # 连续 0.8 秒静音认为语音结束 SILENCE_FRAMES int(SILENCE_DURATION / FRAME_DURATION) MIN_SPEECH_DURATION 0.3 # 最少 0.3 秒才认为是有效语音 MIN_SPEECH_FRAMES int(MIN_SPEECH_DURATION / FRAME_DURATION) # ChatTTS 配置 CHATTTS_MODEL_PATH D:/Pycharm/Project/ChatTTS/chattts_models #填写本地模型的地址 # 初始化 print(加载 ChatTTS 模型...) chat ChatTTS.Chat() chat.load(sourcecustom, custom_pathCHATTTS_MODEL_PATH, compileTrue) print(✅ ChatTTS 加载完成) print(加载 Whisper 模型...) whisper_model WhisperModel( model_size_or_pathWHISPER_MODEL_SIZE, deviceWHISPER_DEVICE, compute_typeCOMPUTE_TYPE, cpu_threadsCPU_THREADS, num_workersNUM_WORKERS ) print(✅ Whisper 加载完成) # 队列与同步信号 audio_queue queue.Queue(maxsize30) text_queue queue.Queue() play_queue queue.Queue() stop_event threading.Event() playback_active threading.Event() # 播放期间为 True录音线程丢弃音频防回环 # 线程函数 # ---- 录音线程实时 VAD 切分语音停顿即送队列 ---- def record_microphone(): p pyaudio.PyAudio() stream p.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferFRAME_SIZE ) print( 录音开始语音停顿时自动切分CtrlC 停止...) speech_buffer [] # 累积语音帧 silence_count 0 # 连续静音帧计数 speech_count 0 # 当前语音段的总语音帧数 try: while not stop_event.is_set(): try: audio_bytes stream.read(FRAME_SIZE, exception_on_overflowFalse) except Exception as e: print(f录音读取错误: {e}) break audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 energy np.sqrt(np.mean(audio_np ** 2)) if len(audio_np) 0 else 0.0 if energy ENERGY_THRESHOLD: # 语音帧 silence_count 0 speech_buffer.append(audio_np) speech_count 1 else: # 静音帧 if speech_count 0: speech_buffer.append(audio_np) silence_count 1 if silence_count SILENCE_FRAMES: # 语音结束 if speech_count MIN_SPEECH_FRAMES: if playback_active.is_set(): # 正在播放扬声器输出丢弃以防止回环 print( 正在播放丢弃本段录音防回环) else: full_speech np.concatenate(speech_buffer) speech_bytes (np.clip(full_speech, -1.0, 1.0) * 32767).astype(np.int16).tobytes() try: audio_queue.put(speech_bytes, timeout1) except queue.Full: print(⚠️ 音频队列已满丢弃该段语音) else: # 语音太短丢弃 pass # 重置状态 speech_buffer [] silence_count 0 speech_count 0 # else: 纯静音不处理 except KeyboardInterrupt: pass finally: stream.stop_stream() stream.close() p.terminate() print( 录音线程结束) # ---- 转录线程 ---- def transcribe_worker(): while not stop_event.is_set(): try: audio_bytes audio_queue.get(timeout0.5) except queue.Empty: continue if audio_bytes is None: break try: audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 if len(audio_np) int(SAMPLE_RATE * 0.2): audio_queue.task_done() continue segments, _ whisper_model.transcribe( audio_np, languagezh, beam_size5, vad_filterTrue, vad_parametersdict(min_silence_duration_ms500), temperature0.0 ) full_text for seg in segments: try: simplified convert(seg.text, zh-cn) except Exception: simplified seg.text full_text simplified if full_text.strip(): print(f 识别文本: {full_text}) text_queue.put(full_text) except Exception as e: print(f❌ 转录错误: {e}) finally: audio_queue.task_done() # ---- 合成线程 ---- def synthesize_worker(): while not stop_event.is_set(): try: text text_queue.get(timeout0.5) except queue.Empty: continue if text is None: break print(f 开始合成: {text}) try: wavs chat.infer([text]) if not wavs or len(wavs) 0: print(⚠️ ChatTTS 返回空结果) text_queue.task_done() continue audio_data np.asarray(wavs[0]) # 确保一维 if audio_data.ndim 1: audio_data audio_data.flatten() # 归一化并裁剪到 [-1, 1]防止 int16 溢出导致爆音/播放中断 max_val np.max(np.abs(audio_data)) if max_val 1.0: audio_data audio_data / max_val audio_int16 (np.clip(audio_data, -1.0, 1.0) * 32767).astype(np.int16) if len(audio_int16) 0: play_queue.put(audio_int16) print( 合成完成加入播放队列) else: print(⚠️ 合成结果为空) except Exception as e: print(f❌ 合成错误: {e}) finally: text_queue.task_done() # ---- 播放线程播放前加锁防回环播放后解锁 ---- def play_worker(): while not stop_event.is_set(): try: audio_int16 play_queue.get(timeout0.5) except queue.Empty: continue if audio_int16 is None: break try: if len(audio_int16) 0: print(⚠️ 播放音频为空跳过) play_queue.task_done() continue audio_float audio_int16.astype(np.float32) / 32768.0 # 确保一维 if audio_float.ndim 1: audio_float audio_float.flatten() duration len(audio_float) / 24000.0 print(f 开始播放时长 {duration:.2f} 秒) # 上锁通知录音线程丢弃扬声器输出的回声 playback_active.set() sd.play(audio_float, 24000) sd.wait() playback_active.clear() # 播放结束后额外等待让扬声器余音消散 time.sleep(1) print(✅ 播放完成) except Exception as e: print(f❌ 播放错误: {e}) playback_active.clear() finally: play_queue.task_done() # 启动所有线程 threads [] t_rec threading.Thread(targetrecord_microphone) t_rec.start() threads.append(t_rec) t_trans threading.Thread(targettranscribe_worker) t_trans.start() threads.append(t_trans) t_synth threading.Thread(targetsynthesize_worker) t_synth.start() threads.append(t_synth) t_play threading.Thread(targetplay_worker) t_play.start() threads.append(t_play) # 主线程等待退出CtrlC try: while True: time.sleep(0.5) except KeyboardInterrupt: print(\n 正在退出...) # 清理同步信号 playback_active.clear() stop_event.set() # 发送哨兵值让阻塞在 get() 的线程退出 audio_queue.put(None) text_queue.put(None) play_queue.put(None) # 等待各线程结束最多等 5 秒 for t in threads: t.join(timeout5) print(✅ 已退出)四、实现语音控制打开应用1、下载语音转文字模型import os # 环境变量必须在导入模型库之前设置 # Hugging Face 镜像站。国内直连 huggingface.co 会超时WinError 10060 # 走镜像站可以正常下载和校验。 os.environ[HF_ENDPOINT] https://hf-mirror.com # 模型缓存根目录。注意 # 1. 指向 hub 的【父目录】不是 hub 本身 # 2. 必须用 / 作分隔符不能用 \否则会被当成转义字符 # 3. 实际模型会存到 .../语音转文字/hub/models--Systran--faster-whisper-small/ os.environ[HF_HOME] C:/Users/chenzl/Desktop/gitee/ChatTTS/语音转文字 from faster_whisper import WhisperModel # 加载 / 下载模型 # 这行执行时先去 HF_HOME/hub/ 找模型找不到就通过 HF_ENDPOINT 下载 # 下载完自动加载到内存。所以这行跑完模型文件已经落到硬盘上了。 model WhisperModel( model_size_or_pathsmall, # 模型规格或本地路径。 # tiny ~39M 最快准确率最低 # base ~74M 较快 # small ~244M 速度和准确率平衡你现在用的 # medium ~769M 更准明显更慢 # large-v3 ~1.5G 最准CPU 上很慢 # 也可传本地文件夹路径如 D:/models/faster-whisper-small devicecpu, # 运行设备 # cpu 普通 CPU 推理 # cuda 需要 NVIDIA GPU CUDA 环境速度大幅提升 compute_typefloat32, # 计算精度影响速度 / 内存 / 准确率 # int8 内存最小速度较快精度略损CPU 上常用 # float16 半精度GPU 上最快精度损失小 # float32 单精度最准确内存和计算开销最大 cpu_threads4, # CPU 推理线程数。一般设为物理核心数太多反而变慢 num_workers1, # 并发处理的工作线程数。单条音频转写保持 1 即可 ) print(下载完成)2、下载字符串向量化模型import os # 环境变量必须在导入模型库之前设置 # Hugging Face 官方域名在国内经常连不上会报 WinError 10060连接超时。 # 设置镜像站后所有对 huggingface.co 的请求都会被重定向到 hf-mirror.com。 os.environ[HF_ENDPOINT] https://hf-mirror.com # 模型缓存根目录。注意三点 # 1. 指向 hub 的【父目录】不是 hub 本身。 # 实际模型会存到 .../字符转向量打开软件/hub/models--BAAI--bge-m3/ # 2. 推荐用 / 作分隔符这里用 \\ 是因为 Python 字符串里 \ 是转义字符 # 写成 C:\Users 会把 \U 当成 Unicode 转义而出错所以双写 \\。 # 3. 这个变量必须在 import 模型库之前设置导入后再设就不生效了。 os.environ[HF_HOME] C:\\Users\\chenzl\\Desktop\\gitee\\ChatTTS\\字符转向量打开软件 from sentence_transformers import SentenceTransformer # 加载 / 下载 BGE-M3 # 这行执行时先去 HF_HOME/hub/ 找 BAAI/bge-m3 # 找不到 → 通过 HF_ENDPOINT 镜像站下载 → 存到 HF_HOME 并加载到内存 # 已存在 → 直接读本地不联网。 model SentenceTransformer( BAAI/bge-m3 # 模型标识格式为 组织名/模型名。 # BAAI 北京智源人工智能研究院 # bge-m3 多语言嵌入模型中文效果强向量维度 1024 # 也可传本地完整路径如 rD:\models\bge-m3 # 那样就直接读本地完全不碰 HF_HOME、不联网。 )3、最终实现import warnings warnings.filterwarnings(ignore, categoryUserWarning, modulezhconv) import threading import queue import subprocess import webbrowser import numpy as np import pyaudio from faster_whisper import WhisperModel from zhconv import convert from sentence_transformers import SentenceTransformer # 1. 配置 SAMPLE_RATE 16000 CHUNK_DURATION 10 SIM_THRESHOLD 0.7 WHISPER_MODEL_PATH rC:\Users\chenzl\Desktop\gitee\ChatTTS\语音转文字\hub\models--Systran--faster-whisper-small\snapshots\536b0662742c02347bc0e980a01041f333bce120 BGE_MODEL_PATH rC:\Users\chenzl\Desktop\gitee\ChatTTS\字符转向量打开软件\hub\models--BAAI--bge-m3\snapshots\5617a9f61b028005a4858fdac845db406aefb181 # 2. 加载 Whisper print(正在加载 Whisper 模型...) whisper_model WhisperModel( model_size_or_pathWHISPER_MODEL_PATH, devicecpu, compute_typefloat32, cpu_threads4, num_workers1, ) # 3. 加载 BGE-M3 print(正在加载 BGE-M3 模型...) embed_model SentenceTransformer(BGE_MODEL_PATH) # 4. 意图库 intents { open_deepseek: [打开deepseek, 启动ai助手, 我要搜索问题, 把网页打开, 打开网页], open_weixin: [打开微信, 启动微信, 我要聊天, 打开WeChat], open_bilibili: [打开哔哩哔哩, 启动B站, 我要看视频, 打开B站], unknown: [ 今天天气怎么样, 你叫什么名字, 现在几点了, 帮我查一下资料, 这个多少钱, 明天要下雨吗, 随便聊聊, 你好, 谢谢, ], } print(正在向量化意图库...) intent_vectors {} for name, phrases in intents.items(): vecs embed_model.encode(phrases, normalize_embeddingsTrue) intent_vectors[name] vecs.mean(axis0) print(意图库准备完成。) # 5. 匹配函数 def match_intent(text, thresholdSIM_THRESHOLD): q embed_model.encode([text], normalize_embeddingsTrue)[0] best, best_score None, -1 for name, vec in intent_vectors.items(): score float(np.dot(q, vec)) if score best_score: best, best_score name, score return (best, best_score) if best_score threshold else (None, best_score) # 6. 动作 def open_deepseek(): webbrowser.open(https://chat.deepseek.com/) def open_weixin(): subprocess.Popen([rD:\Weixin\Weixin.exe]) def open_bilibili(): subprocess.Popen([rD:\bilibili\哔哩哔哩.exe]) actions { open_deepseek: open_deepseek, open_weixin: open_weixin, open_bilibili: open_bilibili, } # 7. 语音转写线程 audio_queue queue.Queue() # 存放待转写的音频 text_queue queue.Queue() # 存放转写好的文字 def transcribe_worker(): 后台线程不断从 audio_queue 取音频转写后把文字放进 text_queue while True: audio_bytes audio_queue.get() if audio_bytes is None: break try: audio_np np.frombuffer(audio_bytes, dtypenp.int16).astype(np.float32) / 32768.0 segments, info whisper_model.transcribe( audio_np, languagezh, beam_size20, vad_filterTrue, vad_parametersdict(min_silence_duration_ms500), temperature0.0, ) pieces [] for segment in segments: simplified convert(segment.text, zh-cn) pieces.append(simplified) text .join(pieces).strip().replace( , ) if text: text_queue.put(text) except Exception as e: print(f[转写出错] {e}) # 8. 启动录音 audio pyaudio.PyAudio() stream audio.open( formatpyaudio.paInt16, channels1, rateSAMPLE_RATE, inputTrue, frames_per_bufferint(SAMPLE_RATE * CHUNK_DURATION), ) # 启动转写线程 transcribe_thread threading.Thread(targettranscribe_worker, daemonTrue) transcribe_thread.start() print(监听中... (CtrlC 退出)) # 9. 主循环 def record_loop(): 后台线程不断录音把音频塞进 audio_queue try: while True: audio_bytes stream.read(int(SAMPLE_RATE * CHUNK_DURATION), exception_on_overflowFalse) audio_queue.put(audio_bytes) except Exception as e: print(f[录音停止] {e}) record_thread threading.Thread(targetrecord_loop, daemonTrue) record_thread.start() try: while True: # 从 text_queue 取转写结果阻塞等待 try: text text_queue.get(timeout0.5) except queue.Empty: continue print(f\n️ 识别{text}) intent, score match_intent(text) if intent and intent ! unknown: actions[intent]() print(f✅ 执行{intent} (置信度 {score:.3f})) else: print(f❌ 没听懂 (最高分 {score:.3f})) except KeyboardInterrupt: print(\n退出) finally: audio_queue.put(None) # 通知转写线程退出 stream.stop_stream() stream.close() audio.terminate()