数字人语音实时对话中的双流输出终极方案

# Copyright 2025 ByteDance and/or its affiliates.## Licensed under the Apache License, Version 2.0 (the "License");# you may not use this file except in compliance with the License.# You may obtain a copy of the License at##     http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.import osimport torchfrom flask import Flask, Response, requestimport numpy as npimport soundfile as sfimport iofrom tts.infer_cli import MegaTTS3DiTInfer, convert_to_wav, cut_wav# 配置路径BASE_DIR = os.path.dirname(os.path.abspath(__file__))CHECKPOINTS_DIR = os.path.join(BASE_DIR, 'checkpoints')ASSETS_DIR = os.path.join(BASE_DIR, 'assets')app = Flask(__name__)# 初始化 MegaTTS3 模型device = 'cuda' if torch.cuda.is_available() else 'cpu'infer_pipe = MegaTTS3DiTInfer(    device=device,    ckpt_root=CHECKPOINTS_DIR,    dit_exp_name='diffusion_transformer',    frontend_exp_name='aligner_lm',    wavvae_exp_name='wavvae',    dur_ckpt_path='duration_lm',    g2p_exp_name='g2p')# 默认参考音频和潜在文件路径DEFAULT_REF_WAV = os.path.join(ASSETS_DIR, 'Chinese_prompt.wav')DEFAULT_REF_NPY = os.path.join(ASSETS_DIR, 'Chinese_prompt.npy')def generate_audio_stream(text, ref_wav=DEFAULT_REF_WAV, ref_npy=DEFAULT_REF_NPY, time_step=32, p_w=1.6, t_w=2.5):    """    生成音频流，按句子分块处理并返回 WAV 数据。    """    try:        # 确保参考音频是 WAV 格式并裁剪        convert_to_wav(ref_wav)        wav_path = os.path.splitext(ref_wav)[0] + '.wav'        cut_wav(wav_path, max_len=28)        # 读取参考音频        with open(wav_path, 'rb') as file:            file_content = file.read()        # 预处理参考音频        resource_context = infer_pipe.preprocess(file_content, latent_file=ref_npy)        # 分块生成音频        def audio_chunks():            # 按句子分割文本            sentences = text.split('。') if '。' in text else [text]            for sentence in sentences:                if sentence.strip():                    # 生成音频                    wav_bytes = infer_pipe.forward(                        resource_context, sentence, time_step=time_step, p_w=p_w, t_w=t_w                    )                    # 将字节流转换为 WAV 格式的音频数据                    wav_data, _ = sf.read(io.BytesIO(wav_bytes))                    with io.BytesIO() as buf:                        sf.write(buf, wav_data, infer_pipe.sr, format='WAV')                        yield buf.getvalue()        return audio_chunks()    except Exception as e:        print(f"Error generating audio: {str(e)}")        return None@app.route('/stream', methods=['GET'])def stream_audio():    """    HTTP 流式音频接口，接收文本并返回音频流。    """    text = request.args.get('text', '你好，这是一段测试语音。')    if not text:        return Response("No text provided", status=400)    audio_chunks = generate_audio_stream(text)    if audio_chunks is None:        return Response("Audio generation failed", status=500)    return Response(audio_chunks, mimetype='audio/wav')if __name__ == '__main__':    # 确保资产目录存在    os.makedirs(ASSETS_DIR, exist_ok=True)        # 检查默认参考文件是否存在    if not os.path.exists(DEFAULT_REF_WAV):        print(f"Warning: Default reference WAV file not found at {DEFAULT_REF_WAV}")    if not os.path.exists(DEFAULT_REF_NPY):        print(f"Warning: Default reference NPY file not found at {DEFAULT_REF_NPY}")    # 启动 Flask 服务    app.run(host='0.0.0.0', port=5000, debug=True)

