企业官网建设流程全解析

TEN Framework 主控扩展 main_python 源码深度解析：AI Agent 会话编排的核心引擎

【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework

导读

main_python是 TEN Framework 中负责AI Agent 会话编排的核心 Python 扩展（Extension），它像一位"总指挥"，串联起语音识别（ASR）、大语言模型（LLM）、文本转语音（TTS）与实时通信（RTC）等组件，让一次"语音进、语音出"的完整 Agent 交互得以流畅运转。本文以该扩展的 README 为主线，结合其真实源码（事件总线、流式 LLM 执行器、会话状态管理）逐层拆解其数据流、事件模型与配置方式，读完你将掌握如何阅读、配置乃至二次开发一个 TEN 主控类扩展。

概述：什么是 main_python

在 TEN Framework 的 Agent 应用中，main_python承担的是**中央控制逻辑（central control logic）**角色。它以AsyncExtension的形式运行，负责：

• 处理实时语音识别结果（ASR），维护流式文本；
• 与语言模型（LLM）交互，协调自然语言理解与响应生成；
• 向 TTS 发送语音合成请求，完成音频输出；
• 跟踪用户会话状态，管理对话上下文；
• 处理中间结果（intermediate）与最终结果（final），保证交互流畅；
• 生成实时字幕（caption），兼顾无障碍与日志记录。

在 doodler 示例应用（面向儿童的语音绘图应用）中，main_python正是把agora_rtc（语音输入）、openai_asr_python（Whisper 转写）、openai_llm2_python（GPT-4o-mini 理解意图）、openai_gpt_image_python（绘图工具）与message_collector（聊天记录）串成完整工作流的"粘合剂"。

功能特性

按照 README，该扩展提供以下核心能力：

• 实时语音处理（Real-time Speech Processing）：接收并处理 ASR 结果，管理流式文本，区分中间结果与最终结果；
• LLM 集成（LLM Integration）：协调语言模型完成自然语言理解与响应生成；
• TTS 协调（TTS Coordination）：管理文本转语音请求，驱动音频输出；
• 会话管理（Session Management）：跟踪用户上下线（presence），维护会话与轮次状态；
• 流式支持（Streaming Support）：同时处理 final 与 intermediate 结果，保障体验平滑；
• 字幕生成（Caption Generation）：为 accessibility 与日志提供实时字幕。

API 接口：输入、输出与命令

输入数据（Input Data）

ASR 结果：语音识别模块通过名为asr_result的 Data 帧上报识别结果，其 JSON 结构为：

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

字段说明：text为识别出的文本；final标识是否为最终结果（false表示中间结果）；metadata.session_id用于标识来源会话。在 agent.py 的on_data中，扩展会读取该 Data 的名称，若为asr_result则解析 JSON 并封装为ASRResultEvent投入事件队列。

LLM 结果：语言模型以流式方式返回响应，其数据结构为：

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

输出数据（Output Data）

文本数据（Text Data）：主控扩展向字幕/消息收集器输出的结构化文本，包含：

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

其中stream_id由会话 ID 转换而来（源码中stream_id = int(self.session_id)），用于把多条字幕流关联到同一个会话。

命令（Commands）

输入命令：上游（如 RTC 扩展）可通过 Cmd 通知用户状态变化：

• on_user_joined：用户加入会话时触发；
• on_user_left：用户离开会话时触发。

此外，源码中还支持tool_register命令，用于注册 LLM 工具（详见下文"工具注册"小节）。

输出命令：主控扩展对外发送的典型命令为flush，用于向 LLM、TTS 和 RTC 组件下发冲刷/打断信号，终止正在进行的生成。

配置：greeting 参数

扩展支持通过运行时属性（property）进行配置，README 给出的配置示例为：

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

配置参数表：

需要说明的是：README 示例中的默认值为"Hello there, I'm TEN Agent"，而仓库内 config.py 中MainControlConfig的实际默认值是"Hello, I am your AI assistant."。greeting通过 Pydantic 的BaseModel定义，在 extension.py 的on_init中，通过ten_env.get_property_to_json(None)读取运行时属性并调用MainControlConfig.model_validate_json()完成校验与装载。同时在 manifest.json 的api.property.properties中声明了greeting的 JSON Schema 类型为string，供 TEN 框架校验。

