LangChain调用本地TTS：构建离线可用的AI助手

📌 背景与需求：为什么需要离线语音合成？

在当前大模型驱动的智能助手应用中，语音交互已成为提升用户体验的关键环节。然而，大多数语音合成（TTS）服务依赖云端API（如阿里云、百度语音、讯飞等），存在三大痛点：

隐私风险：敏感对话内容需上传至第三方服务器
网络依赖：无网环境下无法使用，限制了工业、车载等场景落地
延迟不可控：网络波动导致响应延迟，影响交互流畅性

为此，构建一个本地化、可集成、支持多情感表达的中文TTS系统，成为实现真正“离线AI助手”的关键一步。

本文将围绕ModelScope 的 Sambert-Hifigan 中文多情感语音合成模型，结合 Flask 封装 API，并重点讲解如何通过LangChain 框架调用本地 TTS 服务，打造一套完整的离线语音交互链路。

🎙️ 核心技术选型：Sambert-Hifigan 为何适合本地部署？

技术背景与优势分析

Sambert-Hifigan 是魔搭（ModelScope）平台推出的端到端中文语音合成方案，由两个核心模块构成：

Sambert：声学模型，负责将文本转换为梅尔频谱图
HifiGan：声码器，将频谱图还原为高质量音频波形

该模型支持多种情感风格（如开心、悲伤、愤怒、平静等），显著提升语音自然度和表现力，适用于虚拟助手、有声书、教育机器人等场景。

✅为何选择它作为本地TTS引擎？
支持标准中文普通话及常见方言变体
多情感控制可通过参数调节，无需额外训练
模型体积适中（约300MB），可在消费级CPU上运行
社区活跃，预训练权重开箱即用

🛠️ 环境封装与服务部署：从模型到Web服务

已解决的核心依赖问题

原始 ModelScope 示例代码在现代Python环境中常因版本冲突报错。我们对以下关键依赖进行了深度兼容性修复：

| 包名 | 固定版本 | 说明 | |------|----------|------| |datasets| 2.13.0 | 避免与tokenizers不兼容导致加载失败 | |numpy| 1.23.5 | 兼容旧版 scipy 和 torch 操作 | |scipy| <1.13.0 | 防止signal.resample接口变更引发错误 | |torch| >=1.13.0 | 确保支持 traced model 推理 |

经过测试验证，上述组合可稳定运行于 Python 3.9+ 环境，杜绝“ImportError”或“RuntimeError”类异常。

Flask WebUI + API 双模式服务架构

项目采用轻量级 Flask 框架封装推理逻辑，提供两种访问方式：

1. 图形化界面（WebUI）

用户可通过浏览器直接输入文本，实时生成并播放语音，支持.wav文件下载。

2. 标准 HTTP API

便于程序化调用，接口设计如下：

