TEN Framework 实时语音助手 main_python 扩展：会话编排中枢的架构与实现解析-育师

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

【免费下载链接】ten-framework

Open-source framework for conversational voice AI agents

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

导读

main_python是 TEN Framework 实时语音助手（voice-assistant-realtime）示例中的核心控制扩展，位于 ai_agents/agents/examples/voice-assistant-realtime/tenapp/ten_packages/extension/main_python 目录，承担着 AI 智能体对话的"总编排"角色：它接收 ASR 语音识别结果、驱动 LLM 语义理解与回复生成、协调 TTS 语音输出，并统一管理会话状态与数据路由。阅读本文后，你将掌握该扩展的 API 接口设计、配置参数、事件驱动架构、与 RTC/LLM/TTS 等组件的协作方式，并能够基于源码理解其底层实现原理，为二次开发实时语音 Agent 提供可直接参考的实战范式。

扩展定位：实时语音 Agent 的控制中枢

在 TEN Framework 中，扩展（extension）是构成图（graph）的最小功能单元，通过命令（Cmd）与数据（Data）进行通信。main_python扩展实现的是AsyncExtension异步接口，其职责定位在 README 中有明确描述：作为 AI Agent 交互的中央控制逻辑，管理语音识别（ASR）、语言模型处理（LLM）与文本转语音（TTS）之间的协同。

从源码结构看，该扩展由以下模块组成：

文件	职责
addon.py	通过`@register_addon_as_extension("main_python")`注册扩展实例
extension.py	`MainControlExtension`主类，实现生命周期与事件消费循环
agent/agent.py	`Agent`类，将底层 Cmd/Data 转换为语义化 AgentEvent 并维护工具注册表
agent/events.py	定义全部 AgentEvent 事件模型
config.py	`MainControlConfig`Pydantic 配置模型
helper.py	`_send_cmd`/`_send_data`等图内消息发送工具

值得强调的是，addon.py 中通过装饰器注册后，on_create_instance在运行时创建MainControlExtension(name)实例，这是 TEN Framework 扩展加载的标准入口模式。

功能特性总览

README 将扩展能力归纳为六个方面，这些能力在源码中均有对应实现：

实时语音处理：接收 ASR 结果并管理流式文本（对应InputTranscriptEvent处理）；
LLM 集成：协调语音到语音（voice-to-voice）模型完成语义理解与回复生成（对应_send_message_item、_send_create_response）；
TTS 协调：管理文本转语音请求，输出音频；
会话管理：跟踪用户在线状态与对话状态（对应_rtc_user_count计数与session_ready标志）；
流式支持：同时处理最终结果与中间结果（final/is_final字段贯穿始终）；
字幕生成：为无障碍与日志提供实时字幕（对应_send_transcript转发到message_collector）。

事件驱动的双层架构设计

从源码结构看，该扩展采用"底层消息 → 语义事件 → 业务分发"的双层架构，这是它区别于简单扩展的核心设计亮点。

第一层：Cmd/Data 到 AgentEvent 的转换（agent.py）

agent/agent.py 中的Agent类负责第一层转换：

on_cmd：将on_user_joined、on_user_left、tool_register等命令转换为对应事件并放入asyncio.Queue事件队列；
on_data：将来自 MLLM 服务端的数据（如DATA_MLLM_OUT_REQUEST_TRANSCRIPT输入转写、DATA_MLLM_OUT_RESPONSE_TRANSCRIPT输出转写、DATA_MLLM_OUT_SESSION_READY会话就绪、DATA_MLLM_OUT_INTERRUPTED服务中断、DATA_MLLM_OUT_FUNCTION_CALL函数调用）解析为语义事件。

这些事件全部继承自 agent/events.py 中定义的AgentEventBase（含type与name字段），并汇总为AgentEventUnion 类型，共覆盖八种事件：UserJoinedEvent、UserLeftEvent、ToolRegisterEvent、SessionReadyEvent、ServerInterruptEvent、InputTranscriptEvent、OutputTranscriptEvent、FunctionCallEvent。

第二层：事件消费循环（extension.py）

extension.py 中的_consume_agent_events是核心事件循环，通过await self.agent.get_event()持续从队列取事件，并使用 Python 3.10+ 的match模式匹配分发：

