尧图网站建设 
TEN 框架数字人扩展开发指南基于 AsyncAvatarBaseExtension 实现自定义虚拟形象扩展【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework导读本文以 TEN 框架中spatius_avatar_python扩展包内提供的AsyncAvatarBaseExtension基类为主线系统讲解如何在 TEN 框架面向对话式语音 AI Agent 的开源框架中开发数字人/虚拟形象Avatar扩展。阅读本文后你将掌握基类自动托管的生命周期模型、7 个必须实现的业务方法、音频队列与采样率校验机制、flush/finalize 消息语义以及错误处理协议并能够基于仓库内的 Spatius 参考实现快速接入自己的数字人服务商。概述为什么需要 Avatar 基类数字人扩展的职责是把 TTS 产出的音频流转发给第三方数字人服务同时响应会话中的打断flush与收尾finalize事件。这类扩展的样板逻辑高度相似生命周期管理、音频入队与消费、采样率校验、错误上报、消息分发若每个服务商扩展各自实现一遍既容易出错也难以维护。AsyncAvatarBaseExtension正是为此设计的基类它位于 avatar_base.py自动处理以下全部工作生命周期管理on_init、on_start、on_stop、on_deinit全部由基类托管音频队列与处理循环内置无界asyncio.Queue与后台消费协程采样率验证按目标采样率白名单校验音频帧不支持的采样率被拒绝并上报错误错误处理与日志记录统一的日志前缀、标准化的ModuleError错误载荷消息处理flush命令与finalizedata 的语义化处理可选音频转储便于调试的 PCM 落盘能力。开发者只需实现 7 个业务方法其余全部交给基类。快速开始7 个必需方法创建自定义数字人扩展只需继承AsyncAvatarBaseExtension并实现以下 7 个抽象方法。它们定义于 avatar_base.py以下逐一说明用途、调用时机与实现要点。1.validate_config(ten_env) - bool—— 加载并校验配置用途加载并验证扩展配置。调用时机on_init()期间自动调用。返回值配置有效返回True否则返回False。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 True从源码看on_init中基类会执行config_valid await self.validate_config(ten_env)并用返回值同时设置_config_valid与_audio_processing_enabled两个内部状态avatar_base.py。也就是说配置校验失败会直接禁用音频处理并阻止后续的连接动作。2.get_target_sample_rate() - list[int]—— 声明支持的采样率用途指定数字人服务支持的音频采样率白名单。调用时机每帧音频到达时都会读取。返回值支持的采样率列表单位 Hz。def get_target_sample_rate(self) - list[int]: return [24000] # Spatius 支持 24kHz # return [16000] # Sensetime 支持 16kHz # return [24000, 48000] # 支持多个采样率基类在on_audio_frame中对每一帧执行source_rate not in supported_rates白名单检查avatar_base.py因此采样率列表必须与上游 TTS 实际输出保持一致。3.connect_to_avatar(ten_env) - None—— 建立服务连接用途创建数字人服务客户端并建立连接。调用时机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)源码中on_start对连接调用做了try/except包裹捕获异常后先通过_send_error上报标准化错误code-1、vendor_codetype(e).__name__随后raise重新抛出从而让扩展启动失败avatar_base.py。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—— 发送音频用途把音频数据发送给数字人服务。调用时机由音频处理循环自动调用。注意音频按原样发送基类不做重采样。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—— 发送音频流结束信号用途通知数字人服务音频流已结束。调用时机收到finalizedata 时自动调用详见下文消息处理。async def send_eof_to_avatar(self) - None: await self.client.send_eof()7.interrupt_avatar() - None—— 立即中断数字人用途立即中断数字人当前语音输出。调用时机收到flush命令时自动调用。async def interrupt_avatar(self) - None: if self.client: await self.client.interrupt()可选方法get_dump_config() - tuple[bool, str]用途启用音频转储用于调试。返回值(should_dump, dump_path)。默认值(False, )即默认不转储见 avatar_base.py。def get_dump_config(self) - tuple[bool, str]: if self.config.dump: return (True, self.config.dump_path) return (False, )转储实现位于_dump_audio当开关打开时会在dump_path下自动os.makedirs创建目录并把进入队列前的原始 PCM 以追加模式写入{extension_name}_in.pcm文件avatar_base.py非常适合排查音频没到达服务端一类的问题。完整示例一个最小可用的数字人扩展把 7 个方法组合起来就是一个完整的扩展实现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, )注意示例中MyAvatarClient需要你自行实现对接真实服务商 SDK基类只负责编排调用时机。自动生命周期基类托管的全流程基类把完整的生命周期封装在on_init/on_start/on_stop/on_deinit中avatar_base.py流程如下1. on_init() └─ validate_config() 2. on_start() └─ connect_to_avatar() └─ 启动音频处理循环 3. 音频处理自动 └─ on_audio_frame() 接收音频 └─ 检查采样率 └─ 音频入队 └─ 从循环中调用 send_audio_to_avatar() 4. 消息处理自动 └─ flush 命令 → interrupt_avatar() └─ finalize data → send_eof_to_avatar() 5. on_stop() └─ 停止音频处理循环 └─ disconnect_from_avatar()你不需要重写on_init()、on_start()或on_stop()从源码细节看on_start在配置有效的前提下会依次完成连接服务→创建_process_audio_loop后台任务两个动作on_stop则会先置_audio_processing_enabled False取消音频处理任务吞掉asyncio.CancelledError再执行断开连接最后调用_clear_request_context()清理请求级上下文。音频处理细节采样率验证每帧音频都会根据get_target_sample_rate()的返回值做白名单检查不支持的采样率会被拒绝并向图内发送errordatacode1001错误只发送一次_sample_rate_error_sent标志位避免日志刷屏该标志会在 flush/finalize 处理后被_clear_request_context()重置从而允许下一轮会话再次报错。对应实现见 avatar_base.py。音频队列音频通过put_nowait进入无界asyncio.Queue队列中的元素为QueuedAudioFrame(audiobytes)数据类后台_process_audio_loop循环逐帧取出并调用send_audio_to_avatar()收到flush命令时队列会被清空_clear_audio_queue逐项get_nowait丢弃。音频转储通过get_dump_config()返回(True, /path/to/dump)启用音频保存到{dump_path}/{extension_name}_in.pcm用于调试音频链路问题例如确认上游是否真的把 PCM 帧送进了扩展。错误处理基类把错误处理策略固化在生命周期各处子类无需重复实现场景基类行为配置验证失败validate_config返回False记录错误、禁用音频处理_audio_processing_enabled False、不调用connect_to_avatar()连接错误connect_to_avatar抛异常记录错误并发送errordatacode-1随后重新抛出扩展启动失败断开连接错误disconnect_from_avatar抛异常记录错误异常不传播清理流程继续音频处理错误send_audio_to_avatar抛异常记录错误并上报code-1循环继续处理下一帧错误上报统一走_send_error方法它构造ModuleError载荷moduleavatar附带调用方自定义的vendor、vendor_code、vendor_message并把请求元数据turn_id等与厂商脱敏信息合并进vendor_metadata最后通过Data.create(error)发送到图内avatar_base.py。此外基类还提供了三个可选的厂商信息钩子get_vendor_name()默认返回扩展名、get_vendor_metadata()返回脱敏后的厂商维度与get_reporting_session_id()返回会话标识用于标准化错误载荷中的厂商维度。消息处理flush 与 finalize基类在on_cmd与on_data中拦截两条关键消息avatar_base.py常量分别来自ten_ai_base的CMD_IN_FLUSH与模块内定义的DATA_IN_FINALIZE finalize。Flush 命令会话打断收到flush命令时基类依次执行清空音频队列丢弃所有待发送的 PCM 帧调用interrupt_avatar()立即打断数字人当前语音向下游转发flush命令Cmd.create(CMD_OUT_FLUSH)最后向调用方返回StatusCode.OK的CmdResult。这保证了用户插话 → 停止数字人说话 → 清空积压音频的会话语义。Finalize Data音频流收尾收到finalizedata 时基类把 EOF 哨兵队列中的None元素排到所有待处理音频之后入队音频处理循环消费到该哨兵时调用send_eof_to_avatar()随后清理请求上下文表示 TTS 播放音频已 finalize数字人侧可据此做语音合成收尾。从_handle_finalize的实现看若配置校验失败导致音频处理被禁用finalize 会被直接忽略不会发送 EOFavatar_base.py。仓库参考实现Spatius 数字人扩展文档提到的spatius_avatar_python/extension.py是上述基类的完整落地参考实现了 7 个必需方法并接入 Spatius 数字人服务extension.py关键文件avatar_base.py —— 基类实现extension.py —— Spatius 扩展实现addon.py —— Addon 注册入口register_addon_as_extension(spatius_avatar_python)。SpatiusConfig参数到配置的归一化参考实现定义了一个值得借鉴的模式SpatiusConfig继承BaseConfig内部保留一份用户级paramsTypedDict通过update_params()把用户传入参数归一化到强类型字段再由validate_params()做必填校验必填项spatius_api_key、spatius_app_id、spatius_avatar_id、agora_uid、agora_appid、agora_channelagora_token与agora_appcert二者必填其一sample_rate必须大于 0audio_format必须是AudioFormat枚举值之一且选择ogg_opus时采样率必须落在{8000, 12000, 16000, 24000, 48000}agora_uid必须是整数字符串session_expire_minutes必须大于 0。其对应的默认配置见 property.json敏感字段支持${env:SPATIUS_API_KEY|}形式的环境变量注入例如{ 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 } }各参数含义sample_rate默认 24000Spatius 支持的采样率audio_format默认ogg_opussession_expire_minutes默认 30用于生成 Agora token 的过期时间region留空时使用 SDK 默认区域。扩展还实现了resolve_agora_token()优先使用显式配置的agora_token否则基于agora_appid、agora_appcert、agora_channel、agora_uid与过期时间调用RtcTokenBuilder.buildTokenWithUid动态生成。7 个方法的 Spatius 落地方式参考实现中的方法映射extension.pyvalidate_config创建配置 →update_params()→validate_params()失败时上报code-1012的错误并返回False同时用encrypt()对 api_key、token 等敏感字段做脱敏后才写入日志get_target_sample_rate返回[self.config.sample_rate]connect_to_avatar构造AgoraEgressConfigchannel/token/uid调用new_avatar_session(...)创建会话await session.init()获取鉴权 token再await session.start()建立 WebSocket 连接拿到connection_iddisconnect_from_avatarawait session.close()并清空会话引用send_audio_to_avatarsession.send_audio(bytes(audio_data), endFalse)send_eof_to_avatarsession.send_audio(b, endTrue)以空包携带结束标记interrupt_avatarsession.interrupt()失败仅告警不抛出。扩展同时实现了get_vendor_name()返回spatius、get_reporting_session_id()返回agora_uid与get_vendor_metadata()脱敏后的厂商维度过滤空值。测试用例行为契约的验证仓库配套的测试tests/test_basic.py用 TEN 的AsyncExtensionTester验证了两条关键契约test_basic以单扩展模式启动spatius_avatar_python发送hello_world命令并断言返回StatusCode.OK验证扩展能正常加载启动test_config_error_uses_module_error在缺少配置时断言扩展发出errordata且载荷满足module avatar、code -1012、vendor spatius、vendor_code ValueError同时vendor_metadata不包含url与key等敏感字段——这正是_send_error标准化错误协议的回归保障。测试的 fixturetests/conftest.py通过一个后台线程运行FakeApp在on_init中释放事件锁配合pytest完成整个 TEN 运行时环境的搭建与销毁。扩展包的声明与依赖manifest.json 明确了扩展的类型extension、名称spatius_avatar_python、版本0.2.4依赖ten_runtime_python0.11与ten_ai_base0.7.40并声明了图内协议入向audio_frame_inpcm_frame、cmd_inflush、data_infinalize出向data_outerror。运行依赖见 requirements.txt 与 pyproject.tomlagora-token-builder1.0.0生成 Agora token、spatius[opus]1.0.4Spatius 数字人 SDKopus可选依赖用于 OggOpus 编码、protobuf5.28.3,6。开发新数字人扩展时把其中的spatius换成你自己的服务商 SDK 即可复用整套骨架。总结开发自己的数字人扩展要基于AsyncAvatarBaseExtension实现自己的数字人扩展按以下步骤即可✅ 创建包含服务参数的配置类继承BaseConfig可参考SpatiusConfig的params归一化模式✅ 继承AsyncAvatarBaseExtension✅ 实现 7 个必需方法validate_config、get_target_sample_rate、connect_to_avatar、disconnect_from_avatar、send_audio_to_avatar、send_eof_to_avatar、interrupt_avatar✅ 使用一致的日志前缀基类通过LOG_PREFIX统一日志格式子类可覆盖✅ 使用不同采样率进行测试验证采样率白名单行为✅ 优雅地处理错误连接失败要抛出以阻止启动断开失败要吞掉异常保证清理音频发送失败要记录并继续。基类会自动处理生命周期、音频队列、采样率校验、flush/finalize 消息与错误上报等所有样板逻辑你只需要专注对接数字人服务的业务代码。【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 TEN 框架数字人扩展开发指南：基于 AsyncAvatarBaseExtension 实现自定义虚拟形象扩展
TEN 框架数字人扩展开发指南基于 AsyncAvatarBaseExtension 实现自定义虚拟形象扩展【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework导读本文以 TEN 框架中spatius_avatar_python扩展包内提供的AsyncAvatarBaseExtension基类为主线系统讲解如何在 TEN 框架面向对话式语音 AI Agent 的开源框架中开发数字人/虚拟形象Avatar扩展。阅读本文后你将掌握基类自动托管的生命周期模型、7 个必须实现的业务方法、音频队列与采样率校验机制、flush/finalize 消息语义以及错误处理协议并能够基于仓库内的 Spatius 参考实现快速接入自己的数字人服务商。概述为什么需要 Avatar 基类数字人扩展的职责是把 TTS 产出的音频流转发给第三方数字人服务同时响应会话中的打断flush与收尾finalize事件。这类扩展的样板逻辑高度相似生命周期管理、音频入队与消费、采样率校验、错误上报、消息分发若每个服务商扩展各自实现一遍既容易出错也难以维护。AsyncAvatarBaseExtension正是为此设计的基类它位于 avatar_base.py自动处理以下全部工作生命周期管理on_init、on_start、on_stop、on_deinit全部由基类托管音频队列与处理循环内置无界asyncio.Queue与后台消费协程采样率验证按目标采样率白名单校验音频帧不支持的采样率被拒绝并上报错误错误处理与日志记录统一的日志前缀、标准化的ModuleError错误载荷消息处理flush命令与finalizedata 的语义化处理可选音频转储便于调试的 PCM 落盘能力。开发者只需实现 7 个业务方法其余全部交给基类。快速开始7 个必需方法创建自定义数字人扩展只需继承AsyncAvatarBaseExtension并实现以下 7 个抽象方法。它们定义于 avatar_base.py以下逐一说明用途、调用时机与实现要点。1.validate_config(ten_env) - bool—— 加载并校验配置用途加载并验证扩展配置。调用时机on_init()期间自动调用。返回值配置有效返回True否则返回False。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 True从源码看on_init中基类会执行config_valid await self.validate_config(ten_env)并用返回值同时设置_config_valid与_audio_processing_enabled两个内部状态avatar_base.py。也就是说配置校验失败会直接禁用音频处理并阻止后续的连接动作。2.get_target_sample_rate() - list[int]—— 声明支持的采样率用途指定数字人服务支持的音频采样率白名单。调用时机每帧音频到达时都会读取。返回值支持的采样率列表单位 Hz。def get_target_sample_rate(self) - list[int]: return [24000] # Spatius 支持 24kHz # return [16000] # Sensetime 支持 16kHz # return [24000, 48000] # 支持多个采样率基类在on_audio_frame中对每一帧执行source_rate not in supported_rates白名单检查avatar_base.py因此采样率列表必须与上游 TTS 实际输出保持一致。3.connect_to_avatar(ten_env) - None—— 建立服务连接用途创建数字人服务客户端并建立连接。调用时机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)源码中on_start对连接调用做了try/except包裹捕获异常后先通过_send_error上报标准化错误code-1、vendor_codetype(e).__name__随后raise重新抛出从而让扩展启动失败avatar_base.py。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—— 发送音频用途把音频数据发送给数字人服务。调用时机由音频处理循环自动调用。注意音频按原样发送基类不做重采样。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—— 发送音频流结束信号用途通知数字人服务音频流已结束。调用时机收到finalizedata 时自动调用详见下文消息处理。async def send_eof_to_avatar(self) - None: await self.client.send_eof()7.interrupt_avatar() - None—— 立即中断数字人用途立即中断数字人当前语音输出。调用时机收到flush命令时自动调用。async def interrupt_avatar(self) - None: if self.client: await self.client.interrupt()可选方法get_dump_config() - tuple[bool, str]用途启用音频转储用于调试。返回值(should_dump, dump_path)。默认值(False, )即默认不转储见 avatar_base.py。def get_dump_config(self) - tuple[bool, str]: if self.config.dump: return (True, self.config.dump_path) return (False, )转储实现位于_dump_audio当开关打开时会在dump_path下自动os.makedirs创建目录并把进入队列前的原始 PCM 以追加模式写入{extension_name}_in.pcm文件avatar_base.py非常适合排查音频没到达服务端一类的问题。完整示例一个最小可用的数字人扩展把 7 个方法组合起来就是一个完整的扩展实现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, )注意示例中MyAvatarClient需要你自行实现对接真实服务商 SDK基类只负责编排调用时机。自动生命周期基类托管的全流程基类把完整的生命周期封装在on_init/on_start/on_stop/on_deinit中avatar_base.py流程如下1. on_init() └─ validate_config() 2. on_start() └─ connect_to_avatar() └─ 启动音频处理循环 3. 音频处理自动 └─ on_audio_frame() 接收音频 └─ 检查采样率 └─ 音频入队 └─ 从循环中调用 send_audio_to_avatar() 4. 消息处理自动 └─ flush 命令 → interrupt_avatar() └─ finalize data → send_eof_to_avatar() 5. on_stop() └─ 停止音频处理循环 └─ disconnect_from_avatar()你不需要重写on_init()、on_start()或on_stop()从源码细节看on_start在配置有效的前提下会依次完成连接服务→创建_process_audio_loop后台任务两个动作on_stop则会先置_audio_processing_enabled False取消音频处理任务吞掉asyncio.CancelledError再执行断开连接最后调用_clear_request_context()清理请求级上下文。音频处理细节采样率验证每帧音频都会根据get_target_sample_rate()的返回值做白名单检查不支持的采样率会被拒绝并向图内发送errordatacode1001错误只发送一次_sample_rate_error_sent标志位避免日志刷屏该标志会在 flush/finalize 处理后被_clear_request_context()重置从而允许下一轮会话再次报错。对应实现见 avatar_base.py。音频队列音频通过put_nowait进入无界asyncio.Queue队列中的元素为QueuedAudioFrame(audiobytes)数据类后台_process_audio_loop循环逐帧取出并调用send_audio_to_avatar()收到flush命令时队列会被清空_clear_audio_queue逐项get_nowait丢弃。音频转储通过get_dump_config()返回(True, /path/to/dump)启用音频保存到{dump_path}/{extension_name}_in.pcm用于调试音频链路问题例如确认上游是否真的把 PCM 帧送进了扩展。错误处理基类把错误处理策略固化在生命周期各处子类无需重复实现场景基类行为配置验证失败validate_config返回False记录错误、禁用音频处理_audio_processing_enabled False、不调用connect_to_avatar()连接错误connect_to_avatar抛异常记录错误并发送errordatacode-1随后重新抛出扩展启动失败断开连接错误disconnect_from_avatar抛异常记录错误异常不传播清理流程继续音频处理错误send_audio_to_avatar抛异常记录错误并上报code-1循环继续处理下一帧错误上报统一走_send_error方法它构造ModuleError载荷moduleavatar附带调用方自定义的vendor、vendor_code、vendor_message并把请求元数据turn_id等与厂商脱敏信息合并进vendor_metadata最后通过Data.create(error)发送到图内avatar_base.py。此外基类还提供了三个可选的厂商信息钩子get_vendor_name()默认返回扩展名、get_vendor_metadata()返回脱敏后的厂商维度与get_reporting_session_id()返回会话标识用于标准化错误载荷中的厂商维度。消息处理flush 与 finalize基类在on_cmd与on_data中拦截两条关键消息avatar_base.py常量分别来自ten_ai_base的CMD_IN_FLUSH与模块内定义的DATA_IN_FINALIZE finalize。Flush 命令会话打断收到flush命令时基类依次执行清空音频队列丢弃所有待发送的 PCM 帧调用interrupt_avatar()立即打断数字人当前语音向下游转发flush命令Cmd.create(CMD_OUT_FLUSH)最后向调用方返回StatusCode.OK的CmdResult。这保证了用户插话 → 停止数字人说话 → 清空积压音频的会话语义。Finalize Data音频流收尾收到finalizedata 时基类把 EOF 哨兵队列中的None元素排到所有待处理音频之后入队音频处理循环消费到该哨兵时调用send_eof_to_avatar()随后清理请求上下文表示 TTS 播放音频已 finalize数字人侧可据此做语音合成收尾。从_handle_finalize的实现看若配置校验失败导致音频处理被禁用finalize 会被直接忽略不会发送 EOFavatar_base.py。仓库参考实现Spatius 数字人扩展文档提到的spatius_avatar_python/extension.py是上述基类的完整落地参考实现了 7 个必需方法并接入 Spatius 数字人服务extension.py关键文件avatar_base.py —— 基类实现extension.py —— Spatius 扩展实现addon.py —— Addon 注册入口register_addon_as_extension(spatius_avatar_python)。SpatiusConfig参数到配置的归一化参考实现定义了一个值得借鉴的模式SpatiusConfig继承BaseConfig内部保留一份用户级paramsTypedDict通过update_params()把用户传入参数归一化到强类型字段再由validate_params()做必填校验必填项spatius_api_key、spatius_app_id、spatius_avatar_id、agora_uid、agora_appid、agora_channelagora_token与agora_appcert二者必填其一sample_rate必须大于 0audio_format必须是AudioFormat枚举值之一且选择ogg_opus时采样率必须落在{8000, 12000, 16000, 24000, 48000}agora_uid必须是整数字符串session_expire_minutes必须大于 0。其对应的默认配置见 property.json敏感字段支持${env:SPATIUS_API_KEY|}形式的环境变量注入例如{ 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 } }各参数含义sample_rate默认 24000Spatius 支持的采样率audio_format默认ogg_opussession_expire_minutes默认 30用于生成 Agora token 的过期时间region留空时使用 SDK 默认区域。扩展还实现了resolve_agora_token()优先使用显式配置的agora_token否则基于agora_appid、agora_appcert、agora_channel、agora_uid与过期时间调用RtcTokenBuilder.buildTokenWithUid动态生成。7 个方法的 Spatius 落地方式参考实现中的方法映射extension.pyvalidate_config创建配置 →update_params()→validate_params()失败时上报code-1012的错误并返回False同时用encrypt()对 api_key、token 等敏感字段做脱敏后才写入日志get_target_sample_rate返回[self.config.sample_rate]connect_to_avatar构造AgoraEgressConfigchannel/token/uid调用new_avatar_session(...)创建会话await session.init()获取鉴权 token再await session.start()建立 WebSocket 连接拿到connection_iddisconnect_from_avatarawait session.close()并清空会话引用send_audio_to_avatarsession.send_audio(bytes(audio_data), endFalse)send_eof_to_avatarsession.send_audio(b, endTrue)以空包携带结束标记interrupt_avatarsession.interrupt()失败仅告警不抛出。扩展同时实现了get_vendor_name()返回spatius、get_reporting_session_id()返回agora_uid与get_vendor_metadata()脱敏后的厂商维度过滤空值。测试用例行为契约的验证仓库配套的测试tests/test_basic.py用 TEN 的AsyncExtensionTester验证了两条关键契约test_basic以单扩展模式启动spatius_avatar_python发送hello_world命令并断言返回StatusCode.OK验证扩展能正常加载启动test_config_error_uses_module_error在缺少配置时断言扩展发出errordata且载荷满足module avatar、code -1012、vendor spatius、vendor_code ValueError同时vendor_metadata不包含url与key等敏感字段——这正是_send_error标准化错误协议的回归保障。测试的 fixturetests/conftest.py通过一个后台线程运行FakeApp在on_init中释放事件锁配合pytest完成整个 TEN 运行时环境的搭建与销毁。扩展包的声明与依赖manifest.json 明确了扩展的类型extension、名称spatius_avatar_python、版本0.2.4依赖ten_runtime_python0.11与ten_ai_base0.7.40并声明了图内协议入向audio_frame_inpcm_frame、cmd_inflush、data_infinalize出向data_outerror。运行依赖见 requirements.txt 与 pyproject.tomlagora-token-builder1.0.0生成 Agora token、spatius[opus]1.0.4Spatius 数字人 SDKopus可选依赖用于 OggOpus 编码、protobuf5.28.3,6。开发新数字人扩展时把其中的spatius换成你自己的服务商 SDK 即可复用整套骨架。总结开发自己的数字人扩展要基于AsyncAvatarBaseExtension实现自己的数字人扩展按以下步骤即可✅ 创建包含服务参数的配置类继承BaseConfig可参考SpatiusConfig的params归一化模式✅ 继承AsyncAvatarBaseExtension✅ 实现 7 个必需方法validate_config、get_target_sample_rate、connect_to_avatar、disconnect_from_avatar、send_audio_to_avatar、send_eof_to_avatar、interrupt_avatar✅ 使用一致的日志前缀基类通过LOG_PREFIX统一日志格式子类可覆盖✅ 使用不同采样率进行测试验证采样率白名单行为✅ 优雅地处理错误连接失败要抛出以阻止启动断开失败要吞掉异常保证清理音频发送失败要记录并继续。基类会自动处理生命周期、音频队列、采样率校验、flush/finalize 消息与错误上报等所有样板逻辑你只需要专注对接数字人服务的业务代码。【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考