mirror of
https://github.com/ooyinet/WeClone.git
synced 2026-08-31 00:00:53 +08:00
小铭
@@ -3,7 +3,7 @@
|
||||
## 核心功能✨
|
||||
- 💬 使用微信聊天记录微调LLM
|
||||
- 🎙️ 使用微信语音消息➕0.5B大模型实现高质量声音克隆 👉[WeClone-audio](https://github.com/xming521/WeClone/tree/master/WeClone-audio)
|
||||
- 🔗 绑定到微信机器人，实现自己的数字分身
|
||||
- 🔗 绑定到微信、QQ、Telegram、企微、飞书机器人，实现自己的数字分身
|
||||
|
||||
## 特性与说明📋
|
||||
|
||||
@@ -197,4 +197,15 @@ python ./src/wechat_bot/main.py
|
||||
<br>
|
||||
<br>
|
||||
|
||||
## ⭐ Star History
|
||||
> [!TIP]
|
||||
> 如果本项目对您有帮助，或者您关注本项目的未来发展，请给项目 Star，谢谢
|
||||
|
||||
<div align="center">
|
||||
|
||||
[](https://www.star-history.com/#xming521/WeClone&Date)
|
||||
|
||||
</div>
|
||||
|
||||
|
||||
<div align="center"> 克隆我们，保留那灵魂的芬芳 </div>
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
API_KEY=your_api_key_here
|
||||
PORT=5050
|
||||
|
||||
DEFAULT_VOICE=en-US-AvaNeural
|
||||
DEFAULT_RESPONSE_FORMAT=mp3
|
||||
DEFAULT_SPEED=1.0
|
||||
|
||||
DEFAULT_LANGUAGE=en-US
|
||||
|
||||
REQUIRE_API_KEY=True
|
||||
|
||||
REMOVE_FILTER=False
|
||||
|
||||
EXPAND_API=True
|
||||
@@ -0,0 +1,62 @@
|
||||
import re
|
||||
import emoji
|
||||
|
||||
def prepare_tts_input_with_context(text: str) -> str:
|
||||
"""
|
||||
Prepares text for a TTS API by cleaning Markdown and adding minimal contextual hints
|
||||
for certain Markdown elements like headers. Preserves paragraph separation.
|
||||
|
||||
Args:
|
||||
text (str): The raw text containing Markdown or other formatting.
|
||||
|
||||
Returns:
|
||||
str: Cleaned text with contextual hints suitable for TTS input.
|
||||
"""
|
||||
|
||||
# Remove emojis
|
||||
text = emoji.replace_emoji(text, replace='')
|
||||
|
||||
# Add context for headers
|
||||
def header_replacer(match):
|
||||
level = len(match.group(1)) # Number of '#' symbols
|
||||
header_text = match.group(2).strip()
|
||||
if level == 1:
|
||||
return f"Title — {header_text}\n"
|
||||
elif level == 2:
|
||||
return f"Section — {header_text}\n"
|
||||
else:
|
||||
return f"Subsection — {header_text}\n"
|
||||
|
||||
text = re.sub(r"^(#{1,6})\s+(.*)", header_replacer, text, flags=re.MULTILINE)
|
||||
|
||||
# Announce links (currently commented out for potential future use)
|
||||
# text = re.sub(r"\[([^\]]+)\]\((https?:\/\/[^\)]+)\)", r"\1 (link: \2)", text)
|
||||
|
||||
# Remove links while keeping the link text
|
||||
text = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", text)
|
||||
|
||||
# Describe inline code
|
||||
text = re.sub(r"`([^`]+)`", r"code snippet: \1", text)
|
||||
|
||||
# Remove bold/italic symbols but keep the content
|
||||
text = re.sub(r"(\*\*|__|\*|_)", '', text)
|
||||
|
||||
# Remove code blocks (multi-line) with a description
|
||||
text = re.sub(r"```([\s\S]+?)```", r"(code block omitted)", text)
|
||||
|
||||
# Remove image syntax but add alt text if available
|
||||
text = re.sub(r"!\[([^\]]*)\]\([^\)]+\)", r"Image: \1", text)
|
||||
|
||||
# Remove HTML tags
|
||||
text = re.sub(r"</?[^>]+(>|$)", '', text)
|
||||
|
||||
# Normalize line breaks
|
||||
text = re.sub(r"\n{2,}", '\n\n', text) # Ensure consistent paragraph separation
|
||||
|
||||
# Replace multiple spaces within lines
|
||||
text = re.sub(r" {2,}", ' ', text)
|
||||
|
||||
# Trim leading and trailing whitespace from the whole text
|
||||
text = text.strip()
|
||||
|
||||
return text
|
||||
@@ -0,0 +1,5 @@
|
||||
flask
|
||||
gevent
|
||||
python-dotenv
|
||||
edge-tts
|
||||
emoji
|
||||
@@ -0,0 +1,167 @@
|
||||
# server.py
|
||||
|
||||
from flask import Flask, request, send_file, jsonify
|
||||
from gevent.pywsgi import WSGIServer
|
||||
from dotenv import load_dotenv
|
||||
import os
|
||||
|
||||
from handle_text import prepare_tts_input_with_context
|
||||
from tts_handler import generate_speech, get_models, get_voices
|
||||
from utils import getenv_bool, require_api_key, AUDIO_FORMAT_MIME_TYPES
|
||||
|
||||
app = Flask(__name__)
|
||||
load_dotenv()
|
||||
|
||||
API_KEY = os.getenv('API_KEY', 'your_api_key_here')
|
||||
PORT = int(os.getenv('PORT', 5050))
|
||||
|
||||
DEFAULT_VOICE = os.getenv('DEFAULT_VOICE', 'en-US-AvaNeural')
|
||||
DEFAULT_RESPONSE_FORMAT = os.getenv('DEFAULT_RESPONSE_FORMAT', 'mp3')
|
||||
DEFAULT_SPEED = float(os.getenv('DEFAULT_SPEED', 1.0))
|
||||
|
||||
REMOVE_FILTER = getenv_bool('REMOVE_FILTER', False)
|
||||
EXPAND_API = getenv_bool('EXPAND_API', True)
|
||||
|
||||
# DEFAULT_MODEL = os.getenv('DEFAULT_MODEL', 'tts-1')
|
||||
|
||||
@app.route('/v1/audio/speech', methods=['POST'])
|
||||
@app.route('/audio/speech', methods=['POST']) # Add this line for the alias
|
||||
@require_api_key
|
||||
def text_to_speech():
|
||||
data = request.json
|
||||
if not data or 'input' not in data:
|
||||
return jsonify({"error": "Missing 'input' in request body"}), 400
|
||||
|
||||
text = data.get('input')
|
||||
|
||||
if not REMOVE_FILTER:
|
||||
text = prepare_tts_input_with_context(text)
|
||||
|
||||
# model = data.get('model', DEFAULT_MODEL)
|
||||
voice = data.get('voice', DEFAULT_VOICE)
|
||||
|
||||
response_format = data.get('response_format', DEFAULT_RESPONSE_FORMAT)
|
||||
speed = float(data.get('speed', DEFAULT_SPEED))
|
||||
|
||||
mime_type = AUDIO_FORMAT_MIME_TYPES.get(response_format, "audio/mpeg")
|
||||
|
||||
# Generate the audio file in the specified format with speed adjustment
|
||||
output_file_path = generate_speech(text, voice, response_format, speed)
|
||||
|
||||
# Return the file with the correct MIME type
|
||||
return send_file(output_file_path, mimetype=mime_type, as_attachment=True, download_name=f"speech.{response_format}")
|
||||
|
||||
@app.route('/v1/models', methods=['GET', 'POST'])
|
||||
@app.route('/models', methods=['GET', 'POST'])
|
||||
@require_api_key
|
||||
def list_models():
|
||||
return jsonify({"data": get_models()})
|
||||
|
||||
@app.route('/v1/voices', methods=['GET', 'POST'])
|
||||
@app.route('/voices', methods=['GET', 'POST'])
|
||||
@require_api_key
|
||||
def list_voices():
|
||||
specific_language = None
|
||||
|
||||
data = request.args if request.method == 'GET' else request.json
|
||||
if data and ('language' in data or 'locale' in data):
|
||||
specific_language = data.get('language') if 'language' in data else data.get('locale')
|
||||
|
||||
return jsonify({"voices": get_voices(specific_language)})
|
||||
|
||||
@app.route('/v1/voices/all', methods=['GET', 'POST'])
|
||||
@app.route('/voices/all', methods=['GET', 'POST'])
|
||||
@require_api_key
|
||||
def list_all_voices():
|
||||
return jsonify({"voices": get_voices('all')})
|
||||
|
||||
"""
|
||||
Support for ElevenLabs and Azure AI Speech
|
||||
(currently in beta)
|
||||
"""
|
||||
|
||||
# http://localhost:5050/elevenlabs/v1/text-to-speech
|
||||
# http://localhost:5050/elevenlabs/v1/text-to-speech/en-US-AndrewNeural
|
||||
@app.route('/elevenlabs/v1/text-to-speech/<voice_id>', methods=['POST'])
|
||||
@require_api_key
|
||||
def elevenlabs_tts(voice_id):
|
||||
if not EXPAND_API:
|
||||
return jsonify({"error": f"Endpoint not allowed"}), 500
|
||||
|
||||
# Parse the incoming JSON payload
|
||||
try:
|
||||
payload = request.json
|
||||
if not payload or 'text' not in payload:
|
||||
return jsonify({"error": "Missing 'text' in request body"}), 400
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"Invalid JSON payload: {str(e)}"}), 400
|
||||
|
||||
text = payload['text']
|
||||
|
||||
if not REMOVE_FILTER:
|
||||
text = prepare_tts_input_with_context(text)
|
||||
|
||||
voice = voice_id # ElevenLabs uses the voice_id in the URL
|
||||
|
||||
# Use default settings for edge-tts
|
||||
response_format = 'mp3'
|
||||
speed = DEFAULT_SPEED # Optional customization via payload.get('speed', DEFAULT_SPEED)
|
||||
|
||||
# Generate speech using edge-tts
|
||||
try:
|
||||
output_file_path = generate_speech(text, voice, response_format, speed)
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"TTS generation failed: {str(e)}"}), 500
|
||||
|
||||
# Return the generated audio file
|
||||
return send_file(output_file_path, mimetype="audio/mpeg", as_attachment=True, download_name="speech.mp3")
|
||||
|
||||
# tts.speech.microsoft.com/cognitiveservices/v1
|
||||
# https://{region}.tts.speech.microsoft.com/cognitiveservices/v1
|
||||
# http://localhost:5050/azure/cognitiveservices/v1
|
||||
@app.route('/azure/cognitiveservices/v1', methods=['POST'])
|
||||
@require_api_key
|
||||
def azure_tts():
|
||||
if not EXPAND_API:
|
||||
return jsonify({"error": f"Endpoint not allowed"}), 500
|
||||
|
||||
# Parse the SSML payload
|
||||
try:
|
||||
ssml_data = request.data.decode('utf-8')
|
||||
if not ssml_data:
|
||||
return jsonify({"error": "Missing SSML payload"}), 400
|
||||
|
||||
# Extract the text and voice from SSML
|
||||
from xml.etree import ElementTree as ET
|
||||
root = ET.fromstring(ssml_data)
|
||||
text = root.find('.//{http://www.w3.org/2001/10/synthesis}voice').text
|
||||
voice = root.find('.//{http://www.w3.org/2001/10/synthesis}voice').get('name')
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"Invalid SSML payload: {str(e)}"}), 400
|
||||
|
||||
# Use default settings for edge-tts
|
||||
response_format = 'mp3'
|
||||
speed = DEFAULT_SPEED
|
||||
|
||||
if not REMOVE_FILTER:
|
||||
text = prepare_tts_input_with_context(text)
|
||||
|
||||
# Generate speech using edge-tts
|
||||
try:
|
||||
output_file_path = generate_speech(text, voice, response_format, speed)
|
||||
except Exception as e:
|
||||
return jsonify({"error": f"TTS generation failed: {str(e)}"}), 500
|
||||
|
||||
# Return the generated audio file
|
||||
return send_file(output_file_path, mimetype="audio/mpeg", as_attachment=True, download_name="speech.mp3")
|
||||
|
||||
print(f" Edge TTS (Free Azure TTS) Replacement for OpenAI's TTS API")
|
||||
print(f" ")
|
||||
print(f" * Serving OpenAI Edge TTS")
|
||||
print(f" * Server running on http://localhost:{PORT}")
|
||||
print(f" * TTS Endpoint: http://localhost:{PORT}/v1/audio/speech")
|
||||
print(f" ")
|
||||
|
||||
if __name__ == '__main__':
|
||||
http_server = WSGIServer(('0.0.0.0', PORT), app)
|
||||
http_server.serve_forever()
|
||||
@@ -0,0 +1,133 @@
|
||||
import edge_tts
|
||||
import asyncio
|
||||
import tempfile
|
||||
import subprocess
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# Language default (environment variable)
|
||||
DEFAULT_LANGUAGE = os.getenv('DEFAULT_LANGUAGE', 'en-US')
|
||||
|
||||
# OpenAI voice names mapped to edge-tts equivalents
|
||||
voice_mapping = {
|
||||
'alloy': 'en-US-AvaNeural',
|
||||
'echo': 'en-US-AndrewNeural',
|
||||
'fable': 'en-GB-SoniaNeural',
|
||||
'onyx': 'en-US-EricNeural',
|
||||
'nova': 'en-US-SteffanNeural',
|
||||
'shimmer': 'en-US-EmmaNeural'
|
||||
}
|
||||
|
||||
def is_ffmpeg_installed():
|
||||
"""Check if FFmpeg is installed and accessible."""
|
||||
try:
|
||||
subprocess.run(['ffmpeg', '-version'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
return True
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return False
|
||||
|
||||
async def _generate_audio(text, voice, response_format, speed):
|
||||
"""Generate TTS audio and optionally convert to a different format."""
|
||||
# Determine if the voice is an OpenAI-compatible voice or a direct edge-tts voice
|
||||
edge_tts_voice = voice_mapping.get(voice, voice) # Use mapping if in OpenAI names, otherwise use as-is
|
||||
|
||||
# Generate the TTS output in mp3 format first
|
||||
temp_output_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
|
||||
|
||||
# Convert speed to SSML rate format
|
||||
try:
|
||||
speed_rate = speed_to_rate(speed) # Convert speed value to "+X%" or "-X%"
|
||||
except Exception as e:
|
||||
print(f"Error converting speed: {e}. Defaulting to +0%.")
|
||||
speed_rate = "+0%"
|
||||
|
||||
# Generate the MP3 file
|
||||
communicator = edge_tts.Communicate(text=text, voice=edge_tts_voice, rate=speed_rate)
|
||||
await communicator.save(temp_output_file.name)
|
||||
|
||||
# If the requested format is mp3, return the generated file directly
|
||||
if response_format == "mp3":
|
||||
return temp_output_file.name
|
||||
|
||||
# Check if FFmpeg is installed
|
||||
if not is_ffmpeg_installed():
|
||||
print("FFmpeg is not available. Returning unmodified mp3 file.")
|
||||
return temp_output_file.name
|
||||
|
||||
# Create a new temporary file for the converted output
|
||||
converted_output_file = tempfile.NamedTemporaryFile(delete=False, suffix=f".{response_format}")
|
||||
|
||||
# Build the FFmpeg command
|
||||
ffmpeg_command = [
|
||||
"ffmpeg",
|
||||
"-i", temp_output_file.name, # Input file
|
||||
"-c:a", {
|
||||
"aac": "aac",
|
||||
"mp3": "libmp3lame",
|
||||
"wav": "pcm_s16le",
|
||||
"opus": "libopus",
|
||||
"flac": "flac"
|
||||
}.get(response_format, "aac"), # Default to AAC if unknown
|
||||
"-b:a", "192k" if response_format != "wav" else None, # Bitrate not needed for WAV
|
||||
"-f", {
|
||||
"aac": "mp4", # AAC in MP4 container
|
||||
"mp3": "mp3",
|
||||
"wav": "wav",
|
||||
"opus": "ogg",
|
||||
"flac": "flac"
|
||||
}.get(response_format, response_format), # Default to matching format
|
||||
"-y", # Overwrite without prompt
|
||||
converted_output_file.name # Output file
|
||||
]
|
||||
|
||||
try:
|
||||
# Run FFmpeg command and ensure no errors occur
|
||||
subprocess.run(ffmpeg_command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"FFmpeg error during audio conversion: {e}")
|
||||
|
||||
# Clean up the original temporary file
|
||||
Path(temp_output_file.name).unlink(missing_ok=True)
|
||||
|
||||
return converted_output_file.name
|
||||
|
||||
def generate_speech(text, voice, response_format, speed=1.0):
|
||||
return asyncio.run(_generate_audio(text, voice, response_format, speed))
|
||||
|
||||
def get_models():
|
||||
return [
|
||||
{"id": "tts-1", "name": "Text-to-speech v1"},
|
||||
{"id": "tts-1-hd", "name": "Text-to-speech v1 HD"}
|
||||
]
|
||||
|
||||
async def _get_voices(language=None):
|
||||
# List all voices, filter by language if specified
|
||||
all_voices = await edge_tts.list_voices()
|
||||
language = language or DEFAULT_LANGUAGE # Use default if no language specified
|
||||
filtered_voices = [
|
||||
{"name": v['ShortName'], "gender": v['Gender'], "language": v['Locale']}
|
||||
for v in all_voices if language == 'all' or language is None or v['Locale'] == language
|
||||
]
|
||||
return filtered_voices
|
||||
|
||||
def get_voices(language=None):
|
||||
return asyncio.run(_get_voices(language))
|
||||
|
||||
def speed_to_rate(speed: float) -> str:
|
||||
"""
|
||||
Converts a multiplicative speed value to the edge-tts "rate" format.
|
||||
|
||||
Args:
|
||||
speed (float): The multiplicative speed value (e.g., 1.5 for +50%, 0.5 for -50%).
|
||||
|
||||
Returns:
|
||||
str: The formatted "rate" string (e.g., "+50%" or "-50%").
|
||||
"""
|
||||
if speed < 0 or speed > 2:
|
||||
raise ValueError("Speed must be between 0 and 2 (inclusive).")
|
||||
|
||||
# Convert speed to percentage change
|
||||
percentage_change = (speed - 1) * 100
|
||||
|
||||
# Format with a leading "+" or "-" as required
|
||||
return f"{percentage_change:+.0f}%"
|
||||
@@ -0,0 +1,38 @@
|
||||
# utils.py
|
||||
|
||||
from flask import request, jsonify
|
||||
from functools import wraps
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
def getenv_bool(name: str, default: bool = False) -> bool:
|
||||
return os.getenv(name, str(default)).lower() in ("yes", "y", "true", "1", "t")
|
||||
|
||||
API_KEY = os.getenv('API_KEY', 'your_api_key_here')
|
||||
REQUIRE_API_KEY = getenv_bool('REQUIRE_API_KEY', True)
|
||||
|
||||
def require_api_key(f):
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
if not REQUIRE_API_KEY:
|
||||
return f(*args, **kwargs)
|
||||
auth_header = request.headers.get('Authorization')
|
||||
if not auth_header or not auth_header.startswith('Bearer '):
|
||||
return jsonify({"error": "Missing or invalid API key"}), 401
|
||||
token = auth_header.split('Bearer ')[1]
|
||||
if token != API_KEY:
|
||||
return jsonify({"error": "Invalid API key"}), 401
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
# Mapping of audio format to MIME type
|
||||
AUDIO_FORMAT_MIME_TYPES = {
|
||||
"mp3": "audio/mpeg",
|
||||
"opus": "audio/ogg",
|
||||
"aac": "audio/aac",
|
||||
"flac": "audio/flac",
|
||||
"wav": "audio/wav",
|
||||
"pcm": "audio/L16"
|
||||
}
|
||||
@@ -1,131 +0,0 @@
|
||||
import os
|
||||
from transformers import AutoTokenizer, AutoModelForCausalLM
|
||||
import torch
|
||||
import soundfile as sf
|
||||
from xcodec2.modeling_xcodec2 import XCodec2Model
|
||||
import torchaudio
|
||||
|
||||
|
||||
class TextToSpeech:
|
||||
def __init__(self, sample_audio_path, sample_audio_text):
|
||||
self.sample_audio_text = sample_audio_text
|
||||
# 初始化模型
|
||||
llasa_3b = "HKUSTAudio/Llasa-3B"
|
||||
xcodec2 = "HKUSTAudio/xcodec2"
|
||||
|
||||
self.tokenizer = AutoTokenizer.from_pretrained(llasa_3b)
|
||||
self.llasa_3b_model = AutoModelForCausalLM.from_pretrained(
|
||||
llasa_3b,
|
||||
trust_remote_code=True,
|
||||
device_map="auto",
|
||||
)
|
||||
self.llasa_3b_model.eval()
|
||||
|
||||
self.xcodec_model = XCodec2Model.from_pretrained(xcodec2)
|
||||
self.xcodec_model.eval().cuda()
|
||||
|
||||
# 处理音频
|
||||
waveform, sample_rate = torchaudio.load(sample_audio_path)
|
||||
if len(waveform[0]) / sample_rate > 15:
|
||||
print("已将音频裁剪至前15秒。")
|
||||
waveform = waveform[:, : sample_rate * 15]
|
||||
|
||||
# 检查音频是否为立体声
|
||||
if waveform.size(0) > 1:
|
||||
waveform_mono = torch.mean(waveform, dim=0, keepdim=True)
|
||||
else:
|
||||
waveform_mono = waveform
|
||||
|
||||
self.prompt_wav = torchaudio.transforms.Resample(
|
||||
orig_freq=sample_rate, new_freq=16000
|
||||
)(waveform_mono)
|
||||
|
||||
# Encode the prompt wav
|
||||
vq_code_prompt = self.xcodec_model.encode_code(input_waveform=self.prompt_wav)
|
||||
vq_code_prompt = vq_code_prompt[0, 0, :]
|
||||
self.speech_ids_prefix = self.ids_to_speech_tokens(vq_code_prompt)
|
||||
self.speech_end_id = self.tokenizer.convert_tokens_to_ids("<|SPEECH_GENERATION_END|>")
|
||||
|
||||
def ids_to_speech_tokens(self, speech_ids):
|
||||
speech_tokens_str = []
|
||||
for speech_id in speech_ids:
|
||||
speech_tokens_str.append(f"<|s_{speech_id}|>")
|
||||
return speech_tokens_str
|
||||
|
||||
def extract_speech_ids(self, speech_tokens_str):
|
||||
speech_ids = []
|
||||
for token_str in speech_tokens_str:
|
||||
if token_str.startswith("<|s_") and token_str.endswith("|>"):
|
||||
num_str = token_str[4:-2]
|
||||
num = int(num_str)
|
||||
speech_ids.append(num)
|
||||
else:
|
||||
print(f"Unexpected token: {token_str}")
|
||||
return speech_ids
|
||||
|
||||
@torch.inference_mode()
|
||||
def infer(self, target_text):
|
||||
if len(target_text) == 0:
|
||||
return None
|
||||
elif len(target_text) > 300:
|
||||
print("文本过长，请保持在300字符以内。")
|
||||
target_text = target_text[:300]
|
||||
|
||||
input_text = self.sample_audio_text + " " + target_text
|
||||
|
||||
formatted_text = (
|
||||
f"<|TEXT_UNDERSTANDING_START|>{input_text}<|TEXT_UNDERSTANDING_END|>"
|
||||
)
|
||||
|
||||
chat = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Convert the text to speech:" + formatted_text,
|
||||
},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "<|SPEECH_GENERATION_START|>"
|
||||
+ "".join(self.speech_ids_prefix),
|
||||
},
|
||||
]
|
||||
|
||||
input_ids = self.tokenizer.apply_chat_template(
|
||||
chat, tokenize=True, return_tensors="pt", continue_final_message=True
|
||||
)
|
||||
input_ids = input_ids.to("cuda")
|
||||
|
||||
outputs = self.llasa_3b_model.generate(
|
||||
input_ids,
|
||||
max_length=2048,
|
||||
eos_token_id=self.speech_end_id,
|
||||
do_sample=True,
|
||||
top_p=1,
|
||||
temperature=0.8,
|
||||
)
|
||||
generated_ids = outputs[0][input_ids.shape[1] - len(self.speech_ids_prefix): -1]
|
||||
|
||||
speech_tokens = self.tokenizer.batch_decode(
|
||||
generated_ids, skip_special_tokens=True
|
||||
)
|
||||
|
||||
speech_tokens = self.extract_speech_ids(speech_tokens)
|
||||
speech_tokens = torch.tensor(speech_tokens).cuda().unsqueeze(0).unsqueeze(0)
|
||||
|
||||
gen_wav = self.xcodec_model.decode_code(speech_tokens)
|
||||
gen_wav = gen_wav[:, :, self.prompt_wav.shape[1]:]
|
||||
|
||||
return (16000, gen_wav[0, 0, :].cpu().numpy())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 如果遇到问题，请尝试将参考音频转换为WAV或MP3格式，将其裁剪至15秒以内，并缩短提示文本。
|
||||
sample_audio_text = "对，这就是我万人敬仰的太乙真人，虽然有点婴儿肥，但也掩不住我逼人的帅气。"
|
||||
sample_audio_path = os.path.join(os.path.dirname(__file__), "sample.wav")
|
||||
|
||||
tts = TextToSpeech(sample_audio_path, sample_audio_text)
|
||||
target_text = "晚上好啊，吃了吗您"
|
||||
result = tts.infer(target_text)
|
||||
sf.write(os.path.join(os.path.dirname(__file__), "output.wav"), result[1], result[0])
|
||||
target_text = "我是老北京正黄旗！"
|
||||
result = tts.infer(target_text)
|
||||
sf.write(os.path.join(os.path.dirname(__file__), "output1.wav"), result[1], result[0])
|
||||
Reference in New Issue
Block a user