UserJoinedEvent：用户数加一，触发_greeting_if_ready()欢迎逻辑；
UserLeftEvent：用户数减一；
ToolRegisterEvent：调用agent.register_tool将工具元数据转发给 MLLM；
FunctionCallEvent：调用agent.call_tool执行工具调用；
InputTranscriptEvent：提取session_id元数据并转发用户转写；
OutputTranscriptEvent：转发助手回复转写；
ServerInterruptEvent：触发打断逻辑_interrupt()；
SessionReadyEvent：置位session_ready并尝试发送欢迎语。

这种双层设计让"框架层的 Cmd/Data"与"业务层的语义事件"解耦，扩展主类只关注事件语义，逻辑清晰且易于扩展新事件类型。

API 接口规范

输入数据

ASR 结果（输入转写）

扩展从 MLLM 服务端接收的输入转写数据格式如下（对应 agent/agent.py 中MLLMServerInputTranscript的解析）：

{ "text": "string", "final": "bool", "metadata": { "session_id": "string" } }

final字段区分中间结果与最终结果：为false时表示流式过程中的增量转写，为true时表示一段语音的最终识别结果。metadata.session_id用于多会话场景下区分不同用户或通道。

LLM 结果（输出转写）

{ "text": "string", "end_of_segment": "bool" }

end_of_segment标识一段回复是否结束。在源码中，InputTranscriptEvent 与 OutputTranscriptEvent 分别额外携带delta（增量片段）、content（累计内容）与metadata字段，其中OutputTranscriptEvent还提供is_final标记流式回复是否终结。

输出数据

文本数据（转发给 message_collector）

{ "text": "string", "is_final": "bool", "end_of_segment": "bool", "stream_id": "uint32" }

该结构由 _send_transcript 方法组装并发送。注意源码中实际发送的载荷还包含data_type: "transcribe"、role（user/assistant）、text_ts（毫秒级时间戳）等字段，stream_id默认取session_id的整数值（用户转写）或固定值100（助手回复），用于流式字幕的排序与去抖。

命令接口

输入命令

命令	触发时机	对应事件
`on_user_joined`	用户加入会话	`UserJoinedEvent`
`on_user_left`	用户离开会话	`UserLeftEvent`
`tool_register`	其他扩展注册工具	`ToolRegisterEvent`

输出命令

flush：发送刷新命令到 LLM、TTS 与 RTC 组件，用于打断当前生成。源码 _interrupt 中向agora_rtc发送flush命令实现打断。

配置详解

扩展的配置在 manifest.json 的api.property.properties中声明为字符串类型，实际解析由 config.py 中的 Pydantic 模型完成：

{ "greeting": "Hello there, I'm TEN Agent" }

配置参数说明

参数	类型	默认值	说明
`greeting`	string	`"Hello, I am your AI assistant."`（源码 config.py 中的默认值）	首个用户加入时发送的欢迎语

需要说明两点与 README 的差异：

默认值差异：README 示例写为"Hello there, I'm TEN Agent"，而源码 config.py 中 Pydantic 模型的默认值实为"Hello, I am your AI assistant."，实际生效值以运行时属性为准；
欢迎机制：欢迎语并非直接显示，而是通过 _greeting_if_ready 构造一条"say {greeting} to me"的用户消息注入 LLM，并触发_send_create_response()让模型以语音方式说出欢迎语——即"让模型把欢迎语说给用户听"。

配置加载发生在on_init生命周期中（extension.py）：通过ten_env.get_property_to_json(None)读取运行时属性，再用MainControlConfig.model_validate_json校验解析。由于扩展目录下 property.json 内容为空对象{}，实际配置值来自应用图的属性注入（见下文 graph 配置）。

依赖与运行环境

manifest.json 声明了两项系统级依赖：

ten_runtime_python（版本 0.11）：TEN Framework 的 Python 运行时，提供AsyncExtension、AsyncTenEnv、Cmd、Data等核心 API（README 中标注为 0.10，仓库实际 manifest 已升级为 0.11）；
ten_ai_base（版本 0.7）：AI 基础能力库，提供 MLLM 客户端/服务端数据结构（MLLMClientMessageItem、MLLMServerInputTranscript等）与消息常量（DATA_MLLM_IN_*/DATA_MLLM_OUT_*）。

