ComfyUI与ChatTTS实战：构建高并发多人对话系统的架构设计与实现

最近在做一个多人实时语音对话的项目，遇到了不少头疼的问题。用户一多，音频就开始卡顿、延迟，甚至出现说话顺序错乱的情况。经过一番折腾，我们最终基于 ComfyUI 的工作流编排能力和 ChatTTS 的语音合成技术，搭建了一套还算稳定的高并发对话系统。今天就来分享一下我们的架构设计和实现细节，希望能给遇到类似问题的朋友一些参考。

背景痛点：当并发遇上实时音频

多人实时语音对话听起来很酷，但一旦用户量上来，各种问题就暴露无遗。我们最初的原型在几十个并发用户时还能勉强运行，但当模拟测试到一两百人同时在线对话时，系统几乎崩溃。主要问题集中在三个方面：

1. 音频延迟与卡顿：这是最直观的感受。用户A说完话，用户B要等好几秒才能听到，对话节奏完全被打乱。这背后是音频数据在网络传输、服务器处理、合成排队等多个环节的累积延迟。
2. 语音片段乱序：在高并发下，来自不同用户或同一用户不同句子的音频数据包，到达服务器的顺序可能和发送顺序不一致。如果不加处理，播放出来的对话就会前言不搭后语，逻辑混乱。
3. 资源竞争与瓶颈：语音合成（TTS）和语音识别（ASR）都是计算密集型任务。当大量请求同时涌向有限的 GPU 或 CPU 资源时，会造成任务堆积，响应时间急剧上升。同时，大量的持久连接（如 WebSocket）也会消耗大量服务器内存和端口资源。

这些问题不是简单升级服务器配置就能解决的，必须从架构层面进行重新设计。

技术选型：为什么是 ComfyUI + ChatTTS + WebSocket？

在技术选型上，我们对比了几种常见的方案。

传输层协议对比：gRPC vs WebSocket对于实时音频流，低延迟和双向通信是刚需。

• gRPC：基于 HTTP/2，支持流式传输，性能很高，尤其是在需要强类型定义和多种语言客户端的场景下。但它对移动端和 Web 端的支持不如 WebSocket 原生和简单，且一些防火墙策略可能会拦截 gRPC 流量。
• WebSocket：专为全双工通信设计，协议简单，被所有现代浏览器原生支持，非常适合 Web 前端与后端进行实时数据交换。虽然它本身是文本协议，但通过传输二进制帧（Binary Frame）来传递音频数据毫无压力。

考虑到我们的场景以 Web 应用为主，需要快速建立稳定的双向通道，并且希望客户端实现尽可能轻量，我们最终选择了WebSocket。

核心组件选择：ComfyUI 与 ChatTTS

• ComfyUI：它不仅仅是一个 Stable Diffusion 的 UI。其核心价值在于可视化的工作流编排和强大的异步执行引擎。我们可以将一次语音对话请求拆解成多个节点任务（如：文本接收 -> 情感分析 -> TTS 合成 -> 音频编码），由 ComfyUI 的调度器高效、可靠地执行。这为我们构建复杂的音频处理流水线提供了绝佳的基础框架。
• ChatTTS：这是一个效果不错的开源 TTS 模型，在中文对话场景下自然度很高。它支持细粒度的控制，如笑声、停顿等，非常适合构建富有表现力的对话机器人。将其作为 ComfyUI 工作流中的一个“TTS 节点”来调用，非常方便。

组合起来，ComfyUI 负责任务调度与流程编排，ChatTTS 负责核心语音合成，WebSocket 负责实时数据传输，形成了一个清晰、解耦且易于扩展的架构。

核心实现：从架构到代码

1. 异步音频处理流水线

核心思路是将每个用户的对话请求视为一个任务，放入异步队列中处理，避免阻塞。我们利用 Python 的asyncio和queue模块构建了一个双缓冲区（Double Buffer）流水线。

一个缓冲区用于接收和排序音频片段，另一个缓冲区用于向客户端推送。这样做的好处是生产和消费可以异步进行，平滑流量峰值，避免卡顿。