源码验证：当第一位用户加入且配置了问候语时，_on_user_joined会同时向 TTS 发送问候语音（_send_to_tts）并向 message_collector 发送assistant角色的字幕（_send_transcript），实现"开口即欢迎"。

依赖

README 声明扩展依赖两个 TEN 系统包：

• ten_runtime_python：TEN Framework 核心运行时（README 标注 0.10，当前 manifest.json 实际声明为 0.11）；
• ten_ai_base：AI 基础能力库，提供LLMToolMetadata、LLMRequest/LLMResponse等结构（README 标注 0.6.9，manifest 实际声明为 0.7）。

依赖声明方式：在扩展的manifest.json的dependencies数组中以system类型声明，安装时由 TEN 包管理器自动解析。

安装与集成

安装

该扩展随 TEN Framework 分发布，可通过 TEN 包管理器安装：

ten install main_python

在 doodler 示例中，扩展位于 ten_packages/extension/main_python，其中addon.py通过@register_addon_as_extension("main_python")完成 Addon 注册，on_create_instance负责实例化MainControlExtension。

集成组件

该扩展被设计为与以下 TEN 组件协同工作：

• ASR 扩展：提供语音识别结果；
• LLM 扩展：处理自然语言并生成响应；
• TTS 扩展：将文本转为语音；
• RTC 扩展：处理实时通信（含用户上下线通知与打断）；
• Message Collector（消息收集器）：捕获并展示会话数据（字幕、推理过程等）。

在 doodler 的图配置（tenapp/property.json）中，这些组件通过有向连接与main_python组成 Agent 图。

工作流：一次完整对话的五个阶段

README 将一次典型交互拆解为 5 步：

1. 用户加入（User Joins）：用户上线触发on_user_joined，若配置了问候语则先送出欢迎语；
2. 语音处理（Speech Processing）：ASR 结果到达后被处理并生成字幕；
3. LLM 处理（LLM Processing）：最终语音段（final=true）被送入 LLM 处理；
4. 响应生成（Response Generation）：LLM 响应被转为语音并作为字幕展示；
5. 流式输出（Streaming）：中间与最终结果均被妥善处理，保证交互平滑。

源码级数据流佐证

结合 extension.py 的事件处理器，上述流程的底层实现如下：

• ASR 结果处理（_on_asr_result）：更新session_id与stream_id；当final=true或文本长度大于 2 时先调用_interrupt()打断旧输出；final=true时turn_id += 1并将文本通过agent.queue_llm_input()送入 LLM；随后无论中间还是最终结果，都会调用_send_transcript("user", ...)发送用户字幕。
• 打断机制（_interrupt）：清空句子碎片，调用agent.flush_llm()冲刷 LLM 队列并取消进行中的任务，同时向tts发送tts_flushData、向agora_rtc发送flushCmd——这是实现"用户一开口就立刻打断机器说话"的关键。
• LLM 响应处理（_on_llm_response）：中间message结果累积到sentence_fragment；最终结果时通过_select_tts_text提取首个完整句子（借助helper.parse_sentences按中英文标点切句，无有效内容时回退为默认提示语"Got it. Drawing now!"），并保证每轮只发送一次 TTS（通过_tts_sent_turn_id与turn_id比对）；推理过程（reasoning类型）则以data_type="raw"的 JSON 形式发送给 message_collector。
• 字幕发送（_send_transcript）：统一构造messageData 发往message_collector，携带data_type、role、text、text_ts、is_final、stream_id等字段。

架构：事件驱动的 Agent 内核

README 指出该扩展实现了AsyncExtension接口，提供生命周期管理、事件处理、状态管理与数据路由能力。从源码看，其内核是一个轻量级事件总线 + 双异步队列的 Agent 框架：

1. 事件模型（agent/events.py）

所有事件继承自 Pydantic 的AgentEventBase，分为 Cmd 类与 Data 类：

