基于LiveKit Agents的实时转录与迟到参与者总结工具

目前已有大量智能体与人工智能服务存在，因此这里提出一个简单实用的想法：一种能在音频通话中协助用户的智能体。它具备三项功能：将会议内容完整转录为文本文件保存；为通话中的所有人实时生成字幕；当有用户迟到时，向其发送其错过的内容摘要。

该项目的实现基于本地兼容 LiveKit 的媒体服务器以及用于构建智能体的 LiveKit Agents Python 框架。二者共同处理实时音频传输的复杂环节，这样智能体自身的功能就成为唯一需要编写的代码部分。

组成部件与数据流

整个系统由三部分组成：

• 音频通话室：用户与智能体在此进行交流，部署在本地服务器上，无需注册即可使用。

转录代理，这是一个小型 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上，其余操作则由命令行工具处理：

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 等待明确的调度请求。如果没有名称，该智能体会自动加入每个新房间，这正适合用于文字转录。

后台的语音转文本功能

为便于切换服务提供商，相关选项被隐藏在一个小函数中。无论是 Vosk、OpenAI 还是 AWS，所有 LiveKit 语音转文本插件都提供相同的 .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))

每个发言者都有独立的音频流和文本转录流，同时通过 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 也位于某个“工厂”背后。每个 LiveKit LLM 都提供相同的 .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 文件、从 CDN 获取的 LiveKit 浏览器 SDK 以及一些 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模式影响。