6. Audio Integration

Description

The AudioModule adds the sound dimension to the Document Analyzer. After processing a document, extracting data and generating a summary, the system can convert that summary into audio — useful for accessibility, on-the-go consumption, or simply as an alternative delivery format. Optionally, it can also accept questions by audio (STT → Q&A → TTS), creating a complete conversational interface.

Why it matters: Not every user wants to read. An executive who receives 20 documents a day would rather listen to the summaries while driving. A person with a visual impairment needs audio as the primary output. Adding TTS turns the Document Analyzer from a text tool into a genuinely multimodal tool in its output.

Connection with the module: In Module 5 you learned to use OpenAI's TTS API to turn text into speech, and Whisper to transcribe audio. Here you integrate both capabilities into the Document Analyzer: TTS to generate spoken summaries, and optionally STT to receive questions by audio.

AudioModule Architecture

Responsibilities

AudioModule
├── generate_summary_audio()     → Convert a text summary into an audio file
├── generate_long_audio()        → Handle summaries that exceed the TTS limit
├── transcribe_question()        → (Optional) Convert an audio question to text
└── audio_qa_pipeline()          → (Optional) Audio question → text → Q&A → audio answer

Audio pipeline

Text summary
    │
    ├── Fewer than 4096 chars? ── Yes ──→ Direct TTS → .mp3 file
    │
    └── More than 4096 chars? ─── Yes ──→ Split into chunks
                                             │
                                             ▼
                                        TTS per chunk → multiple .mp3
                                             │
                                             ▼
                                        Concatenate with pydub → final .mp3 file

Audio Q&A pipeline (optional)

Audio question (.mp3/.wav)
    │
    ▼
Whisper STT → Question text
    │
    ▼
RAGModule.query() → Text answer
    │
    ▼
TTS → Answer audio (.mp3)

Implementation: AudioModule

Complete class

import logging
import os
import tempfile
from pathlib import Path
from typing import Optional

from openai import OpenAI

logger = logging.getLogger(__name__)

TTS_CHAR_LIMIT = 4096
SUPPORTED_VOICES = {"alloy", "echo", "fable", "onyx", "nova", "shimmer"}
TTS_MODEL = "tts-1"
TTS_MODEL_HD = "tts-1-hd"
STT_MODEL = "whisper-1"


