A Live Transcription and Late-Join Summary Agent with LiveKit Agents

Plenty of agents and AI services exist already, so here's something small and concrete: an agent that assists people during an audio call. It does three things. It keeps a full transcript of the meeting in a text file. It sends live captions to everyone in the call. And when someone joins late, it sends that person a private summary of what they missed.

The build uses a local LiveKit-compatible media server and the LiveKit Agents Python framework for the agent. Together they cover the hard parts of real-time audio transport, which leaves the agent's own features as the only code to write.

The parts and the data flow

There are three pieces:

• An audio room where people and the bot meet, hosted on a local server so no sign-up is needed.
• The transcriber agent, a small Python program that enters every room as an invisible participant, converts each person's speech to text and saves it.
• A small web app, a plain page for joining a room, talking and watching the transcript appear.

Data moves between them like this:

Running the demo

Requirements are Python 3.10 or newer and, for summaries, an OpenAI or AWS account. The demo runs without an account if LLM_PROVIDER stays empty: captions and the saved transcript still work, and only the catch-up summaries are lost.

Start by cloning the repository:

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

Then create a virtual environment and install dependencies:

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

Next, edit the .env file to pick the transcription provider (STT_PROVIDER) and the summarization provider (LLM_PROVIDER):

# 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

Run three components in separate terminals. The first is the local media server, installed by following its self-hosting instructions. The second is the transcriber agent:

python main.py dev

The third is the web app:

python app/server.py

Open http://localhost:8080, enter a name and join, then speak for a moment and the transcript starts filling in. Load the page in another tab with a different name, and a yellow panel shows up shortly after, recapping what the first tab said. The complete timestamped log is saved under transcripts/.

With AWS, the credentials need permission to use Amazon Transcribe and Amazon Bedrock, for instance via this policy:

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

The agent entrypoint

A LiveKit agent is a function that runs once for each room. It's registered on an AgentServer and the CLI handles the rest:

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)

Notice that @server.rtc_session() has no agent_name. Naming an agent makes LiveKit wait for an explicit dispatch request. Without a name, the agent joins every new room automatically, which suits a transcriber.

Speech to text behind a factory

To allow swapping providers, the choice hides behind one small function. Every LiveKit STT plugin, whether Vosk, OpenAI or AWS, offers the same .stream() interface, so this is the single place in the agent that names a provider:

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}")

Imports sit inside each branch, so someone choosing AWS installs only livekit-plugins-aws, since the other branches never run. For Vosk, partial_results=False drops the interim mid-word guesses and keeps finished sentences only.

One recognizer is shared by all speakers, and a handler fires when an audio track is subscribed:

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))

Each speaker gets their own audio and STT stream, and two jobs run together with asyncio.gather: pushing frames in and reading transcripts out.

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())

feed_audio pushes frames into the recognizer, while emit_transcripts pulls out completed sentences, and the async with block takes care of cleanup. The code reacts only to FINAL_TRANSCRIPT events, which are whole sentences, not each interim guess. Whatever it receives goes to record_line, which saves the line to disk, pushes it to the browser and remembers it for later summaries.

Recording each line

record_line makes a finished utterance permanent. It appends to the file on disk, keeps a copy in memory for the summary and publishes the text to the browser:

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()

The resulting file looks like a normal meeting transcript:

[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

The publishing step doesn't need a custom channel. LiveKit's built-in transcription feature, the reserved lk.transcription topic, does the job. With sender_identity set to the speaker, any LiveKit client receives the text as an ordinary, correctly attributed transcription segment.

Summaries for latecomers

This is the most interesting feature. When someone joins a call in progress, the agent summarizes what they missed and sends it only to them. LiveKit fires participant_connected for every arrival:

@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],
    )

The LLM sits behind a factory too. Every LiveKit LLM exposes the same .chat(), so summarize() never branches:

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()

SUMMARY_PROMPT is a single friendly line asking for a three-to-four-sentence catch-up. .to_str_iterable() converts the streamed reply into text chunks that get joined, the same way for every provider and without a per-vendor SDK.

Two details make delivery tidy. First, waiting two seconds avoids a race: participant_connected can fire just ahead of the moment the new person's browser sets up its listeners. Second, addressing the message with destination_identities means only that one person receives the recap, so everyone already talking is left alone.

Since both plugins come from LiveKit, a given vendor supplies speech recognition and the LLM from one package with one set of credentials. With openai, a single OPENAI_API_KEY handles transcription and summaries alike. With aws, one IAM key pair covers Transcribe and Bedrock.

An offline, free setup is also possible: use vosk for STT and leave LLM_PROVIDER empty. Transcription then runs locally, and the summary feature is disabled and replaced by the full transcript.

The browser side

The front end is intentionally plain: one HTML file, the LiveKit browser SDK from a CDN and some JavaScript, with no framework or bundler.

Browsers can't create their own access tokens, since that would expose the API secret to every visitor. A tiny token server built on Python's standard library and livekit-api fills the gap:

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()
)

On the page, joining takes three calls: fetch a token, connect and enable the microphone:

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);

Displaying the transcript means subscribing to LiveKit's reserved lk.transcription text stream. The handler receives a reader plus info about the sender, which supplies the speaker's name, and an attribute named lk.transcription_final lets it ignore text that isn't finished yet:

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());
});

Because this is the standard channel, the same handler works in any LiveKit client. The catch-up summary travels on a custom summary topic, and when it arrives the page drops the recap into a highlighted box. That's the entire client.

What you end up with

The finished agent handles anything from a two-person chat to a full meeting. It writes a clean transcript to disk, shows live captions and gives latecomers a recap, all in roughly two hundred lines. Vosk, OpenAI or AWS can power it, and a single .env setting flips between them because both halves are wrapped in make_stt() and make_llm() factories. Obvious next steps are storing transcripts, live translation, or turning summaries into action items.

It stays short because the difficult parts are already built: real-time audio transport, track subscriptions and the AI provider communication layers. LiveKit Agents and its plugins supply the building blocks, so the code focuses on agent logic instead of plumbing.

For production, a self-hosted platform built on the same LiveKit-compatible core lets the agent run unchanged while the platform deals with autoscaling, high availability, TURN relaying for restrictive networks, recording management and observability, all inside your own infrastructure with data kept private and costs that don't depend on per-minute SaaS pricing.