TEN Framework — 架构与原理

30 秒导读: TEN 是一个搭「实时多模态语音 agent」的开源框架。你想做一个能打电话进去、边说边被打断、还能查天气的语音助手——TEN 不让你手写一大坨 asyncio 胶水,而是让你把 ASR、LLM、TTS、RTC 当成一个个乐高积木(extension),在一份 JSON 里画一张 graph 把它们连起来。框架的核心价值有两层:底层一个 C/Rust 写的消息路由引擎负责按 graph 把消息投递到对的 extension;上层一个纯 Python 的编排 extension 用异步队列和可取消任务,实现「用户一开口就瞬间掐掉正在播的 AI 语音」这种真实语音产品最难做对的事。

1. 这是什么(零基础也能懂)

一句话定义: TEN 是一个实时对话 AI 的运行时 + 编排框架——把语音链路里的每个环节(听、想、说、传输)做成可热插拔的模块,用配置而不是代码把它们组装成一个能双向说话、能被打断的 agent。

解决什么问题 / 给谁用:

假设你要做一个电话客服机器人,要求是:用户说话时,机器人正在播的话要立刻停下来(专业叫 barge-in / 打断),而不是把整句蠢蠢地说完。你会发现难点根本不在「调 OpenAI」,而在:

音频是流式的,ASR 一边听一边吐字,LLM 一边想一边吐 token,TTS 一边合成一边出声——四个环节全在同时跑。
用户一开口,你得同时中止 LLM 生成、清空 TTS 队列、丢弃已经发到网络上的音频,一个都不能漏,还要快。
换一个 TTS 供应商(ElevenLabs → Deepgram)不该让你重写整条链路。

TEN 就是为这些痛点而生,面向做实时语音/多模态 agent 的开发者与研究者。

它能做什么:

用一份 property.json 里的 graph 声明,把 ASR/LLM/TTS/RTC/工具/头像等 90+ 现成 extension 连成一条流水线。
跨语言:运行时是 C + Rust,extension 可以用 Python / Go / Node.js / C++ 写,同一张 graph 里混用。
内置打断、工具调用(function calling)、流式转写等语音产品必备能力。
一个 Go HTTP 服务器按会话拉起 worker 进程,天然多路并发。

用起来什么样:

你几乎不写编排代码,只写一段 graph 声明(摘自 ai_agents/agents/examples/voice-assistant/tenapp/property.json:8-102),节点是模块、连线是消息流:

"nodes": [
  {"name": "stt", "addon": "deepgram_asr_python", "property": {...}},
  {"name": "llm", "addon": "openai_llm2_python",  "property": {...}},
  {"name": "tts", "addon": "elevenlabs_tts2_python", "property": {...}},
  {"name": "main_control", "addon": "main_python", "property": {...}}
],
"connections": [
  {"extension": "main_control",
   "data": [{"name": "asr_result", "source": [{"extension": "stt"}]}]}
]

想把 Deepgram TTS 换成 xAI?改一个 addon 字段就行——同一份文件里就并排放着 voice_assistant、voice_assistant_xai_tts、voice_assistant_deepgram_tts 好几个 graph(property.json:204-1197)。

一句话直觉/类比: 把它想成语音版的音频「模块化合成器」:每个 extension 是一个效果器盒子,graph 里的连线是插在盒子之间的跳线;框架本身是那块背板,负责让信号(消息)按跳线流到对的盒子。你搭机器,不焊电路。

2. 顶层全景(它大概怎么转)

TEN 有两个层次要分清:外层是「谁来跑这些 agent」(Go 服务器 + worker 进程),内层是「一个 agent 内部怎么流动」(graph + 消息引擎)。

怎么读下面这张图: 从上到下是「一次通话」的生命周期——浏览器发 /start,Go 拉起一个 worker 进程,worker 里由 C/Rust 引擎按 graph 路由消息,音频在 RTC ↔ STT ↔ 控制器 ↔ LLM ↔ TTS 之间转一圈。