import asyncio import json from typing import Optional import numpy as np class AudioPipeline: def __init__(self, user_id: str, websocket): self.user_id = user_id self.ws = websocket # 接收缓冲区 (用于排序和暂存) self.receive_buffer = asyncio.Queue(maxsize=50) # 发送缓冲区 (用于向网络流式发送) self.send_buffer = asyncio.Queue(maxsize=30) self.processing_task: Optional[asyncio.Task] = None self.sending_task: Optional[asyncio.Task] = None async def start(self): """启动管道的处理与发送任务""" self.processing_task = asyncio.create_task(self._process_audio_chunks()) self.sending_task = asyncio.create_task(self._send_audio_chunks()) print(f"Audio pipeline started for user: {self.user_id}") async def put_chunk(self, chunk_id: int, audio_data: bytes): """向接收缓冲区放入音频数据块""" if self.receive_buffer.full(): # 缓冲区满，丢弃最旧的数据或采取其他策略（如日志警告） print(f"Warning: Receive buffer full for user {self.user_id}") await self.receive_buffer.put((chunk_id, audio_data)) async def _process_audio_chunks(self): """处理接收缓冲区的数据：排序、解码、重采样等（示例为简单转发）""" buffer_dict = {} expected_chunk_id = 0 while True: try: chunk_id, audio_data = await asyncio.wait_for(self.receive_buffer.get(), timeout=1.0) buffer_dict[chunk_id] = audio_data # 按顺序处理已缓存的块 while expected_chunk_id in buffer_dict: processed_data = await self._mock_processing(buffer_dict.pop(expected_chunk_id)) if not self.send_buffer.full(): await self.send_buffer.put(processed_data) else: print(f"Warning: Send buffer full, dropping chunk {expected_chunk_id}") expected_chunk_id += 1 except asyncio.TimeoutError: # 超时检查，用于维持循环或执行其他清理 continue except Exception as e: print(f"Error in processing task for {self.user_id}: {e}") break async def _mock_processing(self, audio_data: bytes) -> bytes: """模拟音频处理过程，如降噪、增益等""" # 此处可集成 FFmpeg 或 audioop 进行实际处理 await asyncio.sleep(0.001) # 模拟处理耗时 return audio_data async def _send_audio_chunks(self): """从发送缓冲区取出数据，通过 WebSocket 发送""" while True: try: audio_data = await self.send_buffer.get() if self.ws and not self.ws.closed: # 以二进制帧形式发送 await self.ws.send_bytes(audio_data) else: print(f"WebSocket closed for {self.user_id}, stopping sender.") break except Exception as e: print(f"Error in sending task for {self.user_id}: {e}") break async def stop(self): """停止管道""" if self.processing_task: self.processing_task.cancel() if self.sending_task: self.sending_task.cancel() print(f"Audio pipeline stopped for user: {self.user_id}")

2. 分布式锁控制语音顺序

在分布式部署中，同一个用户的请求可能被负载均衡到不同的服务器实例。为了确保全局的语音片段顺序，我们使用 Redis 分布式锁来控制对“用户当前播放序列号”这个关键状态的更新。

import aioredis import asyncio import uuid from contextlib import asynccontextmanager from typing import Optional class DistributedSequenceLock: def __init__(self, redis_client, user_id: str, lock_timeout: int = 3): self.redis = redis_client self.user_id = user_id self.lock_key = f"seq_lock:{user_id}" self.lock_timeout = lock_timeout # 锁超时时间，防止死锁 async def acquire_and_get_next_seq(self) -> Optional[int]: """获取锁并返回下一个可用的序列号""" lock_identifier = str(uuid.uuid4()) # 尝试获取锁，使用 SET NX EX 命令保证原子性 acquired = await self.redis.set( self.lock_key, lock_identifier, ex=self.lock_timeout, nx=True ) if not acquired: # 获取锁失败，可能其他实例正在处理该用户的上一个请求 return None try: # 在锁的保护下，获取并递增序列号 seq_key = f"user_seq:{self.user_id}" # 使用 Redis 的 INCR 命令，原子性递增 next_seq = await self.redis.incr(seq_key) return next_seq finally: # 释放锁，通过 Lua 脚本确保只有锁的持有者才能删除 lua_script = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ await self.redis.eval(lua_script, 1, self.lock_key, lock_identifier) # 使用示例 async def handle_user_audio_request(user_id: str, audio_text: str): redis = await aioredis.create_redis_pool('redis://localhost') seq_lock = DistributedSequenceLock(redis, user_id) next_seq = await seq_lock.acquire_and_get_next_seq() if next_seq is None: # 未获取到锁，说明系统繁忙，可以告知客户端稍后重试或丢弃此请求 return {"status": "busy", "message": "System is processing previous request."} # 成功获取序列号 next_seq try: # 将序列号 next_seq 和文本 audio_text 一起送入 ComfyUI 工作流进行处理 # ... 调用 ComfyUI API ... pass finally: redis.close() await redis.wait_closed()

TTL 设置最佳实践：锁的超时时间（TTL）设置非常关键。太短，可能在任务完成前锁就释放了，导致顺序混乱；太长，如果持有锁的进程崩溃，其他进程需要等待很久才能继续。我们的经验是，TTL 应略大于单个语音请求的平均处理时间。可以通过监控系统统计这个时间，并动态调整。例如，平均处理时间为 1.2 秒，TTL 可以设置为 2-3 秒。

性能优化：让系统跑得更快更稳

1. 音频帧压缩

原始 TTS 输出的音频数据量很大，直接传输会占用大量带宽。我们使用 FFmpeg 在服务端对音频进行实时转码压缩，例如从 PCM 的 WAV 格式转换为 OPUS 编码的 OGG 格式。OPUS 在低比特率下仍有很好的语音质量。

我们做了简单的量化测试：一段 10 秒、16kHz、16bit 单声道的 PCM 音频，体积约为 320KB。经过 FFmpeg 转码为 OPUS（比特率 16kbps）后，体积仅为~20KB，压缩比达到16:1，而人耳感知的语音质量损失很小。这极大地减轻了网络压力。

