Strona główna / Artykuły

livekitagentsspeech-to-textllmpythonvideo-conferencing

Agent transkrypcji na żywo oraz podsumowania dla uczestników dołączających późno z LiveKit Agents

Stwórz agenta w Pythonie, który zapisuje transkrypcję rozmowy, przesyła na żywo napisy do przeglądarki oraz w sposób prywatny podsumowuje spotkanie dla osób, które dołączają późno.

Wydawca: REACTAPP.TOP8 min czytania1790 słów

Źródła: Building an AI agent for transcribing and summarizing audio calls

Istnieje już wiele agentów i usług AI, więc oto coś małego i konkretnego: agent, który pomaga ludziom podczas rozmowy dźwiękowej. Wykonuje trzy zadania – przechowuje pełny transkrypt spotkania w pliku tekstowym, wysyła podgląd napisów na żywo do wszystkich uczestników rozmowy, a gdy ktoś dołącza późno, wysyła tej osobie prywatny podsumowanie tego, co przegapiła.

W tym rozwiązaniu używany jest lokalny serwer mediów kompatybilny z LiveKit oraz framework Python LiveKit Agents do tworzenia agenta. Razem pokrywają one trudne aspekty transmisji dźwięku w czasie rzeczywistym, dzięki czemu jedynym kodem do napisania pozostają funkcje samego agenta.

Części i przepływ danych