import React, { useState, useRef } from 'react';import axios from 'axios';const App = () => {  const [inputText, setInputText] = useState('');  const [messages, setMessages] = useState([]);  const [isLoading, setIsLoading] = useState(false);  const audioContextRef = useRef(new AudioContext());  // 处理文本输入  const handleInputChange = (e) => setInputText(e.target.value);  // 调用 Dify 接口并处理流式文本  const fetchStreamText = async (text) => {    setIsLoading(true);    const response = await fetch('http://localhost:5001/v1/chat-messages', {      method: 'POST',      headers: {        'Content-Type': 'application/json',        'Authorization': 'Bearer YOUR_DIFY_API_KEY', // 替换为你的 Dify API Key      },      body: JSON.stringify({        inputs: { text },        query: text,        response_mode: 'streaming',        user: 'user123',      }),    });    const reader = response.body.getReader();    const decoder = new TextDecoder();    let buffer = '';    while (true) {      const { done, value } = await reader.read();      if (done) break;      const chunk = decoder.decode(value);      const lines = chunk.split('\n');      for (const line of lines) {        if (line.startsWith('data: ')) {          const data = JSON.parse(line.slice(6));          if (data.event === 'message') {            buffer += data.answer;            // 检查是否到达句子结束            if (/[。！？]/.test(buffer)) {              const sentences = buffer.split(/(?<=[。！？])/);              for (let i = 0; i < sentences.length - 1; i++) {                const sentence = sentences[i];                setMessages((prev) => [...prev, { text: sentence, isUser: false }]);                await fetchAudio(sentence); // 调用 MegaTTS3 生成音频              }              buffer = sentences[sentences.length - 1];            }          }        }      }    }    // 处理剩余的缓冲区内容    if (buffer) {      setMessages((prev) => [...prev, { text: buffer, isUser: false }]);      await fetchAudio(buffer);    }    setIsLoading(false);  };  // 调用 MegaTTS3 接口生成音频并播放  const fetchAudio = async (text) => {    const response = await fetch(`http://localhost:5000/stream?text=${encodeURIComponent(text)}`);    const reader = response.body.getReader();    let audioBufferQueue = [];    let isPlaying = false;    const processChunk = async () => {      const { done, value } = await reader.read();      if (done) return;      const audioBuffer = await audioContextRef.current.decodeAudioData(value.buffer);      audioBufferQueue.push(audioBuffer);      if (!isPlaying) {        playNextBuffer();      }      processChunk();    };    const playNextBuffer = () => {      if (audioBufferQueue.length > 0) {        const buffer = audioBufferQueue.shift();        const source = audioContextRef.current.createBufferSource();        source.buffer = buffer;        source.connect(audioContextRef.current.destination);        source.onended = playNextBuffer;        source.start();        isPlaying = true;      } else {        isPlaying = false;      }    };    processChunk();  };  // 处理发送按钮点击  const handleSend = () => {    if (!inputText.trim()) return;    setMessages((prev) => [...prev, { text: inputText, isUser: true }]);    fetchStreamText(inputText);    setInputText('');  };  return (    <div className="min-h-screen bg-gradient-to-br from-blue-100 to-purple-100 flex items-center justify-center p-4">      <div className="w-full max-w-2xl bg-white rounded-lg shadow-xl p-6">        <h1 className="text-3xl font-bold text-center text-gray-800 mb-6">语音对话助手h1>        {/* 消息显示区域 */}        <div className="h-96 overflow-y-auto mb-4 p-4 bg-gray-50 rounded-lg border border-gray-200">          {messages.map((msg, index) => (            <div              key={index}              className={`mb-2 p-3 rounded-lg max-w-[80%] ${                msg.isUser                  ? 'bg-blue-500 text-white ml-auto'                  : 'bg-gray-200 text-gray-800 mr-auto'              }`}            >              {msg.text}            div>          ))}          {isLoading && (            <div className="text-gray-500 text-center">正在生成...div>          )}        div>        {/* 输入框和发送按钮 */}        <div className="flex gap-2">          <input            type="text"            value={inputText}            onChange={handleInputChange}            placeholder="输入你的消息..."            className="flex-1 p-3 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500"            onKeyPress={(e) => e.key === 'Enter' && handleSend()}          />          <button            onClick={handleSend}            disabled={isLoading}            className="px-6 py-3 bg-blue-600 text-white rounded-lg hover:bg-blue-700 disabled:bg-gray-400 transition-colors"          >            发送          button>        div>      div>    div>  );};export default App;

import React from 'react';import ReactDOM from 'react-dom';import './index.css';import App from './App';ReactDOM.render(  <React.StrictMode>    <App />  React.StrictMode>,  document.getElementById('root'));

import asyncioimport jsonfrom aiohttp import webfrom aiortc import RTCPeerConnection, RTCSessionDescriptionpcs = set()async def offer(request):    params = await request.json()    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])    # 创建 PeerConnection    pc = RTCPeerConnection()    pcs.add(pc)    # 设置远程描述并生成应答    await pc.setRemoteDescription(offer)    answer = await pc.createAnswer()    await pc.setLocalDescription(answer)    return web.Response(        content_type="application/json",        text=json.dumps({            "sdp": pc.localDescription.sdp,            "type": pc.localDescription.type        })    )async def on_shutdown(app):    # 关闭所有连接    coros = [pc.close() for pc in pcs]    await asyncio.gather(*coros)    pcs.clear()app = web.Application()app.on_shutdown.append(on_shutdown)app.router.add_post("/offer", offer)if __name__ == "__main__":    web.run_app(app, host="0.0.0.0", port=8080)

