Text Chunking for TTS

# For more Python SDK migration guides, visit:
# https://github.com/deepgram/deepgram-python-sdk/tree/main/docs
import re
def chunk_by_sentence(text):
    # Split text at sentence boundaries (periods, question marks, exclamation points)
    # while preserving the punctuation
    sentences = re.split(r'(?<=[.!?])\s+', text)
    # Remove any empty chunks
    return [sentence for sentence in sentences if sentence]
# Example usage
text = "Hello, welcome to Deepgram. This is an example of text chunking. How does it sound?"
chunks = chunk_by_sentence(text)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}: {chunk}")
# Output:
# Chunk 1: Hello, welcome to Deepgram.
# Chunk 2: This is an example of text chunking.
# Chunk 3: How does it sound?

# For more Python SDK migration guides, visit:
# https://github.com/deepgram/deepgram-python-sdk/tree/main/docs
import re
import asyncio
from deepgram import AsyncDeepgramClient
class SimpleTextChunker:
    def __init__(self, deepgram_client):
        self.queue = []  # Queue to store incoming paragraph chunks
        self.deepgram_client = deepgram_client
        self.chunk_number = 0
        self.pending_text = ""
    async def synthesize_and_save(self, text):
        audio_response = self.deepgram_client.speak.v1.audio.generate(
            text=text,
            model="aura-2-thalia-en"
        )
        audio_data = b"".join([chunk async for chunk in audio_response])
        self.chunk_number += 1
        filename = f"chunk_{self.chunk_number}.mp3"
        with open(filename, "wb") as audio_file:
            audio_file.write(audio_data)
        print(f"Audio saved to {filename}")
    async def process_text_stream(self, paragraph):
        """Process an array of paragraph chunks, each containing 1-2 sentences"""
        # Queue paragraph as it arrives (simulating fast reception)
        self.queue.append(paragraph)
        print(f"Received and queued paragraph: {paragraph}")
        # You could preprocess paragraphs here and split them by more than just sentence boundaries
        # Process the queue
        while self.queue:
            # Get the next paragraph from the queue
            paragraph = self.queue.pop(0)
            self.pending_text += paragraph
            matches = list(re.finditer(r'[^.!?]+[.!?]', self.pending_text))
            if not matches:
                continue
            self.pending_text = self.pending_text[matches[-1].end():].lstrip()
            # Process each sentence
            for match in matches:
                sentence = match.group().strip()
                # Send the sentence to TTS
                print(f"Sending sentence to TTS: {sentence}")
                await self.synthesize_and_save(sentence)
    async def flush(self):
        if self.pending_text.strip():
            await self.synthesize_and_save(self.pending_text.strip())
            self.pending_text = ""
# Example usage with an array of paragraph chunks
async def main():
    # This simulates text coming in as paragraph chunks from an LLM
    paragraph_chunks = [
        "Hello",
        " world. Deepgram's TTS API offers low latency.",
        "It works great for voice agents.",
        "This approach simulates receiving chunks as paragraphs. Each paragraph may contain one or two sentences.",
        "Try it today! You'll be impressed with the results."
    ]
    # Set up TTS client
    deepgram = AsyncDeepgramClient()
    chunker = SimpleTextChunker(deepgram)
    # Process each paragraph sequentially
    for paragraph in paragraph_chunks:
        await chunker.process_text_stream(paragraph)
    await chunker.flush()
# Run the example
if __name__ == "__main__":
    asyncio.run(main())

# For more Python SDK migration guides, visit:
# https://github.com/deepgram/deepgram-python-sdk/tree/main/docs
async def process_chunks_sequential(chunks, tts_function):
    results = []
    for i, chunk in enumerate(chunks):
        # You might prioritize the first chunk for faster response
        result = await tts_function(chunk)
        results.append(result)
    return results