from fish_audio_sdk import WebSocketSession, TTSRequest# Create WebSocket sessionws_session = WebSocketSession("your_api_key")# Define text generatordef text_stream(): yield "Hello, " yield "this is " yield "streaming text!"# Stream and save audiowith ws_session: with open("output.mp3", "wb") as f: for audio_chunk in ws_session.tts( TTSRequest(text=""), # Empty text for streaming text_stream() ): f.write(audio_chunk)
Set text="" in TTSRequest when streaming. The actual text comes from your text_stream generator.
def generate_text(): # Simulate dynamic text generation responses = [ "Processing your request...", "Here's what I found:", "The answer is 42." ] for response in responses: # Split into smaller chunks for smoother streaming words = response.split() for word in words: yield word + " "with ws_session: for audio_chunk in ws_session.tts( TTSRequest(text=""), generate_text() ): # Process audio in real-time pass
import openaidef stream_chatgpt_response(prompt): response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], stream=True ) for chunk in response: if content := chunk.choices[0].delta.get("content"): yield contentwith ws_session: for audio_chunk in ws_session.tts( TTSRequest(text=""), stream_chatgpt_response("Tell me a joke") ): # Play or save audio pass
def read_file_lines(filepath): with open(filepath, "r") as f: for line in f: yield line.strip() + " "with ws_session: for audio_chunk in ws_session.tts( TTSRequest(text=""), read_file_lines("story.txt") ): # Process each chunk pass
from fish_audio_sdk.exceptions import WebSocketErrtry: with ws_session: for audio_chunk in ws_session.tts( TTSRequest(text=""), text_stream() ): # Process audio passexcept WebSocketErr: print("WebSocket connection failed") # Fallback to regular TTS or retry