Практычныя прытамулкі: стварэнне агента AI для транскрыбавання і стварэння кіява аудыя

Наступныя прыміткі паказваюць практычны шлях адбудовы АІ-агента для транскрыбавання і стварэння кантакту аудыякалоў. Акцэнс ставіцца на контракты, перакананняя і месцы для коду, а не на мотывацыйныя аспекты. Калі працуеце над стадзіяй агледжэння, спачатку запісайце контракт: неабяжныя даны, сігнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае залишыцца чыстым пад час пазнейшых змян у кодзе. Запісвайце часы рэкордавання і косты токеноў або запытак праза функцыйнае рэзультат. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмаверсіі ў спяльныя среды.

Што мы будуем ствараць

Этап «Тое, што мы ствараем» працюе найкраща, калі яго спрыяваць як мерыемую паверхню. Запісаце адна «золатая» версія даных, адзін прыклад неудачы і запіс пра вярнэнне да поперадней версіі перш чым расширваць масштабы. Храніце настройкі параду ад коду прыемліка. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавляць контроль без неабяжнага чытання всіх дадзеных. Храніце стан графа ў простам і типаваным формате. Вярнутыя блокі дадзеных маскуюць інфармацыю пра тое, канфігурацыйны вузел запісаў канкрэтны поле, і спакойваюць працу пасля перарываў.

Адкліканне дэмы

Этап адміністрування дэмы працюе найкраща, калі яго розглядаць як меравальную паверхню. Запісаце адна ідеальная версія, адзин случай неудачы і прыметкі па поверненню да пачатковага стану пры розширэнні масштаба. Дакументавайце як успешны, так і варыянты вярнення да нормальнага стану. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є часткай продукту, а не элементамі пазнейшага доўрабатвання. Храніце стан графаў у простам і типаваным формате. Вярнутыя структуры дакументаў маскуюць інфармацыю пра тое, який вузел запісаў канкрэтны поле, і спакойваюць працу пасля перарываў.

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

Розумеўце код нашага агента

Розумэнне стадзіі нашага агента працуе найэфектывней, калі яго спрыяюць як мерыемую паверхню. Зафіксавайце адны ідеальны прыклад роботы, адну справу з неудачай і прыметку па повернэнню да пярвоначальнага стану, прычым расшырюючы сферу дзеяння. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшоў, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны ланцужок дзеяння. Рэгулюйце стан графа так, каб ён быў простым і з адначыяным типам дадзеных. Вкладаныя структуры дадзеных маскуюць інфармацыю пра тое, канкрэтны вузел запісаў канкрэтнае поле, і спакойваюць працу пасля перарываў.

Крок 1: Агент, які слухае ўсіх

Этап 1: Стадія агента працюе найэфективней, калі яе розглядаць як вимерную паверхню. Зафіксавце адзін «золаты» транскрыпт, адзін прыклад неудачы і запіску пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Разглядзіце гэты этап як кантракт межа вхіднымі даннымі і паверыжанымі выхіднымі рэзультатамі. Дайце назвы артыфактам, задаце критэрыя успеху і не прабуйце завершыць задачу часткова без адзінаго заўважэння. Зберагачыце стан графа ў простам і типаванам формате. Вкладзеныя блокі маскуюць інфармацыю пра тое, калькі вузел запісаў калькі поль, і спакойваюць працу пасля перарываў.

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

Этап 2: Запіс транскрыпта у файл

Этап 2, якім ёсць напісання коду, працюе найэфективнейша, калі яго розглядаць як параметрызаванную плошчу. Запісаце адна ідеальная версія коду, адзін прыклад неудачы і прыметкі па поверненню да попярэдня стану пры расшырэнні масштаба. Запісвайце часы виконання і косты токеноў або запытак праза функцыйнае рэзультаты. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмовай среды ў спяльнаныя сераверы. Храніце стан графа у простам і типаваным формате. Вкладныя структуры маскуюць інфармацыю пра тое, який вузел запісаў канкрэтны поле, і спакшваюць продажчыку роботу пасля перерываў.

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

Этап 3: Дапамога тым, хто запазніўся, за дапамогою LLM

