TEN Framework 中的 AWS ASR Python 扩展：基于 Amazon Transcribe 流式 API 的实时语音识别实战指南-平芜编程栈

人工智能
AI Agent
多模态
语音
AI 应用

【免费下载链接】ten-framework

Open-source framework for conversational voice AI agents

项目地址：https://gitcode.com/TEN-framework/ten-framework

导读

本文围绕 TEN Framework 开源仓库中的aws_asr_python扩展（位于 ai_agents/agents/ten_packages/extension/aws_asr_python）展开，系统讲解如何通过该扩展接入 AWS Transcribe 流式转录 API，为对话式语音 AI Agent 提供低延迟的实时语音转文本能力。读完本文，你将掌握该扩展的完整配置项与默认值、异步生命周期方法、自动重连与指数退避策略、音频转储调试手段，以及基于源码与测试用例的底层实现原理，可直接在 TEN Framework 的语音场景中正确部署与排障。

扩展定位：为 TEN Framework 提供 AWS 实时 ASR 能力

aws_asr_python是一个用 Python 编写的 ASR（自动语音识别）扩展，其核心职责是把输入音频流实时转换为文本，并作为 TEN Framework 图（Graph）中的一个标准 ASR 节点参与语音 Agent 的数据流。根据扩展说明（见 docs/README.zh-CN.md），它具备以下关键能力：

完全异步支持：整体采用异步架构，基于asyncio实现高性能、非阻塞的语音识别处理；
实时流式处理：使用 AWS Transcribe 的流式 API 进行低延迟音频流转录；
音频转储：可选的 PCM 音频录制功能，用于调试和分析识别质量；
错误处理：全面的错误处理与详细日志记录，通过模块错误码、AWS 特定错误详情和优雅降级机制反馈问题；
多语言支持：通过 AWS Transcribe 支持en-US、zh-CN等多种语言代码；
重连管理：内置自动重连机制，保障服务稳定性；
会话管理：支持会话 ID 与音频时间线（timeline）管理。

从清单文件 manifest.json 可以看到，该扩展版本为0.2.3，依赖ten_runtime_python（0.11）与ten_ai_base（0.7）两个系统包，并通过api.interface引用asr-interface.json，即遵循 TEN Framework 标准的 ASR 接口约定；addon.py中使用@register_addon_as_extension("aws_asr_python")将扩展注册到运行时，因此它可以被 TEN Framework 的图配置直接实例化。

配置指南：必填项、可选项与完整示例

扩展的配置通过 TEN Framework 的 property 机制注入，在on_init阶段由AWSASRConfig.model_validate_json()完成解析与校验（见 extension.py）。配置主体是一个params对象，内含认证信息与转录参数。

必需参数

params：AWS Transcribe 配置对象，包含认证信息和转录设置。在 config.py 中，以下字段被声明为必填（Pydantic 中无默认值的字段）：

字段	说明	示例
`region`	AWS 区域	`us-west-2`
`access_key_id`	AWS 访问密钥 ID	`your_aws_access_key_id`
`secret_access_key`	AWS 秘密访问密钥	`your_aws_secret_access_key`
`language_code`	语言代码	`en-US`、`zh-CN`
`media_sample_rate_hz`	音频采样率（Hz）	`16000`
`media_encoding`	音频编码格式	`pcm`

扩展级可选参数

dump：是否启用音频转储（默认false）；
dump_path：转储音频文件路径（默认：扩展目录下的aws_asr_in.pcm）；
log_level：日志级别（默认INFO）；
finalize_mode：完成模式，可选disconnect或mute_pkg（默认disconnect）；
mute_pkg_duration_ms：静音包持续时间（毫秒，默认800）。

AWS Transcribe 可选参数（params 内）

以下字段在 config.py 中均为可选（默认None），并会在启动转录时透传给start_stream_transcription：

字段	说明
`vocabulary_name`	自定义词汇表名称，参考 AWS 文档 custom-vocabulary
`session_id`	会话 ID
`vocab_filter_method`	词汇过滤方法
`vocab_filter_name`	词汇过滤器名称
`show_speaker_label`	是否显示说话人标签
`enable_channel_identification`	是否启用声道识别
`number_of_channels`	声道数量
`enable_partial_results_stabilization`	是否启用部分结果稳定化
`partial_results_stability`	部分结果稳定性设置（如`HIGH`）
`language_model_name`	语言模型名称

完整配置示例

文档给出的完整配置（docs/README.zh-CN.md）：

{ "params": { "region": "us-west-2", "access_key_id": "your_aws_access_key_id", "secret_access_key": "your_aws_secret_access_key", "language_code": "en-US", "media_sample_rate_hz": 16000, "media_encoding": "pcm", "vocabulary_name": "custom-vocabulary", "show_speaker_label": true, "enable_partial_results_stabilization": true, "partial_results_stability": "HIGH" }, "dump": false, "log_level": "INFO", "finalize_mode": "disconnect", "mute_pkg_duration_ms": 800 }

