Агент для прямой транскрипции и краткого обзора участников, пришедших позже, с использованием LiveKit Agents

Уже существует множество агентов и сервисов ИИ, поэтому вот небольшая конкретная идея: агент, который помогает людям во время аудиозвонка. Он выполняет три функции. Во-первых, он сохраняет полный текст записи встречи в текстовом файле. Во-вторых, он отправляет субтитры в реальном времени всем участникам звонка. А когда кто-то присоединяется позже, он отправляет этому человеку закрытый краткий отчет о том, что он пропустил.

Для реализации используется локальный медиасервер, совместимый с LiveKit, и Python-фреймворк LiveKit Agents для создания агента. Вместе они обеспечивают обработку сложных аспектов передачи аудио в реальном времени, что позволяет сосредоточиться только на разработке собственных функций агента.

Компоненты и поток данных

Существует три основных компонента:

• Аудиокомната, где встречаются люди и бот, размещенная на локальном сервере, поэтому регистрация не требуется.

Агент для транскрипции — небольшая программа на Python, которая входит в каждую комнату в качестве невидимого участника, преобразует речь каждого человека в текст и сохраняет его.

Небольшое веб-приложение — простая страница для входа в комнату, разговора и просмотра появляющейся транскрипции.

Данные передаются между ними следующим образом:

Запуск демо-версии

Требуется Python 3.10 или новее, а для генерации кратких сводок — учётная запись в OpenAI или AWS. Демо-версия может работать без учётной записи, если переменная LLM_PROVIDER остаётся пустой: субтитры и сохранённая транскрипция всё равно будут работать, но пропадут только сводки с последними событиями.

Начните с клонирования репозитория:

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

Далее отредактируйте файл .env, чтобы указать поставщика транскрипции (STT_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

Запустите три компонента в отдельных терминалах. Первый — локальный медиа-сервер, установленный согласно инструкциям по самостоятельной настройке. Второй — агент для транскрипции:

python main.py dev

Третьим является веб-приложение:

python app/server.py

Откройте http://localhost:8080, введите имя и присоединитесь, затем говорите некоторое время — и начнет формироваться транскрипт. Загрузите страницу в другой вкладке с другим именем, и вскоре появится желтая панель с кратким обзором того, что было сказано в первой вкладке. Полный лог с временными метками сохраняется в папке transcripts/.

Для работы с AWS у учетных данных должны быть разрешения на использование Amazon Transcribe и Amazon Bedrock, например, согласно этой политике:

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

Точка входа агента

Агент LiveKit — это функция, которая запускается один раз для каждой комнаты. Она регистрируется в AgentServer, а остальное обрабатывается через CLI:

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)

Обратите внимание, что в @server.rtc_session() отсутствует параметр agent_name. Указание имени агента заставляет LiveKit ожидать явного запроса на отправку данных. Без имени агент автоматически присоединяется ко всем новым комнатам, что подходит для переписчика.

Преобразование речи в текст

Чтобы можно было менять поставщиков услуг, выбор скрыт за одной небольшой функцией. Каждый плагин LiveKit для преобразования речи в текст, будь то Vosk, OpenAI или AWS, предоставляет одинаковый интерфейс .stream(); именно здесь в агенте указывается поставщик услуг:

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

Импорты находятся внутри каждой ветки, поэтому тот, кто выбирает AWS, устанавливает только livekit-plugins-aws, поскольку остальные ветки никогда не запускаются. Для Vosk параметр partial_results=False удаляет промежуточные предположения по словам и сохраняет только завершённые предложения.

Один рекогнайзер используется всеми дикторами, а обработчик запускается при подписке на аудиопоток:

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

Каждый оратор получает собственный аудио- и STT-стрим, причем два задания выполняются одновременно с помощью asyncio.gather: загрузка кадров и чтение транскрипций.

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 отправляет кадры в рекогнайзер, в то время как emit_transcripts извлекает готовые предложения, а блок async with занимается очисткой. Код реагирует только на события FINAL_TRANSCRIPT, которые представляют собой полные предложения, а не каждую промежуточную догадку. Все, что он получает, передается в функцию record_line, которая сохраняет текст на диске, отправляет его в браузер и хранит для последующих кратких сводок.

Запись каждой строки

record_line делает завершённую фразу постоянной. Она добавляет её в файл на диске, хранит копию в памяти для составления сводки и отправляет текст в браузер:

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

