基于 TEN Framework 的 AsyncAvatarBaseExtension 基类：7 个方法快速实现数字人/虚拟形象扩展

人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载导读AsyncAvatarBaseExtension是 TEN Framework 为「数字人 / 虚拟形象Avatar / Digital Human」类扩展提供的一个异步基类它将生命周期管理、音频队列与处理循环、采样率校验、错误上报、flush/finalize消息处理等繁琐逻辑全部封装起来开发者只需要实现7 个业务方法即可接入任意 Avatar 服务商。本文以仓库中的 AVATAR_BASE_README.md 为主线结合 avatar_base.py 源码与 Spatius 参考实现讲清基类的设计原理、每个回调方法的职责与调用时机并给出可直接复制运行的完整示例。一、基类能为你做什么基类的核心价值在于把「与具体 Avatar 服务商无关」的通用逻辑全部接管让子类只关注「服务商特有的业务」✅ 自动生命周期管理on_init/on_start/on_stop/on_deinit✅ 内置音频队列asyncio.Queue与异步处理循环✅ 采样率校验不支持的采样率直接拒绝并上报错误✅ 统一的错误处理与日志输出支持标准化的ModuleError错误载荷✅ 消息处理flush命令、finalize数据✅ 可选的音频落盘audio dumping便于排查音频问题从源码看基类继承自AsyncExtension与ABC通过abstractmethod强制子类实现业务方法同时把on_init、on_start、on_stop等生命周期钩子标记为「由基类管理不要覆写」见 avatar_base.py。子类唯一要做的事就是补上 7 个抽象方法。二、快速上手必须实现的 7 个方法创建自己的 Avatar 扩展只需继承AsyncAvatarBaseExtension并实现以下 7 个方法。1.validate_config(ten_env) - bool加载并校验配置调用时机在on_init()阶段由基类自动调用await self.validate_config(ten_env)。返回值配置有效返回True否则返回False。失败后果返回False时基类会记录错误日志、禁用音频处理并且不会调用connect_to_avatar()扩展处于「可运行但不工作」的安全状态。async def validate_config(self, ten_env: AsyncTenEnv) - bool: self.config await MyAvatarConfig.create_async(ten_env) if not self.config.api_key: ten_env.log_error(api_key is required) return False return True2.get_target_sample_rate() - list[int]声明支持的采样率调用时机每当有音频帧到达时基类会用该方法的返回值校验帧采样率。返回值支持的服务商采样率列表单位 Hz。拒绝策略不在列表中的采样率会被拒绝并上报一次错误code1001且错误只发送一次以避免刷屏。def get_target_sample_rate(self) - list[int]: return [24000] # Spatius 支持 24kHz # return [16000] # Sensetime 支持 16kHz # return [24000, 48000] # 支持多个采样率注意基类不做重采样音频数据会原样as-is交给服务商因此你必须保证上游音频帧的采样率与get_target_sample_rate()声明一致或由上游如 TTS 扩展负责转换。3.connect_to_avatar(ten_env) - None建立与 Avatar 服务的连接调用时机配置校验通过后在on_start()阶段由基类调用。失败行为如果此方法抛出异常基类会记录错误、发送标准错误载荷然后重新抛出异常—— 扩展将启动失败。async def connect_to_avatar(self, ten_env: AsyncTenEnv) - None: self.client MyAvatarClient(self.config) await self.client.connect() ten_env.log_info(Connected to avatar service)4.disconnect_from_avatar(ten_env) - None断开连接并释放资源调用时机在on_stop()阶段由基类自动调用。失败行为与连接相反这里抛出的异常只记录日志、不向上传播保证清理流程继续执行。async def disconnect_from_avatar(self, ten_env: AsyncTenEnv) - None: if self.client: await self.client.disconnect() ten_env.log_info(Disconnected from avatar service)5.send_audio_to_avatar(audio_data: bytes) - None发送音频调用时机由音频处理循环自动调用一帧一调。注意音频为原始 PCM 字节流不做重采样如果服务商要求 base64 等编码在本方法内自行转换。async def send_audio_to_avatar(self, audio_data: bytes) - None: # 示例如果服务商要求 base64 编码 base64_audio base64.b64encode(audio_data).decode(utf-8) await self.client.send_audio(base64_audio)6.send_eof_to_avatar() - None通知音频流结束调用时机当收到finalize数据时基类会把 EOF 哨兵队列中的None项排到待发送音频之后由处理循环自动调用本方法确保「先发完已有音频再发 EOF」。async def send_eof_to_avatar(self) - None: await self.client.send_eof()7.interrupt_avatar() - None立即打断当前播报调用时机收到flush命令时调用用于立即停止 Avatar 当前正在进行的语音播报。async def interrupt_avatar(self) - None: if self.client: await self.client.interrupt()三、可选方法get_dump_config()音频落盘调试除 7 个必选方法外还有一个可选方法用于调试def get_dump_config(self) - tuple[bool, str]: 返回值(是否落盘, 落盘目录) if self.config.dump: return (True, self.config.dump_path) return (False, ) # 默认不落盘默认值(False, )即不落盘。落盘规则音频被保存为{dump_path}/{扩展名}_in.pcm追加写模式目录不存在时会自动创建见 avatar_base.py。这对排查「音频没发出去 / 波形异常 / 采样率不对」等问题非常有用。四、完整示例一个可直接运行的 Avatar 扩展以下代码综合了文档示例与基类约定是可复制运行的完整骨架from ten_runtime import AsyncTenEnv from ten_ai_base.config import BaseConfig from avatar_base import AsyncAvatarBaseExtension from dataclasses import dataclass import base64 dataclass class MyAvatarConfig(BaseConfig): api_key: str avatar_id: str default sample_rate: int 24000 dump: bool False dump_path: str class MyAvatarExtension(AsyncAvatarBaseExtension): def __init__(self, name: str): super().__init__(name) self.config: MyAvatarConfig | None None self.client None # 1. 校验配置 async def validate_config(self, ten_env: AsyncTenEnv) - bool: self.config await MyAvatarConfig.create_async(ten_env) if not self.config.api_key: ten_env.log_error([MyAvatar] api_key is required) return False ten_env.log_info(f[MyAvatar] Config validated (avatar{self.config.avatar_id})) return True # 2. 目标采样率 def get_target_sample_rate(self) - list[int]: return [self.config.sample_rate] # 3. 连接服务 async def connect_to_avatar(self, ten_env: AsyncTenEnv) - None: ten_env.log_info([MyAvatar] Connecting...) self.client MyAvatarClient(self.config) await self.client.connect() ten_env.log_info([MyAvatar] Connected) # 4. 断开连接 async def disconnect_from_avatar(self, ten_env: AsyncTenEnv) - None: if self.client: await self.client.disconnect() ten_env.log_info([MyAvatar] Disconnected) # 5. 发送音频 async def send_audio_to_avatar(self, audio_data: bytes) - None: if self.client: base64_audio base64.b64encode(audio_data).decode(utf-8) await self.client.send_audio(base64_audio) # 6. 发送 EOF async def send_eof_to_avatar(self) - None: if self.client: await self.client.send_eof() # 7. 打断播报 async def interrupt_avatar(self) - None: if self.client: await self.client.interrupt() # 可选音频落盘 def get_dump_config(self) - tuple[bool, str]: if self.config: return (self.config.dump, self.config.dump_path) return (False, )五、自动生命周期你不需要覆写任何生命周期钩子基类把完整生命周期封装成一条固定流水线1. on_init() └─ validate_config() 2. on_start() └─ connect_to_avatar() └─ 启动音频处理循环asyncio.create_task 3. 音频处理自动 └─ on_audio_frame() 接收音频 └─ 校验采样率 └─ 入队unbounded queue └─ 处理循环调用 send_audio_to_avatar() 4. 消息处理自动 └─ flush 命令 → interrupt_avatar() └─ finalize 数据 → send_eof_to_avatar() 5. on_stop() └─ 取消音频处理任务 └─ disconnect_from_avatar()你不需要覆写on_init()、on_start()或on_stop()从 avatar_base.py 可以看到基类在这些钩子内部完成了配置校验on_init、连接与任务启动on_start、任务取消与断开on_stop并且on_stop中还会用asyncio.CancelledError妥善收尾音频任务。六、音频处理细节采样率校验每帧音频先取source_rate audio_frame.get_sample_rate()再与get_target_sample_rate()返回的列表比对见 avatar_base.py。不支持的采样率会被拒绝并通过_send_error上报code1001的错误数据。_sample_rate_error_sent标志保证同一轮请求只报一次错避免日志刷屏该标志在_clear_request_context()中重置。音频队列音频帧被包装为QueuedAudioFrame(audiobytes)放入asyncio.Queue无界队列见 avatar_base.py。处理循环_process_audio_loop逐个取出并调用send_audio_to_avatar()循环内对CancelledError单独处理其他异常记录日志并通过_send_error上报后继续处理下一帧。收到flush命令时队列会被清空_clear_audio_queue会统计并记录清除了多少帧。音频落盘通过get_dump_config()返回(True, /path/to/dump)开启。音频写入{dump_path}/{扩展名}_in.pcm如spatius_avatar_python_in.pcm适用于排查「上游是否真的发来了音频」「PCM 内容是否正确」等问题。七、错误处理策略基类对四类错误采用分层策略这是它最值得借鉴的设计之一错误场景处理方式结果配置校验失败validate_config返回False记录错误日志禁用音频处理connect_to_avatar()不会被调用连接失败connect_to_avatar抛异常记录错误 发送错误载荷异常向上传播扩展启动失败断开失败disconnect_from_avatar抛异常只记录错误日志异常不传播清理流程继续音频发送失败send_audio_to_avatar抛异常记录错误 发送错误载荷处理循环继续处理下一帧其中连接与音频发送失败时基类会调用_send_error构造一个标准的ModuleError载荷moduleavatar、携带vendor/vendor_code/vendor_message等字段通过名为error的Data消息发送出去见 avatar_base.py。这套标准化错误协议在 tests/test_basic.py 中有对应的测试用例验证测试断言错误载荷的module avatar、vendor spatius、且vendor_metadata中不包含空值。八、消息处理flush 与 finalizeflush 命令当收到flush命令CMD_IN_FLUSH时基类依次执行清空音频队列丢弃未发送的积压音频调用interrupt_avatar()打断当前播报将flush命令转发给下游ten_env.send_cmd(Cmd.create(CMD_OUT_FLUSH))返回StatusCode.OK的CmdResult。finalize 数据当收到名为finalize的数据时基类将 EOF 哨兵None放入音频队列尾部排在所有待发送音频之后由处理循环顺序消费到哨兵时调用send_eof_to_avatar()。这样既保证了「TTS 播报音频已全部发送完毕」的语义又不会打断正在发送的音频流。九、参考实现Spatius Avatar 扩展仓库中的spatius_avatar_python包是一个完整的参考实现它通过 Spatius SDK 驱动真实数字人服务并演示了基类的全部用法avatar_base.py —— 基类实现本文主体。extension.py —— Spatius 参考实现实现 7 个必选方法与get_dump_config()等可选方法。addon.py —— 通过register_addon_as_extension(spatius_avatar_python)注册扩展。manifest.json —— 声明扩展的 API 契约audio_frame_in、cmd_in、data_in、data_out及全部配置属性。property.json —— 默认配置支持${env:VAR|}环境变量注入如SPATIUS_API_KEY、AGORA_APP_ID。tests/test_basic.py —— 单元测试覆盖基础命令往返与配置错误的标准载荷。Spatius 实现中的几个亮点1. 配置归一化与校验。SpatiusConfig通过update_params()把用户可见的params字典复制到规范化字段再用validate_params()检查必填项、采样率范围Ogg Opus 仅支持8000/12000/16000/24000/48000Hz与 Agora token 二选一agora_token或agora_appcert至少提供一个。2. Token 自动生成。若只配置了agora_appcertresolve_agora_token()会用agora-token-builder的RtcTokenBuilder.buildTokenWithUid结合session_expire_minutes默认 30 分钟自动生成 RTC Token。3. 敏感信息脱敏。日志与vendor_metadata中的 API Key、App Cert 等均通过encrypt()脱敏输出get_vendor_metadata()还会剔除空值字段避免把空串上报出去。4. 连接流程。connect_to_avatar()用new_avatar_session(...)创建会话随后await session.init()获取鉴权 token、await session.start()建立 WebSocket 连接并拿到connection_idsend_audio_to_avatar通过session.send_audio(bytes, endFalse)发送send_eof_to_avatar则用session.send_audio(b, endTrue)表示流结束。配置参考property.json{ dump: false, dump_path: , channel: , agora_uid: , agora_token: , agora_appid: , agora_appcert: , agora_channel: , params: { spatius_api_key: ${env:SPATIUS_API_KEY|}, spatius_app_id: ${env:SPATIUS_APP_ID|}, spatius_avatar_id: , agora_uid: , agora_token: , agora_appid: ${env:AGORA_APP_ID|}, agora_appcert: ${env:AGORA_APP_CERTIFICATE|}, agora_channel: , region: , sample_rate: 24000, session_expire_minutes: 30, audio_format: ogg_opus } }其中agora_token与agora_appcert至少配置一个sample_rate默认24000与get_target_sample_rate()返回[self.config.sample_rate]保持一致audio_format默认ogg_opus。十、落地建议与总结实现自己的 Avatar 扩展遵循以下清单即可✅ 创建继承自BaseConfig的配置类dataclass并在validate_config()中加载与校验✅ 继承AsyncAvatarBaseExtension✅ 实现 7 个必选方法外加可选的get_dump_config()✅ 使用统一的日志前缀基类LOG_PREFIX默认[Spatius]子类可按需覆盖保证日志可检索✅ 用不同采样率的音频测试get_target_sample_rate()的校验逻辑✅ 妥善处理错误连接失败要让扩展启动失败发送失败要保证循环继续。其余一切 —— 生命周期、队列、循环、采样率校验、flush/finalize、错误上报 —— 都由AsyncAvatarBaseExtension自动完成。这种「基类兜底通用逻辑、子类只写业务差异」的模板方法设计让接入新的数字人服务商从「理解整个框架消息流」简化为「实现 7 个方法」是 TEN Framework 在语音 Agent 扩展开发上的一个值得复用的范式。赞分享人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载相关推荐TEN 框架数字人扩展开发指南基于 AsyncAvatarBaseExtension 实现自定义虚拟形象扩展TEN 框架数字人扩展开发指南基于 AsyncAvatarBaseExtension 实现自定义虚拟形象扩展 导读 本文以 TEN 框架中 spatius_a人工智能AI Agent多模态语音AI 应用昇腾C BatchNorm Tiling APIBatchNorm Tiling 功能说明 BatchNorm Tiling API用于获取BatchNorm kernel计算时所需的Tiling参数。获取T人工智能深度学习算子库CANNAscend数字人Live2D终极指南快速打造你的专属虚拟形象数字人Live2D终极指南快速打造你的专属虚拟形象 想要拥有一个会说话、会互动的数字人伙伴吗《Awesome Digital Human Live2D》项目人工智能AI 应用数字人语音AI Agent交互助手上一篇Open Images Dataset 终极上手指南从零开始构建图像识别模型下一篇WebAuthn实战教程从零开始实现安全的用户注册流程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考