最近在项目里用 ChatTTS 做长文本语音合成发现生成速度是个大问题。特别是处理几百上千字的文章时等待时间长得让人有点着急。经过一番折腾总算把生成速度提上来了这里把优化思路和实战代码整理一下希望能帮到有同样困扰的朋友。一、问题到底出在哪一开始我简单地用 ChatTTS 的原始接口去合成一篇 1000 字左右的文章发现耗时接近 30 秒。这显然没法用在需要快速响应的场景里。于是我做了一些 profiling性能分析发现几个主要的瓶颈自回归Autoregressive解码延迟TTS 模型生成音频是“一个字一个字”往外蹦的后一个 token 的生成依赖于前一个。这个过程是串行的文本越长耗时自然线性增长。单次推理Single Inference效率低默认的调用方式是一次只处理一个请求。当有多个请求过来时要么排队要么开多个进程前者慢后者资源消耗大。显存GPU Memory利用不充分每次推理都要重新加载模型计算图中间激活值Activation反复分配和释放产生了显存碎片也浪费了计算资源。计算精度冗余模型默认使用 FP32单精度浮点数进行计算这对于语音合成任务来说很多时候精度是过剩的却消耗了更多的计算量和显存。简单说就是“单线程、高精度、反复折腾”导致了速度上不去。二、我们的优化“三板斧”针对上面这些问题我主要从三个方向入手让模型能“批量干活”、让计算能“偷点懒”、让计算本身“轻量化”。1. 动态批处理Dynamic Batching—— 让等待合批一起出发核心思想是把短时间内到来的多个文本合成请求动态地组合成一个批次Batch送给模型计算。这能极大提升 GPU 的利用率。PyTorch 的DataLoader本身是为训练设计的我们需要改造一下用于推理服务。下面是一个简化版的动态批处理推理器实现import torch import threading from queue import Queue, Empty from typing import List, Optional, Tuple from dataclasses import dataclass from torch.utils.data import Dataset, DataLoader dataclass class TTSRequest: TTS 请求数据结构 text: str request_id: str # 其他参数如说话人ID、语速等 speaker_id: Optional[int] None class DynamicBatchDataset(Dataset): 模拟动态请求队列的数据集 def __init__(self, request_queue: Queue): self.queue request_queue def __len__(self): # 返回一个很大的数让 DataLoader 持续等待 return 1000000 def __getitem__(self, idx): try: # 设置超时避免无限阻塞 request self.queue.get(timeout0.1) return request except Empty: # 队列为空时返回 Nonecollate_fn 会处理 return None def collate_fn(batch: List[Optional[TTSRequest]]) - Tuple[List[TTSRequest], List[str]]: 整理批次数据过滤掉 None valid_requests [item for item in batch if item is not None] texts [req.text for req in valid_requests] return valid_requests, texts class TTSBatchInferenceEngine: TTS 批量推理引擎 def __init__(self, model, batch_size: int 8, max_wait_ms: int 50): self.model model self.model.eval() # 切换到评估模式 self.batch_size batch_size self.request_queue Queue() self.dataset DynamicBatchDataset(self.request_queue) # 使用 DataLoader 进行批处理num_workers0 因为在同一个进程 self.dataloader DataLoader( self.dataset, batch_sizebatch_size, collate_fncollate_fn, num_workers0, ) self.dataloader_iter iter(self.dataloader) self.lock threading.Lock() def add_request(self, request: TTSRequest): 客户端调用此方法添加请求 self.request_queue.put(request) def run_batch_inference(self) - List[Tuple[str, torch.Tensor]]: 执行一批推理返回 (request_id, audio_tensor) 列表 with torch.no_grad(), self.lock: # 加锁保证线程安全 try: batch_requests, batch_texts next(self.dataloader_iter) except StopIteration: # 重置迭代器 self.dataloader_iter iter(self.dataloader) return [] if not batch_texts: return [] # 这里是核心将多个文本向量化后堆叠成批次Tensor # 假设 text_to_tensor 是你的文本预处理函数 batch_tensors [self._text_to_tensor(text) for text in batch_texts] padded_batch torch.nn.utils.rnn.pad_sequence(batch_tensors, batch_firstTrue) # 模型前向传播一次性处理整个批次 batch_audio self.model(padded_batch) # 将结果拆解对应回每个请求 results [] # 注意这里需要根据实际模型输出和填充方式正确截取每个样本的音频 for i, req in enumerate(batch_requests): # audio_length 需要根据实际模型输出或原始文本长度推断 audio_length self._get_audio_length_for_sample(i, batch_tensors, batch_audio) single_audio batch_audio[i, :audio_length] results.append((req.request_id, single_audio)) return results def _text_to_tensor(self, text: str) - torch.Tensor: 将文本转换为模型输入张量示例函数 # 这里应实现你的文本 tokenization 和向量化逻辑 # 例如使用模型的 tokenizer # return torch.tensor([...]) pass def _get_audio_length_for_sample(self, idx: int, input_tensors: List[torch.Tensor], output_tensor: torch.Tensor) - int: 根据输入确定每个样本输出的音频长度示例函数 # 这是一个复杂点需要根据模型特性实现。 # 一种常见方法是模型输出与某个输入维度相关或者模型本身会输出长度信息。 # 这里返回一个固定值仅为示例。 return output_tensor.size(1) # 假设所有样本输出长度相同通常不成立关键点我们用Queue接收实时请求用DataLoader来组织批次的获取逻辑。collate_fn负责将一批零散的请求对象整理成模型可接受的批次化张量主要是文本的 padding 和对齐。通过设置DataLoader的batch_size和timeout逻辑实现了“攒够一批就推理”或“等待超时即使不满一批也推理”的动态策略。注意TTS 模型输出的音频长度可能各不相同pad_sequence用于输入输出后需要根据实际长度信息如果模型提供或通过其他方式将每个样本的音频正确切分出来这是实现中的难点。2. KV Cache 共享 —— 让计算能“偷懒”在自回归生成中每次生成新 token 时Self-Attention自注意力层都需要计算当前 token 与之前所有 token 的关联。KV CacheKey-Value 缓存技术可以缓存之前时间步计算好的 Key 和 Value 向量避免重复计算。对于 ChatTTS在流式生成或同一批处理内生成多个片段时我们可以尝试共享或复用部分KV Cache。例如对于一段文本的前缀部分如果多个生成任务都用到它其对应的KV Cache理论上可以只计算一次。import torch.nn as nn class KVCacheManager: 简单的 KV Cache 管理类概念示例 def __init__(self, model: nn.Module): self.model model self.cache_dict {} # 用于缓存不同文本前缀的 KV 状态 def get_cached_kv(self, text_prefix: str, layer_idx: int): 获取某个文本前缀在特定层的缓存 KV 状态 key f{text_prefix}_layer{layer_idx} return self.cache_dict.get(key, None) def set_cached_kv(self, text_prefix: str, layer_idx: int, k_cache: torch.Tensor, v_cache: torch.Tensor): 设置缓存 key f{text_prefix}_layer{layer_idx} self.cache_dict[key] (k_cache.detach().clone(), v_cache.detach().clone()) # 在模型前向传播中假设我们有一个 generate 函数 def generate_with_kv_cache(self, input_ids, use_cacheFalse, past_key_valuesNone): # ... 模型内部逻辑 ... for i, layer in enumerate(self.decoder_layers): # 如果提供了 past_key_values并且当前步不是第一步则使用缓存的 KV if past_key_values is not None: layer_past past_key_values[i] # 将缓存的 K, V 与当前步计算的 K, V 拼接起来 key torch.cat([layer_past[0], current_key], dim2]) # 假设 dim2 是序列长度维 value torch.cat([layer_past[1], current_value], dim2]) else: key, value current_key, current_value # ... 计算 attention ... new_past_key_values[i] (key, value) # 存储当前步的完整 KV 供下一步使用 # ... 返回输出和新的 past_key_values ...注意KV Cache的共享策略需要非常小心因为它严重依赖于模型架构和生成任务。在 ChatTTS 中如果多个请求的文本开头完全相同例如相同的提示词共享前缀部分的KV Cache才有效。实现时需要深入模型代码并做好缓存失效和内存管理。3. 模型量化与加速推理 —— 让计算“轻量化”将模型从 FP32 转换为 FP16 或 INT8可以显著减少显存占用和加速计算。PyTorch 提供了简单的amp自动混合精度模块。from torch.cuda.amp import autocast def inference_with_amp(model, input_tensor): 使用混合精度进行推理 model.half() # 将模型参数转换为 FP16 (谨慎使用可能影响精度) with torch.no_grad(), autocast(): # autocast 上下文管理器自动选择精度 # 输入数据最好也转为 FP16 output model(input_tensor.half()) return output.float() # 将输出转回 FP32 用于后续处理对于追求极致性能的生产环境可以考虑使用TensorRT或ONNX Runtime进行部署。它们会对计算图进行深度优化、算子融合并利用 Tensor Core 等硬件特性。将 PyTorch 模型转为 TensorRT 的过程大致如下# 这是一个概念性流程具体代码依赖 tensorrt 库版本 import tensorrt as trt # 1. 将 PyTorch 模型导出为 ONNX 格式 torch.onnx.export(pytorch_model, dummy_input, chattts.onnx, ...) # 2. 使用 TensorRT 的解析器构建优化引擎 logger trt.Logger(trt.Logger.WARNING) builder trt.Builder(logger) network builder.create_network(1 int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)) parser trt.OnnxParser(network, logger) with open(chattts.onnx, rb) as f: parser.parse(f.read()) config builder.create_builder_config() config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 30) # 1GB config.set_flag(trt.BuilderFlag.FP16) # 启用 FP16 推理 serialized_engine builder.build_serialized_network(network, config) with open(chattts.engine, wb) as f: f.write(serialized_engine) # 3. 加载引擎并进行推理 runtime trt.Runtime(logger) with open(chattts.engine, rb) as f: engine_data f.read() engine runtime.deserialize_cuda_engine(engine_data)对比PyTorch FP16实现简单兼容性好通常有 1.5-2 倍加速。TensorRT需要转换步骤优化更彻底可能获得 2-5 倍甚至更高的加速比但动态形状支持如可变长度文本可能更复杂。三、实践中的“避坑指南”优化路上踩了不少坑这里分享几点多线程/进程下的模型状态管理PyTorch 模型本身不是线程安全的。如果使用动态批处理并在多线程环境下调用model.forward()需要加锁如上文代码中的self.lock或者为每个工作进程复制一份模型副本消耗更多显存。显存溢出OOM回退机制动态批处理时如果一次性组到一个非常长的文本批次可能会撑爆显存。好的实践是预估批次的总 token 数设定一个上限。当批次过大时自动拆分成更小的子批次顺序执行。在代码中捕获torch.cuda.OutOfMemoryError异常触发回退逻辑如清空缓存、减少批次大小重试。def safe_batch_inference(self, batch_tensors): max_tokens 5000 # 根据你的 GPU 设置一个安全阈值 total_tokens sum(t.size(0) for t in batch_tensors) if total_tokens max_tokens: # 将批次拆分成更小的块 return self._split_and_infer(batch_tensors, max_tokens) try: with torch.cuda.amp.autocast(): return self.model(batch_tensors) except torch.cuda.OutOfMemoryError: torch.cuda.empty_cache() # 降级策略使用更小的批次或 FP32 模式重试 return self.model(batch_tensors.to(torch.float32))流式输出Streaming的首包延迟TTFT优化对于需要边生成边播放的流式 TTS用户感知的首包时间至关重要。除了上述优化还可以分块生成将长文本按句子或标点切分提前生成并缓存前面几块的音频快速返回第一块。低延迟模式模型可能提供一种牺牲一点质量换取更快生成速度的模式。四、优化效果怎么样我们在同一台机器RTX 4090上对 1000 个字符的文本进行合成测试对比优化前后的效果优化策略耗时 (秒)相对原始速度比备注原始方案 (FP32, Single)28.51.0x基线 动态批处理 (Batch4)18.21.57x模拟4个并发请求 FP16 混合精度11.72.44x在批处理基础上 TensorRT 部署 (FP16)7.53.8x最终优化方案RTF (Real Time Factor)也从最初的约 0.035即生成1秒音频需35秒优化到了约 0.009。吞吐量每秒处理的字符数提升了近 4 倍。五、延伸思考这套方法能用在 Whisper 上吗Whisper 是语音识别ASR模型而 ChatTTS 是语音合成TTS模型。虽然任务相反但同属序列生成模型很多优化思路是相通的动态批处理完全适用。可以同时转录多段音频显著提升 GPU 利用率。KV Cache同样适用。Whisper 的解码器也是自回归的KV Cache能有效减少长音频转录时的重复计算。模型量化与 TensorRT适用。FP16/INT8 量化和推理引擎优化能大幅加速 Whisper 的编码器和解码器计算。主要区别在于输入差异TTS 输入是文本离散ASR 输入是音频连续。动态批处理时音频需要处理的是不同长度的 mel-spectrogram 或 raw audiopadding 策略和计算量评估会更复杂。注意力模式Whisper 使用了交叉注意力Cross-Attention连接编码器和解码器其KV Cache的管理可能涉及编码器输出的缓存策略上略有不同。总结优化 ChatTTS 的生成速度是一个系统工程从简单的精度降低到复杂的动态批处理和推理引擎更换每一步都带来相应的收益。对于生产环境我建议的路径是先上 FP16 混合精度获得即时收益 - 引入动态批处理应对并发 - 在性能瓶颈明显时考虑 TensorRT 等重型优化。代码的健壮性也很重要尤其是异常处理和资源管理。希望这篇笔记里的代码片段和思路能为你自己的项目提供一些参考。最终我们成功将合成速度提升了近 4 倍用户体验流畅多了。如果你有更好的点子欢迎一起交流。
ChatTTS 生成速度优化实战：从并发瓶颈到高效推理
最近在项目里用 ChatTTS 做长文本语音合成发现生成速度是个大问题。特别是处理几百上千字的文章时等待时间长得让人有点着急。经过一番折腾总算把生成速度提上来了这里把优化思路和实战代码整理一下希望能帮到有同样困扰的朋友。一、问题到底出在哪一开始我简单地用 ChatTTS 的原始接口去合成一篇 1000 字左右的文章发现耗时接近 30 秒。这显然没法用在需要快速响应的场景里。于是我做了一些 profiling性能分析发现几个主要的瓶颈自回归Autoregressive解码延迟TTS 模型生成音频是“一个字一个字”往外蹦的后一个 token 的生成依赖于前一个。这个过程是串行的文本越长耗时自然线性增长。单次推理Single Inference效率低默认的调用方式是一次只处理一个请求。当有多个请求过来时要么排队要么开多个进程前者慢后者资源消耗大。显存GPU Memory利用不充分每次推理都要重新加载模型计算图中间激活值Activation反复分配和释放产生了显存碎片也浪费了计算资源。计算精度冗余模型默认使用 FP32单精度浮点数进行计算这对于语音合成任务来说很多时候精度是过剩的却消耗了更多的计算量和显存。简单说就是“单线程、高精度、反复折腾”导致了速度上不去。二、我们的优化“三板斧”针对上面这些问题我主要从三个方向入手让模型能“批量干活”、让计算能“偷点懒”、让计算本身“轻量化”。1. 动态批处理Dynamic Batching—— 让等待合批一起出发核心思想是把短时间内到来的多个文本合成请求动态地组合成一个批次Batch送给模型计算。这能极大提升 GPU 的利用率。PyTorch 的DataLoader本身是为训练设计的我们需要改造一下用于推理服务。下面是一个简化版的动态批处理推理器实现import torch import threading from queue import Queue, Empty from typing import List, Optional, Tuple from dataclasses import dataclass from torch.utils.data import Dataset, DataLoader dataclass class TTSRequest: TTS 请求数据结构 text: str request_id: str # 其他参数如说话人ID、语速等 speaker_id: Optional[int] None class DynamicBatchDataset(Dataset): 模拟动态请求队列的数据集 def __init__(self, request_queue: Queue): self.queue request_queue def __len__(self): # 返回一个很大的数让 DataLoader 持续等待 return 1000000 def __getitem__(self, idx): try: # 设置超时避免无限阻塞 request self.queue.get(timeout0.1) return request except Empty: # 队列为空时返回 Nonecollate_fn 会处理 return None def collate_fn(batch: List[Optional[TTSRequest]]) - Tuple[List[TTSRequest], List[str]]: 整理批次数据过滤掉 None valid_requests [item for item in batch if item is not None] texts [req.text for req in valid_requests] return valid_requests, texts class TTSBatchInferenceEngine: TTS 批量推理引擎 def __init__(self, model, batch_size: int 8, max_wait_ms: int 50): self.model model self.model.eval() # 切换到评估模式 self.batch_size batch_size self.request_queue Queue() self.dataset DynamicBatchDataset(self.request_queue) # 使用 DataLoader 进行批处理num_workers0 因为在同一个进程 self.dataloader DataLoader( self.dataset, batch_sizebatch_size, collate_fncollate_fn, num_workers0, ) self.dataloader_iter iter(self.dataloader) self.lock threading.Lock() def add_request(self, request: TTSRequest): 客户端调用此方法添加请求 self.request_queue.put(request) def run_batch_inference(self) - List[Tuple[str, torch.Tensor]]: 执行一批推理返回 (request_id, audio_tensor) 列表 with torch.no_grad(), self.lock: # 加锁保证线程安全 try: batch_requests, batch_texts next(self.dataloader_iter) except StopIteration: # 重置迭代器 self.dataloader_iter iter(self.dataloader) return [] if not batch_texts: return [] # 这里是核心将多个文本向量化后堆叠成批次Tensor # 假设 text_to_tensor 是你的文本预处理函数 batch_tensors [self._text_to_tensor(text) for text in batch_texts] padded_batch torch.nn.utils.rnn.pad_sequence(batch_tensors, batch_firstTrue) # 模型前向传播一次性处理整个批次 batch_audio self.model(padded_batch) # 将结果拆解对应回每个请求 results [] # 注意这里需要根据实际模型输出和填充方式正确截取每个样本的音频 for i, req in enumerate(batch_requests): # audio_length 需要根据实际模型输出或原始文本长度推断 audio_length self._get_audio_length_for_sample(i, batch_tensors, batch_audio) single_audio batch_audio[i, :audio_length] results.append((req.request_id, single_audio)) return results def _text_to_tensor(self, text: str) - torch.Tensor: 将文本转换为模型输入张量示例函数 # 这里应实现你的文本 tokenization 和向量化逻辑 # 例如使用模型的 tokenizer # return torch.tensor([...]) pass def _get_audio_length_for_sample(self, idx: int, input_tensors: List[torch.Tensor], output_tensor: torch.Tensor) - int: 根据输入确定每个样本输出的音频长度示例函数 # 这是一个复杂点需要根据模型特性实现。 # 一种常见方法是模型输出与某个输入维度相关或者模型本身会输出长度信息。 # 这里返回一个固定值仅为示例。 return output_tensor.size(1) # 假设所有样本输出长度相同通常不成立关键点我们用Queue接收实时请求用DataLoader来组织批次的获取逻辑。collate_fn负责将一批零散的请求对象整理成模型可接受的批次化张量主要是文本的 padding 和对齐。通过设置DataLoader的batch_size和timeout逻辑实现了“攒够一批就推理”或“等待超时即使不满一批也推理”的动态策略。注意TTS 模型输出的音频长度可能各不相同pad_sequence用于输入输出后需要根据实际长度信息如果模型提供或通过其他方式将每个样本的音频正确切分出来这是实现中的难点。2. KV Cache 共享 —— 让计算能“偷懒”在自回归生成中每次生成新 token 时Self-Attention自注意力层都需要计算当前 token 与之前所有 token 的关联。KV CacheKey-Value 缓存技术可以缓存之前时间步计算好的 Key 和 Value 向量避免重复计算。对于 ChatTTS在流式生成或同一批处理内生成多个片段时我们可以尝试共享或复用部分KV Cache。例如对于一段文本的前缀部分如果多个生成任务都用到它其对应的KV Cache理论上可以只计算一次。import torch.nn as nn class KVCacheManager: 简单的 KV Cache 管理类概念示例 def __init__(self, model: nn.Module): self.model model self.cache_dict {} # 用于缓存不同文本前缀的 KV 状态 def get_cached_kv(self, text_prefix: str, layer_idx: int): 获取某个文本前缀在特定层的缓存 KV 状态 key f{text_prefix}_layer{layer_idx} return self.cache_dict.get(key, None) def set_cached_kv(self, text_prefix: str, layer_idx: int, k_cache: torch.Tensor, v_cache: torch.Tensor): 设置缓存 key f{text_prefix}_layer{layer_idx} self.cache_dict[key] (k_cache.detach().clone(), v_cache.detach().clone()) # 在模型前向传播中假设我们有一个 generate 函数 def generate_with_kv_cache(self, input_ids, use_cacheFalse, past_key_valuesNone): # ... 模型内部逻辑 ... for i, layer in enumerate(self.decoder_layers): # 如果提供了 past_key_values并且当前步不是第一步则使用缓存的 KV if past_key_values is not None: layer_past past_key_values[i] # 将缓存的 K, V 与当前步计算的 K, V 拼接起来 key torch.cat([layer_past[0], current_key], dim2]) # 假设 dim2 是序列长度维 value torch.cat([layer_past[1], current_value], dim2]) else: key, value current_key, current_value # ... 计算 attention ... new_past_key_values[i] (key, value) # 存储当前步的完整 KV 供下一步使用 # ... 返回输出和新的 past_key_values ...注意KV Cache的共享策略需要非常小心因为它严重依赖于模型架构和生成任务。在 ChatTTS 中如果多个请求的文本开头完全相同例如相同的提示词共享前缀部分的KV Cache才有效。实现时需要深入模型代码并做好缓存失效和内存管理。3. 模型量化与加速推理 —— 让计算“轻量化”将模型从 FP32 转换为 FP16 或 INT8可以显著减少显存占用和加速计算。PyTorch 提供了简单的amp自动混合精度模块。from torch.cuda.amp import autocast def inference_with_amp(model, input_tensor): 使用混合精度进行推理 model.half() # 将模型参数转换为 FP16 (谨慎使用可能影响精度) with torch.no_grad(), autocast(): # autocast 上下文管理器自动选择精度 # 输入数据最好也转为 FP16 output model(input_tensor.half()) return output.float() # 将输出转回 FP32 用于后续处理对于追求极致性能的生产环境可以考虑使用TensorRT或ONNX Runtime进行部署。它们会对计算图进行深度优化、算子融合并利用 Tensor Core 等硬件特性。将 PyTorch 模型转为 TensorRT 的过程大致如下# 这是一个概念性流程具体代码依赖 tensorrt 库版本 import tensorrt as trt # 1. 将 PyTorch 模型导出为 ONNX 格式 torch.onnx.export(pytorch_model, dummy_input, chattts.onnx, ...) # 2. 使用 TensorRT 的解析器构建优化引擎 logger trt.Logger(trt.Logger.WARNING) builder trt.Builder(logger) network builder.create_network(1 int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)) parser trt.OnnxParser(network, logger) with open(chattts.onnx, rb) as f: parser.parse(f.read()) config builder.create_builder_config() config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 30) # 1GB config.set_flag(trt.BuilderFlag.FP16) # 启用 FP16 推理 serialized_engine builder.build_serialized_network(network, config) with open(chattts.engine, wb) as f: f.write(serialized_engine) # 3. 加载引擎并进行推理 runtime trt.Runtime(logger) with open(chattts.engine, rb) as f: engine_data f.read() engine runtime.deserialize_cuda_engine(engine_data)对比PyTorch FP16实现简单兼容性好通常有 1.5-2 倍加速。TensorRT需要转换步骤优化更彻底可能获得 2-5 倍甚至更高的加速比但动态形状支持如可变长度文本可能更复杂。三、实践中的“避坑指南”优化路上踩了不少坑这里分享几点多线程/进程下的模型状态管理PyTorch 模型本身不是线程安全的。如果使用动态批处理并在多线程环境下调用model.forward()需要加锁如上文代码中的self.lock或者为每个工作进程复制一份模型副本消耗更多显存。显存溢出OOM回退机制动态批处理时如果一次性组到一个非常长的文本批次可能会撑爆显存。好的实践是预估批次的总 token 数设定一个上限。当批次过大时自动拆分成更小的子批次顺序执行。在代码中捕获torch.cuda.OutOfMemoryError异常触发回退逻辑如清空缓存、减少批次大小重试。def safe_batch_inference(self, batch_tensors): max_tokens 5000 # 根据你的 GPU 设置一个安全阈值 total_tokens sum(t.size(0) for t in batch_tensors) if total_tokens max_tokens: # 将批次拆分成更小的块 return self._split_and_infer(batch_tensors, max_tokens) try: with torch.cuda.amp.autocast(): return self.model(batch_tensors) except torch.cuda.OutOfMemoryError: torch.cuda.empty_cache() # 降级策略使用更小的批次或 FP32 模式重试 return self.model(batch_tensors.to(torch.float32))流式输出Streaming的首包延迟TTFT优化对于需要边生成边播放的流式 TTS用户感知的首包时间至关重要。除了上述优化还可以分块生成将长文本按句子或标点切分提前生成并缓存前面几块的音频快速返回第一块。低延迟模式模型可能提供一种牺牲一点质量换取更快生成速度的模式。四、优化效果怎么样我们在同一台机器RTX 4090上对 1000 个字符的文本进行合成测试对比优化前后的效果优化策略耗时 (秒)相对原始速度比备注原始方案 (FP32, Single)28.51.0x基线 动态批处理 (Batch4)18.21.57x模拟4个并发请求 FP16 混合精度11.72.44x在批处理基础上 TensorRT 部署 (FP16)7.53.8x最终优化方案RTF (Real Time Factor)也从最初的约 0.035即生成1秒音频需35秒优化到了约 0.009。吞吐量每秒处理的字符数提升了近 4 倍。五、延伸思考这套方法能用在 Whisper 上吗Whisper 是语音识别ASR模型而 ChatTTS 是语音合成TTS模型。虽然任务相反但同属序列生成模型很多优化思路是相通的动态批处理完全适用。可以同时转录多段音频显著提升 GPU 利用率。KV Cache同样适用。Whisper 的解码器也是自回归的KV Cache能有效减少长音频转录时的重复计算。模型量化与 TensorRT适用。FP16/INT8 量化和推理引擎优化能大幅加速 Whisper 的编码器和解码器计算。主要区别在于输入差异TTS 输入是文本离散ASR 输入是音频连续。动态批处理时音频需要处理的是不同长度的 mel-spectrogram 或 raw audiopadding 策略和计算量评估会更复杂。注意力模式Whisper 使用了交叉注意力Cross-Attention连接编码器和解码器其KV Cache的管理可能涉及编码器输出的缓存策略上略有不同。总结优化 ChatTTS 的生成速度是一个系统工程从简单的精度降低到复杂的动态批处理和推理引擎更换每一步都带来相应的收益。对于生产环境我建议的路径是先上 FP16 混合精度获得即时收益 - 引入动态批处理应对并发 - 在性能瓶颈明显时考虑 TensorRT 等重型优化。代码的健壮性也很重要尤其是异常处理和资源管理。希望这篇笔记里的代码片段和思路能为你自己的项目提供一些参考。最终我们成功将合成速度提升了近 4 倍用户体验流畅多了。如果你有更好的点子欢迎一起交流。