打包配置中package.include包含manifest.json、property.json、**.tent、**.py、README.md与tests/**，意味着该扩展以标准 TEN 包形式分发。

使用指南

安装

扩展是 TEN Framework 的一部分，可通过 TEN 包管理器安装：

ten install main_python

集成组件

main_python设计为与实时语音助手图中的其他组件协同工作。从示例应用 tenapp/manifest.json 的依赖列表可以看到完整组件阵容：

ASR/语音输入：通过agora_rtc（版本0.23.9-t1）实时采集音频；
LLM（voice-to-voice）：openai_mllm_python、azure_mllm_python、gemini_mllm_python、glm_mllm_python、stepfun_mllm_python等多供应商 MLLM 扩展，支持 OpenAI GPT Realtime、Azure Voice AI、Gemini 2.0 Flash、GLM、StepFun 等模型；
TTS：由 voice-to-voice 模型直接输出语音，无需独立 TTS 链路；
RTC：agora_rtc承担实时通信，streamid_adapter负责流 ID 适配；
消息收集器：message_collector2接收转写字幕数据；
工具扩展：weatherapi_tool_python提供天气查询工具，演示工具调用链路。

工作流

结合源码，完整对话流程如下：

用户加入：RTC 通道建立后发送on_user_joined命令，扩展计数用户数，并在session_ready就绪时通过 LLM 触发欢迎语；
语音处理：ASR 结果经 MLLM 服务端以输入转写数据下发，扩展生成字幕并转发给message_collector；
LLM 处理：最终语音片段（final=true）作为完整消息送入 LLM，模型以语音对语音方式直接生成回复；
回复生成：LLM 输出转写流式返回，扩展实时转发为字幕，同时音频经 RTC 推流给用户；
流式交互：中间结果（final=false）与最终结果（final=true）分开处理，保证低延迟体验；检测到用户再次说话时，ServerInterruptEvent触发flush打断当前生成。

工具调用

main_python还内置了完整的工具调用（Function Calling）链路：

工具扩展通过tool_register命令注册工具，Agent.register_tool将其记录在tool_registry并转发给 MLLM；
当 MLLM 发起DATA_MLLM_OUT_FUNCTION_CALL时，Agent.call_tool根据工具名从注册表找到来源扩展，向其发送tool_call命令；
工具执行结果若为llmresult类型，则通过DATA_MLLM_IN_FUNCTION_CALL_OUTPUT回传给 MLLM 继续生成。

构建与测试

扩展使用 TEN Framework 标准构建系统：

ten build main_python

运行扩展测试：

ten test main_python

在示例应用层面，tenapp/manifest.json 的 scripts 中，build对应scripts/install_python_deps.sh（安装 Python 依赖），start对应scripts/start.sh。

架构要点总结

从源码抽象出的架构特征可归纳为四点：

生命周期管理：实现on_init（加载配置、创建 Agent、启动事件循环任务）、on_start（预留初始上下文注入点）、on_stop（置位停止标志并调用agent.stop()清空事件队列）；
异步事件处理：基于asyncio.Queue的事件队列与match模式匹配分发，命令与数据处理均为异步；
状态管理：_rtc_user_count追踪用户数、session_ready标记会话就绪、current_metadata维护当前会话元数据；
数据路由：通过 _send_cmd / _send_data 在应用图内定向投递消息，目标地址由Loc("", "", dest)指定扩展名完成解耦通信。

此外，helper.py 还提供了parse_sentences中英文标点断句工具（支持中英文逗号、句号、问号、感叹号），可用于按句子粒度切分流式文本，这在字幕渲染与分段 TTS 场景中非常实用。

在示例应用中的实际配置

main_python的实例化与配置发生在应用级 tenapp/property.json 的predefined_graphs中。以voice_assistant_realtime图为例，agora_rtc节点配置 Agora 凭据与频道参数（subscribe_audio、publish_audio、publish_data均开启），v2v节点选用openai_mllm_python并配置model: "gpt-realtime"、voice: "alloy"、language: "en"、vad_type: "semantic_vad"、vad_eagerness: "auto"等语义级 VAD 参数，实现端到端的极低延迟语音对话。greeting等扩展属性同样可在此处按需注入，覆盖 config.py 中的默认值。

运行该示例前需准备 Agora 凭据（AGORA_APP_ID必填）与至少一个语音模型提供商的 API Key（OpenAI/Azure/Gemini/GLM/StepFun 任选其一），可参考示例目录下 README.md 的环境变量清单与task install/task run启动流程。