Ein Agent für Live-Transkription und Zusammenfassung von Spätzugängen mit LiveKit Agents
Erstellen Sie einen Python-Agenten, der eine Anruftranskript speichert, Live-Untertitel ins Browser anzeigt und für Personen, die zu spät hinzukommen, eine private Zusammenfassung des Meetings erstellt.
Quellen: Building an AI agent for transcribing and summarizing audio calls
Es gibt bereits zahlreiche Agenten- und KI-Dienste, daher hier eine kleine, konkrete Idee: Ein Agent, der Menschen während eines Audioanrufs unterstützt. Er erledigt drei Aufgaben. Er speichert eine vollständige Transkription des Treffens in einer Textdatei. Er sendet Echtzeit-Untertitel an alle Teilnehmer des Anrufs. Und wenn jemand verspätet hinzukommt, erhält diese Person eine private Zusammenfassung dessen, was sie verpasst hat.
Die Umsetzung verwendet einen lokalen, LiveKit-kompatiblen Medienserver sowie das LiveKit Agents Python-Framework für den Agenten. Zusammen übernehmen sie die schwierigen Aspekte der Echtzeit-Audio-Übertragung, wodurch nur noch die eigenen Funktionen des Agents als Code geschrieben werden müssen.
Die Komponenten und der Datenfluss
Es gibt drei Bestandteile:
- Einen Audio-Raum, in dem Menschen und der Bot aufeinandertreffen, der auf einem lokalen Server gehostet wird, sodass keine Anmeldung erforderlich ist.
Daten wechseln zwischen ihnen auf diese Weise:
Ausführung der Demo
Benötigt wird Python 3.10 oder neuer sowie, für Zusammenfassungen, ein OpenAI- oder AWS-Konto. Die Demo läuft auch ohne Konto, wenn LLM_PROVIDER leer bleibt: Untertitel und die gespeicherte Transkription funktionieren weiterhin, nur die Nachholzusammenfassungen gehen verloren.
Beginnen Sie damit, das Repository zu klonen:
git clone https://github.com/openvidu-labs/transcriber-summarizer-agent.git
cd transcriber-summarizer-agent
Dann erstellen Sie eine virtuelle Umgebung und installieren die Abhängigkeiten:
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
Dann bearbeiten Sie die .env-Datei, um den Transkriptionsanbieter (STT_PROVIDER) sowie den Zusammenfassungsanbieter (LLM_PROVIDER) auszuwählen:
# 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
Führen Sie drei Komponenten in separaten Terminals aus. Die erste ist der lokale Medienserver, der gemäß den Anweisungen zur Selbsthosting-Installation installiert wird. Die zweite ist der Transkriptionsagent:
python main.py dev
Drittens ist da die Webanwendung:
python app/server.py
Öffnen Sie http://localhost:8080, geben Sie einen Namen ein und treten Sie bei. Sprechen Sie anschließend für einen Moment, und die Transkription wird automatisch erstellt. Laden Sie die Seite in einem anderen Tab mit einem anderen Namen, und kurz darauf erscheint ein gelbes Feld, das zusammenfasst, was im ersten Tab gesagt wurde. Das vollständige, mit Zeitstempeln versehene Protokoll wird unter transcripts/ gespeichert.
Mit AWS benötigen die Zugangsdaten Berechtigungen, um beispielsweise Amazon Transcribe und Amazon Bedrock nutzen zu können – etwa über diese Richtlinie:
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "TranscriberSummarizer",
"Effect": "Allow",
"Action": [
"transcribe:StartStreamTranscription",
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": "*"
}
]
}
Der Eingangspunkt des Agents
Ein LiveKit-Agent ist eine Funktion, die einmal pro Raum ausgeführt wird. Er wird auf einem AgentServer registriert und die CLI kümmert sich um den 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)
Achten Sie darauf, dass @server.rtc_session() keinen agent_name besitzt. Die Angabe eines Agentennamens veranlasst LiveKit, auf eine explizite Dispatch-Anfrage zu warten. Ohne Namen tritt der Agent automatisch in jeden neuen Raum ein, was für einen Transkriptor geeignet ist.
Sprachzu-Text-Wandlung hinter einer Schicht
Um den Wechsel der Anbieter zu ermöglichen, wird die Auswahl hinter einer kleinen Funktion versteckt. Jedes LiveKit-STT-Plugin – egal ob Vosk, OpenAI oder AWS – bietet dieselbe .stream()-Schnittstelle, sodass dies der einzige Punkt im Agenten ist, an dem ein Anbieter benannt wird:
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}")
Die Imports befinden sich innerhalb jeder Branch, sodass jemand, der AWS wählt, nur livekit-plugins-aws installiert, da die anderen Branches niemals ausgeführt werden. Bei Vosk führt partial_results=False dazu, dass vorläufige Zwischenschätzungen weggelassen werden und nur fertige Sätze behalten bleiben.
Ein Erkennungstool wird von allen Sprechern gemeinsam genutzt, und ein Handler wird ausgelöst, wenn eine Audiodatei abonniert wird:
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))
Jeder Sprecher erhält seinen eigenen Audio- und STT-Stream, und zwei Aufgaben laufen mit asyncio.gather gleichzeitig ab: das Hineinschicken von Frames und das Auslesen der Transkripte.
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 schickt Frames an den Erkennungsalgorithmus, während emit_transcripts fertige Sätze extrahiert und der async with-Block für die Aufräumarbeiten sorgt. Der Code reagiert nur auf FINAL_TRANSCRIPT-Ereignisse, also auf vollständige Sätze und nicht auf jede vorläufige Schätzung. Alles, was er erhält, wird an record_line weitergeleitet, welcher die Zeile auf dem Datenträger speichert, sie an den Browser sendet und sie für spätere Zusammenfassungen merkt.
Jede Zeile aufzeichnen
record_line macht eine abgeschlossene Äußerung dauerhaft. Er fügt sie dem Dateiinhalt auf dem Datenträger hinzu, behält eine Kopie im Speicher für die Zusammenfassung und gibt den Text an den Browser weiter:
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()
Die resultierende Datei sieht wie ein normales Sitzungsprotokoll aus:
[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
Der Veröffentlichungsschritt erfordert keinen benutzerdefinierten Kanal. Die eingebaute Transkriptionsfunktion von LiveKit, das reservierte lk.transcription-Thema, erledigt diese Aufgabe. Wenn sender_identity auf den Sprecher gesetzt ist, erhält jeder LiveKit-Client den Text als gewöhnliches, korrekt zugeordnetes Transkriptionssegment.
Zusammenfassungen für Spätkommer
Das ist die interessanteste Funktion. Wenn jemand zu einem bereits laufenden Anruf stößt, fasst der Agent zusammen, was verpasst wurde, und sendet dies nur an diese Person. LiveKit löst für jeden Neuankömmling participant_connected aus:
@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],
)
Auch das LLM befindet sich hinter einer Fabrik. Jedes LiveKit LLM stellt dieselbe .chat()-Funktion bereit, sodass summarize() niemals unterschiedliche Wege einschlägt:
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 ist eine einzige, freundliche Zeile, die um einen Überblick in drei bis vier Sätzen bittet. .to_str_iterable() wandelt die als Stream übermittelte Antwort in Textblöcke um, die anschließend miteinander verbunden werden – genauso bei jedem Anbieter und ohne ein spezielles SDK des jeweiligen Herstellers.
Zwei Aspekte sorgen dafür, dass die Übermittlung ordentlich abläuft. Erstens verhindert das Warten von zwei Sekunden einen Konflikt: participant_connected kann ausgelöst werden, kurz bevor der Browser des neuen Teilnehmers seine Listener einrichtet. Zweitens bedeutet die Adressierung der Nachricht mit destination_identities, dass nur diese eine Person den Überblick erhält, sodass alle bereits im Gespräch Beteiligten ungestört bleiben.
Da beide Plugins von LiveKit stammen, liefert derselbe Anbieter die Spracherkennung sowie das LLM in einem Paket mit einer einzigen Zugangsdatensammlung. Bei openai kümmert sich eine einzige OPENAI_API_KEY sowohl um die Transkription als auch um Zusammenfassungen. Bei aws deckt ein einziger IAM-Schlüsselpaar sowohl Transcribe als auch Bedrock ab.
Eine offline-basierte, kostenlose Einrichtung ist ebenfalls möglich: Verwenden Sie vosk für die Spracherkennung und lassen Sie LLM_PROVIDER leer. Die Transkription erfolgt dann lokal, wobei die Zusammenfassungsfunktion deaktiviert und durch die vollständige Transkription ersetzt wird.
Die Browser-Seite
Die Benutzeroberfläche ist absichtlich schlicht: eine HTML-Datei, das LiveKit-Browser-SDK von einem CDN sowie etwas JavaScript – ohne Frameworks oder Bundler.
Browser können keine eigenen Zugriffstoken erstellen, da dies das API-Secret jedem Besucher preisgeben würde. Ein kleiner Token-Server, der auf der Standardbibliothek von Python sowie livekit-api basiert, schließt diese Lücke:
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()
)
Auf der Seite erfordert die Verbindung drei Aufrufe: Ein Token abrufen, sich verbinden und das Mikrofon aktivieren:
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);
Das Anzeigen der Transkription setzt eine Abonnement auf LiveKits reservierten lk.transcription-Textstrom voraus. Der Handler erhält einen Leser sowie Informationen über den Sender, der den Namen des Sprechers liefert, und ein Attribut mit dem Namen lk.transcription_final ermöglicht es ihm, Texte zu ignorieren, die noch nicht abgeschlossen sind:
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());
});
Weil es sich um den Standardkanal handelt, funktioniert derselbe Handler in jedem LiveKit-Client. Die Zusammenfassung der nachgehenden Inhalte wird über ein benutzerdefiniertes summary-Thema übertragen, und sobald sie ankommt, zeigt die Seite die Zusammenfassung in einem hervorgehobenen Feld an. Das ist der gesamte Client.
Was am Ende dabei herauskommt
Der fertige Agent kann alles von einem Gespräch zwischen zwei Personen bis hin zu einer vollständigen Besprechung bewältigen. Er schreibt eine saubere Transkription auf die Festplatte, zeigt Live-Untertitel an und bietet Spätankömmlingen eine Zusammenfassung – alles in etwa zweihundert Zeilen. Vosk, OpenAI oder AWS können ihn antreiben, und eine einzige .env-Einstellung wechselt zwischen ihnen, da beide Teile in den Factory-Funktionen make_stt() und make_llm() eingebettet sind. Offensichtliche weitere Schritte sind das Speichern von Transkriptionen, die Live-Übersetzung oder die Umwandlung von Zusammenfassungen in Handlungsanweisungen.
Es bleibt kurz, weil die schwierigen Teile bereits implementiert sind: der Echtzeit-Audio-Transport, die Abonnements für Spuren sowie die Kommunikationsschichten zum AI-Anbieter. LiveKit Agents und seine Plugins liefern die Bausteine, sodass sich der Code auf die Agentenlogik statt auf die technischen Infrastrukturaspekte konzentrieren kann.
Zur Produktion ermöglicht eine selbst gehostete Plattform, die auf dem gleichen LiveKit-kompatiblen Kern basiert, es dem Agenten, unverändert zu laufen, während die Plattform sich um Automaskierung, hohe Verfügbarkeit, TURN-Relaying für restriktive Netzwerke, Aufnahmeverwaltung und Überwachungskapazitäten kümmert – alles innerhalb Ihrer eigenen Infrastruktur, wobei die Daten privat bleiben und die Kosten nicht von der pro Minute berechneten SaaS-Preisstruktur abhängen.