仓库自带的 property.json 演示了使用环境变量占位符注入敏感凭据的推荐做法：

{ "params": { "region": "${env:AWS_ASR_REGION|}", "access_key_id": "${env:AWS_ASR_ACCESS_KEY_ID|}", "secret_access_key": "${env:AWS_ASR_SECRET_ACCESS_KEY|}", "language_code": "en-US", "media_sample_rate_hz": 16000, "media_encoding": "pcm" } }

即通过AWS_ASR_REGION、AWS_ASR_ACCESS_KEY_ID、AWS_ASR_SECRET_ACCESS_KEY三个环境变量提供配置值（占位符中|后的空串为缺省值），避免把密钥硬编码进配置。测试目录中的 property_zh.json 还给出了zh-CN语言代码的变体。

凭据序列化保护

值得注意的细节：config.py通过encrypting_serializer("access_key_id", "secret_access_key")（实现见 utils.py）为两个密钥字段注册了 Pydantic JSON 序列化器。当调用model_dump_json()输出配置日志时，密钥只会显示为前缀***后缀的掩码形式（仅取字符串前 1/5 与后 1/5 字符，最长 5 字符），防止密钥在日志中泄露，同时不影响内部使用。

核心 API 与异步生命周期

该扩展实现了AsyncASRBaseExtension接口（源自ten_ai_base.asr），核心方法均以异步方式定义，由 TEN 运行时在合适的时机回调：

方法	职责
`on_init()`	读取并校验配置，初始化 AWS ASR 客户端、音频转储器与重连管理器
`start_connection()`	建立与 AWS Transcribe 服务的连接并启动转录事件消费协程
`stop_connection()`	关闭与 ASR 服务的连接（并停止音频转储）
`send_audio()`	发送音频帧进行识别
`finalize()`	完成当前识别会话（按`finalize_mode`分派）
`is_connected()`	检查连接状态

内部实现方法还包括：

_handle_transcript_event()：处理转录事件；
_disconnect_aws()：断开 AWS 连接；
_reconnect_aws()：重新连接 AWS；
_handle_finalize_disconnect()：处理“断开连接”完成模式；
_handle_finalize_mute_pkg()：处理“静音包”完成模式。

连接建立流程

start_connection()（extension.py）的实际调用链如下：

使用TranscribeStreamingClient(**self.config.params.to_client_params())创建客户端。to_client_params()返回region与StaticCredentialResolver(access_key_id, secret_access_key)，即使用静态凭据解析器完成认证；
调用client.start_stream_transcription(**self.config.params.to_transcription_params())建立转录流。to_transcription_params()会过滤掉region、密钥、log_level、finalize_mode、mute_pkg_duration_ms等客户端级字段，仅把language_code、media_sample_rate_hz、media_encoding及所有可选转录参数透传给 AWS API；
连接成功后connected = True，并通知ReconnectManager.mark_connection_successful()重置重连计数；
通过asyncio.create_task(_handle_events())启动事件消费协程，异步遍历stream.output_stream，对TranscriptEvent调用_handle_transcript_event()；若流被关闭（异常），则置connected = False并触发_reconnect_aws()自动重连。

音频发送与格式约束

send_audio()接收AudioFrame，取出缓冲区字节后：

若启用了dump，将原始字节推入音频转储器；
依据采样率计算本次音频时长并记入audio_timeline（len(buf) / (sample_rate / 1000 * 2)，即 16-bit 采样每样本 2 字节）；
通过stream.input_stream.send_audio_event(audio_chunk=bytes(buf))送入 AWS 转录流；
当流已关闭时会抛出IOError，此时扩展将其视为需要重连的信号，置connected = False并调用_reconnect_aws()，返回False。

音频格式约束：PCM16（16 位 PCM）、支持多种采样率（如 16000 Hz）、单声道。从时长计算公式可以看出，扩展按 2 字节/样本的 16 位 PCM 单声道布局处理输入音频，配置时需保证输入帧格式与media_encoding、media_sample_rate_hz一致。

转录结果处理

_handle_transcript_event()逐条遍历transcript.results，对每个候选结果：

若结果非 partial（is_partial == False）且之前有未完成的finalize计时，则发送send_asr_finalize_end()通知会话结束；
提取alternatives[0]中的词级信息（content、stable、start_time、end_time），结合audio_timeline把 AWS 相对时间戳换算为全局音频时间线偏移，构造ASRWord列表；
最终通过send_asr_result()向下游输出ASRResult，包含text、final、start_ms、duration_ms、language与words等字段。

is_connected()的实现同时检查self.stream非空、底层输入流未关闭且self.connected为真，保证状态判断与实际可用性一致。

finalize 完成模式：disconnect 与 mute_pkg