主要部件一句话职责:

部件	干什么	在哪个文件
Go HTTP Server	接 `/start /stop /ping`,一会话拉起一个 worker 进程,并把 channel 等动态值注入 graph	`ai_agents/server/internal/http_server.go`
Worker	跑 `tman run start`,加载 graph、跑所有 extension 的实际 OS 进程	`ai_agents/server/internal/worker_linux.go:28`
消息引擎(Rust)	定义 graph 数据结构(nodes/connections)、校验、把「谁连谁」变成路由表	`core/src/ten_rust/src/graph/mod.rs`
消息引擎(C)	引擎主循环、extension 线程、消息(cmd/data/audio_frame)的实际投递	`core/src/ten_runtime/engine/engine.c`、`core/src/ten_runtime/extension_thread/`
`main_control`	编排大脑:消费 ASR 结果、驱动 LLM、切句喂 TTS、处理打断	`.../main_python/extension.py:28` (`MainControlExtension`)
`Agent`	编排大脑内部的事件循环:异步队列 + 可取消任务	`.../main_python/agent/agent.py:10` (`Agent`)
`LLMExec`	对 LLM 的流式收发、上下文管理、工具调用、abort	`.../main_python/agent/llm_exec.py:32` (`LLMExec`)

(上表 main_python 路径均在 ai_agents/agents/examples/voice-assistant/tenapp/ten_packages/extension/ 下。)

主线走一遍(高层,不进代码):

浏览器 POST /start,带上 channel 名和 graph 名 → handlerStart(http_server.go:236)拉起一个 worker。
worker 里 C/Rust 引擎读 graph,把 agora_rtc 收到的音频 pcm_frame 按连线送到 stt。
stt 流式吐出 asr_result(有 partial / final),经 graph 送到 main_control。
main_control 攒够一句就把文本 chat_completion 给 llm,LLM 流式吐 delta。
main_control 把 LLM 的 delta 按标点切成整句,逐句发 tts_text_input 给 tts。
tts 合成的 pcm_frame 送回 agora_rtc 播给用户。
只要用户又开口,main_control 立刻三路 flush,把 4-6 步正在进行的一切掐掉,回到第 3 步。

一个关键认知:框架和业务是分开的。 C/Rust 引擎只懂「按 graph 投递消息」,它完全不知道什么是 ASR / LLM;所有语音智能都在 extension 里、尤其在 main_control 这个纯 Python 编排 extension 里。这就是为什么换供应商只改 JSON——业务逻辑和运行时是解耦的。

3. 核心机制

下面钻三个子系统:graph 消息路由(框架的骨架)、语音编排循环与打断(价值最密的地方)、以及外层的 worker 进程模型。

3.1 Graph:用一份 JSON 声明消息怎么流

它要解决的小问题: 四个模块要互相收发消息,但你不想让 stt 里硬编码「把结果发给叫 main_control 的那个」——那样就没法复用、没法换。TEN 的答案是:连线不写在代码里,写在 graph 声明里,由框架在启动时读成一张路由表。

思路/直觉: 一张 graph = 节点(nodes)+ 连线(connections)。节点是 extension 实例,连线声明「某条消息从哪个 source 流到哪个 dest」。消息分四类,各有独立的连线数组。

TEN 把消息按类型分成四种,这是理解整个 graph 的基础:

消息类型	载荷	典型例子
`cmd`	具名命令,有返回值(request/response)	`chat_completion`、`flush`、`tool_call`、`on_user_joined`
`data`	具名数据,单向	`asr_result`、`tts_text_input`、`tts_audio_end`
`audio_frame`	PCM 音频流	`pcm_frame`
`video_frame`	视频流	头像 / 视觉场景

真实实现: graph 的数据模型在 Rust 里,一张图就是 nodes + connections 两个数组(core/src/ten_rust/src/graph/mod.rs:172-178,struct Graph)。每条连接 GraphConnection 恰好按四种消息类型各开一个可选数组(core/src/ten_rust/src/graph/connection/mod.rs:195-208):