Для шага публикации не требуется специальный канал. Встроенная функция транскрипции LiveKit, а именно предназначенная для этой цели тема lk.transcription, справляется с задачей. При установке значения sender_identity на идентификатор оратора любой клиент LiveKit получает текст в виде обычного, правильно атрибутированного фрагмента транскрипции.

Краткое изложение для тех, кто опоздал

Это самая интересная функция. Когда к уже идущему звонку присоединяется кто-то, оператор составляет краткое изложение того, что он пропустил, и отправляет его только ему. LiveKit генерирует сигнал participant_connected при каждом присоединении:

@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 также находится за сервером. Каждый LLM от LiveKit предоставляет один и тот же метод .chat(), поэтому метод summarize() никогда не делает разветвлений:

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_str_iterable() преобразует потоковый ответ в фрагменты текста, которые затем объединяются; это делается одинаково для всех поставщиков без необходимости использования специального SDK от конкретного производителя.

Два фактора способствуют чёткой передаче информации. Во-первых, ожидание двух секунд предотвращает конфликты: сигнал participant_connected может поступить незадолго до того, как браузер нового участника настроит свои слушатели. Во-вторых, использование destination_identities для адресации сообщения гарантирует, что его получит только один человек, тем самым не беспокоя остальных участников, уже ведущих разговор.

Поскольку оба плагина разработаны компанией LiveKit, один поставщик предоставляет функции распознавания речи и ИИ-модель в одном пакете с одним набором учетных данных. С использованием openai один параметр OPENAI_API_KEY обеспечивает как транскрипцию, так и создание кратких сводок. С aws одна пара ключей IAM используется как для сервиса Transcribe, так и для Bedrock.

Также возможна бесплатная настройка в офлайн-режиме: для распознавания речи используется vosk, а поле LLM_PROVIDER оставляется пустым. В этом случае транскрипция генерируется локально, а функция создания сводок отключается и заменяется полной транскрипцией.

Браузерная часть

Фронтенд специально оставлен простым: один файл HTML, браузерный SDK LiveKit с CDN и немного JavaScript — без использования каких-либо фреймворков или инструментов для сборки кода.

Браузеры не могут создавать собственные токены доступа, поскольку это приведёт к раскрытию секрета API перед каждым посетителем. Небольшой сервер токенов, созданный на основе стандартной библиотеки Python и livekit-api, заполняет эту пробел:

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

Отображение транскрипции подразумевает подписку на зарезервированный LiveKit текстовый поток lk.transcription. Обработчик получает объект для чтения вместе с информацией о отправителе, содержащей имя говорящего, а атрибут lk.transcription_final позволяет ему игнорировать текст, который ещё не завершён:

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

Поскольку это стандартный канал, один и тот же обработчик работает в любом клиенте LiveKit. Краткое содержание просмотра передается по специальной теме summary, и по её прибытии страница отображает краткое содержание в выделенном блоке. Вот и всё, что требуется от клиента.

Что в итоге получается

Готовый агент может обрабатывать что угодно — от чата между двумя людьми до полноценной встречи. Он записывает чистый текст переписки на диск, отображает субтитры в реальном времени и предоставляет тем, кто опоздал, краткое содержание просмотра; всё это занимает примерно двести строк. Для его работы могут использоваться Vosk, OpenAI или AWS, причём одно настройка файла .env позволяет переключаться между ними, поскольку обе части реализованы с помощью функций make_stt() и make_llm(). Очевидными следующими шагами являются хранение текстов переписок, реальное время перевода или преобразование кратких содержаний в планы действий.

Он остается простым, поскольку сложные компоненты уже реализованы: передача аудио в реальном времени, подписка на треки и слои связи с поставщиками ИИ. LiveKit Agents и его плагины обеспечивают необходимые элементы, поэтому код сосредотачивается на логике агента, а не на внутренних механизмах работы.

Для производственного использования самостоятельно развернутая платформа на основе того же совместимого ядра LiveKit позволяет агенту работать без изменений, в то время как платформа занимается автомасштабированием, высокой доступностью, ретрансляцией через TURN в ограниченных сетях, управлением записями и мониторингом — всё это внутри вашей собственной инфраструктуры, при этом данные остаются конфиденциальными, а затраты не зависят от тарифов SaaS за минуту использования.