Practical notes: Building an AI agent for transcribing and summarizing audio

The following notes reconstruct a practical path around “Building an AI agent for transcribing and summarizing audio calls”. Emphasis stays on contracts, checks, and drop-in code placeholders rather than motivational framing. When working through the Overview stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments.

What we are building

The What we are building stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Running the demo

The Running the demo stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

git clone https://github.com/openvidu-labs/transcriber-summarizer-agent.git
cd transcriber-summarizer-agent

python -m venv venv && source venv/bin/activate
pip install -r requirements.txt

# STT_PROVIDER: openai | aws | vosk (offline)
# LLM_PROVIDER: openai | aws | (empty for no summarization)
STT_PROVIDER=
LLM_PROVIDER=

# Required when "openai" is selected for STT_PROVIDER or LLM_PROVIDER
OPENAI_API_KEY=
# Required when "aws" is selected for STT_PROVIDER or LLM_PROVIDER
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=
AWS_DEFAULT_REGION=us-east-1

python main.py dev

python app/server.py

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "TranscriberSummarizer",
            "Effect": "Allow",
            "Action": [
                "transcribe:StartStreamTranscription",
                "bedrock:InvokeModel",
                "bedrock:InvokeModelWithResponseStream"
            ],
            "Resource": "*"
        }
    ]
}

Understanding our agent’s code

The Understanding our agent s stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Step 1: An agent that listens to everyone

The Step 1 An agent stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

from livekit.agents import AgentServer, JobContext, cli

server = AgentServer()
@server.rtc_session()
async def entrypoint(ctx: JobContext):
    await ctx.connect()
    room = ctx.room
if __name__ == "__main__":
    cli.run_app(server)

def make_stt():
    if STT_PROVIDER == "vosk":
        from livekit.plugins import vosk
        return vosk.STT(model_path=VOSK_MODEL_PATH, language="en-US", partial_results=False)
    if STT_PROVIDER == "openai":
        from livekit.plugins import openai
        return openai.STT(model="gpt-4o-mini-transcribe")
    if STT_PROVIDER == "aws":
        from livekit.plugins import aws
        return aws.STT()
    raise ValueError(f"Unknown STT_PROVIDER {STT_PROVIDER!r}")

speech_to_text = make_stt()   # one engine, shared by every speaker

@room.on("track_subscribed")
def _on_track_subscribed(track, publication, participant):
    if track.kind == rtc.TrackKind.KIND_AUDIO:
        asyncio.create_task(transcribe_track(participant, track))

async def transcribe_track(participant, track):
    audio = rtc.AudioStream(track, sample_rate=16000, num_channels=1)

    async with speech_to_text.stream() as stt_stream:
        async def feed_audio():
            async for event in audio:
                stt_stream.push_frame(event.frame)
            stt_stream.end_input()       # no more audio: let the recognizer finish
        async def emit_transcripts():
            async for event in stt_stream:
                if event.type == stt_api.SpeechEventType.FINAL_TRANSCRIPT and event.alternatives:
                    text = event.alternatives[0].text.strip()
                    if text:
                        await record_line(participant, track, text)
        await asyncio.gather(feed_audio(), emit_transcripts())

Step 2: Writing the transcript to a file

The Step 2 Writing the stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

conversation = []  # in-memory history, used for the summary

async def record_line(participant, track, text):
    speaker = participant.name or participant.identity
    timestamp = datetime.datetime.now().strftime("%H:%M:%S")
    conversation.append(f"{speaker}: {text}")
    with open(transcript_path, "a", encoding="utf-8") as f:
        f.write(f"[{timestamp}] {speaker}: {text}\n")
    # Publish on LiveKit's built-in transcription channel, attributed to the speaker.
    writer = await room.local_participant.stream_text(
        topic=TOPIC_TRANSCRIPTION,                       # "lk.transcription"
        sender_identity=participant.identity,
        attributes={
            ATTRIBUTE_TRANSCRIPTION_FINAL: "true",
            ATTRIBUTE_TRANSCRIPTION_TRACK_ID: track.sid,
            ATTRIBUTE_TRANSCRIPTION_SEGMENT_ID: utils.shortuuid("SG_"),
        },
    )
    await writer.write(text)
    await writer.aclose()

[14:02:11] Alice: should we ship the release today
[14:02:15] Bob: yes but let us wait for the tests to pass
[14:02:20] Alice: agreed lets do it after lunch

Step 3: Catching latecomers up with an LLM

The Step 3 Catching latecomers stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Budget tokens per turn and per session. Agentic tools expand context aggressively; hard caps keep demos from becoming surprise invoices.

@room.on("participant_connected")
def _on_participant_connected(participant):
    asyncio.create_task(summarize_for(participant))

    async def summarize_for(participant):
    await asyncio.sleep(2)          # let the newcomer's browser get ready
    if not conversation:
        return                       # nothing said yet, nothing to summarize

    summary = await summarize(conversation)
    await room.local_participant.send_text(
        summary,
        topic="summary",
        destination_identities=[participant.identity],
    )

def make_llm():
    model = os.getenv("SUMMARY_MODEL")   # optional override; default per provider
    if LLM_PROVIDER == "openai":
        from livekit.plugins import openai
        return openai.LLM(model=model or "gpt-4.1")
    if LLM_PROVIDER == "aws":
        from livekit.plugins import aws
        return aws.LLM(model=model or "us.amazon.nova-2-lite-v1:0")
    raise ValueError(f"Unknown LLM_PROVIDER {LLM_PROVIDER!r}")

async def summarize(conversation):
    ctx = llm.ChatContext.empty()
    ctx.add_message(role="system", content=SUMMARY_PROMPT)
    ctx.add_message(role="user", content="Transcript so far:\n" + "\n".join(conversation))
    chunks = [c async for c in make_llm().chat(chat_ctx=ctx).to_str_iterable()]
    return "".join(chunks).strip()

One key for both halves

The One key for both stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Step 4: A dead-simple frontend

The Step 4 A dead-simple stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts. The Step 4 A dead-simple stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments.

from livekit.api import AccessToken, VideoGrants

token = (
    AccessToken(API_KEY, API_SECRET)
    .with_identity(identity)
    .with_name(name)
    .with_grants(VideoGrants(room_join=True, room=room))
    .to_jwt()
)

const { token, url } = await (await fetch(`/token?room=${room}&identity=${id}&name=${name}`)).json();
const room = new LivekitClient.Room();
await room.connect(url, token);
await room.localParticipant.setMicrophoneEnabled(true);

room.registerTextStreamHandler("lk.transcription", async (reader, participantInfo) => {
  if (reader.info.attributes?.["lk.transcription_final"] !== "true") return;
  const text = await reader.readAll();
  addLine(nameFor(participantInfo?.identity), text, new Date().toLocaleTimeString());
});

Where to go from here

For the Where to go from stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Operational checklist

For the Operational checklist stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state.

Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.

Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Write a short runbook: how to rotate keys, how to drain the queue, how to roll back the last ingest.

Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments.

Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Before promoting the stack, freeze versions, capture a golden transcript for the critical path, and confirm rollback steps. Shared environments need rate limits, tenancy checks, and a clear owner for secret rotation. Prefer boring reliability over clever one-off demos.

Batch note for d9d69769604b: keep provider keys out of the repo, set a per-session token ceiling, and store transcripts next to the eval fixtures so later model swaps stay comparable.