pub struct GraphConnection {
    pub loc: GraphLoc,                              // 这条连接属于哪个 extension
    pub cmd: Option<Vec<GraphMessageFlow>>,         // 四种消息各一个流数组
    pub data: Option<Vec<GraphMessageFlow>>,
    pub audio_frame: Option<Vec<GraphMessageFlow>>,
    pub video_frame: Option<Vec<GraphMessageFlow>>,
}

每个 GraphMessageFlow 用 name/names(消息名)+ dest/source(目的地/来源)把「哪个消息流向谁」说清楚(connection/mod.rs:271-284)。启动时 validate_and_complete 会递归校验每条 flow、每个 dest(connection/mod.rs:223-264),把不合法的 graph 提前挡掉。

关键细节/坑:

节点定位是三选一。 GraphLoc 里 extension / subgraph / selector 互斥,get_node_type 靠哪个字段非空来判断类型(connection/mod.rs:73-83)。这让一个节点既能是普通 extension,也能是子图(复用一整块 graph)。
支持跨进程/跨机的多 app graph。 GraphLoc 有个可选 app 字段(URI),同一张 graph 的节点可以分布在不同 app 上;AppUriDeclarationState 枚举把「全不声明 / 全一致 / 混合」三态区分开(mod.rs:94-112),localhost 在多 app 模式下被禁止,避免别的 app 不知道往哪连(connection/mod.rs:369-388)。这是「单机 demo」到「分布式部署」不改业务代码的地基。

3.2 语音编排循环与「三路打断」(全项目最精华)

它要解决的小问题: 真实对话里,AI 话说到一半用户插嘴,AI 必须立刻闭嘴。但此刻有三处正在「往外吐东西」:LLM 在流式生成、TTS 在合成、已经生成的音频在 RTC 上往用户耳朵里灌。要干净地打断,三处都得同时叫停——漏一处,用户就会听到一句「幽灵尾音」。

思路/直觉: 编排大脑 main_control 内部用一个 Agent 对象跑事件循环。它把 ASR 和 LLM 事件分别放进两条 asyncio 队列,由两个消费者协程处理;关键是 LLM 事件的处理被包成一个可以随时 cancel() 的 task——打断 = 取消这个 task + 清空队列 + 向下游发 flush。

图示(打断时发生了什么):

原理演示(把核心想法演出来,示意,非源码):

# 演示:为什么 LLM 处理要包成可取消的 task
async def consume_llm(queue):
    while True:
        event = await queue.get()
        active = asyncio.create_task(dispatch(event))  # 单独 task
        try:
            await active
        except asyncio.CancelledError:
            pass  # 被打断时,这里安静地结束

async def interrupt(active, queue):
    active.cancel()                 # 掐掉正在跑的这一句
    while not queue.empty():        # 清掉排队等着说的后续句子
        queue.get_nowait()
# 重点看:打断不是「设个标志位等它自己停」,而是真的 cancel + 清队列,所以是即时的

真实实现: 打断的总入口是 _interrupt,一次干三件事(.../main_python/extension.py:203-213):

async def _interrupt(self):
    self.sentence_fragment = ""
    await self.agent.flush_llm()                                   # ① 停 LLM
    await _send_data(self.ten_env, "tts_flush", "tts", {...})      # ② 清 TTS
    await _send_cmd(self.ten_env, "flush", "agora_rtc")           # ③ 丢 RTC 音频

① 停 LLM 在 Agent.flush_llm(agent/agent.py:192-214):先 llm_exec.flush(),再把 _llm_queue 里排队的响应事件全 get_nowait() 清掉,最后 cancel() 正在跑的 _llm_active_task。而 LLMExec.flush(agent/llm_exec.py:66-79)会向 llm extension 发一个 abort cmd 带上 request_id,让底层真正停止那次流式请求。
② 清 TTS / ③ 丢 RTC 是发 tts_flush data 和 flush cmd 给下游 extension,由它们各自清掉自己的合成队列 / 网络缓冲。

