Youtube Live Transcription API

Oscar Martin Bahamonde Muñoz — Mon, 28 Jul 2025 04:51:17 +0000

This is a submission for the AssemblyAI Voice Agents Challenge

What I Built

Real-time Youtube Transcription App addressing 2nd prompt of Assembly AI Challenge (Real-Time Voice Performance)

Demo

App is currently on production stage:

http://aai.oscarbahamonde.com

GitHub Repository

https://github.com/bahamondeX/aai-challenge-2

Technical Implementation & AssemblyAI Integration

I used universal streaming AssemblyAI API to leverage it's top notch performance for real-time use cases and expressive API to handle multiple speakers with turns formatting:


def handler(url: str, is_live: bool = False) -> tp.Generator[bytes, None, None]:
    return get_live_audio_optimized(url) if is_live else get_audio_stream_pcm_s16le_optimized(url)


def on_begin(self: StreamingClient, event: BeginEvent):
    logger.info(f"Begin event: {event}")


def on_turn(self: StreamingClient, event: TurnEvent):
    self.queue.put_nowait(event)  # type: ignore


def on_termination(self: StreamingClient, event: TerminationEvent):
    logger.info(f"Termination event: {event}")


def on_error(self: StreamingClient, event: StreamingError):
    logger.error(f"Error event: {event}")


async def transcriber(url: str, is_live: bool = False):
    client = StreamingClient(StreamingClientOptions(api_key=AAI_API_KEY))
    client.queue = asyncio.Queue[TurnEvent]()  # type: ignore

    client.on(StreamingEvents.Begin, on_begin)  # type: ignore
    client.on(StreamingEvents.Turn, on_turn)  # type: ignore
    client.on(StreamingEvents.Termination, on_termination)  # type: ignore
    client.on(StreamingEvents.Error, on_error)  # type: ignore

    client.connect(StreamingParameters(sample_rate=16000))

    async def stream_audio():
        try:
            await asyncio.get_event_loop().run_in_executor(
                None, lambda: client.stream(handler(url, is_live))
            )
        except Exception as e:
            logger.error(f"Streaming error: {e}")

    asyncio.create_task(stream_audio())

    while True:
        try:
            data = await client.queue.get()  # type: ignore
            content = data.model_dump_json()
            yield content
        except asyncio.TimeoutError:
            logger.warning("Transcription timeout")
            break
        except Exception as e:
            logger.error(f"Transcription error: {e}")
            break

Forem: Oscar Martin Bahamonde Muñoz

Youtube Live Transcription API

What I Built

Demo

GitHub Repository

Technical Implementation & AssemblyAI Integration