• 统一管道

统一管道

一轮对话的整条链：opus → vad → asr → turn → ling → tts。节点是纯流变换，观察者由引擎在节点边界自动广播；Round 消费内层广播，Session 只订阅 Round 的生命周期事件。 这条链为什么长成这样、替代了哪些形态，见管道演进记录。

设计约束

• 每个 Round 一条链：opus→vad→asr→turn→ling→tts 首尾相接，TurnComplete 在链内进入 ling→tts。
• 所有节点同构：统一为单一 Node::stream(upstream, ctx) 流变换，没有能力子 trait、没有 InputChain，ling/tts 不享有特权。
• 观察者内建抽象层：NodeContext 携带观察者注册点（Round 注入 EventSink）与 session_id（供节点日志关联 / 状态归属）；业务节点完全无感，广播由引擎自动处理。
• 引擎取引用 + 池化：ASR 共享引擎（create_stream() 天然可并发）；VAD 是串行状态机，用对象池复用（取用前 clear()）。
• 上一轮 Round 默认继续跑：仅当新回合需占用输出通道 / 打断正在说的话 / 抢 websocket 时才中断它（epoch 防串扰）。

角色与归属

Session(总调度) —— 管理 Round 生命周期；转发 Frame::Voice → active Round 链首；注册为 Round 的外层观察者（RoundEvent）
   Frame::Voice ──> PipelineEvent::AudioFrame ──> Round 链首
                        ↓
   Round(唯一处理器) 拥有单条链(内部实现):
     opus → vad → asr → turn ──(TurnComplete 链内自动进)──> ling → tts
       引擎在节点边界自动广播(event 带时机标签 Before/After)
            │ 内层：Round 作为广播观察者订阅，统一消费（barge-in 检测/TurnComplete/AudioOut 逐句转发/TTS 状态机/tail Err）
            ▼
        Round → RoundEvent 通道（SpeechStarted / TurnComplete / EmptyTurn / SpokenEnd）
            │ 外层：Session 注册为 Round 的观察者，据此做 shadow→running 升级 / 相位切换 / epoch / ListenStop
            ▼
        OutputMessage ──> 客户端

单链 + 观察者协议

文件：service/src/pipeline/mod.rs

pub enum PipelineEvent {
    // 感知
    AudioFrame(Vec<u8>),                   // Opus 原始音频，源自 Frame::Voice
    PcmFrame(Vec<f32>, u32),               // Opus 解码后的样本 + 采样率
    SpeechStarted,                         // VAD 上升沿
    SpeechEnded,                           // VAD 下降沿
    PartialTranscript(String),
    TurnText { text: String, prob: f32 },  // 关卡：一轮识别完成（ASR 产出）
    TurnComplete { text: String, prob: f32 }, // 回合边界收尾（turn 节点把 TurnText 标为回合结束，边界判定在 AsrNode + Session）
    Configure(AudioParam),                 // 下行音频参数：经流内事件触发重配
    FinishTurn,                            // 内部控制：请求 ASR 立即 finish
    // 表达
    TextChunk { text: String, emotion: Option<String> },
    AudioOut { audio: Vec<Vec<u8>>, is_first: bool, is_last: bool },
}

pub struct NodeContext {
    pub cancel: CancellationToken,
    pub emit: EventSink,          // Round 注入的观察者广播发送端
    pub session_id: String,       // 会话标识：Round 注入，供节点日志关联/状态归属
}

// 统一节点协议 —— 业务节点是无感的纯变换：从上游流读取关心变体，透传其余，可叠加新事件。
// **原型（Prototype）**：Node 既是模板又是实例，自带 `new_instance` 以自身为配方克隆新实例；
// Session 持一组模板，每 Round 让每个模板 `new_instance` 出独立实例（Opus/VAD 等有状态节点须新鲜实体）。
/// 会话级能力值：任何 `Send + Sync + 'static` 的具体类型（`NodeCapability = dyn Any + Send + Sync`）
/// 包装为 `Box<NodeCapability>` 即可被节点声明、被 Session 以 `downcast_ref::<T>()` 按类型 look up。
pub type NodeCapability = dyn Any + Send + Sync;

pub enum ReleaseMode {
    /// 节点进程完成（其流消费到 `None`）即释放：适合无跨事件状态的节点。
    Immediate,
    /// 整轮 round 结束才释放：适合贯穿一轮识别/表达的有状态节点。
    Deferred,
}