触发打断的判据也很实用(extension.py:83-94,_on_asr_result):不是等 ASR 出 final 才打断——只要 partial 结果的文本长度 > 2 就先打断,抢那几百毫秒的响应速度;等 final 才真正把用户这句话 queue_llm_input 送进 LLM。

关键细节/坑:

ASR 队列和 LLM 队列分开,消费方式也不同。 ASR 消费者只是顺序 dispatch(agent.py:82-85);LLM 消费者却把每次 dispatch 包成独立 task 并保存句柄(agent.py:87-97),就是为了能在打断时精确 cancel 这一个。两条队列职责不同,不能混。
切句是打断能干净的前提。 LLM 的 delta 先经 parse_sentences 按中英文标点切成整句再喂 TTS(.../main_python/helper.py:18-31,extension.py:98-108),没切完的半句留在 sentence_fragment。打断时第一件事就是把 sentence_fragment 清空,避免残句被误发。
取消要能优雅落地。 LLMExec._process_input_queue 专门 catch asyncio.CancelledError,把当前已生成的文本作为 final 回吐一次再退出(llm_exec.py:113-118),不让上下文半途撕裂。

3.3 Worker 进程模型:一会话一进程

它要解决的小问题: 一台机器要同时服务很多路通话,彼此不能互相拖垮。Python 的 GIL、某个 extension 崩了不该带崩别人——最省心的隔离办法就是一路会话一个 OS 进程。

思路/直觉: Go 写的 HTTP 服务器是「调度层」,它自己不跑 agent,只负责按 /start 把 worker 进程拉起来、按 /stop 杀掉、按 /ping 续命。

真实实现:

handlerStart(ai_agents/server/internal/http_server.go:236-321)校验 channel 非空、未超并发上限、channel 未占用,然后 worker.start。
worker 用 exec.Command 直接跑框架命令,用分参数而非 shell 拼接来防注入(worker_linux.go:29-31):

// Use separate arguments to avoid shell injection
cmd := exec.Command("tman", "run", "start", "--", "--property", w.PropertyJsonFile)

动态属性注入是这层的巧思:服务器在写出 property.json 前,会把请求里的 channel 注入到任何声明了 channel 属性的节点(http_server.go:706-725),再把 user_uid/bot_uid 等按 startPropMap 注入(http_server.go:678-681),还能把 req.Properties[extensionName] 深度合并进指定节点(http_server.go:644-676)。所以新增一个带 channel 的头像 extension,不用改服务器代码就能拿到动态 channel。

关键细节/坑:

安全上有意收紧。 请求里带的 tenapp_dir 会被忽略,强制用启动时的目录,防路径穿越(http_server.go:288-293);channel 名要过 sanitizeChannelName 校验,防路径注入(http_server.go:809-813)。
针对特定供应商限流。 graph 名里含 gemini 的,单 graph 并发 worker 数被硬上限卡住(http_server.go:267-284)——一个很务实的「防某家 API 配额被打爆」补丁。

4. 巧妙之处与局限

值得偷师的设计:

业务逻辑 = 一个普通 extension。 最反直觉也最漂亮的一点:编排大脑 main_control 不是框架的一部分,而是和 ASR/TTS 平级的一个 Python extension,它靠 @register_addon_as_extension("main_python") 注册进来(.../main_python/addon.py:13-19)。这意味着「对话怎么编排」完全是用户代码,框架零耦合,你可以整个换掉。
打断做成「三路 flush + 可取消 task」而非标志位轮询。 很多语音 demo 用 self.interrupted = True 然后到处检查——延迟高、容易漏。TEN 直接 task.cancel() + 清队列 + 向下游发 flush,即时且覆盖全链路(extension.py:203-213)。这是做实时语音最该抄的一段。
消息四分类是恰到好处的抽象。 cmd(带返回值的请求)/data(单向数据)/audio_frame/video_frame 四类刚好覆盖对话 agent 的全部通信,又都在 graph 里用同一套 source/dest 声明(connection/mod.rs:195-208),既统一又表达力足够。
配置即产品线。 同一份 property.json 里并排放着六七个 graph 变体(不同 ASR/TTS 组合),换供应商 = 换 graph 名,不碰代码(property.json:204-1197)。

