企业官网建设流程全解析

TEN Framework 语音助手中枢扩展 main_python：从 Twilio 通话到 ASR/LLM/TTS 的完整编排实践

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

导读

main_python是 TEN Framework 语音助手示例（voice-assistant-sip-twilio）中的核心控制扩展，承担 AI 智能体对话的"总指挥"角色：它接收 ASR 识别结果、驱动 LLM 生成回复、把文本交给 TTS 合成语音，并负责用户会话状态管理与跨组件数据路由。本文基于该扩展的源码与配置，完整讲解其事件驱动架构、输入输出接口、配置参数、Twilio 媒体流集成原理与音频采样率转换细节，读完即可理解并二次开发一套可运行的电话语音助手中枢逻辑。

扩展定位：为什么需要一个"主控制"扩展

在 TEN Framework 的图（graph）编排模型中，语音助手通常由 STT、LLM、TTS、RTC 等多个扩展协作完成。main_python不是这些能力的实现者，而是它们的协调者——正如其文档所述：它"充当 AI 智能体对话的编排器，处理实时语音处理、LLM 交互和 TTS 输出，并管理用户会话状态"。

从源码结构看（extension.py），MainControlExtension继承AsyncExtension，其核心职责可归纳为：

• 生命周期管理：在on_init中加载配置并启动内置的 Twilio 呼叫服务器，在on_stop中结束所有活动通话并清理资源；
• 事件路由：把框架下达的Cmd与Data转交给内部Agent事件循环处理；
• 音频转发：on_audio_frame将 TEN 框架产出的 PCM 音频下发到所有活跃 Twilio 通话的 WebSocket；
• 打断（Interrupt）：检测到用户新语音时，向 LLM、TTS、RTC 发送 flush 指令，实现"边说边打断"的自然对话体验。

在示例图的编排中（见 tenapp/property.json），main_control节点使用main_pythonaddon，与deepgram_asr_python、openai_llm2_python、elevenlabs_tts2_python共同组成voice_assistant图。

事件驱动内核：Agent 类与事件队列

main_python的内部编排基于一个轻量级事件系统，核心实现在 agent/agent.py 与 agent/events.py。

事件类型（AgentEvent）

事件基类AgentEventBase定义了type（cmd/data）与name，派生出五类事件：

双队列消费模型

Agent内部维护两条asyncio.Queue：

• _asr_queue：ASR 结果按序消费，由_consume_asr任务派发；
• _llm_queue：LLM 流式输出按序消费，_consume_llm会把当前事件包装为独立asyncio.Task，这样当用户中途说话触发打断时，可以取消正在执行的 LLM 处理任务（对应flush_llm中_llm_active_task.cancel()的逻辑），避免过时回复继续播放。

装饰器注册机制

扩展在on_init中扫描自身方法，凡带有_agent_event_type属性的方法都会通过self.agent.on(event_type, fn)自动注册为事件处理器（见 extension.py）。注册语法由 agent/decorators.py 提供，典型处理器包括：

• _on_asr_result：更新session_id、递增turn_id、必要时触发打断、把最终结果送入 LLM、并发送转写文本；
• _on_llm_response：将 LLM 增量文本按句子切分后逐句送入 TTS（_send_to_tts），同时以message/reasoning两种类型发送转写。

接口契约：输入数据、输出数据与命令

README 文档明确给出了该扩展面向框架的接口契约，以下是结合源码的完整解读。

输入数据

ASR Result（Data 名为asr_result，由Agent.on_data解析）：

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

final为false时表示中间识别结果（用于实时字幕），为true时表示最终结果（会递增turn_id并送入 LLM）。注意_on_asr_result中有一个细节：当event.final为真或文本长度大于 2 时都会触发_interrupt()，这是为了让用户一开口就立刻打断当前播放。

LLM Result（由LLMExec回调产生，type可为message或reasoning）：

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

在事件模型中对应LLMResponseEvent的delta（增量片段）、text（累计文本）、is_final（是否最终）与type字段。

输出数据

Text Data：扩展通过_send_transcript把转写文本发送给message_collector，字段如下：

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

实际发送的负载（见 extension.py）还包含data_type（transcribe或raw）、role（user/assistant）、text_ts（毫秒时间戳）等字段，reasoning 内容会以 JSON 字符串包装后作为raw类型发送。

此外扩展还会向 TTS 发送两类数据：

• tts_text_input（目标tts）：携带request_id、text、text_input_end（是否本段结束）与metadata；
• tts_flush（目标tts）：携带flush_id用于清空 TTS 播放队列。

命令（Commands）

• 输入命令：on_user_joined（用户加入会话）、on_user_left（用户离开会话），以及tool_register（注册 LLM 工具，见 agent/agent.py）；
• 输出命令：flush，由_interrupt发出，目标为agora_rtc（_send_cmd(self.ten_env, "flush", "agora_rtc")），用于清空 RTC 音频缓冲，实现"说话即打断"。