import subprocessdef encode_to_opus(pcm_data, sample_rate=16000):    command = [        'ffmpeg',        '-f', 's16le',           # 输入格式为 16 位 PCM        '-ar', str(sample_rate), # 采样率        '-i', 'pipe:0',          # 从管道读取输入        '-c:a', 'libopus',       # 使用 Opus 编码        '-b:a', '16k',           # 比特率        '-f', 'opus',            # 输出格式        'pipe:1'                 # 输出到管道    ]    process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)    opus_data, _ = process.communicate(input=pcm_data)    return opus_data

from aiortc import MediaStreamTrackimport asyncioimport numpy as npclass MegaTTS3AudioTrack(MediaStreamTrack):    kind = "audio"    def __init__(self, model, text, ref_wav):        super().__init__()        self.model = model  # MegaTTS3 模型        self.text = text    # 输入文本        self.ref_wav = ref_wav  # 参考音频        self.queue = asyncio.Queue()        asyncio.create_task(self.generate_audio())    async def generate_audio(self):        # 按句子分块生成音频，实现流式输出        sentences = self.text.split('。')        for sentence in sentences:            if sentence.strip():                audio = self.model.inference(sentence, self.ref_wav)  # 生成音频                pcm_data = (audio * 32767).astype(np.int16).tobytes()  # 转换为 PCM                opus_data = encode_to_opus(pcm_data)  # 编码为 Opus                await self.queue.put(opus_data)  # 放入队列    async def recv(self):        # 从队列中获取数据        opus_data = await self.queue.get()        # 这里需要将 Opus 数据封装为 WebRTC 所需的格式        # 具体实现可能需要借助 aiortc 的内部工具，暂略        return opus_data

async function start() {    const pc = new RTCPeerConnection();    // 创建并设置本地 offer    const offer = await pc.createOffer();    await pc.setLocalDescription(offer);    // 发送 offer 到后端    const response = await fetch('http://localhost:8080/offer', {        method: 'POST',        headers: { 'Content-Type': 'application/json' },        body: JSON.stringify({            sdp: pc.localDescription.sdp,            type: pc.localDescription.type        })    });    const answer = await response.json();    await pc.setRemoteDescription(new RTCSessionDescription(answer));    // 监听音频流并播放    pc.ontrack = (event) => {        const stream = event.streams[0];        const audio = document.createElement('audio');        audio.srcObject = stream;        audio.autoplay = true;        document.body.appendChild(audio);    };}start();