pub trait Node: Send + Sync {
    /// 克隆：以自身为模板产出一个新的、可跑在链里的实例（供每个 Round 独立使用）。
    fn new_instance(&self) -> Arc<dyn Node>;
    fn stream(&self, upstream: EventStream, ctx: &NodeContext) -> EventStream;
    /// 下行配置（模板级）：Session 在 hello 时把传输 `Configure` 转发给模板一次，
    /// 更新模板自身状态；后续每轮 `new_instance` 从中继承。默认空实现 = 不接受配置。
    fn on_configure(&self, _event: &PipelineEvent) {}
    /// 泛化能力 look up：节点声明的会话级能力集合（空 = 无）。Session 按类型 downcast 检索；
    /// 新增能力只需 `Box::new(具体类型)` 并入 `capabilities()`，不改 Node trait。
    fn capabilities(&self) -> Vec<Box<NodeCapability>> { Vec::new() }
    /// 释放策略（框架强制）：默认 `Deferred`；Immediate 节点在其流消费到 `None` 时被框架立即 `on_release`。
    fn release_mode(&self) -> ReleaseMode { ReleaseMode::Deferred }
    /// 轮次生命周期：round 开始时被 `NodeChain::begin` 驱动（默认 no-op）。
    fn on_acquire(&self) {}
    /// 轮次生命周期：Immediate 节点在 `with_observer` 流末触发，Deferred 节点在 `NodeChain::finish` 触发。
    fn on_release(&self) {}
}

观察者/广播（时机标签）——附在广播载荷，不是 NodeContext 字段：

pub enum TapPoint { Before, After }        // 进入节点输入前 / 节点产出后
pub struct Tapped { pub point: TapPoint, pub event: PipelineEvent }

• EventSink：Round 注入的可 clone 广播发送端（broadcast::Sender<Tapped> 或双端 unbounded）。
• 广播由引擎自动做：在 compose/NodeChain 驱动层，于每个节点边界把 Tapped{Before, ev} 与 Tapped{After, ev} 都发到 emit。业务节点不感知观察者存在—— 它只写纯 stream 变换，不调 ctx.emit。

语义：

• 单一 Node trait：能力一律经 capabilities() 申报，不用能力子 trait 区分节点。reset 不是公共协议—— VAD per-Round（池取用前 clear()）；reconfigure 分两层：Session 在 hello 时把流内 Configure 事件转发给模板（Node::on_configure），模板 new_instance 把配置态克隆给每轮实例。
• NodeContext 承载 cancel + emit（观察者入口）+ session_id（Round 注入，供节点日志关联 / 状态归属）； 不含任何「时机」配置——Before/After 是 广播载荷的标签，由引擎贴，不与上下文耦合。
• 错误复用 AppError，经 TryStreamExt 自动短路透传 Err；不塞 PipelineEvent::Error。

两个时机的用途（互补）：

• Before（进入前）：Round 感知原始输入事件——SpeechStarted（barge-in 打断）、音频到达 （重置空闲计时）。这类信号在链内会被下游变换吞噬，只有 Before 广播能让 Round 及时拿到。
• After（产出后）：Round 感知变换结果——TurnComplete（推进说段）、AudioOut（逐句转发）、 Err（超时/中断）。

下行配置（模板级，Configure → Node::on_configure）

语义：Configure(AudioParam) 是传输层（客户端 hello audio_params）下发的运行时参数，当前只被 OpusDecodeNode 消费（解码采样率）。它不逐轮发送，而是——Session 在 handle_connect 时把该事件 转发给模板一次（templates.iter().for_each(|t| t.on_configure(&Configure(params)))）；OpusDecodeNode 覆写 on_configure 更新模板态，且 new_instance 从 self 继承已配置采样率 → 每轮新实例解码器/重采样器 天然正确。首轮因 on_idle 先建 round 后 connect，仍补一次流内 Configure；无持久化字段、无每轮重喂。

采样率归一（opus → 16k）：OpusDecodeNode 负责 AudioFrame→PcmFrame，解码后把 PCM 重采样到 16000 再输出（PcmFrame(_, 16000) 恒 16k）。因为下游 VAD（Earshot）与 ASR（sherpa X-ASR）都硬编码 假设 16k 输入、链上无其它重采样点；客户端按 hello 声明异于 16k 的上行采样率（如 24k）时，统一在此降 16k， VAD/ASR 总拿到正确采样率。复用 rubato（api 已有依赖，matcha 同款 Fft::new + 分块处理模式）。

