Un agent de transcription en direct et un agent de résumé pour les participants arrivant tard avec LiveKit Agents
Créez un agent Python qui enregistre la transcription d’un appel, diffuse des sous-titres en direct dans le navigateur et résume en privé la réunion pour ceux qui y participent en retard.
Sources: Building an AI agent for transcribing and summarizing audio calls
De nombreux agents et services d’IA existent déjà, alors voici une idée simple et concrète : un agent qui assiste les personnes lors d’un appel audio. Il effectue trois tâches. Il conserve une transcription complète de la réunion dans un fichier texte. Il envoie des sous-titres en direct à tous les participants à l’appel. Et lorsque quelqu’un rejoint en retard, il lui envoie un résumé privé de ce qu’il a manqué.
La mise en œuvre utilise un serveur multimédia local compatible LiveKit ainsi que le framework Python LiveKit Agents pour l’agent. Ensemble, ils gèrent les aspects complexes du transport audio en temps réel, ce qui ne laisse que les fonctionnalités propres à l’agent à coder.
Les composants et le flux de données
Il y a trois éléments :
- Une salle audio où les personnes et le bot se rencontrent, hébergée sur un serveur local, ce qui évite toute nécessité d’inscription.
Les données circulent entre eux de cette manière :
Exécution de la démonstration
Les exigences sont Python 3.10 ou une version plus récente, ainsi que, pour les résumés, un compte OpenAI ou AWS. La démonstration fonctionne sans compte si LLM_PROVIDER reste vide : les sous-titres et la transcription enregistrée fonctionnent toujours, seuls les résumés de rattrapage sont perdus.
Commencez par cloner le répertoire :
git clone https://github.com/openvidu-labs/transcriber-summarizer-agent.git
cd transcriber-summarizer-agent
Créez ensuite un environnement virtuel et installez les dépendances :
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
Ensuite, modifiez le fichier .env pour choisir le fournisseur de transcription (STT_PROVIDER) et le fournisseur de résumé (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
Exécutez trois composants dans des terminaux séparés. Le premier est le serveur multimédia local, installé en suivant ses instructions d’hébergement autonome. Le second est l’agent de transcription :
python main.py dev
Le troisième est l’application web :
python app/server.py
Ouvrez http://localhost:8080, saisissez un nom et rejoignez la session, parlez ensuite pendant un moment et le transcript commence à se remplir. Chargez la page dans une autre fenêtre avec un nom différent ; un panneau jaune apparaîtra peu après pour résumer ce qui a été dit dans la première fenêtre. Le journal complet, avec des timestamps, est enregistré dans le répertoire transcripts/.
Avec AWS, les identifiants doivent disposer des autorisations nécessaires pour utiliser Amazon Transcribe et Amazon Bedrock, par exemple via cette politique :
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "TranscriberSummarizer",
"Effect": "Allow",
"Action": [
"transcribe:StartStreamTranscription",
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": "*"
}
]
}
Point d'entrée de l'agent
Un agent LiveKit est une fonction qui s'exécute une fois pour chaque pièce. Il est enregistré sur un AgentServer et la CLI s'occupe du reste :
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)
Remarquez que @server.rtc_session() ne comporte pas de agent_name. En donnant un nom à l’agent, LiveKit attend une demande de diffusion explicite. Sans nom, l’agent rejoint automatiquement chaque nouvelle salle, ce qui convient à un transcriveur.
Reconnaissance vocale en arrière-plan
Afin de permettre le changement de fournisseur, ce choix se cache derrière une petite fonction. Chaque plugin STT de LiveKit, que ce soit Vosk, OpenAI ou AWS, propose la même interface .stream() ; c’est donc là que l’agent définit le fournisseur :
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}")
Les imports se trouvent à l’intérieur de chaque branche, de sorte que celui qui choisit AWS n’installe que livekit-plugins-aws, car les autres branches ne sont jamais exécutées. Pour Vosk, partial_results=False élimine les prédictions intermédiaires au milieu des mots et ne conserve que les phrases complètes.
Un même reconnaisseur est partagé par tous les locuteurs, et un gestionnaire est déclenché lorsque l’on s’abonne à une piste 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))
Chaque orateur dispose de sa propre chaîne audio et de sa propre chaîne STT, et deux tâches s’exécutent simultanément à l’aide de asyncio.gather : l’envoi des frames et la lecture des transcriptions.
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 envoie des frames au système de reconnaissance, tandis que emit_transcripts extrait les phrases complètes, et le bloc async with s’occupe du nettoyage. Le code ne réagit qu’aux événements FINAL_TRANSCRIPT, qui représentent des phrases entières et non chacune des prédictions intermédiaires. Tout ce qui est reçu est envoyé à record_line, qui enregistre la phrase sur disque, la transmet au navigateur et la mémorise pour des résumés ultérieurs.
Enregistrement de chaque phrase
record_line rend permanente une phrase complètement enregistrée. Il l’ajoute au fichier sur disque, conserve une copie en mémoire pour les résumés et publie le texte au navigateur :
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()
Le fichier résultant ressemble à un compte rendu de réunion normal :
[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
L’étape de publication ne nécessite pas de canal personnalisé. La fonction de transcription intégrée à LiveKit, le sujet réservé lk.transcription, s’en charge. Lorsque sender_identity est défini sur l’orateur, tout client LiveKit reçoit le texte sous forme de segment de transcription ordinaire et correctement attribué.
Résumés pour ceux qui arrivent en retard
C’est la fonctionnalité la plus intéressante. Lorsqu’une personne rejoint une appel en cours, l’agent résume ce qu’elle a manqué et ne l’envoie qu’à elle. LiveKit déclenche participant_connected à chaque arrivée :
@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],
)
Le LLM se trouve également derrière une interface utilisateur. Chaque LLM LiveKit expose la même fonction .chat(), de sorte que la fonction summarize() ne prend jamais de chemin différent :
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 est une seule ligne amicale demandant un résumé de trois à quatre phrases. .to_str_iterable() convertit la réponse en flux en blocs de texte qui sont ensuite assemblés, de la même manière pour chaque fournisseur et sans nécessiter un SDK spécifique à chacun.
Deux détails permettent une transmission ordonnée. Premièrement, attendre deux secondes évite tout conflit : participant_connected peut être déclenché juste avant que le navigateur de la nouvelle personne n’installe ses écouteurs. Deuxièmement, l’adressage du message via destination_identities garantit que seule cette personne reçoive le résumé, laissant ainsi intacts tous ceux qui discutent déjà.
Puisque les deux plugins proviennent de LiveKit, un même fournisseur fournit la reconnaissance vocale ainsi que l’LLM dans un seul package, accompagné d’un ensemble de identifiants. Avec openai, une seule clé OPENAI_API_KEY gère à la fois la transcription et les résumés. Avec aws, une seule paire de clés IAM couvre les services Transcribe et Bedrock.
Il est également possible d’effectuer une configuration hors ligne et gratuite : utilisez vosk pour la reconnaissance vocale en temps réel et laissez LLM_PROVIDER vide. La transcription s’effectue alors localement, la fonction de résumé étant désactivée au profit du texte transcrit intégral.
Le côté navigateur
L’interface utilisateur est délibérément simple : un seul fichier HTML, le SDK LiveKit pour navigateur provenant d’un CDN ainsi que du JavaScript, sans framework ni outil de bundling.
Les navigateurs ne peuvent pas créer leurs propres jetons d’accès, car cela exposerait le secret de l’API à tous les visiteurs. Un petit serveur de jetons, construit à partir de la bibliothèque standard de Python et de livekit-api, comble cette lacune :
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()
)
Sur cette page, la connexion nécessite trois appels : récupérer un jeton, se connecter et activer le micro :
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);
Afficher la transcription signifie s’abonner au flux de texte réservé lk.transcription de LiveKit. Le gestionnaire reçoit un lecteur ainsi que des informations sur l’expéditeur, qui fournissent le nom du locuteur, et une propriété nommée lk.transcription_final lui permet d’ignorer les textes qui ne sont pas encore terminés :
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());
});
Puisqu’il s’agit du canal standard, le même gestionnaire fonctionne dans n’importe quel client LiveKit. Le résumé des événements manqués est transmis via un sujet personnalisé summary, et une fois reçu, la page affiche ce résumé dans une boîte mise en évidence. C’est tout ce que nécessite le client.
Résultat final
L’agent ainsi créé peut gérer tout, d’une conversation à deux personnes jusqu’à une réunion complète. Il enregistre un transcript propre sur disque, affiche des sous-titres en direct et fournit un résumé aux arrivants en retard, tout cela en environ deux cents lignes. Vosk, OpenAI ou AWS peuvent le faire fonctionner, et une seule configuration .env permet de choisir entre eux, car les deux parties sont encapsulées dans les fonctions make_stt() et make_llm(). Les prochaines étapes évidentes consistent à stocker les transcripts, à assurer une traduction en temps réel ou à transformer les résumés en actions concrètes.
Il reste court car les parties difficiles ont déjà été développées : le transport audio en temps réel, les abonnements aux pistes et les couches de communication avec les fournisseurs d’IA. Les LiveKit Agents et leurs plugins fournissent les éléments de base, permettant ainsi au code de se concentrer sur la logique des agents plutôt que sur les aspects techniques complexes.
Pour un usage en production, une plateforme auto-hébergée basée sur le même noyau compatible avec LiveKit permet aux agents de fonctionner sans modification, tandis que la plateforme gère l’auto-échelle, la haute disponibilité, le relais TURN pour les réseaux restrictifs, la gestion des enregistrements et l’observabilité, tout cela au sein de votre propre infrastructure, avec des données restées privées et des coûts qui ne dépendent pas d’un tarif SaaS par minute.