class AudioModule:
    def __init__(self, output_dir: str = "/tmp/audio"):
        self.client = OpenAI()
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def generate_summary_audio(
        self,
        text: str,
        filename: str = "summary.mp3",
        voice: str = "nova",
        hd: bool = False
    ) -> str:
        if voice not in SUPPORTED_VOICES:
            logger.warning(f"Voice '{voice}' not recognized, using 'nova'")
            voice = "nova"

        if len(text) > TTS_CHAR_LIMIT:
            return self.generate_long_audio(text, filename, voice, hd)

        output_path = os.path.join(self.output_dir, filename)
        model = TTS_MODEL_HD if hd else TTS_MODEL

        response = self.client.audio.speech.create(
            model=model,
            voice=voice,
            input=text
        )
        response.stream_to_file(output_path)

        file_size = os.path.getsize(output_path)
        logger.info(
            f"Audio generated: {output_path} "
            f"({file_size / 1024:.1f} KB, {len(text)} chars)"
        )

        return output_path

    def generate_long_audio(
        self,
        text: str,
        filename: str = "summary.mp3",
        voice: str = "nova",
        hd: bool = False
    ) -> str:
        try:
            from pydub import AudioSegment
        except ImportError:
            logger.warning(
                "pydub not installed. Truncating summary to 4096 chars."
            )
            return self.generate_summary_audio(
                text[:TTS_CHAR_LIMIT], filename, voice, hd
            )

        chunks = self._split_text_for_tts(text)
        model = TTS_MODEL_HD if hd else TTS_MODEL
        temp_files = []

        try:
            for i, chunk in enumerate(chunks):
                temp_path = os.path.join(self.output_dir, f"_temp_part_{i}.mp3")
                response = self.client.audio.speech.create(
                    model=model,
                    voice=voice,
                    input=chunk
                )
                response.stream_to_file(temp_path)
                temp_files.append(temp_path)
                logger.info(f"Part {i+1}/{len(chunks)} generated ({len(chunk)} chars)")

            segments = [AudioSegment.from_mp3(f) for f in temp_files]

            pause = AudioSegment.silent(duration=500)
            combined = segments[0]
            for seg in segments[1:]:
                combined = combined + pause + seg

            output_path = os.path.join(self.output_dir, filename)
            combined.export(output_path, format="mp3")

            total_duration = len(combined) / 1000
            logger.info(
                f"Long audio generated: {output_path} "
                f"({len(chunks)} parts, {total_duration:.1f}s)"
            )

            return output_path

        finally:
            for f in temp_files:
                try:
                    os.remove(f)
                except OSError:
                    pass

    def transcribe_question(self, audio_path: str) -> str:
        with open(audio_path, "rb") as f:
            transcript = self.client.audio.transcriptions.create(
                model=STT_MODEL,
                file=f,
                language="es"
            )
        logger.info(f"Transcription: '{transcript.text[:100]}...'")
        return transcript.text

    def audio_qa_pipeline(
        self,
        audio_question_path: str,
        rag_module,
        doc_id: Optional[str] = None,
        voice: str = "nova"
    ) -> dict:
        question_text = self.transcribe_question(audio_question_path)

        qa_result = rag_module.query(question=question_text, doc_id=doc_id)

        answer_audio_path = self.generate_summary_audio(
            text=qa_result.answer,
            filename=f"qa_answer_{doc_id or 'all'}.mp3",
            voice=voice
        )

        return {
            "question_text": question_text,
            "answer_text": qa_result.answer,
            "answer_audio_path": answer_audio_path,
            "sources": qa_result.sources,
            "confidence": qa_result.confidence
        }

    def _split_text_for_tts(self, text: str) -> list[str]:
        if len(text) <= TTS_CHAR_LIMIT:
            return [text]

        chunks = []
        sentences = text.replace(". ", ".\n").split("\n")
        current_chunk = ""

        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue

            if len(current_chunk) + len(sentence) + 1 <= TTS_CHAR_LIMIT:
                current_chunk += (" " + sentence if current_chunk else sentence)
            else:
                if current_chunk:
                    chunks.append(current_chunk)
                if len(sentence) > TTS_CHAR_LIMIT:
                    for i in range(0, len(sentence), TTS_CHAR_LIMIT):
                        chunks.append(sentence[i:i + TTS_CHAR_LIMIT])
                    current_chunk = ""
                else:
                    current_chunk = sentence

        if current_chunk:
            chunks.append(current_chunk)

        return chunks

    def estimate_cost(self, text: str) -> dict:
        char_count = len(text)
        cost_per_char = 15.0 / 1_000_000  # $15 per 1M chars for tts-1
        cost_per_char_hd = 30.0 / 1_000_000  # $30 per 1M chars for tts-1-hd

        return {
            "characters": char_count,
            "chunks_needed": max(1, (char_count + TTS_CHAR_LIMIT - 1) // TTS_CHAR_LIMIT),
            "cost_tts1": round(char_count * cost_per_char, 4),
            "cost_tts1_hd": round(char_count * cost_per_char_hd, 4)
        }

Voice Configuration

Voices available in OpenAI TTS

Voice selection by document type

VOICE_BY_DOCTYPE = {
    "invoice": "alloy",      # neutral for financial data
    "contract": "onyx",      # formal for legal documents
    "manual": "fable",       # warm for instructions
    "report": "nova",        # friendly for summaries
}


def get_voice_for_document(doc_type: str, user_preference: Optional[str] = None) -> str:
    if user_preference and user_preference in SUPPORTED_VOICES:
        return user_preference
    return VOICE_BY_DOCTYPE.get(doc_type, "nova")

TTS Models: Standard vs HD

Comparison

When to use HD

def should_use_hd(doc_type: str, text_length: int) -> bool:
    if doc_type in ("contract", "report") and text_length < 2000:
        return True
    return False

For the Document Analyzer, tts-1 is enough in most cases. Use tts-1-hd only for important documents where audio quality is critical.

Integration with the Document Analyzer

Full flow with audio

def analyze_with_audio(
    processor: DocumentProcessor,
    analyzer: VisionAnalyzer,
    summarizer,
    audio: AudioModule,
    file_path: str,
    voice: str = "nova",
    hd: bool = False
) -> dict:
    content = processor.process(file_path)
    doc_type = analyzer.classify(content)
    summary = summarizer.summarize(content)

    cost_estimate = audio.estimate_cost(summary)
    logger.info(f"Estimated TTS cost: ${cost_estimate['cost_tts1']:.4f}")

    audio_path = audio.generate_summary_audio(
        text=summary,
        filename=f"{Path(file_path).stem}_summary.mp3",
        voice=voice,
        hd=hd
    )

    return {
        "summary": summary,
        "audio_path": audio_path,
        "audio_cost": cost_estimate,
        "document_type": doc_type
    }

Serving Audio from FastAPI

Endpoint for audio

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()


@app.get("/audio/{filename}")
async def serve_audio(filename: str):
    audio_dir = "/tmp/audio"
    file_path = os.path.join(audio_dir, filename)

    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="Audio not found")

    if not filename.endswith((".mp3", ".wav")):
        raise HTTPException(status_code=400, detail="Unsupported audio format")

    return FileResponse(
        file_path,
        media_type="audio/mpeg",
        filename=filename
    )

