Text Chunking for TTS

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

1
# For more Python SDK migration guides, visit:
2
# https://github.com/deepgram/deepgram-python-sdk/tree/main/docs
3
4
import re
5
import asyncio
6
from deepgram import AsyncDeepgramClient
7
8
def chunk_by_sentence(text):
9
    # Split text at sentence boundaries (periods, question marks, exclamation points)
10
    # while preserving the punctuation
11
    sentences = re.split(r'(?<=[.!?])\s+', text)
12
13
    # Remove any empty chunks
14
    return [sentence for sentence in sentences if sentence]
15
16
class SimpleTextChunker:
17
    def __init__(self, deepgram_client):
18
        self.queue = []  # Queue to store incoming paragraph chunks
19
        self.processed_sentences = set()
20
        self.deepgram_client = deepgram_client
21
22
    async def process_text_stream(self, paragraph):
23
        """Process an array of paragraph chunks, each containing 1-2 sentences"""
24
25
        # Queue paragraph as it arrives (simulating fast reception)
26
        self.queue.append(paragraph)
27
        print(f"Received and queued paragraph: {paragraph}")
28
29
        # You could preprocess paragraphs here and split them by more than just sentence boundaries
30
31
        # Process the queue
32
        while self.queue:
33
            # Get the next paragraph from the queue
34
            paragraph = self.queue.pop(0)
35
36
            # Split paragraph into sentences using our chunk_by_sentence function
37
            sentences = chunk_by_sentence(paragraph)
38
39
            # Process each sentence
40
            for sentence in sentences:
41
                if sentence and sentence not in self.processed_sentences:
42
                    # Send the sentence to TTS
43
                    print(f"Sending sentence to TTS: {sentence}")
44
                    audio_response = await self.deepgram_client.speak.v1.audio.generate(
45
                        text=sentence,
46
                        model="aura-2-thalia-en",
47
                        sample_rate=24000
48
                    )
49
                    # In a real app, you would play this audio immediately
50
                    self.processed_sentences.add(sentence)
51
52
# Example usage with an array of paragraph chunks
53
async def main():
54
    # This simulates text coming in as paragraph chunks from an LLM
55
    paragraph_chunks = [
56
        "Deepgram's TTS API offers low latency. It works great for voice agents.",
57
        "This approach simulates receiving chunks as paragraphs. Each paragraph may contain one or two sentences.",
58
        "Try it today! You'll be impressed with the results."
59
    ]
60
61
    # Set up TTS client
62
    deepgram = AsyncDeepgramClient()
63
64
    # Set up listeners for TTS events to handle audio data and connection status
65
66
    chunker = SimpleTextChunker(deepgram)
67
    # Process each paragraph sequentially
68
    for paragraph in paragraph_chunks:
69
        await chunker.process_text_stream(paragraph)
70
71
# Run the example
72
if __name__ == "__main__":
73
    asyncio.run(main())

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