Istnieją trzy komponenty:

  • Pomieszczenie dźwiękowe, w którym spotykają się ludzie i bot, hostowane na lokalnym serwerze, więc nie jest potrzebne rejestracja.
  • Agent transkrybujący – mały program w Pythonie, który wchodzi do każdego pokoju jako niewidzialny uczestnik, przekształca mowę każdej osoby na tekst i go zapisuje.
  • Mała aplikacja internetowa – prosta strona do wejścia do pokoju, rozmawiania oraz obserwowania pojawiania się transkrypcji.

    • Dane przemieszczają się między nimi w ten sposób:

    Ruch demo

    Wymagane jest Python 3.10 lub nowszy, a do tworzenia streszczeń konieczny jest konto OpenAI lub AWS. Demo działa bez konta, jeśli pole LLM_PROVIDER pozostanie puste: napisy i zapisana transkrypcja nadal funkcjonują, tracą się jedynie streszczenia uzupełniające.

    Najpierw sklonuj repozytorium:

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

    Następnie utwórz środowisko wirtualne i zainstaluj zależności:

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

    Następnie edytuj plik .env, aby wybrać dostawcę transkrypcji (STT_PROVIDER) oraz dostawcę streszczania (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

    Zapuść trzy komponenty w oddzielnych terminalach. Pierwszym jest lokalny serwer mediów, zainstalowany zgodnie z instrukcjami self-hosting. Drugim jest agent transkrypcji:

    python main.py dev

    Trzecim jest aplikacja internetowa:

    python app/server.py

    Otwórz http://localhost:8080, wpisz nazwę i dołącz, następnie mów przez chwilę – wtedy zaczyna się wypełniać transkrypcja. Załaduj stronę w innej karcie pod inną nazwą; wkrótce po tym pojawi się żółty panel, który podsumowuje to, co powiedziano w pierwszej karcie. Pełny zapis z datami i godzinami jest zapisywany w katalogu transcripts/.

    W przypadku AWS dane dostępowe wymagają uprawnień do korzystania z Amazon Transcribe i Amazon Bedrock, na przykład poprzez tę politykę:

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

    Punkt wejścia agenta

    Agent LiveKit to funkcja, która jest wykonywana raz dla każdego pokoju. Jest on rejestrowany w AgentServer, a resztą zajmuje się interfejs wiersza poleceń:

    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)

    Zauważ, że @server.rtc_session() nie zawiera pola agent_name. Określenie nazwy agenta sprawia, że LiveKit czeka na wyraźną prośbę o wysłanie danych. Bez nazwy agent automatycznie dołącza do każdego nowego pokoju, co nadaje się dla osoby przepisującej tekst.

    Tłumaczenie mowy na tekst za pomocą specjalistycznego oprogramowania

    Aby umożliwić zmianę dostawców, wybór ukryty jest za pomocą jednej małej funkcji. Każdy plugin STT od LiveKit, niezależnie od tego, czy to Vosk, OpenAI czy AWS, oferuje tę samą interfejs .stream(), więc to właśnie tutaj w agentzie określany jest dostawca:

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

    Importy znajdują się wewnątrz poszczególnych gałęzi, więc osoba wybierająca AWS instaluje tylko livekit-plugins-aws, ponieważ pozostałe gałęzie nigdy nie są uruchamiane. W przypadku Vosk parametr partial_results=False odrzuca tymczasowe przypuszczenia dotyczące środkowych słów i zachowuje jedynie ukończone zdania.

    Jeden rozpoznawacz jest wspólny dla wszystkich mówców, a obsługa zostaje uruchomiona po subskrypcji ścieżki audio:

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

    Każdy mówca otrzymuje własny strumień audio oraz strumień STT, a dwie operacje są wykonywane jednocześnie za pomocą asyncio.gather: wprowadzanie klatek oraz odczytywanie transkrypcji.

    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 wysyła ramki do rozpoznawacza, podczas gdy emit_transcripts wyodrębnia ukończone zdania, a blok async with zajmuje się sprzątaniem. Kod reaguje tylko na zdarzenia FINAL_TRANSCRIPT, które to są pełne zdania, a nie poszczególne tymczasowe przypuszczenia. Wszystko, co otrzymuje, trafia do record_line, który zapisuje tekst na dysku, wysyła go do przeglądarki i przechowuje go do późniejszych streszczeń.

    Zapisywanie każdej linii

    record_line utrwala ukończoną wypowiedź. Dodaje ją do pliku na dysku, przechowuje kopię w pamięci do celów streszczenia i publikuje tekst w przeglądarce:

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

    Utworzony plik wygląda jak zwykły protokół spotkania:

    [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 publikacji nie wymaga specjalnego kanału. Wbudowana funkcja transkrypcji LiveKit, czyli temat lk.transcription, sprosta temu zadaniu. Gdy sender_identity jest ustawiony na mówcę, każdy klient LiveKit otrzymuje tekst jako zwykły, poprawnie opisany segment transkrypcji.

    Streszczenia dla tych, którzy przychodzą późno

    To najciekawsza funkcja. Gdy ktoś dołącza do trwającej rozmowy, agent streszcza to, co przegapił, i wysyła to wyłącznie do niego. LiveKit uruchamia sygnał participant_connected przy każdym dołączeniu:

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

    LLM znajduje się również w tle aplikacji. Każdy LLM z LiveKit oferuje ten sam metodę .chat(), więc summarize() nigdy nie rozwidla się:

    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 to jedna przyjazna linijka, która prosi o krótki opis sytuacji składający się z trzech do czterech zdań. Funkcja .to_str_iterable() przekształca strumieniową odpowiedź na fragmenty tekstu, które są następnie łączone ze sobą – w ten sam sposób dla każdego dostawcy i bez konieczności używania specjalnego SDK.

    Dwa elementy sprawiają, że proces przekazywania informacji jest uporządkowany. Po pierwsze, oczekiwanie dwóch sekund zapobiega konfliktom: participant_connected może zostać wywołany tuż przed tym, jak przeglądarka nowej osoby uruchomi swoje słuchawki. Po drugie, adresowanie wiadomości za pomocą destination_identities oznacza, że tylko ta jedna osoba otrzymuje podsumowanie, dzięki czemu wszyscy pozostali uczestnicy rozmowy nie są przeszkadzani.

    Ponieważ oba pluginy pochodzą z LiveKit, dany dostawca zapewnia rozpoznawanie mowy oraz model językowy z jednego pakietu, wykorzystując jeden zestaw uprawnień. W przypadku openai pojedynczy klucz OPENAI_API_KEY służy zarówno do transkrypcji, jak i tworzenia streszczeń. Z kolei w przypadku aws jeden parzysty kluczy IAM obejmuje funkcje Transcribe i Bedrock.

    Możliwa jest również bezpłatna konfiguracja offline: użyj vosk do przetwarzania mowy na tekst i pozostaw pole LLM_PROVIDER puste. Wtedy transkrypcja odbywa się lokalnie, a funkcja streszczeń jest wyłączona i zastępowana pełną transkrypcją.

    Część przeglądarkowa

    Frontend jest celowo prosty: jeden plik HTML, SDK LiveKit dostępne przez CDN oraz trochę JavaScriptu – bez żadnych frameworków ani narzędzi do pakowania kodu.

    Przeglądarki nie mogą tworzyć własnych tokenów dostępu, ponieważ to ujawniłoby sekret API wszystkim odwiedzającym. Mały serwer tokenów, zbudowany na standardowej bibliotece Pythona oraz livekit-api, wypełnia tę lukę:

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

    Na tej stronie połączenie wymaga trzech kroków: pobranie tokena, połączenie oraz włączenie mikrofonu:

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

    Wyświetlanie transkrypcji oznacza subskrypcję na zarezerwowany przez LiveKit strumień tekstu lk.transcription. Obsługa otrzymuje czytnik wraz z informacjami o nadawcy, które podają imię mówcy, a atrybut o nazwie lk.transcription_final pozwala jej ignorować tekst, który jeszcze nie został ukończony:

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

    Ponieważ jest to kanał standardowy, ten sam obsługujący element działa we wszystkich klientach LiveKit. Streszczenie nadrobek informacji jest przesyłane przez specjalny temat summary, a po jego przybyciu strona umieszcza streszczenie w wyróżnionym polu. To właśnie stanowi cały klient.

    Co otrzymujemy ostatecznie

    Gotowy agent radzi sobie z czymkolwiek – od rozmowy między dwiema osobami po pełne spotkanie. Zapisuje czysty transkrypt na dysku, pokazuje napisy na żywo oraz dostarcza streszczenie osobom, które przybyły późno, wszystko to w około dwustu liniach kodu. Może być napędzany przez Vosk, OpenAI lub AWS, a jedna ustawienie pliku .env pozwala przełączać się między nimi, ponieważ obie części są otoczone funkcjami make_stt() i make_llm(). Oczywistymi kolejnymi krokami są przechowywanie transkryptów, tłumaczenie na żywo lub przekształcanie streszczeń w konkretne zadania.

    Jest on krótki, ponieważ trudne elementy zostały już zaimplementowane: transmisja audio w czasie rzeczywistym, subskrypcje na ścieżki dźwiękowe oraz warstwy komunikacji z dostawcami AI. LiveKit Agents wraz z jego pluginami dostarczają niezbędne elementy budulcowe, dzięki czemu kod koncentruje się na logice agenta, a nie na szczegółach technicznych.

    Dla użycia w produkcji platforma self-hosted, zbudowana na tym samym komercie kompatybilnym z LiveKit, umożliwia działanie agenta bez zmian, podczas gdy platforma zajmuje się automatycznym skalowaniem, wysoką dostępnością, przekierowywaniem sygnałów przez TURN w sieciach o restrykcyjnych zasadach, zarządzaniem nagraniami oraz możliwościami monitorowania – wszystko to w ramach Twojej własnej infrastruktury, przy zachowaniu poufności danych i kosztach niezależnych od cen modelu SaaS opartego na liczeniu minut.

    Literatura pokrewna

    ← Wszystkie artykułyWyślij brief