辅助函数_send_cmd、_send_data定义在 helper.py 中，它们通过Loc("", "", dest)直接指定图内目标扩展发送，省去了手动建立连接的开销（注释也提醒：这类写法仅适用于该图内部逻辑，通用扩展应避免）。

配置体系：从 property.json 到 Pydantic 模型

配置参数总览

README 中列出的配置只有greeting一项，但仓库实际配置远不止于此。扩展的配置由 Pydantic 模型MainControlConfig定义（见 config.py），在on_init中通过ten_env.get_property_to_json(None)读取运行时属性并校验。

配置文件 property.json 展示了实际用法——敏感信息全部通过${env:XXX}占位符从环境变量注入：

{ "greeting": "Hello, I am your AI assistant.", "twilio_account_sid": "${env:TWILIO_ACCOUNT_SID}", "twilio_auth_token": "${env:TWILIO_AUTH_TOKEN}", "twilio_from_number": "${env:TWILIO_FROM_NUMBER}", "twilio_server_port": 9000, "twilio_public_server_url": "${env:TWILIO_PUBLIC_SERVER_URL}", "twilio_use_https": false, "twilio_use_wss": false }

注意：示例的 property.json 中twilio_use_https/twilio_use_wss为false，这是配合 ngrok 本地开发的选择——ngrok 负责 SSL 终结，本地服务器保持 HTTP/WS；若直接部署在带证书的公网服务器上，则建议开启。twilio_public_server_url为空时媒体流与状态回调不会启用，因此它是媒体流能力的开关。

对应的运行时 schema 声明在 manifest.json 的api.property.properties中，类型包括 string/int32/bool，供 TEN 运行时做配置校验。

环境变量准备

在示例根目录的.env中配置（详见 voice-assistant-sip-twilio/README.md）：

# Twilio（必填） TWILIO_ACCOUNT_SID=your_twilio_account_sid_here TWILIO_AUTH_TOKEN=your_twilio_auth_token_here TWILIO_FROM_NUMBER=+1234567890 TWILIO_PUBLIC_SERVER_URL=https://your-domain.com # Deepgram（STT，必填） DEEPGRAM_API_KEY=your_deepgram_api_key_here # OpenAI（LLM，必填） OPENAI_API_KEY=your_openai_api_key_here OPENAI_MODEL=gpt-4 # ElevenLabs（TTS，必填） ELEVENLABS_TTS_KEY=your_elevenlabs_tts_key_here # 可选 WEATHERAPI_API_KEY=your_weather_api_key_here NGROK_AUTHTOKEN=your_ngrok_auth_token_here

对话工作流：从电话接通到语音回复

综合 README 的 Workflow 与源码实现，一条完整通话的处理链路如下：

1. 用户加入：Twilio 媒体流 WebSocket 建立后，服务器收到start事件，记录callSid与streamSid，随即调用on_websocket_connected——扩展会立即把配置的greeting文本送入 TTS（_send_to_tts(greeting_text, True)），实现"接听即问候"。
2. 语音上行：Twilio 把用户语音以 μ-law 编码的 base64 负载通过 WebSocket 的media事件推送到/media端点；TwilioCallServer解码后调用_forward_audio_to_ten（见 server.py 与 extension.py）：
   • base64.b64decode还原 μ-law 数据；
   • audioop.ulaw2lin(mulaw_data, 2)转换为 16-bit PCM；
   • 构造AudioFrame("pcm_frame")，标注 8000 Hz、单声道、2 字节/样本、INTERLEAVE 格式，stream_id固定为 54321，目的地指向streamid_adapter扩展，最终经ten_env.send_audio_frame送入 STT 处理。
3. ASR 与打断：STT 返回的asr_result数据触发_on_asr_result；一旦检测到用户说话（final 或文本超 2 字符），立即调用_interrupt()：清空句子缓存、flush_llm、向 TTS 发tts_flush、向 RTC 发flush，保证用户新输入优先。
4. LLM 处理：最终 ASR 文本经agent.queue_llm_input进入 LLM 上下文（LLMExec实现在 agent/llm_exec.py）。
5. TTS 合成与下行：LLM 流式增量经parse_sentences按中英文标点（,，.。?？!！）切句，完整句子即时送入 TTS 以降低首包延迟；is_final的剩余片段在结尾补发（text_input_end=True）。TTS 生成的 16 kHz PCM 音频帧回到on_audio_frame，经send_audio_to_twilio下发。
6. 音频下行：send_audio_to_twilio中完成"16 kHz → 8 kHz"降采样与 PCM→μ-law 编码，再以{"event": "media", "streamSid": ..., "media": {"payload": base64}}的格式写入通话对应的 WebSocket（见 extension.py）。
7. 通话结束：Twilio 推送stop事件或用户挂断后，扩展通过DELETE /api/call/{call_sid}结束通话，清理active_call_sessions与音频转储文件。

Twilio 媒体流集成：内置服务器详解