Response from the /analyze endpoint with audio

When generate_audio_summary=True, the response includes the audio URL:

{
    "success": true,
    "summary": "Invoice FAC-2025-0042...",
    "audio_summary_url": "/audio/invoice_march_summary.mp3",
    "metadata": {
        "audio_duration_estimate_seconds": 15.5,
        "audio_cost_usd": 0.0075
    }
}

Handling Long Summaries

The problem

OpenAI TTS has a limit of 4096 characters per request. A summary of a 30-page report can have 6000+ characters.

Splitting strategy

The _split_text_for_tts() function splits by sentences, not by arbitrary characters:

6000-char text
    │
    ├── Split by sentences (". " as separator)
    │
    ├── Group sentences up to 4096 chars per chunk
    │   ├── Chunk 1: sentences 1-15 (3800 chars)
    │   └── Chunk 2: sentences 16-22 (2200 chars)
    │
    ├── Generate audio per chunk
    │   ├── Chunk 1 → part_0.mp3
    │   └── Chunk 2 → part_1.mp3
    │
    └── Concatenate with a 500ms pause between chunks
        └── summary_final.mp3

Why split by sentences

Splitting mid-sentence produces audio with abrupt cuts. Splitting by sentences ensures each chunk is a complete semantic unit:

BAD:  "...the invoice total is $1,74"  |  "0.00 mexican pesos..."
GOOD: "...the invoice total is $1,740.00 mexican pesos."  |  "The items include..."

Audio Duration Estimation

Approximate formula

A standard English voice speaks ~150 words per minute. An average word in English is about ~5 characters.

def estimate_audio_duration(text: str) -> float:
    word_count = len(text.split())
    words_per_minute = 150
    return round(word_count / words_per_minute * 60, 1)

Reference table

Troubleshooting

"The audio sounds cut off or has glitches"

Probable cause: The text has special characters that TTS doesn't handle well (emojis, strange Unicode characters, long URLs).

Solution: Clean the text before TTS:

import re


def clean_text_for_tts(text: str) -> str:
    text = re.sub(r'https?://\S+', 'web link', text)
    text = re.sub(r'[^\w\s.,;:!?¿¡()\-\'"áéíóúüñÁÉÍÓÚÜÑ$%]', '', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

"Error concatenating audio with pydub"

Probable cause: ffmpeg is not installed (required by pydub).

Solution:

# Mac
brew install ffmpeg

# Ubuntu/Debian
apt-get install ffmpeg

# Docker (add to the Dockerfile)
RUN apt-get update && apt-get install -y ffmpeg

"The audio is too long (>5 minutes)"

Solution: Summarize more aggressively before TTS:

def compact_summary_for_tts(summary: str, max_chars: int = 2000) -> str:
    if len(summary) <= max_chars:
        return summary

    from openai import OpenAI
    client = OpenAI()

    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": (
                f"Compact this summary to a maximum of {max_chars} characters "
                f"while keeping the key points:\n\n{summary}"
            )
        }],
        max_tokens=500
    )
    return r.choices[0].message.content

"High TTS cost in production"

Solution: Implement an audio cache:

import hashlib


class AudioCache:
    def __init__(self, cache_dir: str = "/tmp/audio_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def get_or_generate(
        self,
        text: str,
        audio_module: AudioModule,
        voice: str = "nova"
    ) -> str:
        cache_key = hashlib.md5(f"{text}_{voice}".encode()).hexdigest()
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.mp3")

        if os.path.exists(cache_path):
            logger.info(f"Audio cache hit: {cache_key}")
            return cache_path

        return audio_module.generate_summary_audio(
            text=text,
            filename=f"{cache_key}.mp3",
            voice=voice
        )

Using the AudioModule

Complete example

audio = AudioModule(output_dir="./audio_output")

summary = """
Invoice FAC-2025-0042 from Tech Solutions S.A. for $1,740.00 MXN.
Includes: software license for $1,200.00 and technical support for $300.00.
Subtotal: $1,500.00. Tax 16%: $240.00. Total: $1,740.00.
Issue date: March 15, 2025.
Payment method: Bank transfer.
"""

cost = audio.estimate_cost(summary)
print(f"Estimated cost: ${cost['cost_tts1']:.4f}")
print(f"Chunks needed: {cost['chunks_needed']}")

