From c3d2c7f3a8795f487dc3fae9056ec016d94a6776 Mon Sep 17 00:00:00 2001 From: ylhan Date: Wed, 16 Sep 2026 09:50:38 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20Pipecat=20=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E9=93=BE=E8=B7=AF=E5=B9=B6=E8=A1=A5=E9=BD=90=E7=BD=91?= =?UTF-8?q?=E9=A1=B5=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 39 +++++---- app/tests/test_transport.py | 78 ++++++++++++++++++ client/test_client.py | 87 ++++++++++++-------- requirements.txt | 3 + services/bot_server.py | 154 ++++++++++++++++++++++++++---------- 部署说明.md | 35 +++++--- 6 files changed, 296 insertions(+), 100 deletions(-) create mode 100644 app/tests/test_transport.py diff --git a/README.md b/README.md index c3a380a..424fdb1 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ app/ 助手本体（部署到服务器） │ ├── stt.py sherpa-onnx 流式 Zipformer（CPU） │ ├── tts.py 本地 vits-zh-ll（sherpa-onnx） │ └── remote_tts.py Qwen3-TTS 远端服务适配（当前主用） -└── tests/ 4 个测试，共 88 项断言，全部无需硬件 +└── tests/ 5 个测试脚本，全部无需硬件 services/ ├── tts_server.py Qwen3-TTS 服务（跑在 ~/tts-qwen，独立 venv） @@ -52,6 +52,7 @@ cd ~/ai-companion .venv/bin/python app/tests/test_memory.py # 28 项 .venv/bin/python app/tests/test_identity.py # 17 项 .venv/bin/python app/tests/test_engine_e2e.py # 16 项，端到端 +.venv/bin/python app/tests/test_transport.py # 3 项，媒体时钟/VAD/静音判定 ``` 端到端测试**不需要麦克风、音箱、摄像头、ESP32**—— @@ -80,28 +81,32 @@ cd ~/ai-companion **引擎与传输解耦。** 不依赖 Pipecat、不依赖任何传输方式，只依赖四个接口。 好处：可以用文件音频端到端测试；接 WebRTC 时只需写薄适配层。 -## 当前卡点：Pipecat 传了但没接上 +## Pipecat 实时链路：已闭环 -方案文档写"Pipecat 作为唯一会话管线框架"，但**代码里它只在 requirements.txt 里**。 +2026-09-15 复查并修复传输层实验： -2026-09-15 做了传输层实验（`services/bot_server.py` + `client/test_client.py`）： +- ✅ 信令、ICE、data channel、双向媒体轨 +- ✅ `PipelineWorker` 自动 `StartFrame`，自定义 `EngineProcessor` 正常收帧 +- ✅ 文件音频 → STT → DeepSeek → Qwen3-TTS → WebRTC 非静音回复 +- ✅ 保留 RTVI 和 `/client/` prebuilt 网页客户端，并提供其需要的 `/start` 入口 -- ✅ **传输层跑通**：信令、ICE、data channel、双向媒体轨 -- ❌ **帧路由未通**：自定义处理器的 `process_frame` 调用次数为 0， - 连自己注入的 StartFrame 都没进去；引擎从未运行 +原来的“`process_frame` 调用次数为 0”并非管线接线或模型不兼容。真实根因是 +Pipecat 启动预热 NLTK `punkt_tab` 时发生运行时下载，20 秒 setup 超时，worker +尚未发送 `StartFrame` 就退出。数据预装后帧路由正常，因此确定采用 A 方案： +Pipecat 管传输、帧与 RTVI，自有 `ConversationEngine` 管会话语义。 -现象是"连接全对、音轨协商成功、但助手永远不说话，且日志里一条错误都没有"。 +自写客户端另有两处测试缺陷也已修复：文件结束后的静音帧没有推进 PTS，导致 VAD +收不到 700ms 尾静音；接收端又把 WebRTC 自动静音误判为回复。现在成功条件是收到 +非静音语音，并按实际 WebRTC 采样率保存 WAV。 -**这直接影响 A/B 架构方案**：A 是"Pipecat 只做传输、逻辑归自有引擎"， -B 是"按 Pipecat 的 service 模型重构"。帧送不进自定义处理器， -说明 A 的核心假设不是开箱即用的。 - -根因未定位，两种可能必须区分：接线错误（接缝浅，A 可行） -vs 模型不兼容（接缝深，应转 B）。 +2026-09-16 又通过 SSH 隧道用 `/client/` 做了真人浏览器测试：补齐 Playground +要求的 `/start` 后，麦克风 → STT → LLM → TTS → 浏览器播放正常，单轮服务端指标 +约 3.0 秒。这个页面是调试工具而非产品 UI；当前连接期间由 VAD 检出每段话后都会 +响应，**尚无唤醒门控**。正式唤醒仍按方案放在 P2 的 N5095 本地端，待机时不上传音频。 完整记录见代码仓库外的 `P0-Pipecat传输层实验报告.md`（在 Obsidian 笔记库里）。 -### 实验挖出的 8 个坑 +### 实验挖出的 10 个坑 每个都表现为"什么都没发生"，极难排查： @@ -109,10 +114,12 @@ vs 模型不兼容（接缝深，应转 B）。 2. 客户端必须建 data channel 并定期发 `"ping"`，否则服务端一帧音频都不推 3. offer 里没有媒体时 aiortc **完全不收集 ICE 候选**（连连通性测试都要加静音轨） 4. ICE 收集完成前发 offer → SDP 里 0 个候选地址 -5. `PipelineTask.run()` 要 `WorkerParams(task_manager=...)`，不是 `PipelineParams` +5. `PipelineWorker.run()` 要 `WorkerParams(task_manager=...)`，不是 `PipelineParams` 6. `TaskManager()` 必须在事件循环内创建 7. **后台任务里的异常被静默吞掉**，日志里查不到 8. 信令回调里必须 `create_task`，不能 `await task.run()` +9. Pipecat 会预热 NLTK `punkt_tab`；生产部署必须预装，不能在首个会话临时下载 +10. `PipelineWorker` 自动发送 `StartFrame`，手工再发会造成重复启动帧 ## 踩过的坑 diff --git a/app/tests/test_transport.py b/app/tests/test_transport.py new file mode 100644 index 0000000..3f2fc15 --- /dev/null +++ b/app/tests/test_transport.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""传输适配层的无硬件回归测试。""" + +from __future__ import annotations + +import asyncio +import tempfile +import time +import wave +from pathlib import Path +import sys + +import numpy as np +from av import AudioFrame + +ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(ROOT)) + +from client.test_client import FileAudioTrack, ReplyRecorder +from services.bot_server import SimpleVAD + + +def _write_test_wav(path: Path) -> None: + samples = np.ones(1600, dtype=np.int16) * 1000 # 100ms 语音 + with wave.open(str(path), "wb") as output: + output.setnchannels(1) + output.setsampwidth(2) + output.setframerate(16000) + output.writeframes(samples.tobytes()) + + +async def test_file_track_clock() -> None: + with tempfile.TemporaryDirectory(prefix="transport_test_") as directory: + path = Path(directory) / "input.wav" + _write_test_wav(path) + track = FileAudioTrack(path, tail_silence_s=0.8) + track._t0 = time.perf_counter() - 10 # 测试不需要真实等待 + pts = [] + while not track.finished: + pts.append((await track.recv()).pts) + + assert len(pts) == 90, len(pts) # 100ms 音频 + 800ms 静音 + assert all(after > before for before, after in zip(pts, pts[1:])), pts[-5:] + + +def test_vad_tail_silence() -> None: + vad = SimpleVAD() + events = [vad.feed(np.ones(320, dtype=np.int16) * 1000, 20) for _ in range(20)] + events += [vad.feed(np.zeros(320, dtype=np.int16), 20) for _ in range(40)] + assert "end" in events + + +def test_reply_requires_non_silent_audio() -> None: + recorder = ReplyRecorder(Path("/tmp/unused-transport-test.wav")) + silence = AudioFrame(format="s16", layout="mono", samples=320) + silence.planes[0].update(np.zeros(320, dtype=np.int16).tobytes()) + silence.sample_rate = 48000 + recorder.feed(silence) + assert not recorder.has_speech + assert recorder.sample_rate == 48000 + + speech = AudioFrame(format="s16", layout="mono", samples=320) + speech.planes[0].update((np.ones(320, dtype=np.int16) * 500).tobytes()) + speech.sample_rate = 48000 + recorder.feed(speech) + assert recorder.has_speech + + +async def main() -> int: + await test_file_track_clock() + test_vad_tail_silence() + test_reply_requires_non_silent_audio() + print("传输适配层回归测试：3/3 PASS") + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/client/test_client.py b/client/test_client.py index 36f4f46..4ff7852 100644 --- a/client/test_client.py +++ b/client/test_client.py @@ -76,7 +76,8 @@ class FileAudioTrack(AudioStreamTrack): kind = "audio" - def __init__(self, wav_path: Path, *, sample_rate: int = 16000): + def __init__(self, wav_path: Path, *, sample_rate: int = 16000, + tail_silence_s: float = 1.2): super().__init__() with wave.open(str(wav_path)) as w: assert w.getnchannels() == 1, "需要单声道" @@ -88,7 +89,10 @@ class FileAudioTrack(AudioStreamTrack): 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._source_pos = 0 + self._timestamp_samples = 0 + self._tail_remaining = int(tail_silence_s * sample_rate) + self.source_finished_at: float | None = None self._t0: float | None = None self._frame_samples = sample_rate // 100 # 10ms @@ -96,19 +100,25 @@ class FileAudioTrack(AudioStreamTrack): """按 10ms 一块、以真实时间节奏产出。""" if self._t0 is None: self._t0 = time.perf_counter() - self._pos = 0 + self._source_pos = 0 - # 音频放完后持续吐静音，让服务端的 VAD 能判到"说完" - if self._pos >= len(self.samples): + frame_pts = self._timestamp_samples + if self._source_pos >= len(self.samples): + if self.source_finished_at is None: + self.source_finished_at = time.perf_counter() audio = np.zeros(self._frame_samples, dtype=np.float32) + self._tail_remaining = max(0, self._tail_remaining - self._frame_samples) else: - audio = self.samples[self._pos:self._pos + self._frame_samples] + audio = self.samples[self._source_pos:self._source_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 + self._source_pos += self._frame_samples + + # 即使源文件已经结束，静音帧也必须持续推进时钟和 PTS。 + self._timestamp_samples += self._frame_samples # 实时节奏 - target = self._t0 + self._pos / self.sample_rate + target = self._t0 + self._timestamp_samples / self.sample_rate delay = target - time.perf_counter() if delay > 0: await asyncio.sleep(delay) @@ -118,33 +128,45 @@ class FileAudioTrack(AudioStreamTrack): 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.pts = int(frame_pts * 90000 / self.sample_rate) f.time_base = fractions.Fraction(1, 90000) return f @property def finished(self) -> bool: - return self._pos >= len(self.samples) + return self._source_pos >= len(self.samples) and self._tail_remaining <= 0 class ReplyRecorder: """收服务端音频，写 wav，并记录首音延迟。""" - def __init__(self, out_path: Path, sample_rate: int = 24000): + def __init__(self, out_path: Path, sample_rate: int = 24000, + speech_rms_threshold: float = 80.0): 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() + self.first_non_silent_at: float | None = None + self.speech_rms_threshold = speech_rms_threshold 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) + channels = len(frame.layout.channels) + if frame.format.is_planar: + raw = arr.astype(np.float32).mean(axis=0) + else: + raw = arr.astype(np.float32).reshape(-1, channels).mean(axis=1) + # WebRTC/Opus 通常以 48kHz 交付，即使服务端原始输出是 24kHz。 + # WAV 必须记录接收帧的实际采样率，否则时长会被算错。 + if not self.chunks: + self.sample_rate = frame.sample_rate + elif frame.sample_rate != self.sample_rate: + raise ValueError( + f"回复音轨采样率发生变化: {self.sample_rate} -> {frame.sample_rate}") + raw = raw.reshape(-1) + rms = float(np.sqrt(np.mean(raw ** 2))) if raw.size else 0.0 + if self.first_non_silent_at is None and rms >= self.speech_rms_threshold: + self.first_non_silent_at = time.perf_counter() + self.chunks.append(raw / 32768.0) def save(self) -> float: if not self.chunks: @@ -160,8 +182,13 @@ class ReplyRecorder: 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 has_speech(self) -> bool: + return self.first_non_silent_at is not None + + def first_audio_ms_from(self, started_at: float | None) -> float | None: + if started_at is None or self.first_non_silent_at is None: + return None + return (self.first_non_silent_at - started_at) * 1000 def post_offer(server: str, sdp: str, type_: str) -> dict: @@ -197,8 +224,6 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, 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: @@ -293,10 +318,10 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, while not track.finished and time.perf_counter() - t0 < timeout_s: await asyncio.sleep(0.2) log.info("音频已发完，等待回复…") - # 等首个回复音频 + # 等首个非静音回复；SmallWebRTC 空闲时也会持续发送静音。 t_wait = time.perf_counter() while time.perf_counter() - t_wait < timeout_s: - if recorder.chunks: + if recorder.has_speech: break await asyncio.sleep(0.1) # 再收一会儿，让回复说完 @@ -308,13 +333,13 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, print() print("=== 结果 ===") - print(f" 连接: {'✅ 成功' if dur > 0 else '⚠️ 无音频返回'}") + print(f" 回复: {'✅ 收到非静音语音' if recorder.has_speech 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 + print(f" 录制音轨: {dur:.2f} 秒 → {out}") + fa = recorder.first_audio_ms_from(track.source_finished_at if track else None) + if fa is not None: + print(f" 首音延迟（源音频结束→首个非静音输出）: {fa:.0f} ms") + return 0 if (connect_only or recorder.has_speech) else 1 def main() -> int: diff --git a/requirements.txt b/requirements.txt index 428e364..3628a25 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,9 @@ # 实测可用组合：Python 3.12（服务器）/ 3.13（N5095），两台机器都验证过。 pipecat-ai==1.10.0 # 会话管线框架（服务器侧用其传输/打断能力） +aiortc==1.15.0 # N5095 文件/声卡客户端与 WebRTC 回归测试 +av==17.1.0 # aiortc 音频帧（客户端直接导入） +nltk==3.10.3 # Pipecat 分句预热；punkt_tab 需在部署阶段预装 sherpa-onnx==1.13.8 # 本地 STT/TTS 运行时（自带 onnxruntime，不依赖 pip 的） numpy==2.5.3 aiohttp==3.14.3 # LLM 流式请求 diff --git a/services/bot_server.py b/services/bot_server.py index 6134d17..c2a93d4 100644 --- a/services/bot_server.py +++ b/services/bot_server.py @@ -32,7 +32,7 @@ import logging import os import sys import time -import uuid +from collections import deque from pathlib import Path import numpy as np @@ -51,7 +51,7 @@ from pipecat.frames.frames import ( TextFrame, ) from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams from pipecat.transports.smallwebrtc.request_handler import ( @@ -73,20 +73,25 @@ logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("bot") +# Pipecat 会在每个 worker 启动时预热 NLTK 分句器。固定到私有、可预装的目录， +# 避免使用权限过宽的 ~/nltk_data，也避免首个会话临时联网下载。 +DEFAULT_NLTK_DATA = Path.home() / ".local/share/ai-companion/nltk_data" +os.environ.setdefault("NLTK_DATA", str(DEFAULT_NLTK_DATA)) + #: 输入音频格式（WebRTC 侧协商后的格式） IN_RATE = 16000 IN_CHANNELS = 1 #: 输出音频格式（Qwen3-TTS 输出 24kHz） OUT_RATE = 24000 -#: 管线参数。给 PipelineTask 构造用。 +#: 管线参数。给 PipelineWorker 构造用。 PIPELINE_PARAMS = PipelineParams( audio_in_sample_rate=IN_RATE, audio_out_sample_rate=OUT_RATE, enable_metrics=True, ) -#: 注意：PipelineTask.run() 要的是 WorkerParams（内含 task_manager）， +#: 注意：PipelineWorker.run() 要的是 WorkerParams（内含 task_manager）， #: **不是** PipelineParams。传错会依次报两个错，且都发生在后台任务里， #: 主日志里什么都看不到： #: 1. 不传 → "PipelineWorker.run() missing 1 required positional argument: 'params'" @@ -121,8 +126,13 @@ class SimpleVAD: self._silence = 0.0 self._speech = 0.0 + @property + def active(self) -> bool: + """是否已经进入一段候选语音。""" + return self._speech > 0.0 + def feed(self, pcm16: np.ndarray, chunk_ms: float) -> str | None: - """喂一块音频，返回 'start' / 'end' / None。""" + """喂一块音频，返回 start / end / discard / None。""" if pcm16.size == 0: return None rms = float(np.sqrt(np.mean(pcm16.astype(np.float32) ** 2))) @@ -138,6 +148,8 @@ class SimpleVAD: if self._silence >= self.silence_ms: if self._speech >= self.min_speech_ms: event = "end" + else: + event = "discard" self.reset() return event @@ -153,13 +165,15 @@ class EngineProcessor(FrameProcessor): def __init__(self, cfg: AppConfig, transport: SmallWebRTCTransport): super().__init__() - log.info("[诊断] EngineProcessor 已创建") self.cfg = cfg self.transport = transport self._vad = SimpleVAD() + self._pre_roll: deque[bytes] = deque(maxlen=50) # 约 1 秒 self._buf = bytearray() # 当前话语的 PCM16 self._turn: asyncio.Task | None = None self._engine: ConversationEngine | None = None + self._memory: SQLiteMemoryStore | None = None + self._llm: FallbackProvider | None = None self._session = "" self._bot_speaking = False # 计时：用于拿真实延迟数字 @@ -170,32 +184,24 @@ class EngineProcessor(FrameProcessor): 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._memory = SQLiteMemoryStore( + Path.home() / ".local/share/ai_assistant/memory.db") + await self._memory.open() + self._llm = FallbackProvider(create_provider(self.cfg.llm), + create_provider(self.cfg.llm_fallback)) self._engine = ConversationEngine( - self.cfg, llm, mem, ManualIdentityProvider(), + self.cfg, self._llm, self._memory, 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 初始化完成") + log.info("会话处理器已就绪") 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) @@ -233,9 +239,27 @@ class EngineProcessor(FrameProcessor): 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) + was_active = self._vad.active + self._pre_roll.append(frame.audio) event = self._vad.feed(pcm16, chunk_ms) + if event == "start": + # 带上说话前约 1 秒，避免 VAD 切掉首字。 + self._buf = bytearray().join(self._pre_roll) + elif was_active: + self._buf.extend(frame.audio) + + # 防止噪声或异常客户端让常驻进程无限积累音频。 + max_bytes = IN_RATE * 2 * 60 + if len(self._buf) > max_bytes: + log.warning("单次话语超过 60 秒，强制截断处理") + event = "end" + self._vad.reset() + + if event == "discard": + self._buf.clear() + return + if event == "end": audio = np.frombuffer(bytes(self._buf), dtype=np.int16).astype(np.float32) / 32768.0 self._buf.clear() @@ -243,15 +267,11 @@ class EngineProcessor(FrameProcessor): 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)) + self._turn = self.create_task(self._run_turn(audio), "conversation-turn") 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 + await self.cancel_task(self._turn, timeout=5.0) self._turn = None # ------------------------------------------------------------ 跑一轮 @@ -295,6 +315,15 @@ class EngineProcessor(FrameProcessor): async def cleanup(self): await self._cancel_turn() + if self._engine is not None: + await self._engine.end_session() + if self._llm is not None: + await self._llm.aclose() + if self._memory is not None: + await self._memory.close() + self._engine = None + self._llm = None + self._memory = None await super().cleanup() @@ -303,7 +332,7 @@ class EngineProcessor(FrameProcessor): app = FastAPI(title="AI Companion · Pipecat experimental server") handler = SmallWebRTCRequestHandler() _cfg: AppConfig | None = None -_tasks: list[PipelineTask] = [] +_tasks: list[PipelineWorker] = [] _bg_tasks: set[asyncio.Task] = set() # 持有后台 task 引用，避免被 GC @@ -329,24 +358,39 @@ async def offer(request: SmallWebRTCRequest): # 打断在 1.10 已移到 pipecat/turns/ 框架。实验阶段我们不用它的 # 轮次框架（方案 §5 的边界决定是"打断由客户端判定"）， # 所以这里只开指标。 - # 注意：这个版本的 PipelineTask.run(params) **要求单独传 params**， - # 只传给构造函数不够。漏传的报错是 + # 注意：PipelineWorker 的音频参数传构造函数；run() 另收 + # WorkerParams（主要提供 task_manager）。漏传 run 参数会报： # "PipelineWorker.run() missing 1 required positional argument: 'params'" # 而这个异常发生在后台任务里，会被静默吞掉—— # 现象是"连接正常但什么反应都没有"，而且日志里一条错误都看不到。 # 所以下面的 _run_task 里必须 log.exception，不能让异常无声消失。 - task = PipelineTask(pipeline) + # 保留 RTVI：网页 prebuilt 客户端用它显示状态、转写和指标； + # N5095 的精简客户端可以忽略不认识的 data-channel 消息。 + task = PipelineWorker( + pipeline, + params=PIPELINE_PARAMS, + enable_rtvi=True, + idle_timeout_secs=None, + ) _tasks.append(task) log.info("新连接已建立，管线启动") + @task.event_handler("on_pipeline_started") + async def _on_pipeline_started(worker, frame): + log.info("管线已就绪") + + @task.event_handler("on_pipeline_timeout") + async def _on_pipeline_timeout(worker, frame): + log.error("管线等待 %s 超时", type(frame).__name__) + + @task.event_handler("on_pipeline_error") + async def _on_pipeline_error(worker, frame): + log.error("管线错误: %s", frame) + @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()]) + # PipelineWorker.run() 会自动发送且只应发送一次 StartFrame。 + log.info("客户端已连接") @transport.event_handler("on_client_disconnected") async def _on_disc(t, client): @@ -360,13 +404,13 @@ async def offer(request: SmallWebRTCRequest): # 于是 on_client_connected 永远不触发、StartFrame 不发、一帧音频都收不到。 # 官方 runner 用的是 background_tasks.add_task(bot, args)，同样的道理。 async def _run_task(): - log.info("[诊断] PipelineTask 开始运行") + log.info("PipelineWorker 开始运行") try: # TaskManager 必须在这里建——模块级没有运行中的事件循环 await task.run(WorkerParams(task_manager=TaskManager())) except Exception as e: # noqa: BLE001 - log.exception("[诊断] PipelineTask 异常退出: %s", e) - log.info("[诊断] PipelineTask 已结束") + log.exception("PipelineWorker 异常退出: %s", e) + log.info("PipelineWorker 已结束") bg = asyncio.create_task(_run_task()) _bg_tasks.add(bg) @@ -378,6 +422,20 @@ async def offer(request: SmallWebRTCRequest): return answer +@app.post("/start") +async def start_web_client(): + """给 Pipecat Playground 返回本服务的 WebRTC 信令入口。 + + prebuilt 网页客户端会先 POST /start，再用响应里的 endpoint 发 SDP offer； + 自写 aiortc/N5095 客户端则可以继续直接调用 /api/offer。 + """ + return { + "webrtcRequestParams": { + "endpoint": "/api/offer", + }, + } + + @app.get("/health") async def health(): return {"status": "ok", "connections": len([t for t in _tasks if not t.has_finished()])} @@ -399,6 +457,19 @@ def mount_client() -> bool: return False +def check_runtime_data() -> None: + """启动前验证运行时数据，避免首个会话里偷偷联网下载。""" + import nltk + + try: + nltk.data.find("tokenizers/punkt_tab") + except LookupError as e: + raise RuntimeError( + "缺少 NLTK punkt_tab。请在部署阶段运行 " + "`python -m nltk.downloader punkt_tab`，并设置 NLTK_DATA。" + ) from e + + def main() -> int: global _cfg p = argparse.ArgumentParser() @@ -406,6 +477,7 @@ def main() -> int: p.add_argument("--port", type=int, default=7860) a = p.parse_args() + check_runtime_data() _cfg = AppConfig.load() log.info("LLM=%s TTS=%s/%s STT=%d线程", _cfg.llm.model, _cfg.voice.tts_backend, _cfg.voice.tts_speaker, diff --git a/部署说明.md b/部署说明.md index 9281103..cb8b904 100644 --- a/部署说明.md +++ b/部署说明.md @@ -61,7 +61,7 @@ DoA 读取和网络桥接，不跑任何推理。 │ ├── memory/ 接口 + SQLite 实现 │ ├── identity/ 接口 + 手动切换实现 │ ├── voice/ STT / TTS（本地 + 远端两种后端） -│ └── tests/ 4 个测试，共 88 项断言 +│ └── tests/ 5 个测试脚本（含传输适配层回归） ├── models -> ~/ai_companion_bench/models 软链，未重复占用 └── .venv/ 608MB（Python 3.12） @@ -100,7 +100,7 @@ TTS 的线程扩展方向又和 STT 相反（越多越快）。 cd ~/tts-qwen && setsid nohup .venv/bin/python tts_server.py --port 18040 \ > logs/tts_service.log 2>&1 < /dev/null & -# 2. （可选）Pipecat 传输层服务端 —— 实验状态，帧路由未通 +# 2. Pipecat 实时服务端（含 /client/ 网页测试入口） 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 & @@ -160,28 +160,39 @@ rsync -az --delete --exclude '__pycache__' --exclude '*.pyc' \ （vLLM 0.28 拉上来的）；源码编译要从 GitHub 拉依赖，而这台机器下 GitHub release 会断连。 -## 当前卡点：Pipecat 帧路由未通 +## Pipecat 实时链路状态 -2026-09-15 的传输层实验：**传输层跑通（信令/ICE/data channel/双向媒体轨）， -但帧送不进自定义处理器**——`process_frame` 调用次数为 0，引擎从未运行。 +2026-09-15 已闭环：`PipelineWorker` 自动启动，音频帧进入自定义处理器，完整 +STT → DeepSeek → Qwen3-TTS → WebRTC 返回非静音语音。保留 RTVI 与 `/client/` +网页测试入口，并补齐 Playground 启动所需的 `/start`；架构确定采用 +“Pipecat 管传输，自有引擎管语义”的 A 方案。 -现象：连接全对、音轨协商成功、助手永远不说话，**日志里一条错误都没有**。 +原帧路由问题的根因是 NLTK `punkt_tab` 首次运行下载拖住 Pipecat 预热，导致 +20 秒 setup timeout；不是框架不支持自定义处理器。服务器已把数据预装到权限 700 的 +`~/.local/share/ai-companion/nltk_data`，服务启动时会 fail-fast 检查。 -这直接影响 A/B 架构方案（A：Pipecat 只做传输；B：按框架 service 模型重构）。 -根因未定位。完整记录见 Obsidian 笔记库里的 `P0-Pipecat传输层实验报告.md`。 +单次文件音频实测：源音频结束到客户端首个非静音约 5.8 秒；其中包含约 0.7 秒 +VAD 等待，当前仍高于 P1 的 2 秒目标，需要后续单独优化。完整记录见 Obsidian +笔记库里的 `P0-Pipecat传输层实验报告.md`。 -**注意** `services/bot_server.py` 里目前还带着 `[诊断]` 前缀的日志， -是留给下一轮排查用的，定位后应清理。 +2026-09-16 通过 SSH 隧道打开 `http://localhost:7860/client/` 真人实测成功；补齐 +Playground 使用的 `/start` 后，浏览器麦克风、RTVI、语音回复均正常，单轮服务端 +指标约 3.0 秒。当前是 P1 调试模式：连接后持续上传并响应每段有效语音，尚未实现 +P2 的本地唤醒和 20 秒连续对话窗口。 ## 未做的事 - **没有常驻服务**：TTS 服务和助手都没有 systemd 开机自启（方案 §10 要求 P2 后做） -- **没接 Pipecat 管线**：引擎与传输解耦，接 WebRTC 时写薄适配层即可 +- **网页 HTTPS 尚未启用**：`/client/` 已挂载且 HTTP 返回正常，但浏览器远程取麦克风 + 需要 HTTPS；Tailscale Serve 仍需在 tailnet 管理页启用 +- **唤醒尚未实现**：当前网页连接后会持续上传麦克风并响应每段有效语音；正式方案 + 是在 N5095 本地做关键词检测，待机时音频不出网 - **设备侧代码不存在**：N5095 的外设桥（USB、WebRTC 客户端、串口）尚未开始 - **STT 没换**：仍是 2023 年的 Zipformer int8，中文识别率从未测过 - **记忆库在服务器上**，而人脸特征按方案必须留在设备端——将来做 "删除我的数据"要跨两台机器协调，这个数据边界问题需要单独想 -- **代码仍在 note_sync 笔记仓库里**，未拆成独立仓库 +- **代码与设计文档分属两个仓库**：实现位于 `ai-companion`，设计记录位于 + `note_sync`，状态变更需同步更新两边 ## 相关文档