1. 从零搭建语音转写流水线:为什么我选择从博客翻译切入
语音转文字这件事,看起来门槛不高,但真要把一条完整的流水线跑通,坑比想象中多得多。我最初接触 AssemblyAI 这个平台,是因为手头有一批英文技术博客的音频版本需要转成文字稿,再翻译成中文归档。当时试过几个方案:本地跑 Whisper 模型、调用某云厂商的语音识别接口、还有 AssemblyAI。最后让我决定深入用下去的,是它在长音频转写上的稳定性和说话人分离的准确度。
但问题来了——官方文档全是英文,社区里的中文资料要么太浅(只讲怎么调 API),要么太旧(接口都变了)。我索性一边用一边把踩过的坑记下来,形成了这个系列。这篇是第一篇,主要解决三个问题:AssemblyAI 到底能做什么、它的核心接口怎么用、以及从音频到中文翻译稿的完整链路怎么搭。
适合谁看?如果你手头有音频转文字的需求,不管是播客整理、会议记录、还是像我这样的技术博客翻译,这篇都能给你一条能直接跑通的路。不需要你有很深的机器学习背景,但基本的 Python 操作和 API 调用概念得有。我会把每一步为什么这么做讲清楚,参数怎么算、坑在哪里、怎么绕过去,都说明白。
先说结论:AssemblyAI 的核心价值在于它把语音识别、说话人分离、内容审核、摘要生成这些能力打包成了几个 REST 接口,你不需要自己训练模型,也不需要维护 GPU 集群。对于中小规模的转写需求,它的性价比和开发效率都很能打。下面我从实际使用角度,把整个链路拆开讲。
2. AssemblyAI 的核心能力边界与选型逻辑
2.1 它到底解决了什么问题
语音转写这个领域,技术路线大致分三类:本地开源模型(如 Whisper)、云厂商通用接口(如各大云平台的语音识别)、垂直领域专用平台(如 AssemblyAI)。三者的取舍逻辑不一样。
本地跑 Whisper 的好处是数据不出本地,隐私性好,而且完全免费。但缺点也很明显:长音频转写速度慢,说话人分离需要额外接 pyannote 之类的工具,而且对硬件有要求。我之前用一台没有独立显卡的机器跑一个两小时的音频,等了将近四十分钟,这还不算说话人分离的后处理。
云厂商通用接口的优势是稳定、便宜,但通常只给转写文本,说话人分离、摘要、情感分析这些要么没有,要么得额外付费开通。而且很多接口对音频格式、时长、并发有严格限制。
AssemblyAI 的定位在两者之间:它提供的是转写加增强的一站式服务。你上传一个音频 URL,它返回的不只是文字,还能带上说话人标签、时间戳、置信度、甚至自动摘要和关键词提取。对于我这种需要把博客音频转成结构化文字稿的场景,省了很多后处理工作。
2.2 核心接口与参数速查
AssemblyAI 的 API 设计很简洁,核心就两个动作:提交转写任务和轮询获取结果。下面这张表是我整理的最常用参数,实际调的时候直接对照着填就行。
| 参数名 | 作用 | 常用取值 | 注意事项 |
|---|---|---|---|
| audio_url | 音频文件地址 | 公网可访问的 URL | 不支持本地文件直传,需先上传到对象存储 |
| speaker_labels | 说话人分离 | true / false | 开启后返回每个词的说话人标签 |
| language_code | 语言代码 | en_us、zh_cn 等 | 不填则自动检测,但自动检测偶尔会误判 |
| punctuate | 自动标点 | true / false | 建议开启,否则返回文本没有标点 |
| summarization | 自动摘要 | true / false | 开启后返回摘要文本 |
| auto_chapters | 自动章节 | true / false | 适合长音频,自动切分章节 |
这里有个细节值得说:language_code如果不填,平台会自动检测语言。我实测下来,纯英文音频检测准确率很高,但中英混杂的音频偶尔会判成英文,导致中文部分转写质量下降。所以如果你的音频语言明确,强烈建议手动指定,别偷懒。
另一个坑是audio_url必须是公网可访问的。很多人第一次用的时候直接传本地路径,结果报错。正确的做法是先把音频传到对象存储(比如 S3 兼容的存储服务),拿到公网 URL 再提交。如果你只是测试,也可以用一些临时文件托管服务,但正式环境还是走对象存储稳妥。
2.3 为什么不用官方 SDK 而直接调 REST
AssemblyAI 提供了 Python 和 JavaScript 的 SDK,但我个人更倾向于直接调 REST 接口。原因有三个:第一,SDK 版本更新有时滞后于 API,新参数可能不支持;第二,直接调 REST 你能清楚看到每个请求和响应,排查问题方便;第三,跨语言场景下,REST 是通用的,不绑定特定语言生态。
当然,如果你项目里就是 Python,用 SDK 也没问题,能省一些样板代码。但理解底层 REST 的调用逻辑,对排查问题很有帮助。下面我会用 Python 的 requests 库来演示,这样你能看到完整的请求构造过程。
3. 从音频 URL 到转写文本的完整实操
3.1 环境准备与依赖安装
开始之前,你需要准备两样东西:一个 AssemblyAI 的 API Key,以及一个能跑 Python 的环境。API Key 在注册后可以在控制台找到,这里不赘述注册流程。
Python 环境建议 3.8 以上,依赖就两个核心库:
pip install requests如果你需要处理音频上传,可能还需要boto3之类的对象存储 SDK,但那是另一回事。这篇聚焦转写本身,假设你已经有了音频的公网 URL。
提示:API Key 不要硬编码在代码里,用环境变量或者配置文件管理。我见过太多人把 Key 直接写在脚本里然后不小心提交到公开仓库,结果被人盗刷额度。
3.2 提交转写任务:请求构造与参数详解
提交任务的核心是构造一个 POST 请求到转写接口。下面是我常用的一个封装函数:
import requests import os API_KEY = os.environ.get("ASSEMBLYAI_API_KEY") BASE_URL = "https://api.assemblyai.com/v2" def submit_transcript(audio_url, speaker_labels=True, language_code="en_us"): headers = { "authorization": API_KEY, "content-type": "application/json" } data = { "audio_url": audio_url, "speaker_labels": speaker_labels, "language_code": language_code, "punctuate": True, "summarization": True } response = requests.post(f"{BASE_URL}/transcript", json=data, headers=headers) response.raise_for_status() return response.json()这个函数返回的 JSON 里有一个id字段,这就是任务 ID,后面轮询结果要用。注意response.raise_for_status()这行,它会在 HTTP 状态码非 200 时抛异常,方便你第一时间发现参数错误或鉴权失败。
这里有个经验:提交任务后不要立刻轮询。转写是异步的,刚提交时状态是queued,你马上查也是这个状态。我一般等个十秒再开始轮询,减少无效请求。
3.3 轮询获取结果:状态机与超时处理
轮询的逻辑就是不断查询任务状态,直到变成completed或error。下面是一个带超时和间隔控制的实现:
import time def poll_transcript(transcript_id, timeout=1800, interval=10): headers = {"authorization": API_KEY} url = f"{BASE_URL}/transcript/{transcript_id}" start = time.time() while time.time() - start < timeout: response = requests.get(url, headers=headers) response.raise_for_status() result = response.json() status = result.get("status") if status == "completed": return result elif status == "error": raise RuntimeError(f"转写失败: {result.get('error')}") time.sleep(interval) raise TimeoutError("轮询超时,任务未在指定时间内完成")几个参数的选择逻辑:timeout设 1800 秒(半小时),是因为一个两小时的音频,转写加说话人分离通常需要几分钟到十几分钟,留足余量。interval设 10 秒,太短了浪费请求,太长了结果延迟高。这个组合我实测下来比较平衡。
注意:轮询频率不要太高,有些平台对请求频率有限制。10 秒一次对于单个任务来说完全够用,也不会触发限流。
3.4 结果解析:从 JSON 到可读文本
任务完成后,返回的 JSON 结构比较丰富。核心字段有这几个:
text:完整转写文本,带标点words:每个词的详细信息,包括开始时间、结束时间、置信度、说话人标签utterances:按说话人分组的语句块,适合做对话稿summary:自动摘要(如果开启了 summarization)
如果你要做对话稿,用utterances最方便。每个 utterance 包含说话人、文本、开始和结束时间。下面是一个把 utterances 转成可读对话稿的示例:
def format_utterances(result): lines = [] for utt in result.get("utterances", []): speaker = utt.get("speaker", "未知") text = utt.get("text", "") start = utt.get("start", 0) / 1000 # 毫秒转秒 lines.append(f"[{start:.1f}s] 说话人{speaker}: {text}") return "\n".join(lines)这里start字段单位是毫秒,转成秒更直观。说话人标签是 A、B、C 这样的字母,不是真实姓名,你需要自己根据内容判断谁是谁。
4. 翻译环节:把英文转写稿变成中文
4.1 翻译方案选型:为什么不用机器翻译直接怼
拿到英文转写稿后,下一步是翻译成中文。很多人第一反应是丢给机器翻译接口,一键搞定。但我实测下来,技术类内容的机器翻译直接输出,可读性很差。术语翻译不统一、长句语序混乱、代码相关的描述经常翻错。
我的做法是机器翻译打底加人工校对。具体来说,先用翻译接口把整段文本翻一遍,然后我自己对照英文原文逐段校对。这样比纯人工翻译快很多,又比纯机器翻译质量高。
翻译接口的选择上,我试过几个主流方案,差异主要在术语准确度和长文本处理上。对于技术博客这种场景,建议选择支持术语库自定义的接口,把常用技术术语提前录入,能显著提升一致性。
4.2 分段翻译与上下文保持
一个容易被忽略的点是:翻译要分段做,但不能分得太碎。如果把每个句子单独翻译,上下文丢失,代词指代会乱。我的做法是按段落翻译,每个段落控制在 500 到 1000 字符之间。
具体操作上,我会先把转写稿按自然段切分,然后逐段调用翻译接口。切分的时候注意不要在句子中间断开,按句号、问号、感叹号这些标点来切。
import re def split_paragraphs(text, max_len=800): sentences = re.split(r'(?<=[.!?])\s+', text) paragraphs = [] current = "" for sent in sentences: if len(current) + len(sent) > max_len and current: paragraphs.append(current.strip()) current = sent else: current += " " + sent if current: paragraphs.append(current.strip()) return paragraphs这个切分逻辑是:按句子边界切,但累积到接近 max_len 才断开。这样既保证了每段不会太长,又尽量保持了语义完整。
4.3 术语一致性维护
技术博客翻译最头疼的就是术语不一致。同一个词,前面翻成"接口",后面翻成"API",读者会困惑。我的解决办法是维护一个术语对照表,翻译前先做术语替换,把关键术语统一成固定译法。
比如:
| 英文术语 | 统一译法 |
|---|---|
| transcript | 转写稿 |
| speaker labels | 说话人分离 |
| utterance | 语句块 |
| endpoint | 接口端点 |
这个表不用很大,二三十个核心术语就够了。翻译前用脚本做一遍预替换,能省很多校对时间。
5. 踩坑实录:那些文档里不会写的问题
5.1 音频格式与编码的隐形坑
AssemblyAI 支持多种音频格式,但不是所有格式都一视同仁。我遇到过最诡异的问题是:同样是一个 MP3 文件,有的能正常转写,有的返回空结果。排查了半天,发现是采样率的问题。部分低采样率的音频,平台虽然接受了,但转写质量极差。
后来我养成了一个习惯:上传前先用 ffprobe 检查音频参数。采样率低于 16kHz 的,先转成 16kHz 以上再传。声道数也建议转成单声道,立体声有时候会导致说话人分离混乱。
ffprobe -v error -show_entries stream=sample_rate,channels -of default=noprint_wrappers=1 input.mp3这个命令能快速看到采样率和声道数。如果采样率是 8000Hz,建议先用 ffmpeg 转一下:
ffmpeg -i input.mp3 -ar 16000 -ac 1 output.mp3-ar 16000指定采样率 16kHz,-ac 1指定单声道。这一步花不了几秒钟,但能避免很多转写质量问题。
5.2 长音频的分片策略
AssemblyAI 对单个音频的时长有限制,具体限制各时期可能不同,但长音频处理是个绕不开的问题。我的策略是:超过一定时长的音频,先分片再分别转写,最后合并。
分片的关键是找到合适的切分点。不能在句子中间切,否则转写结果会在边界处断掉。我的做法是用静音检测来找切分点:音频中能量低于阈值持续超过 0.5 秒的位置,就是天然的切分点。
from pydub import AudioSegment from pydub.silence import detect_silence def split_audio(file_path, min_silence_len=500, silence_thresh=-40): audio = AudioSegment.from_file(file_path) silences = detect_silence(audio, min_silence_len=min_silence_len, silence_thresh=silence_thresh) # 根据静音区间计算切分点,这里简化处理 # 实际使用时需要根据目标分片时长选择最近的静音点 return silencesdetect_silence返回的是静音区间的列表,每个区间是[开始毫秒, 结束毫秒]。你根据目标分片时长,选择最近的静音点作为切分位置。这样切出来的片段,边界处不会有截断的词。
5.3 说话人分离的准确度问题
说话人分离不是万能的。我实测下来,以下情况准确率会明显下降:
- 说话人声音相似(比如同性别、同音域)
- 多人同时说话(重叠语音)
- 音频质量差、背景噪音大
- 说话人频繁切换(比如快速对话)
对于这些情况,我的建议是:不要完全依赖自动分离结果,把它当作初稿,人工过一遍。特别是做对话稿的时候,说话人标错会导致内容归属错误,影响理解。
另外一个小技巧:如果音频里说话人数量已知,可以在提交任务时通过参数提示平台。虽然 AssemblyAI 没有直接的"说话人数量"参数,但你可以通过后处理来校验——如果分离出的说话人数量明显多于预期,说明分离过度了,需要人工合并。
6. 完整链路串联与效率优化
6.1 从音频到中文稿的自动化脚本
把前面各个环节串起来,就是一个完整的流水线。下面是一个简化版的端到端脚本框架:
def full_pipeline(audio_url, target_lang="zh"): # 第一步:提交转写 submit_result = submit_transcript(audio_url, speaker_labels=True, language_code="en_us") transcript_id = submit_result["id"] # 第二步:轮询结果 result = poll_transcript(transcript_id) # 第三步:格式化转写稿 formatted = format_utterances(result) # 第四步:分段翻译 paragraphs = split_paragraphs(formatted) translated = [] for para in paragraphs: # 这里调用翻译接口,具体实现略 translated.append(translate_text(para, target_lang)) # 第五步:合并输出 return "\n\n".join(translated)这个框架里,翻译那一步需要你自己接一个翻译服务。我建议把翻译单独封装成一个函数,方便替换不同的翻译后端。
6.2 批量处理的并发控制
如果你有多个音频要处理,串行跑太慢。但并发也不能无限制,平台通常有并发限制。我的做法是用一个简单的信号量控制并发数:
import threading semaphore = threading.Semaphore(3) # 最多3个并发 def process_one(audio_url): with semaphore: return full_pipeline(audio_url)并发数设多少合适?我一般设 3 到 5。太高了容易触发限流,太低了效率上不去。具体数值你可以根据实际请求的响应时间来调整——如果经常收到 429 状态码,就调低;如果响应很快且没有报错,可以适当调高。
6.3 结果缓存与断点续传
转写和翻译都是耗时操作,如果中途失败,全部重来太浪费。我的做法是每一步的结果都落盘缓存,用任务 ID 作为文件名。下次跑的时候先检查缓存,有就直接读,没有才重新请求。
import json import os def cached_transcript(transcript_id, fetch_func): cache_file = f"cache/{transcript_id}.json" if os.path.exists(cache_file): with open(cache_file, "r") as f: return json.load(f) result = fetch_func(transcript_id) os.makedirs("cache", exist_ok=True) with open(cache_file, "w") as f: json.dump(result, f) return result这个缓存机制看起来简单,但实际用起来能省大量时间。特别是调试阶段,你可能会反复跑同一个音频,有缓存就不用每次都等转写完成。
7. 一些个人体会与后续扩展方向
这套流水线我跑了大概几十个音频,整体稳定。最大的体会是:语音转写这件事,平台能力只占一半,另一半在于你对音频本身的处理。采样率、声道、静音切分这些预处理做得好,转写质量能提升一个档次。
另一个体会是翻译环节不要贪快。机器翻译直接输出看着省事,但校对成本很高。分段翻译加术语表预替换,虽然多了一步,但整体效率反而更高。
后续我打算在这几个方向继续优化:一是把说话人分离的结果和音频时间轴对齐,生成带时间戳的双语对照稿;二是尝试用大语言模型做翻译后编辑,进一步提升中文可读性;三是把整个流水线容器化,方便在不同环境快速部署。
如果你也在做类似的事情,欢迎交流。这个系列我会继续更新,下一篇讲怎么处理多语言混杂的音频,以及如何用自定义词汇表提升专业术语的转写准确率。