实用提示：构建用于音频转录与总结的AI智能体
《实用笔记》操作指南：构建用于音频转录与总结的AI智能体——面向采用该模式的团队提供的合同、检查清单及可直接插入的代码模块。
以下笔记为“构建用于音频通话转录与总结的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": "*"
}
]
}
了解我们代理的代码
将理解智能体所处的阶段视为可测量的对象，这样会更有效。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任主体，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在流程中断后导致无法继续执行。
第一步：一个能监听所有信息的智能体
第一步的代理阶段若被视为可度量的对象，效果最佳。在扩大范围之前，先记录一份优秀的处理结果、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致无法继续处理。
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())
第二步：将处理结果写入文件
将“第二步：编写代码”视为一个可度量的环节会更为有效。在扩大范围之前，先记录一份优秀的实现示例、一个失败案例以及回滚说明。在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据结构会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致流程无法继续。
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
第三步：利用大语言模型帮助落后者赶上
将第三阶段的“捕捉迟到者”流程视为可度量的指标来处理效果最佳。在扩大范围之前，先记录一份成功的案例、一个失败案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 为每轮及每次会话设定预算额度。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程突然产生额外费用。
@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步的简单阶段视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障原因应能指向单一责任主体，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。 在将第4步的简单阶段视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。
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的批处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将日志存储在评估用示例文件旁边，以便后续模型更换时仍能保持对比性。