POST /tts Content-Type: application/json { "text": "你好，我是你的本地AI助手。", "emotion": "happy", // 可选：neutral, sad, angry, surprised 等 "speed": 1.0 // 语速调节 [0.8 ~ 1.5] }

返回结果包含音频 Base64 编码或文件路径：

{ "status": "success", "audio_path": "/static/audio/output_20250405.wav", "download_url": "http://localhost:5000/static/audio/output_20250405.wav" }

🧩 实现细节：Flask服务核心代码解析

以下是服务端关键实现逻辑（精简版）：

# app.py from flask import Flask, request, jsonify, render_template import os import numpy as np import soundfile as sf from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks app = Flask(__name__) app.config['STATIC_FOLDER'] = 'static/audio' # 初始化TTS流水线 tts_pipeline = pipeline( task=Tasks.text_to_speech, model='damo/speech_sambert-hifigan_tts_zh-cn_pretrain_16k')

@app.route('/tts', methods=['POST']) def tts(): data = request.json text = data.get('text', '').strip() emotion = data.get('emotion', 'neutral') speed = float(data.get('speed', 1.0)) if not text: return jsonify({"error": "文本不能为空"}), 400 try: # 执行推理 result = tts_pipeline(input=text, voice=emotion, speed=speed) audio_data = result["output_wav"] # 保存音频 filename = f"output_{int(time.time())}.wav" filepath = os.path.join(app.config['STATIC_FOLDER'], filename) sf.write(filepath, audio_data, 16000) return jsonify({ "status": "success", "audio_path": filepath, "download_url": f"/static/audio/{filename}" }) except Exception as e: return jsonify({"error": str(e)}), 500

@app.route('/') def index(): return render_template('index.html') # 提供可视化界面

🔍注意点：
使用soundfile保存.wav文件时需指定采样率16000
voice参数实际映射到模型的情感嵌入向量，不同模型支持的情感标签略有差异
建议启用缓存机制避免重复合成相同文本

🤖 LangChain 集成：让AI助手“开口说话”

LangChain 本身不内置TTS功能，但我们可以通过自定义OutputParser或CallbackHandler，在其输出阶段触发本地TTS服务。

方案设计思路

我们将实现一个TTSCallback类，在每次 LLM 生成回复后自动调用本地 TTS API 并播放语音。

步骤一：封装TTS客户端

# tts_client.py import requests import pygame import tempfile import time class LocalTTSClient: def __init__(self, api_url="http://localhost:5000/tts"): self.api_url = api_url def speak(self, text: str, emotion="neutral", speed=1.0): try: response = requests.post(self.api_url, json={ "text": text, "emotion": emotion, "speed": speed }, timeout=30) if response.status_code == 200: result = response.json() audio_url = result.get("download_url") # 下载音频 audio_resp = requests.get(audio_url) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: f.write(audio_resp.content) temp_path = f.name # 播放语音 pygame.mixer.init() pygame.mixer.music.load(temp_path) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): time.sleep(0.1) os.unlink(temp_path) # 播放完成后删除临时文件 else: print(f"TTS请求失败: {response.json()}") except Exception as e: print(f"语音合成失败: {e}")

步骤二：集成至 LangChain 流程

# chain_with_tts.py from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_community.llms import Ollama # 示例使用本地Ollama模型 from tts_client import LocalTTSClient # 初始化组件 llm = Ollama(model="qwen:7b") tts_client = LocalTTSClient() # 定义提示模板 template = """你是一位贴心的AI助手，请用自然口语化中文回答用户问题。 用户问：{question} 请简洁回应。""" prompt = PromptTemplate.from_template(template) chain = LLMChain(llm=llm, prompt=prompt) # 交互循环 while True: user_input = input("你说：") if user_input.lower() in ['退出', 'quit', 'exit']: break # 调用LLM生成回复 response = chain.run(question=user_input) print(f"AI答：{response}") # 同步触发语音播报 emotion = "happy" if "?" not in user_input else "neutral" tts_client.speak(response.strip(), emotion=emotion, speed=1.1)

💡情感动态判断建议：
用户提问 →neutral
表达喜悦/赞美 →happy
请求帮助 →gentle
检测负面情绪关键词 →calm或sad

⚙️ 性能优化与工程实践建议

1. 推理加速技巧（CPU环境）

启用 JIT 编译：部分模型支持 TorchScript 导出，减少解释开销
批处理短句：将多个短文本拼接成一句合成，降低调用频率
预加载模型：服务启动时完成初始化，避免首次延迟过高

2. 音频播放稳定性增强

使用pygame替代playsound，避免线程阻塞问题
添加音量控制接口：pygame.mixer.music.set_volume(0.8)
异步播放：使用threading.Thread防止阻塞主对话流

3. 日志与错误降级机制

import logging logging.basicConfig(level=logging.INFO) # 在 speak 方法中添加重试机制 for i in range(3): try: # ... 调用API ... break except requests.ConnectionError: logging.warning("TTS服务未响应，尝试重启或跳过...") time.sleep(1) else: logging.error("TTS服务连续失败，已跳过语音播报")

🧪 实际使用流程演示

启动 Flask 服务：bash python app.py
服务默认监听http://localhost:5000
访问 WebUI：
打开浏览器 → 输入http://localhost:5000
在文本框输入：“今天天气真不错，适合出去散步。”
选择情感为“开心”，点击“开始合成语音”
等待1~3秒，即可在线播放或下载.wav文件
运行 LangChain 助手：bash python chain_with_tts.py
输入：“你好啊！”
AI回复：“你好！有什么我可以帮你的吗？” → 自动语音播报

🔄 架构全景图：离线AI助手完整链路

+------------------+ +--------------------+ +---------------------+ | 用户输入文本 | --> | 本地LLM (如Qwen) | --> | LangChain 处理逻辑 | +------------------+ +--------------------+ +----------+----------+ | v +-----------------------+ | 本地TTS (Sambert-Hifigan)| +----------+------------+ | v +------------------------+ | 播放语音 (pygame/flask) | +------------------------+

所有组件均运行于本地设备，无需联网即可完成“听-思-说”闭环，真正实现隐私安全、低延迟、高可用的离线AI助手。