能力 look up（Node → Session 上行声明，泛化）

与下行配置（Configure(AudioParam)，Session→Node 下发运行时参数）正交，节点还提供上行能力声明。 能力泛化：不穷举为 Node trait 方法，而是以类型别名 NodeCapability = dyn Any + Send + Sync 包装 任意 Send + Sync + 'static 的具体类型即可作为能力被节点声明、被 Session 按类型 downcast_ref::<T>() look up。当前唯一能力成员是 AudioSpec（下行音频格式）：

#[derive(Debug, Clone)]
pub struct AudioSpec { pub sample_rate: u32, pub channel: u32, pub frame_duration_ms: u64 }

// Node trait 默认：capabilities() -> Vec<Box<NodeCapability>> { Vec::new() }
impl Node for TtsNode {
    fn capabilities(&self) -> Vec<Box<NodeCapability>> {
        vec![Box::new(self.tts.audio_spec())]   // 任意具体类型直接 Box 包装
    }
}

• Tts::audio_spec() -> AudioSpec（service component/tts.rs）：引擎自报下行输出格式。TtsMatcha 返回自身 output_sample_rate/output_channel/output_frame_duration；TtsMute（无音频、仅透传 文本）返回默认 (16000, 1, 60)。
• Session 解析：build() 时一次性遍历 node_templates，在每个模板的 capabilities() 里 downcast_ref::<AudioSpec>()，find_map 首个命中，存为 Option<AudioSpec> 字段。
• 语义（不兜底）：无 TTS 节点 ⇒ None——握手回包 audio_params: None（显式"无下行语音能力"， 客户端不再期待音频），pacer 亦不构造（AudioResult 只来自 TTS，天然不触发）。
• 扩展性：新增能力（如上行格式、会话 token 预算）只需 Box::new(具体类型) 并入 capabilities()，不改 Node trait。
• 配置源只有一处：api config::audio::AudioConfig（流向 TtsManager/TtsMatcha）；session 侧不复制音频配置。

节点生命周期（框架强制的资源释放）

有状态节点（VAD / ASR）在一轮识别内持有资源（VAD 状态机取自注入的 pool、ASR 流贯穿喂帧→finish），需要 生命周期钩子统一驱动。释放策略由节点 release_mode() 声明，框架强制：

• Immediate：with_observer 在每个叶子流的 with_observer 包裹层检测到该流消费到 None（进程完成） 时，若该叶子 release_mode()==Immediate 即立即调 leaf.on_release()。因 compose_chain 入口对每个 叶子恰好 with_observer 包一次，触发每叶恰好一次，嵌套 compose 不会重复。
• Deferred：资源贯穿整轮，由 NodeChain 在 round 结束时统一释放。

触发载体——NodeChain 持叶子列表：compose_chain 把传入的有序 Vec<Arc<dyn Node>> 折叠成嵌套 组合，叶子引用被吞没；为驱动生命周期，NodeChain::new(head, leaves) 保留叶子副本：

// session 构链时
let leaves: Vec<Arc<dyn Node>> = templates.iter().map(|t| t.new_instance()).collect();
let head = compose_chain(leaves.clone()).expect("chain");
let chain = NodeChain::new(head, leaves);
chain.begin();   // 对所有叶子逐一 on_acquire()
// round 结束
chain.finish();  // 仅对 Deferred 叶子逐一 on_release()（Immediate 已在流末释放，不重复）

单链组合

compose(a, b) 把 a.stream 输出接成 b.stream 输入，返回新 Node（纯 pipe，不广播）。 统一拼链函数 compose_chain(Vec<Arc<dyn Node>>) 逐叶子用 with_observer 包一次（自动广播）后再 reduce(compose) 折叠成单条链——顺序与数量由外部传入的 Vec 决定：

pub fn compose_chain(nodes: Vec<Arc<dyn Node>>) -> Option<Arc<dyn Node>> {
    nodes.into_iter().map(with_observer).reduce(compose)
}