main_python的一大特点是将 Twilio 集成服务器内嵌到扩展进程中（_start_server在on_init中启动），由 server.py 中的TwilioCallServer基于 FastAPI + uvicorn 实现，同一端口同时提供 HTTP API 与 WebSocket 媒体流：

关键设计点：

• 媒体流地址构造：connect.stream(url=...)的地址由twilio_use_wss决定协议前缀，/media为路径；Webhook 地址由twilio_use_https决定，路径为/webhook/status。二者均基于twilio_public_server_url拼装。
• SSL 处理：start_server中即使配置了 HTTPS/WSS，本地仍以 HTTP 启动 uvicorn，注释明确说明"ngrok 将负责 SSL 终结"——这降低了本地开发门槛。
• 通话状态机：active_call_sessions以call_sid为键维护会话，包含phone_number、message、status、stream_sid、websocket等字段；WebSocketstart事件到达后才绑定流 SID 与 WebSocket 对象。
• 音频转储：扩展支持把通话音频以 PCM 文件转储到audio_dump_directory（默认/tmp/twilio_audio_dumps），便于调试，文件名为twilio_audio_{call_sid}_{timestamp}.pcm。

音频采样率与编码转换细节

Twilio 媒体流固定使用8 kHz、μ-law、单声道，而 TEN 图内 TTS（如示例中的 ElevenLabspcm_16000）输出16 kHz PCM，因此扩展承担了双向转换：

上行（Twilio → TEN）：μ-law → PCM（audioop.ulaw2lin），保持 8 kHz，交由 STT 处理（图中 Deepgram 节点sample_rate配置为 8000，见 tenapp/property.json）。

下行（TEN → Twilio）：先降采样再编码：

1. _downsample_audio处理 16 kHz → 8 kHz：对 16-bit 样本采用"每 2 个样本取 1 个"的简单抽取（decimation），其他采样率组合则回退到audioop.ratecv按最大公约数计算转换比；
2. audioop.lin2ulaw(downsampled, 2)把 PCM 编码为 μ-law；
3. base64.b64encode后装入media事件负载发送。

这一设计保证了电话侧的语音质量与延迟之间的平衡，也让扩展可以直接复用 TEN 生态中标准的 16 kHz TTS 输出。

安装、构建、测试与运行

通过 TEN 包管理器操作

README 给出的标准操作（扩展位于示例 tenapp 内，命令在对应 tenapp 目录下执行）：

# 安装扩展 ten install main_python # 构建扩展 ten build main_python # 运行测试 ten test main_python

运行完整示例

1. 安装依赖并启动（在示例根目录）：

cd ai_agents/agents/examples/voice-assistant-sip-twilio task install task run

2. 本地开发时用 ngrok 暴露端口（start-with-ngrok.sh会自动启动 ngrok 并把公网地址用于TWILIO_PUBLIC_SERVER_URL）：

./start-with-ngrok.sh

3. 访问入口：

   • 前端控制台：http://localhost:3000（支持外呼发起/接听管理，见 frontend）
   • 内置 API/WebSocket 服务器：http://localhost:9000
   • TMAN Designer（可视化改图）：http://localhost:49483

4. 外呼示例（调用扩展内置的 REST API）：

curl -X POST http://localhost:9000/api/call \ -H "Content-Type: application/json" \ -d '{ "phone_number": "+1234567890", "message": "Hello from AI assistant!" }'

依赖版本说明

README 中声明的依赖为ten_runtime_python0.10 与ten_ai_base0.6.9；而仓库当前 manifest.json 实际声明为ten_runtime_python0.11 与ten_ai_base0.7，具体以仓库当前版本为准。代码中大量使用ten_runtime的AsyncExtension、AsyncTenEnv、Cmd、Data、AudioFrame、Loc以及ten_ai_base.types.LLMToolMetadata类型，安装时需保证这两个系统包可达。

架构小结与扩展点

从实现看，main_python采用"事件驱动 + 双队列 + 内置服务器"的架构：Agent类作为纯事件总线（不依赖具体 STT/LLM/TTS 实现），MainControlExtension作为运行时驱动（对接 TEN 框架与 Twilio 媒体流），TwilioCallServer作为通信边界（HTTP API + WebSocket）。三者解耦清晰，使得：

• 替换 STT/LLM/TTS 提供商只需修改图配置（如通过 TMAN Designer 在 http://localhost:49483 调整节点属性），无需改动主控制逻辑；
• 新增会话事件只需定义新的AgentEvent子类并注册处理器；
• 接入其他电话通道（如 voice-assistant-sip-plivo、voice-assistant-sip-telnyx）可复用同样的主控制编排模式。

如果需要了解该示例的完整使用方式（环境变量、ngrok 配置、Docker 打包等），可继续阅读 voice-assistant-sip-twilio/README.md 与 server/README.md。

许可证

main_python扩展属于 TEN Framework 的一部分，遵循 Apache License 2.0（见 LICENSE），欢迎在遵循贡献规范的前提下参与改进。

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