Этап №3, прымкі для тых, хто запаздзеў, работае наякша, калі яго спрыяваць як мерыемую плошчу. Зафіксавайце адны «золаты» прымер, адны кейс неудачы і запіс пра вярнэнне да поперадньего стану, прычым расшырюючы сферу дзеяння. Зберагаўце настройкі праза код аплікацыі. Файлы сяродавішча, хранільнікі секрэтных дадзеных і пазначкі функцый должны знаходзіцца ў аднам месцы, куды аператары можаць адбавляць контроль без неабяжнага чытання всіх дадзеных. Задазвайце ліміты бюджэту на кожны раунд і кожную сесію. Інструменты-агенты агрэсывна расширваюць контекст; строгі ліміты не дазволяюць дэмам ператварыцца на неспакоўныя рахункі.

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

Адны ключ для обох частак

Едын ключ для обох стадзіяў работае лепш, калі яго спрыятаць як вимерную паверхню. Зафіксавайце адны «золаты» прымер, адны кейс неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Дакументавайце як шлях успеху, так і шлях вяснавання проблемы разам. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсць часткай продукту, а не пасляднім дапрацоўкам. Храніце стан графа ў простым і типаваным формате. Вкладзеныя блокі маскуюць, який вузел запісаў канкрэтны поле, і спакшуюць продовжэнне роботы пасля перарываў.

Шаг 4: Чыста простая фронтэнд-частка

Этап 4 A, які ўсё проста, працюе найэфектыўней, калі яго розглядаць як мерыемую паверхню. Зафіксуйце адна ідеальная транскрыпцыю, адзін прыклад неудачы і прыметку па адвярненню роботы перш чым расширваць масштаб. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны ланцюг задач. Рэзультаты графа павінны быць простымі та з адначытаемымі датамі. Вкладзеныя блокі маскуюць інфармацыю пра тое, який вузел запісаў канкрэтны поле, і спакоююць працу пасля перерываў. Этап 4 A, які ўсё проста, працюе найэфектыўней, калі яго розглядаць як мерыемую паверхню. Зафіксуйце адна ідеальная транскрыпцыю, адзін прыклад неудачы і прыметку па адвярненню роботы перш чым расширваць масштаб. Запісвайце часы виконання і косты токеноў або запытак разам з функцыйнальнымі рэзультатамі. Відразувая видавальнасць костаў запобегае неспакойным рахункам, калі процес пераходзіць з дэмаверсіі ў спяльныя сераўы.

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

Куда далей

Для этапа «Куды ісці» неабяжна пазначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры перадзеіснавленні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан.

Чэк-ліст для эксплуатацыі

Для этапа чэк-ліста для эксплуатацыі таксама неабяжна пазначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры перадзеіснавленні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан.

Спрытваце гэты ўражак як кантракт межа вхіднымі дадзеннямі і пасвярджанымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаце критэрыя успеху і не прымайце часткова завершэнне без падтверджэння.

Неабходна людская падтвердзення для тых процесаў, якія витрачаюць грошы або зменяюць данні ў працэсе виробніцтва. Праця ў часе компілявання не абавязкова значыць повную готовнасць до роботы.

Напішыце кароткі посібнік: як зменяць канты, як спрачыць очакваннія ў черзі, як анулюваць пасляпэўныя дзеянні.

Зазначайце час выканання і кост токенаў або запытаў разам з функцыйнальнымі рэзультатамі. Відразлівае відображэння костоў з’являецца прычыной парадоксальных сум у момент пераходу з дэмовай версіі ў спяльныя сераўы.

Неабходна людская падтвердзення для тых процесаў, якія витрачаюць грошы або зменяюць данні ў працэсе виробніцтва. Праця ў часе компілявання не абавязкова значыць повную готовнасць до роботы.

Перш чым запускать стак, заморозьце версіі, зафіксавце «золаты» транскрыпты для критычнага шляху і паказвце способы абяроны. У спільных средах неабходны ліміты частоты запытанняў, пераконтрольванне прав на выкарыстоўвання ресурсаў і чысткі власнік для змены секрэтных даных. Лепш выбіраць простую надзею на надзейнасць, чым хітрыя експерыментальныя дэманстрацыі.

Прыметка для d9d69769604b: не кладзіце ключы прадаўцаў у репазітарый, задаце верхнюю межу токеноў на кожную сесію і зберагачыце транскрыпты празаўсёды з фікстурамі для ацэнкі, каб пазнейшыя замены моделяў заставаліся порównаннэй.