边界与局限(诚实说):

RTC 是一等公民,默认绑定 Agora。 示例 graph 里 agora_rtc 是音频出入口(property.json:11-24),做纯 WebSocket / 本地音频要换传输 extension,不是零成本。
上手门槛高。 运行时是 C + Rust + Go 多语言编译产物,靠 Docker 容器 ten_agent_dev 起,不是 pip install 就能跑;调试要理解 graph 声明 + tman 工具链。
示例代码里能看到未完成的路径。 例如 LLMExec._handle_llm_response 里 requery 分支整段被注释掉(llm_exec.py:257-276),说明工具调用的部分形态仍在演进——照抄前要读清楚。
本文引用的 ASR/LLM/TTS 基类不在本仓库。 AsyncTTS2BaseExtension 等基类来自 ten_ai_base 包,靠 tman 依赖装入,克隆里没有源码;要读基类实现需另取该包(inferred:仓库内 grep 不到 class AsyncTTS2BaseExtension 定义,只有大量 from ten_ai_base import ... 的引用)。

代码地图(导航索引)

关注点	文件	说明
Graph 数据模型	`core/src/ten_rust/src/graph/mod.rs`	`struct Graph`(nodes+connections)、`AppUriDeclarationState` 多 app 三态
连接与消息流	`core/src/ten_rust/src/graph/connection/mod.rs`	`GraphConnection`(cmd/data/audio/video 四类)、`GraphMessageFlow`、`GraphLoc`
引擎主循环(C)	`core/src/ten_runtime/engine/engine.c`	运行时引擎,消息进出口
Extension 线程与投递	`core/src/ten_runtime/extension_thread/`	每个 extension 的线程与消息路由实现
编排大脑	`ai_agents/agents/examples/voice-assistant/tenapp/ten_packages/extension/main_python/extension.py`	`MainControlExtension`、`_interrupt`、`_on_asr_result`、`_send_to_tts`
事件循环	`.../main_python/agent/agent.py`	`Agent`、`_consume_llm`、`flush_llm`、双队列
LLM 流式与工具调用	`.../main_python/agent/llm_exec.py`	`LLMExec`、`flush`(发 abort)、`_send_to_llm`、`_handle_llm_response`
切句 / 发消息辅助	`.../main_python/helper.py`	`parse_sentences`、`_send_cmd`/`_send_data`/`_send_cmd_ex`
Addon 注册	`.../main_python/addon.py`	`@register_addon_as_extension("main_python")`
Graph 声明(示例)	`.../voice-assistant/tenapp/property.json`	多个 `predefined_graphs`,nodes/connections 全景
HTTP 服务器	`ai_agents/server/internal/http_server.go`	`handlerStart`/`handlerStop`/`handlerPing`、`processProperty`、动态注入
Worker 进程	`ai_agents/server/internal/worker_linux.go`	`Worker.start`(`tman run start`)、防注入、pid 管理

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

3. 核心机制​

3.1 Graph:用一份 JSON 声明消息怎么流​

3.2 语音编排循环与「三路打断」(全项目最精华)​

3.3 Worker 进程模型:一会话一进程​

4. 巧妙之处与局限​

代码地图(导航索引)​

1. 这是什么(零基础也能懂)

2. 顶层全景(它大概怎么转)

3. 核心机制

3.1 Graph:用一份 JSON 声明消息怎么流

3.2 语音编排循环与「三路打断」(全项目最精华)

3.3 Worker 进程模型:一会话一进程

4. 巧妙之处与局限

代码地图(导航索引)