产出的链形：opus → vad → asr → turn → ling → tts。 组装者只给裸原型 Vec<Arc<dyn Node>>；with_observer 对组装者完全透明，由 compose_chain 内部统一处理（保证每个叶子恰好广播一次）。

链构建（模板 + 每轮实例）

• 只有三层：模板集合 → 每轮 new_instance → compose_chain。没有 PipelineFactory / NodeFactory / NodeDeps / ChainBuilder 这类中间层。
• Node = 原型：每个节点自带 new_instance(&self) -> Arc<dyn Node>，以自身为模板克隆实例。 模板自持所需共享引用（Arc<VadPool> / Arc<dyn Asr> / Arc<dyn Ling> / Arc<dyn Tts>）， new_instance 时克隆出每 Round 独立的新实体。节点不自存 per-session session_id——会话标识由 Round 注入 NodeContext.session_id，节点仅从 ctx 读取用于日志关联。
• Session 持裸原型集合 Vec<Arc<dyn Node>>（SessionBuilder::with_node_templates），每 Round： templates.iter().map(|t| t.new_instance()) → compose_chain(...) 产链。
• Session 构建时从模板 look up 会话级能力：find_map downcast AudioSpec（见「能力 look up」）， 驱动握手音频声明与 pacer；无 TTS 节点则不声明下行音频（audio_params: None）。
• 调用方（api 站点 ws / matrix / 测试） 按配置条件组装裸原型 Vec<Arc<dyn Node>>：条件 push / 换序即可决定链上有哪些阶段、什么顺序。
• 每 Round 一条链 = 每次 new_instance 从 VAD 池 acquire 一个新 VAD 实例。

节点归属

所有节点为纯变换：不感知观察者/广播；TtsNode 以 FIFO 配平 emotion，编码错误以 Err 透传。

引擎池化

• VAD 对象池：VadPool { free: Mutex<Vec<Box<dyn Vad>>>, config }。acquire() 空则建、 否则取并 clear()；release() 归还。保证同时仅被一个 Round 持有（串行状态机，并发复用破 is_speech）。
• ASR 共享引擎：Arc<dyn Asr> 单例；每 Round 现取 create_stream()，finish() 后终结，不值得池化。

Round 统一消费（两层观察者）

Round 是「内层封装 pipeline 广播 + 外层可观察主体」的双层观察者模型。

• 内层（Round ← pipeline 广播）：Round 拥有单条 NodeChain；start() 时把自己注入为 EventSink 观察者，订阅 Tapped 广播流统一消费单链。按 Tapped 的 point + event 分支：
  • Before SpeechStarted → 判定 barge-in（含 lockout，已过锁定期后）→ 通知 Session 决策
  • After TurnComplete → 发 STT + 通知 Session 升级
  • After AudioOut → 逐句 SentenceStart/Audio/SentenceEnd 转发 + TTS 状态机 + 超时
  • tail Err → LlmNoUsableOutput / TtsEncode / 中断
  • 其他（PcmFrame/PartialTranscript）→ 前向 / 日志
• 外层（Session ← Round 可观察主体）：Round 本身实现观察者模式，暴露 RoundEvent 广播通道； Session 注册订阅（每 Round 唯一订阅者）。Round 判定信号并内联转发输出（STT/LLM/TTS/Audio → output_tx）+ 维护 TTS 状态机；Session 只做生命周期决策（shadow→running 升级 / 相位切换 / epoch / 打断 stop_round / ListenStop）。

文件：service/src/session/round.rs

pub enum RoundEvent {
    SpeechStarted,                             // 语音起始（barge-in 判定已过 lockout）
    TurnComplete { text: String, prob: f32 },  // 一轮识别完成（应升级 shadow→running / Speaking / ListenStop）
    EmptyTurn,                                 // 空输入完成（同 TurnComplete 轮转出新的 shadow，但不产 STT）
    SpokenEnd,                                 // TTS 表达结束（该 round 完成）
}

pub struct Round {
    chain: NodeChain,                                  // 单条链
    round_event_tx: broadcast::Sender<RoundEvent>,     // 外层观察者通道（Round 广播）
    round_event_rx: broadcast::Receiver<RoundEvent>,   // Session 注册订阅
    // ...
}

• Round::event_receiver() → 返回 round_event_rx，Session 注册。
• Round 的 start() 起观察者 task：select! 于 { tail.next()（驱动 + Err） / tap_rx.recv()（控制） / cancel }， 统一消费并负责逐句输出转发。

