Pipecat 传输层实验：传输层通，帧路由未通

新增实验代码
- services/bot_server.py：Pipecat SmallWebRTC 服务端，
  把 ConversationEngine 包成一个 FrameProcessor（检验 A 方案的核心假设）
- client/test_client.py：自写 aiortc 客户端——Pipecat 的客户端 SDK
  覆盖浏览器/移动端/C++/ESP32，没有适配"Linux x86 盒子接 USB 阵列"这种形态

实验结果
- ✅ 传输层跑通：信令（47–280ms 往返）、ICE 连通、data channel、
  双向媒体轨（客户端音轨送达服务端，服务端音轨客户端收到）
- ❌ 帧路由未通：自定义处理器的 process_frame 调用次数为 0，
  连自己注入的 StartFrame 都没进去，引擎从未运行
- 现象：连接全对、音轨协商成功、助手永远不说话，日志里一条错误都没有

对架构决策的影响
A 方案（Pipecat 只做传输、逻辑归自有引擎）的核心假设不是开箱即用的。
B 方案按框架原生 service 重构，绕开了这个接缝。
根因未定位：接线错误（接缝浅，A 可行）vs 模型不兼容（接缝深，应转 B）。

顺带挖出 8 个 Pipecat 集成坑，每个都表现为"什么都没发生"，已记入 README：
客户端需发 data channel ping、offer 无媒体则不收集 ICE 候选、
run() 要 WorkerParams、TaskManager 需在事件循环内建、
后台任务异常被静默吞掉等。

其他
- bot_server.py 保留 [诊断] 前缀日志，供下一轮排查，定位后应清理
- 客户端默认地址改用 tailscale 域名（局域网已于当日重新编址，
  从 192.168.0.x 变为 172.16.1.x）
- 新增依赖 aiortc（含 av 17.x，注意 18.x 与 aiortc 1.15 不兼容）

README.md

@@ -31,7 +31,11 @@ app/                    助手本体（部署到服务器）
 └── tests/              4 个测试，共 88 项断言，全部无需硬件
 
 services/
-└── tts_server.py       Qwen3-TTS 服务（跑在 ~/tts-qwen，独立 venv）
+├── tts_server.py       Qwen3-TTS 服务（跑在 ~/tts-qwen，独立 venv）
+└── bot_server.py       Pipecat SmallWebRTC 服务端（实验，见下）
+
+client/
+└── test_client.py      自写 aiortc 客户端（实验用，也是 N5095 客户端的雏形）
 
 p0/                     P0 阶段的验证与基准脚本（保留用于复现）
 ├── verify_deepseek.py  DeepSeek 四条路径
@@ -76,6 +80,40 @@ cd ~/ai-companion
 **引擎与传输解耦。** 不依赖 Pipecat、不依赖任何传输方式，只依赖四个接口。
 好处：可以用文件音频端到端测试；接 WebRTC 时只需写薄适配层。
 
+## 当前卡点：Pipecat 传了但没接上
+
+方案文档写"Pipecat 作为唯一会话管线框架"，但**代码里它只在 requirements.txt 里**。
+
+2026-09-15 做了传输层实验（`services/bot_server.py` + `client/test_client.py`）：
+
+- ✅ **传输层跑通**：信令、ICE、data channel、双向媒体轨
+- ❌ **帧路由未通**：自定义处理器的 `process_frame` 调用次数为 0，
+  连自己注入的 StartFrame 都没进去；引擎从未运行
+
+现象是"连接全对、音轨协商成功、但助手永远不说话，且日志里一条错误都没有"。
+
+**这直接影响 A/B 架构方案**：A 是"Pipecat 只做传输、逻辑归自有引擎"，
+B 是"按 Pipecat 的 service 模型重构"。帧送不进自定义处理器，
+说明 A 的核心假设不是开箱即用的。
+
+根因未定位，两种可能必须区分：接线错误（接缝浅，A 可行）
+vs 模型不兼容（接缝深，应转 B）。
+
+完整记录见代码仓库外的 `P0-Pipecat传输层实验报告.md`（在 Obsidian 笔记库里）。
+
+### 实验挖出的 8 个坑
+
+每个都表现为"什么都没发生"，极难排查：
+
+1. Pipecat 客户端 SDK 不覆盖"Linux x86 盒子接 USB 阵列"这种形态 → 客户端得自己写
+2. 客户端必须建 data channel 并定期发 `"ping"`，否则服务端一帧音频都不推
+3. offer 里没有媒体时 aiortc **完全不收集 ICE 候选**（连连通性测试都要加静音轨）
+4. ICE 收集完成前发 offer → SDP 里 0 个候选地址
+5. `PipelineTask.run()` 要 `WorkerParams(task_manager=...)`，不是 `PipelineParams`
+6. `TaskManager()` 必须在事件循环内创建
+7. **后台任务里的异常被静默吞掉**，日志里查不到
+8. 信令回调里必须 `create_task`，不能 `await task.run()`
+
 ## 踩过的坑
 
 全部写在对应文件的注释里，这里只列索引，详见 `部署说明.md` §已知的坑：