• UserJoinedEvent/UserLeftEvent（Cmd 类）：用户上下线；
• ToolRegisterEvent（Cmd 类）：携带LLMToolMetadata与来源扩展名，用于注册 LLM 工具；
• ASRResultEvent（Data 类）：携带text、final、metadata；
• LLMResponseEvent（Data 类）：携带delta、text、is_final，并用type区分普通消息（message）与推理过程（reasoning）。

统一类型AgentEvent是上述五者的 Union，作为事件分发与类型匹配的依据。

2. 事件注册与分发（agent/agent.py）

Agent类维护一个_callbacks注册表，支持agent.on(EventType, handler)与@agent.on(EventType)两种注册方式。extension.py在on_init中通过扫描实例方法上的_agent_event_handler装饰器元数据完成自动注册。_dispatch按事件类型匹配（isinstance）并串行执行所有处理器，单处理器异常仅记录日志而不中断整体分发。

3. 双队列消费者（ASR / LLM 分离）

Agent内置_asr_queue与_llm_queue两个asyncio.Queue，并各启动一个消费者任务。LLM 消费端特别设计为可取消：_consume_llm将分发逻辑包装为asyncio.create_task，以便flush_llm()在打断时能取消进行中的 LLM 处理任务并清空队列——这是实现低延迟打断的并发基础。

4. LLM 执行器（agent/llm_exec.py）

LLMExec是 LLM 交互的具体执行者：

• 输入队列：queue_input()将用户文本入队，_process_input_queue循环消费并构造LLMMessageContent(role="user", ...)；
• 流式请求：通过_send_cmd_ex向llm扩展发送chat_completionCmd，streaming=True，模型名传空字符串表示使用 LLM 扩展默认模型，temperature=0.7，并携带tools=self.available_tools；
• 响应分发：_handle_llm_response用match语句分派LLMResponseMessageDelta（增量文本）、LLMResponseMessageDone（结束）、LLMResponseReasoningDelta/Done（推理流）与LLMResponseToolCall（工具调用）；
• 上下文管理：_queue_context/_write_context维护contexts消息列表，连续同角色消息自动合并；
• 工具调用：LLMResponseToolCall触发时，根据tool_registry找到注册该工具的扩展，向其发送tool_callCmd，成功后把function_call与function_call_output写回上下文并继续请求 LLM，形成"工具增强"闭环。

5. 图内通信工具（helper.py）

_send_cmd/_send_data是图内便捷通信函数：通过Loc("", "", dest)指定目标扩展名，即可在无需显式连线的情况下向图内其他扩展发送 Cmd 或 Data——这正体现了主控扩展"假设目标扩展已存在于图中"的编排式设计（源码注释明确提示：此类写法只适用于特定图，通用扩展应避免使用）。

开发：构建与测试

构建

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

ten build main_python

测试

运行扩展测试：

ten test main_python

从 manifest.json 的package.include可见，发布包会包含manifest.json、property.json、**.tent、**.py、README.md与tests/**，测试目录被显式纳入打包范围。

生命周期与状态管理小结

扩展完整实现了AsyncExtension生命周期钩子：

• on_init：加载配置（MainControlConfig）、创建Agent、自动注册事件处理器；
• on_start：记录启动日志；
• on_stop：置stopped标志、调用agent.stop()（内部依次停止 LLMExec、冲刷 LLM 队列并取消 ASR/LLM 消费者任务）；
• on_cmd/on_data：把框架层的 Cmd/Data 透传给Agent，由Agent转换为领域事件。

状态管理方面，扩展用_rtc_user_count跟踪在线用户数（首个用户上线才触发问候）、turn_id标记对话轮次、session_id标识会话、_tts_sent_turn_id防止重复播报——这些字段共同构成了多用户场景下会话编排的状态基础。

许可证

该扩展属于 TEN Framework 的一部分，遵循 Apache License 2.0 开源协议发布；如需贡献，请参阅 TEN Framework 主仓库的贡献指南（AGENTS.md 与 CLAUDE.md）。

【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址: https://gitcode.com/TEN-framework/ten-framework