回合联动

• 回合边界判定在 AsrNode 的静音确认（SILENCE_CONFIRM_MS=200，VAD 报非语音后触发 finish）与 Session 控制事件（FinishTurn：ListenStop / 静音超时 / transport stall）完成；TurnNode 只做 TurnText→TurnComplete 显式收尾标记（D2）。
• 规则收尾（默认）：AsrNode 按静音确认阈值（200ms）收尾一段完整意图——VAD 报非语音后 silence_samples ≥ SILENCE_CONFIRM_MS·sample_rate/1000 即 finish；另有静音超时、transport stall、 FinishTurn 控制事件、prefix。
• 上一轮 Round 的处置：默认继续跑；仅当新回合需占用输出通道 / 打断说话 / 抢 websocket 时才中断—— 复用 RoundStopReason::Upgrade/BargeIn + epoch 防陈旧输出。

Session 总调度（Round 生命周期管理）

• phase 状态机保留：Idle / Listening / Speaking。
• on_listening(Frame::Voice) → 转发 AudioFrame 到 active Round 链首（D1）。
• Session 注册为 current Round 的外层观察者：在 RoundEvent 通道上 select!， 收到信号做生命周期决策：
  • SpeechStarted → barge-in：next_round_epoch() + stop_round(BargeIn)（该打断由当前 shadow 继续听， 其 TurnComplete 再升级，不再 new_shadow_round）
  • TurnComplete{text, prob} → on_turn_complete：shadow→running 升级、相位 Speaking、ListenStop， 并立即新建下一轮 shadow（new_shadow_round）以支持说话中 / 连续输入监听
  • SpokenEnd → 表达结束（shadow 相位处理 / 空闲计时）
• 取输入帧前先消化已入队的 RoundEvent：生命周期事件走 broadcast、该轮输出帧走输出通道， 两条通道各自排队。不先消化，barge-in 可能在 TurnComplete 之前被处理，看到尚未升级的 shadow （running_round == None）就放弃打断，整段已排队 TTS 继续播。
• 节奏门控的是取帧而不是发送：frame_duration_ms 只决定「下一帧何时允许从输出队列取出」， 任意时刻最多一帧在途。先取后等会把整段 TTS 预取进转发循环，让每帧都与输入帧争用 select， barge-in 因此可能被推迟 N 帧。
• Session 不持有 drain_hearing / handle_hearing_event / active_hearing：响应逻辑在 Round 观察者里。
• with_node_templates(Vec<Arc<dyn Node>>) 注入裸原型集合（api 站点动态组，决定链的阶段/顺序）。
• 握手（handle_connect）回包 audio_params 由构建时 look up 的 AudioSpec 决定；无 TTS 节点则下发 audio_params: None（无下行语音能力，不兜底默认）。pacer 节奏取 frame_duration_ms（无 spec 不构造）。
• epoch 防串扰保留；输出转发也算活动（重置 idle_since）。

空输入决策（中枢 gatekeeper）

空输入的判别表、计数与提示语分级归对话流程；这里只留机制：会话进入 Listening 时 Session 依「模式 + 前一 turn」定出 EmptyKind，观察 (After, EmptyInput) / (After, EmptyTurn) 后产出 Prompt{kind, count}（重新注入链首，各节点透传未识别事件）、Silence（不注入表达）或 GiveUp（回 Idle）。Ling 只按 Act 组 prompt，Echo 按 kind/count 取固定句——判别与渲染不交叉。

测试

cargo test -p service --lib pipeline
cargo test -p api --test pipeline_nodes_test
cargo test -p api --test session

相关文件

硬约束

• VAD 池回收：VAD 是串行状态机，acquire/release 严格配对，同一实例同时只能被一个 Round 持有。
• 回合判定归属：边界由 AsrNode 的静音确认与 Session 控制事件（FinishTurn / 静音超时 / transport stall）决定，TurnNode 只标 TurnComplete。
• 广播顺序：Before/After 标签只由引擎在节点边界贴；节点自己不调 ctx.emit，否则观察者会收到错序或重复信号。
• epoch 与终止：epoch 防陈旧输出，不随链结构调整。
• emotion：TTS 的 emotion 靠 FIFO 配平，不做文本键映射。