最近在折腾语音对话系统想把ChatTTS和Ollama这两个好用的工具结合起来搞一个能说会道的AI助手。ChatTTS在中文语音合成上效果很自然而Ollama能方便地本地部署各种大语言模型。但把它们俩“撮合”到一起可不是简单调两个API就完事了中间遇到了不少坑尤其是音频流的延迟和对话上下文的同步问题经常是这边话还没说完那边就开始抢答了体验很割裂。经过一番摸索总算搞出了一套比较稳定的方案这里就把从零搭建的过程和踩过的那些坑记录下来希望能帮到有同样想法的朋友。1. 为什么要把ChatTTS和Ollama放一起简单来说就是为了打造一个完整的、本地化的语音对话链路。Ollama负责“思考”理解你的问题并生成文本回答ChatTTS负责“说话”把文本回答转换成自然流畅的语音。这个组合非常适合做智能音箱、语音助手或者任何需要语音交互的原型。但理想很丰满现实很骨感。直接串起来会遇到几个核心痛点音频流延迟Audio Stream Latency如果等Ollama生成完整文本再交给ChatTTS合成用户会明显感觉到回答前的停顿很不自然。必须实现流式Streaming响应边生成文本边合成语音。上下文同步Context Synchronization语音合成需要时间而Ollama的流式输出是一段一段的。要处理好文本流和音频流之间的节奏避免语音卡顿或中断。资源消耗语音合成比较吃CPU/GPU同时处理流式请求需要好的异步架构不然容易阻塞。2. 技术方案设计与核心架构为了解决上述问题我没有采用简单的“请求-等待-响应”模式而是设计了一个轻量的中间层用FastAPI来搭建。这个中间层核心工作是“协调”和“转发”。整体架构图思路如下用户语音 - [语音识别服务] - 文本 - [Ollama流式API] - 文本流 | v 音频流 - [音频缓冲区] - [ChatTTS流式合成] - 文本块 - [中间层(FastAPI)]中间层作为代理同时连接Ollama和ChatTTS。它从Ollama获取流式文本然后几乎实时地送给ChatTTS合成小段的音频最后将这些音频块拼接并流式返回给客户端。3. 关键代码实现与分步讲解下面我们分步拆解核心代码。我们假设你已经安装了fastapi,aiohttp,soundfile等库。第一步搭建FastAPI应用与异步流处理骨架我们首先创建一个FastAPI应用并定义一个核心的对话端点。这个端点将处理来自客户端的请求并管理整个流式处理的生命周期。from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse import aiohttp import asyncio import json import io from pydantic import BaseModel app FastAPI(titleChatTTS-Ollama Bridge) class ConversationRequest(BaseModel): message: str model: str llama3.2 # 默认使用Ollama中的某个模型 voice: str default # ChatTTS的语音参数 app.post(/api/conversation/stream) async def stream_conversation(request: ConversationRequest): 核心流式对话接口。 1. 将用户消息发送给Ollama获取流式文本响应。 2. 将收到的文本流实时发送给ChatTTS合成语音。 3. 将合成的音频流返回给客户端。 # 这里将实现主要的流式逻辑 pass第二步连接Ollama获取流式文本响应我们需要使用aiohttp向Ollama的流式API发送请求并异步地读取返回的文本块。Ollama的流式API会返回一系列JSON对象每个对象包含文本的一部分。async def fetch_ollama_stream(session: aiohttp.ClientSession, model: str, prompt: str): 从Ollama获取流式文本生成。 ollama_url http://localhost:11434/api/generate # Ollama默认地址 payload { model: model, prompt: prompt, stream: True # 关键参数开启流式 } try: async with session.post(ollama_url, jsonpayload, timeoutaiohttp.ClientTimeout(total300)) as resp: if resp.status ! 200: error_text await resp.text() raise HTTPException(status_coderesp.status, detailfOllama error: {error_text}) # 异步读取每一行一个JSON对象 async for line in resp.content: if line: try: data json.loads(line.decode(utf-8)) # 提取生成的文本片段 text_chunk data.get(response, ) if text_chunk: yield text_chunk # 如果响应结束跳出循环 if data.get(done, False): break except json.JSONDecodeError: continue # 忽略非JSON行 except asyncio.TimeoutError: raise HTTPException(status_code504, detailOllama request timeout)第三步连接ChatTTS进行流式语音合成这是最复杂的一步。ChatTTS通常需要一个完整的句子才能合成出自然的语音。但为了降低延迟我们不能等一个完整的句子。这里采用一个折中方案设置一个小的文本缓冲区当缓冲区积累到一定长度如15个字符或遇到句末标点。时就触发一次语音合成请求。同时我们需要处理音频格式。ChatTTS返回的可能是高采样率如24000Hz的音频而前端播放可能需要16000Hz所以需要进行采样率转换。import numpy as np import soundfile as sf from io import BytesIO # 注意这里需要你安装相应的音频处理库例如librosa或pydub用于采样率转换。 # 以下示例使用一个假设的resample函数。 async def synthesize_with_chattts(session: aiohttp.ClientSession, text: str, voice_params: dict): 向ChatTTS服务发送文本合成音频并返回PCM数据。 假设ChatTTS服务有一个接受文本返回wav的端点。 chattts_url http://your-chattts-server:port/synthesize # 替换为你的ChatTTS服务地址 payload {text: text, **voice_params} async with session.post(chattts_url, jsonpayload) as resp: if resp.status 200: # 假设返回的是WAV格式的二进制数据 audio_data await resp.read() # 将WAV字节流转换为numpy数组并进行采样率转换 # 这里简化处理实际需用soundfile或librosa读取并转换 audio_array, orig_sr sf.read(BytesIO(audio_data)) target_sr 16000 if orig_sr ! target_sr: audio_array resample_audio(audio_array, orig_sr, target_sr) # 假设的函数 return audio_array.tobytes(), target_sr else: raise Exception(fChatTTS synthesis failed: {resp.status})第四步核心协调逻辑——将文本流转换为音频流现在我们需要把第二步和第三步连接起来。我们创建一个异步生成器函数它消费来自Ollama的文本流并产出音频数据块。async def text_to_audio_stream(session: aiohttp.ClientSession, text_stream, voice_params: dict): 协调器消费文本流生产音频流。 text_buffer sentence_enders set(。) async for text_chunk in text_stream: text_buffer text_chunk # 触发合成条件缓冲区达到一定长度或遇到句子结束符 if len(text_buffer) 15 or (text_buffer and text_buffer[-1] in sentence_enders): if text_buffer.strip(): # 避免合成空文本 try: audio_bytes, sample_rate await synthesize_with_chattts(session, text_buffer, voice_params) # 可以在这里将音频字节流分块例如每0.5秒的音频yield出去 # 为了简化我们直接yield整个句子的音频 chunk_size sample_rate * 2 # 例如每2秒的音频作为一个块 for i in range(0, len(audio_bytes), chunk_size): yield audio_bytes[i:ichunk_size] await asyncio.sleep(0.01) # 小延迟模拟流式 except Exception as e: print(fChatTTS synthesis error: {e}) # 可以选择跳过错误继续下一个文本块 text_buffer # 清空缓冲区 # 处理缓冲区中剩余的文本 if text_buffer.strip(): try: audio_bytes, _ await synthesize_with_chattts(session, text_buffer, voice_params) yield audio_bytes except Exception as e: print(fFinal synthesis error: {e})第五步整合到FastAPI端点并返回流式响应最后我们将上述所有组件整合到最初的API端点中。app.post(/api/conversation/stream) async def stream_conversation(request: ConversationRequest): async def audio_generator(): # 创建aiohttp会话用于外部HTTP调用 async with aiohttp.ClientSession() as session: # 1. 获取Ollama文本流 ollama_text_stream fetch_ollama_stream(session, request.model, request.message) # 2. 将文本流转换为音频流 voice_params {voice: request.voice} audio_stream text_to_audio_stream(session, ollama_text_stream, voice_params) # 3. 将音频流以WAV格式流式输出 # 先yield一个WAV头简化版实际需要根据采样率、位数等生成 wav_header generate_wav_header(sample_rate16000, bits_per_sample16, channels1) # 假设的函数 yield wav_header # 然后yield音频数据 async for audio_chunk in audio_stream: yield audio_chunk # 返回音频流响应 return StreamingResponse( audio_generator(), media_typeaudio/wav, headers{ Cache-Control: no-cache, Transfer-Encoding: chunked, } )4. 性能优化与RTF测试系统搭起来后性能是关键。一个重要的指标是RTFReal-Time Factor实时因子它表示处理一段语音所需时间与该段语音时长的比值。RTF 1 表示处理速度快于实时是理想状态。我测试了在不同并发请求数下系统的RTF表现。测试环境Ollama (Llama3.2 8B) 和 ChatTTS 服务分别运行中间层为2核4G的云服务器。并发数1平均RTF约为0.8。这意味着合成10秒的语音系统需要8秒处理体验良好。并发数3平均RTF上升到1.5。处理开始出现排队延迟感明显。并发数5平均RTF超过2.5部分请求超时。优化建议缓存Caching对于常见的、确定的回答如“你好”、“谢谢”可以预先合成音频并缓存直接返回大幅降低RTF。异步合成队列将ChatTTS合成任务放入一个异步队列由单独的worker进程处理避免HTTP请求阻塞。调整文本缓冲区策略根据网络状况和服务器负载动态调整触发合成的文本长度阈值。使用更高效的音频编码比如返回OPUS格式的音频比WAV体积小很多减少网络传输延迟。5. 生产环境检查清单如果你打算把这个原型部署到生产环境请务必检查以下事项必须配置的Ollama参数OLLAMA_NUM_PARALLEL设置Ollama能处理的并行请求数根据你的GPU内存调整。OLLAMA_HOST如果你需要从非本地网络访问记得绑定到0.0.0.0并设置好防火墙。模型预热在服务启动后先发送一个简单请求“预热”模型避免第一次请求响应极慢。音频缓冲区的推荐大小文本缓冲区推荐在10到20个字符之间并配合句末标点检测。太小会导致合成请求过于频繁增加开销太大会增加语音首字延迟。音频输出缓冲区在StreamingResponse中系统有内部缓冲区。如果客户端播放卡顿可以尝试在yield音频数据前稍微增加await asyncio.sleep()的时间但这会增加整体延迟需要权衡。常见的SSL证书错误解决方案自签名证书如果你在内网使用自签名证书部署ChatTTS或Ollama通过HTTPS需要在aiohttp ClientSession中设置verify_sslFalse仅限测试环境生产环境有安全风险。证书验证失败生产环境应使用正规CA签发的证书。如果仍有问题可以尝试将证书文件路径设置为环境变量SSL_CERT_FILE或在创建aiohttp会话时指定connectoraiohttp.TCPConnector(ssl_contextssl.create_default_context(cafile‘path/to/cert.pem’))。折腾完这一套感觉就像搭好了一个精密的管道系统看着文本和音频在其中顺畅地流动最终变成一段段清晰的对话还是挺有成就感的。这套方案虽然还有优化空间比如在语音合成的自然度和流式延迟之间做更精细的权衡但作为一个能在1小时内跑起来的原型它已经足够强大可以支撑起很多有趣的语音应用创意了。希望这篇笔记能帮你避开我踩过的那些坑更快地搭建出自己的语音对话系统。