client/test_client.py

@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""最小 aiortc 客户端（实验用）。
+
+**为什么需要它**：Pipecat 的官方客户端 SDK 覆盖浏览器 / 移动端 / C++ / ESP32，
+唯独没有适配"跑 Linux 的 x86 盒子接 USB 麦克风阵列"——也就是 N5095 的形态。
+所以客户端必须自己写，这个文件是雏形。
+
+它同时是验证工具：不依赖浏览器就能测服务端。用法：
+
+    # 发一段 wav，收回应答音频
+    python client/test_client.py --server http://<服务器>:7860 \
+        --wav p0/stt_tts_bench/wavs/stt_1.wav --out /tmp/reply.wav
+
+    # 只做连接测试，不发音频
+    python client/test_client.py --server http://<服务器>:7860 --connect-only
+
+**它还不是什么**：不做 AEC、不做播客参考、不做打断（那些要真实声卡和硬件），
+只用文件音频验证信令与媒体通路。
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import fractions
+import json
+import logging
+import sys
+import time
+import urllib.request
+import wave
+from pathlib import Path
+
+import numpy as np
+from aiortc import (
+    RTCConfiguration,
+    RTCPeerConnection,
+    RTCSessionDescription,
+)
+from aiortc.mediastreams import AudioStreamTrack, MediaStreamError
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("client")
+
+
+class SilentTrack(AudioStreamTrack):
+    """静音音轨。只为触发 ICE 候选收集，不发真实音频。"""
+
+    kind = "audio"
+
+    def __init__(self, sample_rate: int = 16000):
+        super().__init__()
+        self.sample_rate = sample_rate
+        self._n = 0
+
+    async def recv(self):
+        import fractions as _fr
+        from av import AudioFrame as _AF
+        await asyncio.sleep(0.02)
+        pcm = np.zeros(self.sample_rate // 50, dtype=np.int16)
+        f = _AF(format="s16", layout="mono", samples=len(pcm))
+        f.planes[0].update(pcm.tobytes())
+        f.sample_rate = self.sample_rate
+        f.pts = self._n
+        f.time_base = _fr.Fraction(1, self.sample_rate)
+        self._n += len(pcm)
+        return f
+
+
+class FileAudioTrack(AudioStreamTrack):
+    """把 wav 文件当作麦克风音源，按实时速率吐出来。
+
+    实时速率很重要——如果一次性灌完，服务端的 VAD 会把它当成
+    一段超长的连续语音，测出来的延迟也不真实。
+    """
+
+    kind = "audio"
+
+    def __init__(self, wav_path: Path, *, sample_rate: int = 16000):
+        super().__init__()
+        with wave.open(str(wav_path)) as w:
+            assert w.getnchannels() == 1, "需要单声道"
+            self.src_rate = w.getframerate()
+            pcm = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16)
+        self.samples = pcm.astype(np.float32) / 32768.0
+        self.sample_rate = sample_rate
+        if self.src_rate != sample_rate:
+            n = int(len(self.samples) * sample_rate / self.src_rate)
+            self.samples = np.interp(np.linspace(0, len(self.samples) - 1, n),
+                                     np.arange(len(self.samples)), self.samples)
+        self._pos = 0
+        self._t0: float | None = None
+        self._frame_samples = sample_rate // 100     # 10ms
+
+    async def recv(self):
+        """按 10ms 一块、以真实时间节奏产出。"""
+        if self._t0 is None:
+            self._t0 = time.perf_counter()
+            self._pos = 0
+
+        # 音频放完后持续吐静音，让服务端的 VAD 能判到"说完"
+        if self._pos >= len(self.samples):
+            audio = np.zeros(self._frame_samples, dtype=np.float32)
+        else:
+            audio = self.samples[self._pos:self._pos + self._frame_samples]
+            if len(audio) < self._frame_samples:
+                audio = np.pad(audio, (0, self._frame_samples - len(audio)))
+            self._pos += self._frame_samples
+
+        # 实时节奏
+        target = self._t0 + self._pos / self.sample_rate
+        delay = target - time.perf_counter()
+        if delay > 0:
+            await asyncio.sleep(delay)
+
+        pcm = (np.clip(audio, -1, 1) * 32767).astype(np.int16)
+        from av import AudioFrame
+        f = AudioFrame(format="s16", layout="mono", samples=len(pcm))
+        f.planes[0].update(pcm.tobytes())
+        f.sample_rate = self.sample_rate
+        f.pts = int(self._pos * 90000 / self.sample_rate)
+        f.time_base = fractions.Fraction(1, 90000)
+        return f
+
+    @property
+    def finished(self) -> bool:
+        return self._pos >= len(self.samples)
+
+
+class ReplyRecorder:
+    """收服务端音频，写 wav，并记录首音延迟。"""
+
+    def __init__(self, out_path: Path, sample_rate: int = 24000):
+        self.out_path = out_path
+        self.sample_rate = sample_rate
+        self.chunks: list[np.ndarray] = []
+        self.first_at: float | None = None
+        self.t0 = time.perf_counter()
+
+    def feed(self, frame) -> None:
+        if self.first_at is None:
+            self.first_at = time.perf_counter()
+        arr = frame.to_ndarray()
+        # 多声道取第一路
+        if arr.shape[0] > 1:
+            arr = arr[0:1]
+        self.chunks.append(arr.astype(np.float32).reshape(-1) / 32768.0)
+
+    def save(self) -> float:
+        if not self.chunks:
+            return 0.0
+        audio = np.concatenate(self.chunks)
+        pcm = (np.clip(audio, -1, 1) * 32767).astype(np.int16)
+        self.out_path.parent.mkdir(parents=True, exist_ok=True)
+        with wave.open(str(self.out_path), "wb") as w:
+            w.setnchannels(1)
+            w.setsampwidth(2)
+            w.setframerate(self.sample_rate)
+            w.writeframes(pcm.tobytes())
+        return len(audio) / self.sample_rate
+
+    @property
+    def first_audio_ms(self) -> float | None:
+        return None if self.first_at is None else (self.first_at - self.t0) * 1000
+
+
+def post_offer(server: str, sdp: str, type_: str) -> dict:
+    """把 offer 发给服务端，取回 answer。"""
+    payload = json.dumps({"sdp": sdp, "type": type_, "pc_id": None,
+                          "restart_pc": False, "request_data": None}).encode()
+    req = urllib.request.Request(f"{server.rstrip('/')}/api/offer", data=payload,
+                                 headers={"Content-Type": "application/json"},
+                                 method="POST")
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return json.loads(r.read().decode())
+
+
+async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool,
+              timeout_s: float) -> int:
+    pc = RTCPeerConnection(RTCConfiguration(iceServers=[]))
+    recorder = ReplyRecorder(out)
+    track = None
+    got_audio = asyncio.Event()
+
+    @pc.on("connectionstatechange")
+    async def _state():
+        log.info("连接状态: %s", pc.connectionState)
+        if pc.connectionState in ("failed", "closed"):
+            got_audio.set()
+
+    @pc.on("track")
+    def _track(t):
+        log.info("收到服务端音轨: %s", t.kind)
+
+        async def _pump():
+            try:
+                while True:
+                    frame = await t.recv()
+                    recorder.feed(frame)
+                    if not got_audio.is_set() and recorder.chunks:
+                        pass
+            except (MediaStreamError, asyncio.CancelledError):
+                pass
+            finally:
+                got_audio.set()
+
+        asyncio.ensure_future(_pump())
+
+    if wav and not connect_only:
+        track = FileAudioTrack(wav)
+        pc.addTrack(track)
+        log.info("音源: %s（%d 采样点）", wav.name, len(track.samples))
+    else:
+        # --connect-only 也要加一条（静音）音轨。
+        # **不能省**：aiortc 的 offer 里没有媒体时完全不收集 ICE 候选，
+        # 现象是"信令成功、SDP 交换正常、但 ICE 永远停在 new"。
+        pc.addTrack(SilentTrack())
+        log.info("连接测试模式：已加静音音轨（仅为触发 ICE 收集）")
+
+    # **必须建 data channel 并定期发 "ping"。**
+    # Pipecat 的 SmallWebRTC 用 data channel 上的消息来判断"客户端还活着"：
+    #   if isinstance(message, str) and message.startswith("ping"):
+    #       self._last_received_time = time.time()
+    # 而 connection 的 connected 事件、进而 on_client_connected 回调、
+    # 进而 StartFrame（音频推流的开关）都依赖它。
+    # 不发 ping 的现象是：ICE 连通、音轨协商成功、但**服务端一帧音频都不推**、
+    # 助手永远不说话，而且日志里只有一句 "Data channel not established within 10s"。
+    channel = pc.createDataChannel("pipecat")
+    dc_state = {"ok": False}
+
+    @channel.on("open")
+    def _dc_open():
+        dc_state["ok"] = True
+        log.info("data channel 已打开，开始发心跳")
+
+    async def _keepalive():
+        while pc.connectionState not in ("closed", "failed"):
+            if dc_state["ok"]:
+                try:
+                    channel.send("ping")
+                except Exception:  # noqa: BLE001
+                    pass
+            await asyncio.sleep(1.0)
+
+    asyncio.ensure_future(_keepalive())
+
+    offer = await pc.createOffer()
+    await pc.setLocalDescription(offer)
+
+    # 等 ICE 收集完成再发 offer——**这步不能省**。
+    # 不等的话 SDP 里没有任何候选地址，服务端拿到的是一份"空地址"的 offer，
+    # 信令看着成功，但 ICE 永远连不上（状态停在 new）。
+    # 这叫非 trickle ICE：把候选都塞进 SDP 一次发完。
+    # 真实产品会用 trickle（边收集边发候选），但那样需要额外的信令端点。
+    for _ in range(100):
+        if pc.iceGatheringState == "complete":
+            break
+        await asyncio.sleep(0.1)
+    n_cand = pc.localDescription.sdp.count("candidate:")
+    log.info("ICE 收集: %s，SDP 含 %d 个候选", pc.iceGatheringState, n_cand)
+    if n_cand == 0:
+        log.warning("SDP 里没有候选地址，ICE 很可能连不上")
+
+    log.info("本地 SDP 已生成，发给服务端…")
+
+    t_send = time.perf_counter()
+    answer = post_offer(server, pc.localDescription.sdp, pc.localDescription.type)
+    if answer.get("type") not in ("answer", None):
+        log.error("服务端返回异常: %s", answer)
+        return 1
+    if answer.get("sdp"):
+        await pc.setRemoteDescription(RTCSessionDescription(
+            sdp=answer["sdp"], type=answer["type"]))
+        log.info("远端 SDP 已设置（往返 %.0f ms）", (time.perf_counter() - t_send) * 1000)
+    else:
+        log.info("服务端未返回 SDP（可能是 pc_id 复用）: %s", answer)
+
+    # 等 ICE 连通
+    for _ in range(100):
+        if pc.connectionState in ("connected", "completed"):
+            break
+        if pc.connectionState in ("failed", "closed"):
+            log.error("连接失败")
+            await pc.close()
+            return 1
+        await asyncio.sleep(0.1)
+    log.info("✅ ICE 已连通（%s）", pc.connectionState)
+    print(f"  ✅ 连接建立，状态 {pc.connectionState}")
+
+    if track:
+        # 等音频放完 + 留出响应时间
+        t0 = time.perf_counter()
+        while not track.finished and time.perf_counter() - t0 < timeout_s:
+            await asyncio.sleep(0.2)
+        log.info("音频已发完，等待回复…")
+        # 等首个回复音频
+        t_wait = time.perf_counter()
+        while time.perf_counter() - t_wait < timeout_s:
+            if recorder.chunks:
+                break
+            await asyncio.sleep(0.1)
+        # 再收一会儿，让回复说完
+        await asyncio.sleep(6)
+
+    dur = recorder.save()
+    await pc.close()
+    got_audio.set()
+
+    print()
+    print("=== 结果 ===")
+    print(f"  连接: {'✅ 成功' if dur > 0 else '⚠️ 无音频返回'}")
+    if dur:
+        print(f"  收到音频: {dur:.2f} 秒 → {out}")
+        fa = recorder.first_audio_ms
+        if fa:
+            print(f"  首音延迟（发完到收到首段）: {fa:.0f} ms")
+    return 0 if dur > 0 else 1
+
+
+def main() -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--server", default="http://cadadmin-server.taild0f91d.ts.net:7860",
+                   help="服务端地址。默认用 tailscale 域名——LAN 地址会被重新编址"
+                        "（2026-09-15 实测从 192.168.0.x 变成 172.16.1.x）")
+    p.add_argument("--wav", type=Path)
+    p.add_argument("--out", type=Path, default=Path("/tmp/client_reply.wav"))
+    p.add_argument("--connect-only", action="store_true")
+    p.add_argument("--timeout", type=float, default=30.0)
+    a = p.parse_args()
+    if not a.connect_only and not a.wav:
+        print("需要 --wav 或 --connect-only", file=sys.stderr)
+        return 2
+    return asyncio.run(run(a.server, a.wav, a.out,
+                           connect_only=a.connect_only, timeout_s=a.timeout))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