finalize()根据finalize_mode选择会话收尾方式（extension.py）：

disconnect（默认）：调用_handle_finalize_disconnect()，直接向流发送end_stream()结束转录，等待 AWS 返回最终结果。这种模式适合“说完即断”的回合式交互，能最快获得确定性的最终文本；
mute_pkg：调用_handle_finalize_mute_pkg()，不关闭流，而是构造一段长度为mute_pkg_duration_ms * sample_rate / 1000 * 2字节的全零静音包发送给 AWS，并通过audio_timeline.add_silence_audio()把这段静音计入时间线。静音包会促使 AWS 返回已识别内容的最终结果，同时保持连接复用，适合连续多轮对话场景。默认静音包时长 800ms 可在配置中调整。

自动重连机制与指数退避

扩展内置自动重连机制（实现见 reconnect_manager.py），关键策略与文档描述一致：

最多 5 次重连尝试（max_attempts = 5，可在构造时调整）；
指数退避：基础延迟 300ms，按delay = base_delay * 2^(attempts-1)计算，即依次为300ms、600ms、1.2s、2.4s、4.8s；
连接成功后自动重置计数器：mark_connection_successful()会复位attempts；
详细日志：每次重连前记录Attempting reconnection #N/5 after X seconds delay...，达到上限后通过error_handler发送FATAL_ERROR模块错误。

重连由_reconnect_aws()驱动：先通过can_retry()判断是否还有重试额度，再调用handle_reconnect(connection_func=self.start_connection, error_handler=self.send_asr_error)执行单次带延迟的重连尝试。整个链路保证了在网络抖动或 AWS 侧断流时语音服务可以自动恢复。

开发与测试

构建

扩展作为 TEN Framework 构建系统的一部分进行构建，无需额外的构建步骤（文档说明见 docs/README.zh-CN.md）。依赖声明在 requirements.txt 与 pyproject.toml 中：

amazon-transcribe==0.6.4：AWS Transcribe Python 客户端库；
pydantic（>=2.13.4）：配置校验与数据模型；
typing_extensions（>=4.15.0）：类型提示；
pytest==8.3.4：测试框架（开发依赖）。

Python 版本要求>=3.10。

测试

运行单元测试：

pytest tests/

测试目录提供了完整的可运行验证链路：

test_asr_result.py：通过AsyncExtensionTester以单扩展测试模式加载aws_asr_python，持续发送 16 位 PCM 音频帧（320 字节/帧），并校验下游asr_result数据包含id、text、final、start_ms、duration_ms、language、metadata字段以及session_id回传正确，收到final=true结果后结束测试；
mock.py：通过unittest.mock.patch替换TranscribeStreamingClient，模拟流式转录的输入/输出流，其中MockInputStream会在发送超过 100 个音频块后抛出IOError以模拟流关闭场景，从而验证重连逻辑；
conftest.py：以FakeApp在独立线程中启动一个最小化的 TEN 应用作为测试宿主，为扩展测试提供运行时环境。

使用方法

安装：扩展随 TEN Framework 自动安装；
配置：设置 AWS 凭据（推荐环境变量方式）与 Transcribe 参数；
集成：通过 TEN Framework 的 ASR 接口在图中使用该扩展；
监控：检查日志进行调试与监控，日志类别包含vendor（厂商状态/错误）与key_point（关键节点配置）两类，便于过滤检索。

故障排除

常见问题

连接失败：检查 AWS 凭据是否正确、网络是否可达 AWS Transcribe 服务端点；
认证错误：验证 AWS 访问密钥及 IAM 权限是否包含 Transcribe 流式转录所需的transcribe:StartStreamTranscription权限；
音频质量问题：验证音频格式（PCM16、单声道）与采样率设置是否与media_sample_rate_hz、media_encoding一致；
性能问题：调整缓冲区设置与语言模型（language_model_name）等参数；
日志问题：配置适当的log_level（如DEBUG）以获取更详细的转录事件与重连日志。

调试模式

通过在配置中设置dump: true启用调试模式，把送入 AWS 的原始音频录制为 PCM 文件（默认路径aws_asr_in.pcm，可通过dump_path修改）。需要注意的是，当dump_path不以.pcm结尾时，扩展会将其视为目录并在其下创建aws_asr_in.pcm（见 extension.py）。录制的文件可用于离线回放、人工比对识别文本，或与下游 TTS 输入对照排查链路问题。

许可证

此扩展是 TEN Framework 的一部分，根据 Apache License, Version 2.0 授权。

延伸阅读：想深入了解本扩展所实现的 ASR 接口约定，可查看ten_ai_base系统中的 asr-interface.json（manifest 中引用的接口定义）；想了解如何将该扩展编排进完整的语音 Agent 图，可参考仓库中 ai_agents/agents/examples 下的 voice-assistant 系列示例。