Wskazówki praktyczne: Tworzenie agenta AI do transkrypcji i streszczania dźwięku

Poniższe notatki przedstawiają praktyczny plan działania dotyczący „Tworzenia agenta AI do transkrypcji i streszczania rozmów audio”. Nacisk kładziony jest na umowy, sprawdzenia oraz miejsca zastępcze dla kodu, a nie na motywacyjne aspekty. Podczas przechodzenia przez etap przeglądu najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się od wersji demonstracyjnej do środowisk współdzielonych.

To, co tworzymy

Etap „To, co budujemy” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres projektu. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione bloki danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwę w kontynuacji pracy po zakłóceniach.

Rozpoczynanie dema

Faza testowania demo funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia do analizy. Zapisz jeden udany przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres testów. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach.

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": "*"
        }
    ]
}

Rozumienie kodu naszego agenta

Rozumienie etapu pracy naszego agenta działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Wolimy małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwanie kontynuacji po zakłóceniach.

Krok 1: Agent, który słucha wszystkich

Etap 1 – faza agenta funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej strukturze typów. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.

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

Krok 2: Zapisywanie zapisu do pliku

Etap 2, polegający na pisaniu kodu, funkcjonuje najlepiej, gdy traktowany jest jako coś mierzalnego. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwę w kontynuacji pracy po zakłóceniach.

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

Krok 3: Doprowadzenie do poziomu pozostałych za pomocą LLM

Etap nr 3, polegający na łapaniu tych, którzy przychodzą późno, działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres pracy. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustal limit tokenów na rundę i sesję. Narzędzia typu agentic intensywnie rozszerzają kontekst; sztywne ograniczenia zapobiegają temu, by demonstracje przerodziły się w niespodziewane rachunki.

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

Jedny klucz dla obu części

Klucz do obu etapów działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres projektu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane w danym polu, i powodują przerwanie kontynuacji działania po zakłóceniach.

Krok 4: Bardzo prosty interfejs użytkownika

Etap 4 A – niezwykle prosty etap funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwania w kontynuacji działania po zakłóceniach. Etap 4 A – niezwykle prosty etap funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.

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

Dalej co robić

Na etapie „Dokąd iść”, zanim zmieni się kod, należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić, nie musząc czytać całej struktury.

Lista kontrolna operacyjna

Na etapie Listy kontrolnej operacyjnej, zanim zmieni się kod, należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.

Rozpatruj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania.

Zapewnij ludzką aprobatę dla operacji, które powodują wydanie pieniędzy lub zmianę danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się pełnej kompletności procesu biznesowego.

Napisz krótki przewodnik: jak rotować klucze, jak opróżniać kolejki z zadań, jak cofnąć ostatni proces pobierania danych.

Zapisz czasy wykonywania zadań oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.

Zapewnij ludzką aprobatę dla operacji, które powodują wydanie pieniędzy lub zmianę danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się pełnej kompletności procesu biznesowego.

Zanim zaczniesz promować tę architekturę, zamroź wersje, utwórz dokładny zapis dla kluczowych etapów realizacji oraz potwierdź kroki odwracania zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji przynależności użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację haseł. Wolimy nudną niezawodność od pomysłowych, jednorazowych demonstracji.

Uwaga dotycząca d9d69769604b: unikaj przechowywania kluczy dostawcy w repozytorium, ustaw ograniczenie liczby tokenów na sesję oraz przechowuj zapisy obok plików testowych, aby późniejsze zmiany modeli pozostawały porównywalne.