services/bot_server.py

@@ -0,0 +1,422 @@
+#!/usr/bin/env python3
+"""Pipecat SmallWebRTC 服务端（实验）。
+
+**目的**：验证 Pipecat 传输层与自有引擎能否协同工作，并拿到真实传输下的延迟数字。
+不是产品代码——见文件末尾"这还不是什么"。
+
+结构：
+
+    ┌──────────────── Pipecat ────────────────┐
+    │ transport.input() → EngineProcessor     │
+    │                      ↓                  │
+    │                   transport.output()    │
+    └─────────────────────────────────────────┘
+                          ↑
+                  我们的 ConversationEngine
+                  （STT→记忆→LLM→TTS 全在里面）
+
+EngineProcessor 把整条管线"吞"进一个处理器——这是 A 方案的核心假设：
+**Pipecat 提供传输与帧模型，会话语义归我们的引擎。** 这个实验就是在检验它成不成立。
+
+用法：
+    # 服务器上
+    .venv/bin/python services/bot_server.py --host 0.0.0.0 --port 7860
+    # 浏览器打开 http://<服务器IP>:7860/  点连接后说话
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+import sys
+import time
+import uuid
+from pathlib import Path
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from fastapi import FastAPI
+from fastapi.responses import RedirectResponse
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    InputAudioRawFrame,
+    InterruptionFrame,
+    OutputAudioRawFrame,
+    StartFrame,
+    TextFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.smallwebrtc.request_handler import (
+    SmallWebRTCRequest,
+    SmallWebRTCRequestHandler,
+)
+from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport
+from pipecat.utils.asyncio.task_manager import TaskManager
+from pipecat.workers.base_worker import WorkerParams
+
+from app.config import AppConfig
+from app.engine import ConversationEngine, EventKind
+from app.identity import ManualIdentityProvider
+from app.llm import FallbackProvider, create_provider
+from app.memory import SQLiteMemoryStore
+from app.voice import StreamingSTT, create_tts
+
+logging.basicConfig(level=logging.INFO,
+                    format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("bot")
+
+#: 输入音频格式（WebRTC 侧协商后的格式）
+IN_RATE = 16000
+IN_CHANNELS = 1
+#: 输出音频格式（Qwen3-TTS 输出 24kHz）
+OUT_RATE = 24000
+
+#: 管线参数。给 PipelineTask 构造用。
+PIPELINE_PARAMS = PipelineParams(
+    audio_in_sample_rate=IN_RATE,
+    audio_out_sample_rate=OUT_RATE,
+    enable_metrics=True,
+)
+
+#: 注意：PipelineTask.run() 要的是 WorkerParams（内含 task_manager），
+#: **不是** PipelineParams。传错会依次报两个错，且都发生在后台任务里，
+#: 主日志里什么都看不到：
+#:   1. 不传 → "PipelineWorker.run() missing 1 required positional argument: 'params'"
+#:   2. 传错 → "'PipelineParams' object has no attribute 'task_manager'"
+#: 而且 TaskManager() 内部调用 asyncio.get_running_loop()，
+#: **不能在模块级创建**（会报 "no running event loop"），必须在异步上下文里建。
+#: 参考实现：pipecat/workers/runner.py
+
+
+# ---------------------------------------------------------------- 能量 VAD
+
+class SimpleVAD:
+    """极简能量判定。
+
+    **这是实验用的占位实现**，不是最终方案。方案 §5 把"说完判定"列为需要
+    实测的部分，正式的 VAD 还没选（候选：Pipecat 的 Silero、
+    sherpa-onnx VAD、或客户端 VAD + 服务端二次确认）。
+
+    用它的唯一原因是：先把传输层跑通，VAD 的好坏不影响"Pipecat 能不能
+    当传输层用"这个结论。
+    """
+
+    def __init__(self, *, threshold: float = 350.0,
+                 silence_ms: int = 700, min_speech_ms: int = 300):
+        self.threshold = threshold          # PCM16 的 RMS 阈值
+        self.silence_ms = silence_ms
+        self.min_speech_ms = min_speech_ms
+        self._silence = 0.0
+        self._speech = 0.0
+
+    def reset(self) -> None:
+        self._silence = 0.0
+        self._speech = 0.0
+
+    def feed(self, pcm16: np.ndarray, chunk_ms: float) -> str | None:
+        """喂一块音频，返回 'start' / 'end' / None。"""
+        if pcm16.size == 0:
+            return None
+        rms = float(np.sqrt(np.mean(pcm16.astype(np.float32) ** 2)))
+        event = None
+        if rms >= self.threshold:
+            if self._speech == 0.0:
+                event = "start"
+            self._speech += chunk_ms
+            self._silence = 0.0
+        else:
+            if self._speech > 0:
+                self._silence += chunk_ms
+                if self._silence >= self.silence_ms:
+                    if self._speech >= self.min_speech_ms:
+                        event = "end"
+                    self.reset()
+        return event
+
+
+# ---------------------------------------------------------------- 引擎处理器
+
+class EngineProcessor(FrameProcessor):
+    """把 ConversationEngine 包成一个 Pipecat 处理器。
+
+    A 方案的核心假设在这里被检验：**一个处理器吞掉整条会话逻辑，
+    Pipecat 只负责音频进出和帧路由。**
+    """
+
+    def __init__(self, cfg: AppConfig, transport: SmallWebRTCTransport):
+        super().__init__()
+        log.info("[诊断] EngineProcessor 已创建")
+        self.cfg = cfg
+        self.transport = transport
+        self._vad = SimpleVAD()
+        self._buf = bytearray()          # 当前话语的 PCM16
+        self._turn: asyncio.Task | None = None
+        self._engine: ConversationEngine | None = None
+        self._session = ""
+        self._bot_speaking = False
+        # 计时：用于拿真实延迟数字
+        self._t_speech_end = 0.0
+        self._t_first_audio = 0.0
+
+    # ------------------------------------------------------------ 生命周期
+
+    async def _ensure_engine(self) -> ConversationEngine:
+        if self._engine is None:
+            mem = SQLiteMemoryStore(Path.home() / ".local/share/ai_assistant/memory.db")
+            await mem.open()
+            llm = FallbackProvider(create_provider(self.cfg.llm),
+                                   create_provider(self.cfg.llm_fallback))
+            self._engine = ConversationEngine(
+                self.cfg, llm, mem, ManualIdentityProvider(),
+                create_tts(self.cfg.voice), StreamingSTT(self.cfg.voice))
+            self._session = await self._engine.start_session(user_id="u_ylhan")
+        return self._engine
+
+    async def setup(self, setup):
+        log.info("[诊断] EngineProcessor.setup() 被调用")
+        await super().setup(setup)
+        log.info("[诊断] EngineProcessor 初始化完成")
+
+    async def process_frame(self, frame, direction: FrameDirection):
+        log.info("[诊断·顶] 进入 process_frame: %s", type(frame).__name__)
+        await super().process_frame(frame, direction)
+
+        # [诊断] 记录到达的帧类型分布
+        self._seen = getattr(self, "_seen", {})
+        k = type(frame).__name__
+        self._seen[k] = self._seen.get(k, 0) + 1
+        if sum(self._seen.values()) <= 40 or self._seen[k] == 1:
+            log.info("[诊断] 帧: %s (累计 %d)", k, sum(self._seen.values()))
+
+        if isinstance(frame, StartFrame):
+            await self.push_frame(frame, direction)
+
+        elif isinstance(frame, InputAudioRawFrame):
+            await self._on_audio(frame)
+
+        elif isinstance(frame, InterruptionFrame):
+            # 客户端判定打断后发来的信号（或 Pipecat 内部产生）
+            log.info("收到打断信号，取消当前生成")
+            await self._cancel_turn()
+            await self.push_frame(frame, direction)
+
+        elif isinstance(frame, (EndFrame, CancelFrame)):
+            await self._cancel_turn()
+            await self.push_frame(frame, direction)
+
+        else:
+            await self.push_frame(frame, direction)
+
+    # ------------------------------------------------------------ 音频处理
+
+    async def _on_audio(self, frame: InputAudioRawFrame) -> None:
+        pcm16 = np.frombuffer(frame.audio, dtype=np.int16)
+        chunk_ms = len(pcm16) / IN_RATE * 1000.0
+
+        # 头几帧打日志，用于确认音频真的到了处理器
+        self._n_audio = getattr(self, "_n_audio", 0) + 1
+        if self._n_audio <= 3:
+            rms = float(np.sqrt(np.mean(pcm16.astype(np.float32) ** 2))) if pcm16.size else 0.0
+            log.info("收到音频帧 #%d: %d 采样, %.0fms, RMS=%.0f",
+                     self._n_audio, pcm16.size, chunk_ms, rms)
+
+        # 助手正在说话时收到的音频 = 用户插话的候选。
+        # 这里只记日志，真正的打断判定在客户端（见方案 §5 的边界决定）。
+        if self._bot_speaking and float(np.sqrt(np.mean(pcm16.astype(np.float32) ** 2))) > self._vad.threshold:
+            log.info("检测到用户插话（助手正在播放）")
+
+        self._buf.extend(frame.audio)
+        event = self._vad.feed(pcm16, chunk_ms)
+
+        if event == "end":
+            audio = np.frombuffer(bytes(self._buf), dtype=np.int16).astype(np.float32) / 32768.0
+            self._buf.clear()
+            if len(audio) > IN_RATE * 0.3:      # 太短的不处理
+                self._t_speech_end = time.perf_counter()
+                self._t_first_audio = 0.0
+                await self._cancel_turn()
+                self._turn = asyncio.create_task(self._run_turn(audio))
+
+    async def _cancel_turn(self) -> None:
+        if self._turn and not self._turn.done():
+            self._turn.cancel()
+            try:
+                await self._turn
+            except (asyncio.CancelledError, Exception):  # noqa: BLE001
+                pass
+        self._turn = None
+
+    # ------------------------------------------------------------ 跑一轮
+
+    async def _run_turn(self, audio: np.ndarray) -> None:
+        try:
+            engine = await self._ensure_engine()
+            async for ev in engine.respond_to_audio(audio, session_id=self._session):
+                if ev.kind == EventKind.RESPONSE_DELTA and ev.text:
+                    await self.push_frame(TextFrame(text=ev.text))
+                elif ev.kind == EventKind.SPEAK and ev.audio is not None and len(ev.audio):
+                    if self._t_first_audio == 0.0:
+                        self._t_first_audio = time.perf_counter()
+                        log.info("首音延迟（说完→首段音频）: %.0f ms",
+                                 (self._t_first_audio - self._t_speech_end) * 1000)
+                    await self._push_audio(ev.audio, ev.sample_rate)
+                elif ev.kind == EventKind.METRICS and ev.metrics:
+                    log.info("本轮指标: %s", ev.metrics.summary())
+        except asyncio.CancelledError:
+            log.info("本轮被取消")
+            raise
+        except Exception as e:  # noqa: BLE001
+            log.exception("跑轮失败: %s", e)
+
+    async def _push_audio(self, samples: np.ndarray, sample_rate: int) -> None:
+        """把 float32 音频推给传输层。必要时重采样到 OUT_RATE。"""
+        if sample_rate != OUT_RATE:
+            # 简单线性重采样。实验够用；正式实现应该用 soxr/sox（已装）。
+            n = int(len(samples) * OUT_RATE / sample_rate)
+            samples = np.interp(np.linspace(0, len(samples) - 1, n),
+                                np.arange(len(samples)), samples).astype(np.float32)
+        pcm = (np.clip(samples, -1, 1) * 32767).astype(np.int16)
+        self._bot_speaking = True
+        # 按 20ms 一块推，模拟渐进播放
+        step = OUT_RATE // 50
+        for i in range(0, len(pcm), step):
+            chunk = pcm[i:i + step]
+            await self.push_frame(OutputAudioRawFrame(
+                audio=chunk.tobytes(), sample_rate=OUT_RATE, num_channels=1))
+        self._bot_speaking = False
+
+    async def cleanup(self):
+        await self._cancel_turn()
+        await super().cleanup()
+
+
+# ---------------------------------------------------------------- HTTP 服务
+
+app = FastAPI(title="AI Companion · Pipecat experimental server")
+handler = SmallWebRTCRequestHandler()
+_cfg: AppConfig | None = None
+_tasks: list[PipelineTask] = []
+_bg_tasks: set[asyncio.Task] = set()   # 持有后台 task 引用，避免被 GC
+
+
+@app.post("/api/offer")
+async def offer(request: SmallWebRTCRequest):
+    """WebRTC 信令：客户端发 offer，我们回 answer。"""
+
+    async def on_connection(connection):
+        transport = SmallWebRTCTransport(
+            webrtc_connection=connection,
+            params=TransportParams(
+                audio_in_enabled=True,
+                audio_in_sample_rate=IN_RATE,
+                audio_in_channels=IN_CHANNELS,
+                audio_out_enabled=True,
+                audio_out_sample_rate=OUT_RATE,
+                audio_out_channels=1,
+            ),
+        )
+        engine_proc = EngineProcessor(_cfg, transport)
+        pipeline = Pipeline([transport.input(), engine_proc, transport.output()])
+        # 注意：这个版本里 PipelineParams 没有 allow_interruptions——
+        # 打断在 1.10 已移到 pipecat/turns/ 框架。实验阶段我们不用它的
+        # 轮次框架（方案 §5 的边界决定是"打断由客户端判定"），
+        # 所以这里只开指标。
+        # 注意：这个版本的 PipelineTask.run(params) **要求单独传 params**，
+        # 只传给构造函数不够。漏传的报错是
+        #   "PipelineWorker.run() missing 1 required positional argument: 'params'"
+        # 而这个异常发生在后台任务里，会被静默吞掉——
+        # 现象是"连接正常但什么反应都没有"，而且日志里一条错误都看不到。
+        # 所以下面的 _run_task 里必须 log.exception，不能让异常无声消失。
+        task = PipelineTask(pipeline)
+        _tasks.append(task)
+        log.info("新连接已建立，管线启动")
+
+        @transport.event_handler("on_client_connected")
+        async def _on_conn(t, client):
+            # **这一步不能省。** 传输层的音频推流由 StartFrame 驱动
+            # （base_input.start()），而 PipelineTask.run() 自己不会补发。
+            # 不喂 StartFrame 的话：ICE 连通、信令全对、但一帧音频都收不到，
+            # 表现为"连上了但助手永远不说话"。
+            log.info("客户端已连接，发送 StartFrame 启动音频流")
+            await task.queue_frames([StartFrame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def _on_disc(t, client):
+            log.info("客户端断开")
+            await task.cancel()
+
+        # **必须作为后台任务跑，不能 await。**
+        # 这个回调是被 handle_web_request 调用的，而后者又在 HTTP 请求里被 await。
+        # 如果这里 `await task.run()`，就会在信令请求里阻塞整个会话的生命周期，
+        # 管线初始化的时序会歪掉——实测表现为 setup()/connect() 不被调用，
+        # 于是 on_client_connected 永远不触发、StartFrame 不发、一帧音频都收不到。
+        # 官方 runner 用的是 background_tasks.add_task(bot, args)，同样的道理。
+        async def _run_task():
+            log.info("[诊断] PipelineTask 开始运行")
+            try:
+                # TaskManager 必须在这里建——模块级没有运行中的事件循环
+                await task.run(WorkerParams(task_manager=TaskManager()))
+            except Exception as e:  # noqa: BLE001
+                log.exception("[诊断] PipelineTask 异常退出: %s", e)
+            log.info("[诊断] PipelineTask 已结束")
+
+        bg = asyncio.create_task(_run_task())
+        _bg_tasks.add(bg)
+        bg.add_done_callback(_bg_tasks.discard)
+
+    answer = await handler.handle_web_request(request, on_connection)
+    if answer is None:
+        return {"status": "ok"}
+    return answer
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "connections": len([t for t in _tasks if not t.has_finished()])}
+
+
+@app.get("/")
+async def root():
+    return RedirectResponse("/client/")
+
+
+def mount_client() -> bool:
+    """挂载 prebuilt 网页客户端（用于测试）。"""
+    try:
+        from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
+        app.mount("/client", SmallWebRTCPrebuiltUI, name="client")
+        return True
+    except Exception as e:  # noqa: BLE001
+        log.warning("prebuilt 客户端不可用（不影响服务端）: %s", e)
+        return False
+
+
+def main() -> int:
+    global _cfg
+    p = argparse.ArgumentParser()
+    p.add_argument("--host", default="0.0.0.0")
+    p.add_argument("--port", type=int, default=7860)
+    a = p.parse_args()
+
+    _cfg = AppConfig.load()
+    log.info("LLM=%s  TTS=%s/%s  STT=%d线程",
+             _cfg.llm.model, _cfg.voice.tts_backend, _cfg.voice.tts_speaker,
+             _cfg.voice.stt_threads)
+    mounted = mount_client()
+    log.info("网页客户端: %s", "已挂载 /client/" if mounted else "未挂载")
+
+    import uvicorn
+    uvicorn.run(app, host=a.host, port=a.port, log_level="warning")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

部署说明.md

@@ -51,6 +51,8 @@ DoA 读取和网络桥接，不跑任何推理。
 
 ```
 ~/ai-companion/            助手主体
+├── client/test_client.py  自写 aiortc 客户端（实验用）
+├── services/bot_server.py Pipecat SmallWebRTC 服务端（实验，端口 7860）
 ├── app/                   24 个 py 文件
 │   ├── config.py          配置，支持环境变量覆盖
 │   ├── prompts.py         单条 system 每轮重建
@@ -98,7 +100,11 @@ TTS 的线程扩展方向又和 STT 相反（越多越快）。
 cd ~/tts-qwen && setsid nohup .venv/bin/python tts_server.py --port 18040 \
   > logs/tts_service.log 2>&1 < /dev/null &
 
-# 2. 确认就绪
+# 2. （可选）Pipecat 传输层服务端 —— 实验状态，帧路由未通
+cd ~/ai-companion && setsid nohup .venv/bin/python services/bot_server.py \
+  --host 0.0.0.0 --port 7860 > logs/bot_server.log 2>&1 < /dev/null &
+
+# 3. 确认就绪
 curl -s http://127.0.0.1:18040/health
 
 # 3. 跑测试
@@ -154,6 +160,19 @@ rsync -az --delete --exclude '__pycache__' --exclude '*.pyc' \
     （vLLM 0.28 拉上来的）；源码编译要从 GitHub 拉依赖，而这台机器下
     GitHub release 会断连。
 
+## 当前卡点：Pipecat 帧路由未通
+
+2026-09-15 的传输层实验：**传输层跑通（信令/ICE/data channel/双向媒体轨），
+但帧送不进自定义处理器**——`process_frame` 调用次数为 0，引擎从未运行。
+
+现象：连接全对、音轨协商成功、助手永远不说话，**日志里一条错误都没有**。
+
+这直接影响 A/B 架构方案（A：Pipecat 只做传输；B：按框架 service 模型重构）。
+根因未定位。完整记录见 Obsidian 笔记库里的 `P0-Pipecat传输层实验报告.md`。
+
+**注意** `services/bot_server.py` 里目前还带着 `[诊断]` 前缀的日志，
+是留给下一轮排查用的，定位后应清理。
+
 ## 未做的事
 
 - **没有常驻服务**：TTS 服务和助手都没有 systemd 开机自启（方案 §10 要求 P2 后做）