audio_path = audio.generate_summary_audio(
    text=summary,
    filename="invoice_summary.mp3",
    voice="nova"
)
print(f"Audio generated: {audio_path}")

duration = estimate_audio_duration(summary)
print(f"Estimated duration: {duration}s")

Expected output

Estimated cost: $0.0042
Chunks needed: 1
Audio generated: ./audio_output/invoice_summary.mp3
Estimated duration: 24.0s

Exercises

Exercise 1: Audio with multiple sections and pauses

Modify generate_long_audio() so that, in addition to the pause between chunks, it adds a longer pause (1.5 seconds) when it detects a section change in the text (indicated by double line breaks or headers with #).

def generate_sectioned_audio(
    self,
    text: str,
    filename: str = "summary.mp3",
    voice: str = "nova"
) -> str:
    from pydub import AudioSegment

    sections = re.split(r'\n\n+|(?=^#{1,3}\s)', text, flags=re.MULTILINE)
    sections = [s.strip() for s in sections if s.strip()]

    temp_files = []
    try:
        for i, section in enumerate(sections):
            sub_chunks = self._split_text_for_tts(section)
            for j, chunk in enumerate(sub_chunks):
                temp_path = os.path.join(self.output_dir, f"_sec_{i}_part_{j}.mp3")
                response = self.client.audio.speech.create(
                    model=TTS_MODEL, voice=voice, input=chunk
                )
                response.stream_to_file(temp_path)
                temp_files.append({"path": temp_path, "section_end": j == len(sub_chunks) - 1})

        short_pause = AudioSegment.silent(duration=500)
        long_pause = AudioSegment.silent(duration=1500)

        combined = AudioSegment.from_mp3(temp_files[0]["path"])
        for item in temp_files[1:]:
            pause = long_pause if item.get("section_end") else short_pause
            combined = combined + pause + AudioSegment.from_mp3(item["path"])

        output_path = os.path.join(self.output_dir, filename)
        combined.export(output_path, format="mp3")
        return output_path

    finally:
        for item in temp_files:
            try:
                os.remove(item["path"])
            except OSError:
                pass

Exercise 2: Complete audio Q&A pipeline

Implement the full flow: the user sends an audio file with their question, the system transcribes it, searches the RAG, generates the answer, and converts it to audio. Return both the text and the audio of the answer.

def full_audio_qa(
    audio_module: AudioModule,
    rag_module,
    audio_question_path: str,
    doc_id: Optional[str] = None,
    voice: str = "nova"
) -> dict:
    question_text = audio_module.transcribe_question(audio_question_path)
    logger.info(f"Question transcribed: {question_text}")

    qa_result = rag_module.query(question=question_text, doc_id=doc_id)
    logger.info(f"Answer generated: {qa_result.answer[:100]}...")

    clean_answer = clean_text_for_tts(qa_result.answer)
    answer_filename = f"qa_answer_{doc_id or 'all'}.mp3"
    answer_audio = audio_module.generate_summary_audio(
        text=clean_answer,
        filename=answer_filename,
        voice=voice
    )

    return {
        "question": {
            "audio_path": audio_question_path,
            "text": question_text
        },
        "answer": {
            "text": qa_result.answer,
            "audio_path": answer_audio,
            "sources": qa_result.sources,
            "confidence": qa_result.confidence
        },
        "costs": {
            "stt": 0.006,  # ~$0.006/min for Whisper
            "rag_query": 0.01,
            "tts": audio_module.estimate_cost(clean_answer)["cost_tts1"]
        }
    }


audio = AudioModule()
rag = RAGModule(persist_directory="./chroma_data")

result = full_audio_qa(audio, rag, "question.mp3", doc_id="invoice_001")
print(f"Question: {result['question']['text']}")
print(f"Answer: {result['answer']['text']}")
print(f"Audio: {result['answer']['audio_path']}")

Summary

• The AudioModule converts text summaries into audio with OpenAI TTS.
• It handles long summaries by splitting them into chunks and concatenating with pydub.
• 6 available voices: alloy, echo, fable, onyx, nova, shimmer — selectable by document type.
• Two models: tts-1 (standard, economical) and tts-1-hd (high quality, double the cost).
• Optional audio Q&A pipeline: audio question → transcription → RAG → audio answer.
• Audio cache avoids regenerating the same content, reducing costs in production.
• Integration with FastAPI to serve audio files as endpoints.

Additional Resources

1. OpenAI TTS API — Official documentation
2. OpenAI Whisper API — Transcription
3. pydub Documentation — Audio manipulation
4. Module 5 of this guide — Audio processing foundation