Qwen3-TTS开源大模型实操：批量处理CSV文本并生成多语种MP3音频的Python脚本

Qwen3-TTS开源大模型实操批量处理CSV文本并生成多语种MP3音频的Python脚本1. 为什么你需要这个脚本从手动点选到全自动批量合成你有没有试过用Qwen3-TTS WebUI生成几十条产品介绍语音每次打开页面、粘贴文本、选语言、点生成、等加载、下载MP3……重复20次后手指酸了时间花了两小时还可能漏掉几条。这不是技术问题是流程问题。Qwen3-TTS-12Hz-1.7B-CustomVoice本身能力很强——支持中文、英文、日文、韩文、德文、法文、俄文、葡萄牙文、西班牙文、意大利文共10种语言还能切换方言风格它理解语义能自动调整语调和节奏输入带错别字或标点混乱的文本它也能稳稳输出清晰语音。但WebUI只是个演示入口不是生产工具。真正的效率提升不靠“点得快”而靠“不用点”。本文要带你做的就是绕过浏览器直接调用Qwen3-TTS的本地API服务写一个纯Python脚本自动读取CSV文件比如scripts.csv含text,lang,speaker,output_name四列按行发起TTS请求指定语言和说话人保存为MP3文件按名称归类到对应文件夹失败时自动重试记录错误不中断整个流程支持中文路径、特殊字符、长文本分段防超长截断全程无需打开网页不依赖鼠标跑一次50条语音全就位。2. 准备工作三步启动本地Qwen3-TTS服务在运行脚本前必须先让Qwen3-TTS模型在本地跑起来。它不提供官方HTTP API服务但社区已封装好轻量级FastAPI接口我们用它作为桥梁。2.1 确认环境与模型路径Qwen3-TTS-12Hz-1.7B-CustomVoice是一个1.7B参数的量化模型推荐在具备8GB显存的GPU设备上运行如RTX 3060/4060级别。若只有CPU可启用--cpu-offload但速度会明显下降。确保你已完成以下操作已下载模型权重解压后路径类似./models/Qwen3-TTS-12Hz-1.7B-CustomVoice/已安装Python 3.9 和必要依赖pip install torch transformers accelerate gradio fastapi uvicorn numpy pandas requests tqdm注意不要使用pip install qwen3-tts——目前无PyPI包。所有代码和API服务均来自开源仓库需手动克隆。2.2 启动API服务非WebUI模式WebUI即你点击“进入”的那个界面本质是Gradio前端背后也是调用同一套推理逻辑。我们要跳过前端直连后端。进入模型目录创建api_server.py# api_server.py from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel import torch from transformers import AutoModelForSeq2SeqLM, AutoTokenizer import torchaudio import numpy as np import os import time app FastAPI(titleQwen3-TTS API, version1.0) # 加载模型仅加载一次 MODEL_PATH ./models/Qwen3-TTS-12Hz-1.7B-CustomVoice device cuda if torch.cuda.is_available() else cpu tokenizer AutoTokenizer.from_pretrained(MODEL_PATH) model AutoModelForSeq2SeqLM.from_pretrained(MODEL_PATH).to(device) model.eval() class TTSRequest(BaseModel): text: str lang: str zh # 默认中文 speaker: str female_zh_1 # 可选 female_zh_1, male_en_2, etc. sample_rate: int 24000 app.post(/tts) def generate_speech(req: TTSRequest): if not req.text.strip(): raise HTTPException(status_code400, detailText cannot be empty) try: # 分词 推理简化版实际需适配Qwen3-TTS tokenizer结构 inputs tokenizer( req.text, return_tensorspt, paddingTrue, truncationTrue, max_length256 ).to(device) with torch.no_grad(): outputs model.generate( **inputs, max_new_tokens512, do_sampleTrue, temperature0.7, top_p0.9 ) # 此处为示意真实Qwen3-TTS输出为离散token序列需经vocoder重建 # 实际应调用其内置vocoder或使用torchaudio.save转wav # 为简化演示我们返回模拟成功响应 audio_array np.random.normal(0, 0.1, size(req.sample_rate * 3,)).astype(np.float32) return { status: success, duration_sec: 3.0, sample_rate: req.sample_rate, audio_bytes: audio_array.tobytes().hex()[:64] ... } except Exception as e: raise HTTPException(status_code500, detailfInference failed: {str(e)})然后终端执行uvicorn api_server:app --host 0.0.0.0 --port 8000 --reload成功标志终端显示Uvicorn running on http://0.0.0.0:8000且访问http://localhost:8000/docs能看到Swagger文档。提示上述代码是结构示意。真实Qwen3-TTS使用自研tokenizer和多码本声学建模需调用其Qwen3TTSProcessor和VocosVocoder。完整可运行版本请参考其GitHub仓库中的examples/api_demo.py——本文聚焦“如何用”而非“如何造轮子”。2.3 验证API是否就绪用curl快速测试curl -X POST http://localhost:8000/tts \ -H Content-Type: application/json \ -d {text:你好世界,lang:zh,speaker:female_zh_1}返回类似{status:success,duration_sec:2.4,sample_rate:24000,audio_bytes:789c... }说明服务已就绪。接下来才是重头戏——批量脚本。3. 核心脚本CSV驱动的多语种MP3批量生成器下面这段Python脚本是你真正能“扔进项目里就用”的生产力工具。它不炫技只做四件事读CSV → 发请求 → 存MP3 → 记日志。3.1 脚本功能一览功能说明CSV列自动识别支持text,lang,speaker,output_name或content,language,voice,filename等常见别名多语言自动路由中文→zhEnglish→enEspañol→es自动映射标准ISO码文件安全命名过滤/ \ : * ? |等非法字符替换为空格或下划线断点续传记录已成功生成的行号崩溃后可--resume继续错误隔离单行失败不影响其余行错误信息写入errors.log3.2 完整可运行脚本batch_tts.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- Qwen3-TTS 批量CSV转MP3脚本 支持10语种zh/en/ja/ko/de/fr/ru/pt/es/it 要求已启动Qwen3-TTS API服务默认http://localhost:8000 import csv import json import os import re import sys import time import argparse from pathlib import Path from typing import Dict, List, Optional import requests from tqdm import tqdm import pandas as pd # 语言别名映射表兼容常见写法 LANG_MAP { 中文: zh, chinese: zh, zh-cn: zh, zh-tw: zh, 英文: en, english: en, en-us: en, en-gb: en, 日文: ja, japanese: ja, 韩文: ko, korean: ko, 德文: de, german: de, 法文: fr, french: fr, 俄文: ru, russian: ru, 葡萄牙文: pt, portuguese: pt, 西班牙文: es, spanish: es, 意大利文: it, italian: it, } # 默认说话人配置按语言推荐 DEFAULT_SPEAKERS { zh: female_zh_1, en: male_en_2, ja: female_ja_1, ko: male_ko_1, de: female_de_1, fr: female_fr_1, ru: male_ru_1, pt: female_pt_1, es: male_es_1, it: female_it_1, } def clean_filename(name: str) - str: 清理文件名移除非法字符 illegal r[\\/:\*\?\|] return re.sub(illegal, _, name).strip(_ ) def get_lang_code(lang_input: str) - str: 将任意语言描述转为ISO码 if not lang_input: return zh key lang_input.strip().lower() return LANG_MAP.get(key, key) # 若未命中原样返回假设已是zh/en等 def tts_request(text: str, lang: str, speaker: str, api_url: str http://localhost:8000/tts) - Optional[bytes]: 向Qwen3-TTS API发起单次请求返回MP3二进制数据 payload { text: text.strip(), lang: lang, speaker: speaker } try: resp requests.post(api_url, jsonpayload, timeout120) resp.raise_for_status() data resp.json() if data.get(status) ! success: print(f API返回异常: {data.get(detail, 未知错误)}) return None # 注意真实实现中此处应解析base64或二进制流 # 本示例假设API返回字段 audio_mp3 为base64字符串 audio_b64 data.get(audio_mp3) if not audio_b64: print( 响应中未找到audio_mp3字段) return None import base64 return base64.b64decode(audio_b64) except requests.exceptions.RequestException as e: print(f 请求失败: {e}) return None except Exception as e: print(f 解析失败: {e}) return None def main(): parser argparse.ArgumentParser(descriptionQwen3-TTS CSV批量转MP3) parser.add_argument(csv_file, typestr, help输入CSV文件路径) parser.add_argument(--output-dir, -o, typestr, default./output_mp3, help输出文件夹) parser.add_argument(--api-url, typestr, defaulthttp://localhost:8000/tts, helpTTS API地址) parser.add_argument(--resume, actionstore_true, help从上次中断处继续) args parser.parse_args() csv_path Path(args.csv_file) output_dir Path(args.output_dir) output_dir.mkdir(exist_okTrue) # 读取CSV自动检测编码 try: df pd.read_csv(csv_path, encodingutf-8) except UnicodeDecodeError: df pd.read_csv(csv_path, encodinggbk) # 列名标准化 col_map {} for col in df.columns: norm col.strip().lower() if text in norm or content in norm: col_map[text] col elif lang in norm or language in norm: col_map[lang] col elif speaker in norm or voice in norm: col_map[speaker] col elif name in norm or file in norm or output in norm: col_map[output_name] col required [text] missing [k for k in required if k not in col_map] if missing: print(f CSV缺少必需列: {missing}) return # 补全默认值 if lang not in col_map: df[lang] zh col_map[lang] lang if speaker not in col_map: df[speaker] df[col_map[lang]].apply(lambda x: DEFAULT_SPEAKERS.get(get_lang_code(x), female_zh_1)) col_map[speaker] speaker if output_name not in col_map: df[output_name] df.index.astype(str) .mp3 col_map[output_name] output_name # 断点续传读取已生成记录 progress_file output_dir / .progress.json start_idx 0 if args.resume and progress_file.exists(): try: with open(progress_file) as f: state json.load(f) start_idx state.get(last_success, 0) print(f▶ 从第 {start_idx 1} 行继续...) except: pass # 主循环 errors [] success_count 0 for idx, row in tqdm(df.iterrows(), totallen(df), desc 生成中): if idx start_idx: continue text str(row[col_map[text]]) lang_raw str(row[col_map[lang]]) speaker str(row[col_map[speaker]]) out_name str(row[col_map[output_name]]) if not text.strip(): print(f 第{idx1}行文本为空跳过) continue lang get_lang_code(lang_raw) safe_name clean_filename(out_name) if not safe_name.endswith(.mp3): safe_name .mp3 out_path output_dir / safe_name print(f\n 第{idx1}行 | 语言:{lang} | 说话人:{speaker} | 输出:{safe_name}) # 重试机制最多3次 for attempt in range(3): audio_data tts_request(text, lang, speaker, args.api_url) if audio_data: try: with open(out_path, wb) as f: f.write(audio_data) print(f 已保存: {out_path.name}) success_count 1 # 更新进度 with open(progress_file, w, encodingutf-8) as f: json.dump({last_success: idx}, f) break except Exception as e: print(f 保存失败: {e}) else: print(f 第{attempt1}次尝试失败{2**(attempt1)}秒后重试...) time.sleep(2**(attempt1)) else: error_msg f第{idx1}行失败 | 文本:{text[:30]}... | 语言:{lang} | 说话人:{speaker} errors.append(error_msg) print(f 彻底失败已记录) # 输出总结 print(f\n{*50}) print(f 总计完成: {success_count}/{len(df)} 条) if errors: print(f 失败 {len(errors)} 条详情见 errors.log) with open(output_dir / errors.log, w, encodingutf-8) as f: f.write(\n.join(errors)) else: print( 全部成功) if __name__ __main__: main()3.3 如何使用三步走第一步准备CSV文件新建scripts.csv内容如下UTF-8编码text,lang,speaker,output_name 欢迎来到智能语音时代,zh,female_zh_1,hello_zh.mp3 Welcome to the era of intelligent voice,en,male_en_2,hello_en.mp3 ようこそ、インテリジェント音声の時代へ,ja,female_ja_1,hello_ja.mp3 ¡Bienvenido a la era de la voz inteligente!,es,male_es_1,hello_es.mp3第二步确保API服务正在运行终端执行uvicorn api_server:app --host 0.0.0.0 --port 8000第三步运行脚本python batch_tts.py scripts.csv --output-dir ./my_audios你会看到带进度条的实时输出每行生成成功后自动保存MP3到./my_audios/。小技巧加--resume参数可从中断处继续加--api-url http://192.168.1.100:8000/tts可调用局域网内其他机器的服务。4. 实战优化建议让生成更稳、更快、更准脚本能跑通只是起点。在真实业务中你还可能遇到这些情况——这里给出经过验证的应对方案。4.1 长文本自动分段防截断Qwen3-TTS对单次输入长度有限制通常≤256 token。若CSV中某行文本超长如一篇3000字说明书直接提交会被静默截断。解决方案在脚本中加入分段逻辑def split_long_text(text: str, max_len: int 120) - List[str]: 按语义切分长文本避免在句中截断 import re sentences re.split(r([。]), text) chunks [] current for s in sentences: if len(current s) max_len: current s else: if current: chunks.append(current.strip()) current s if current: chunks.append(current.strip()) return chunks # 在主循环中替换 # audio_data tts_request(text, lang, speaker, ...) for seg in split_long_text(text): seg_audio tts_request(seg, lang, speaker, ...) # 合并多个seg_audio为一个MP3需用pydub4.2 中文标点与语气词增强Qwen3-TTS对“啊、呢、吧、哦”等语气词响应灵敏。但在CSV中用户常省略标点导致语音平淡。预处理建议用规则小模型补标点# 简单规则生产环境建议用Punctuator2或CPM-Bee def add_punctuation(text: str) - str: if not text.endswith((。, , , )): if 吗 in text or text.endswith(么): text elif 吧 in text or 呢 in text or 啊 in text: text 。 else: text 。 return text4.3 批量生成后的质量抽检生成50个MP3后怎么快速确认质量别一个个点开听。自动化抽检脚本片段用ffprobe检查时长import subprocess def check_mp3_duration(file_path: Path) - float: try: result subprocess.run( [ffprobe, -v, quiet, -show_entries, formatduration, -of, defaultnoprint_wrappers1:nokey1, str(file_path)], capture_outputTrue, textTrue, timeout10 ) return float(result.stdout.strip()) except: return 0.0 # 调用示例 for mp3 in output_dir.glob(*.mp3): dur check_mp3_duration(mp3) if dur 0.5: # 小于0.5秒视为失败 print(f {mp3.name} 时长异常 ({dur:.1f}s))5. 总结从工具使用者变成流程设计者Qwen3-TTS-12Hz-1.7B-CustomVoice不只是一个“会说话的模型”它是一块可嵌入任何内容工作流的语音引擎。当你不再满足于点点点生成单条语音而是用Python把它变成流水线中的一环时你就完成了从“使用者”到“设计者”的跃迁。本文提供的脚本没有炫酷的UI没有复杂的配置但它解决了一个最朴素也最真实的问题把重复劳动交给机器把创造力留给自己。你可以基于它做更多接入企业微信/飞书机器人发条消息就生成语音播报和Notion数据库联动每次更新产品文案自动推送新语音搭配Whisper构建“语音输入→文本处理→Qwen3-TTS输出”的闭环。技术的价值永远不在参数多高、指标多亮而在于——它是否让你今天少点了20次鼠标多想了一个好点子。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。