2. WebSocket 连接池与自动扩容

每个 WebSocket 连接都会占用内存和文件描述符。为了支持高并发，我们实现了 WebSocket 连接池，并设计了简单的自动扩容算法。

伪代码逻辑：

初始化连接池，设置最小连接数 min_conn=5，最大连接数 max_conn=100 设置目标利用率阈值（如：high_watermark=0.8, low_watermark=0.3） 定时任务（每30秒执行）： 当前活跃连接数 = 获取当前正在使用的连接数 当前总连接数 = 连接池大小 当前利用率 = 当前活跃连接数 / 当前总连接数 如果 当前利用率 > high_watermark 且 当前总连接数 < max_conn： 需要扩容的连接数 = min( (当前活跃连接数 / high_watermark) - 当前总连接数, max_conn - 当前总连接数 ) 创建新的物理连接，加入连接池 否则如果 当前利用率 < low_watermark 且 当前总连接数 > min_conn： 需要缩容的连接数 = 当前总连接数 - max( min_conn, (当前活跃连接数 / low_watermark) ) 从连接池中移除闲置时间最长的连接

参数调优建议：

• min_conn：根据系统常驻负载设置，避免冷启动时频繁创建连接。
• max_conn：受限于服务器内存和操作系统文件描述符限制，需要压测得出。
• high_watermark：设置得太高（如0.95），可能导致在流量突增时来不及扩容，请求排队。建议设置在 0.7-0.8。
• low_watermark：设置得太高，可能导致连接池频繁缩容。建议设置在 0.2-0.4。
• 扩容/缩容步长：不要一次性增加或减少太多连接，可以采用渐进式，比如每次调整当前连接数的 10%-20%。

避坑指南：那些我们踩过的坑

1. Chrome 浏览器的自动播放策略这是前端常见的坑。Chrome 禁止页面在用户没有交互（如点击）之前自动播放带声音的媒体。我们的语音对话应用一打开就需要播放欢迎语，结果被拦截了。解决方案：

   • 引导用户交互：在应用启动时，设计一个“开始对话”的按钮，用户点击后，不仅开始对话逻辑，也触发了 Web Audio API 的上下文恢复（resume()）。
   • 静音播放：可以先以muted状态播放一个极短的无声音频，解锁自动播放限制，然后在收到真正的音频流时取消静音。但这种方法用户体验不佳。
   • 使用 Web Audio API：在用户手势事件中创建AudioContext，并确保所有play()操作都在该事件回调中或之后进行。这是最推荐的方式。

2. 语音识别模型的内存泄漏长时间运行的 ASR 服务，如果模型加载或推理过程中没有正确释放资源，很容易内存泄漏，最终导致服务 OOM（Out Of Memory）崩溃。监控方案：

   • 进程级监控：使用如psutil库定期（如每分钟）记录服务进程的内存 RSS（Resident Set Size）和 VMS（Virtual Memory Size）。观察其增长趋势，如果呈现持续上升且不回落，很可能存在泄漏。
   • 对象级监控：在 Python 中，可以使用tracemalloc模块来跟踪内存分配的位置。定期打快照并比较，找出哪些对象在持续增加。
   • 压力测试与基线对比：在固定的并发请求压力下，运行服务数小时，记录内存使用曲线，与一个已知无泄漏的基线版本进行对比。
   • 定期重启：作为临时应对措施，可以设置一个内存阈值或运行时间阈值，通过 Supervisor 或 Kubernetes 的 Liveness Probe 来自动重启服务。但这只是治标不治本。

总结与思考

通过 ComfyUI 的流程编排、ChatTTS 的语音合成、WebSocket 的实时通信，再加上异步流水线、分布式锁、连接池等一系列优化，我们最终构建的系统能够相对稳定地支持 500+ 的并发实时对话。整个过程中，解耦、异步、监控是三个最重要的关键词。

当然，系统还有很大的优化空间。例如，我们目前是将每个用户的对话视为独立流。一个很自然的延伸思考是：如何在一个多人语音房间中，实现说话人分离（Speaker Diarization）并与声纹识别（Voiceprint Recognition）融合，从而为不同的说话人生成带有其音色特征的语音回复？

这涉及到更复杂的音频处理流程：首先从混音流中分离出不同说话人的片段，然后提取每个片段的声纹特征进行识别或注册，最后在 TTS 阶段根据识别出的说话人 ID 来调整合成语音的音色。ComfyUI 的节点化工作流非常适合编排这样的复杂管道，可以将 VAD（语音活动检测）、分离、识别、TTS 等模块串联起来。有兴趣的朋友可以深入研究一下pyannote-audio（用于说话人日记）和Resemblyzer（用于声纹提取）等开源工具。

构建高并发实时系统就像搭积木，既要选对坚固的组件（技术选型），也要设计好组件之间的连接方式（架构设计），最后还得不断地测试和加固（性能优化与避坑）。希望我们这次的实践分享，能为你搭建自己的“积木城堡”提供一些有用的图纸。