From bd543738622c155572e6f3ce7ef5e5c8eb67cfdd Mon Sep 17 00:00:00 2001 From: 5a5c91bdef336346 Date: Fri, 21 Aug 2026 12:12:24 +0700 Subject: [PATCH 1/2] add pipecat rt --- CLAUDE.md | 4 + docs/realtime-voice.md | 56 +++ docs/vi/realtime-voice_vi.md | 41 ++ hal/drivers/audio_route.py | 8 + hal/drivers/voice/_internal/config.py | 19 + hal/drivers/voice/aec.py | 324 +++++++++++++ hal/drivers/voice/tts/service.py | 5 + hal/drivers/voice/voice_service.py | 5 + hal/pipecat_rt.py | 662 ++++++++++++++++++++++++++ 9 files changed, 1124 insertions(+) create mode 100644 hal/drivers/voice/aec.py create mode 100644 hal/pipecat_rt.py diff --git a/CLAUDE.md b/CLAUDE.md index 2ed1f1d3a..805ace064 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,6 +2,10 @@ This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. +## Coding convention + +Avoid commenting anything; just drop short notes for critical functions; other functions just need a description, params note (and/or short example); do not provide full payload or usecase to trigger the function; The same expectation for module layer; + ## Multi-IDE Rules (Cursor + Claude Code) This repo is developed in both **Cursor** and **Claude Code**. The following rules (from `.cursor/rules/`) apply to all code changes: diff --git a/docs/realtime-voice.md b/docs/realtime-voice.md index ec3cfc932..4f081c89c 100644 --- a/docs/realtime-voice.md +++ b/docs/realtime-voice.md @@ -147,6 +147,61 @@ which do not exist until the first `connect()` succeeds. No HAL restart or new audio is required; voice turns keep using the main-agent fallback until the connection recovers. +## Echo cancellation (AEC) + +`hal/drivers/voice/aec.py` runs the mic through WebRTC's APM (AEC3) with the +audio being played as the reference. It is **provider-independent**: the +reference is tapped in `_WatchedStream.write` (`tts/service.py`), the single +point every playback path reaches the device through — synthesized speech, the +`speak_queue` drain, and realtime **native audio**. Tapping there rather than at +synthesis is deliberate: TTS renders a sentence far faster than real time, while +the output stream writes at playback rate, which is the timing the mic sees. + +**Off by default** (`HAL_AEC_ENABLED=false`). It needs the +`aec-audio-processing` binding, which is **not** a hal dependency — PyPI ships +no Linux wheels for it, so a device needs a locally built one. When the import +fails, `configure()` logs once and every entry point becomes a no-op; the voice +path behaves exactly as before. + +| Env | Default | Meaning | +|-----|---------|---------| +| `HAL_AEC_ENABLED` | `false` | Master switch | +| `HAL_AEC_DELAY_MS` | `150` | Speaker→mic delay hint. Measured lag on a lamp is ~154 ms; correcting 80→150 took ERLE from 10.9 to 18.6 dB | +| `HAL_AEC_NS` | `true` | Also run APM noise suppression | +| `HAL_AEC_TAIL_S` | `0.5` | Keep cancelling this long after the last speaker write, then bypass the APM | +| `HAL_AEC_DUMP_DIR` | — | Write `aec_mic/ref/out.wav` for offline ERLE analysis | + +Only the **barge-in monitor** and the main VAD loop are wrapped. Because +`_wait_for_tts()` closes the mic while the device speaks, the barge-in monitor +is currently the only path where the mic is open during playback — so it is the +only place AEC can be measured until the mic stays open through playback (full +duplex, not implemented). The reverb gate is deliberately left uncancelled so +its timing is unchanged. + +**Measured on a lamp** (OrangePi sun60 / A523, USB mic + USB speaker — two +independent clock domains), speaking a ~22 s utterance with the mic open: + +| delay hint | ERLE overall | during convergence | converged | +|---|---|---|---| +| 80 ms (old default) | 10.9 dB | 6.5 dB | 18.8 dB | +| **150 ms** | **18.6 dB** | **14.2 dB** | **21.1 dB** | + +Cost is ~3.9 % of one A523 core at realtime. The MacBook reference figure for +the same canceller is ~42 dB; the gap is the hardware — two free-running USB +clocks and a cheap analog path. Whether ~20 dB is enough to stop STT reading the +device's own speech is not yet established: that needs a mic-open turn, which +`_wait_for_tts()` does not currently allow. + +`process()` buffers to the APM's fixed 10 ms frames and returns exactly as many +samples as the caller asked for (priming once with up to 10 ms of silence), so +hal's 64 ms framing is unaffected. ERLE is logged periodically while the +speaker is active — **0 dB means the canceller is doing nothing**. + +> The image already loads PulseAudio's `module-echo-cancel` (`setup.sh`), but +> nothing reaches it: a udev rule sets `PULSE_IGNORE=1` on the speaker codec so +> hal can own it, and capture goes through `arecord -D plughw:` directly. That +> module has no reference and no client; it is not what cancels echo here. + ## Emotion expression (fire-and-forget) If the device declares the `expression` capability @@ -800,3 +855,4 @@ is a top-level `config.json` flag: | `models/`, `enums/` | Input/output/event types, provider + gateway enums | | `resources/` | System prompts (shared + per-provider) | | `../voice/voice_service.py` | Integration: streams mic audio, consumes output, routes delegate/handled | +| `../voice/aec.py` | WebRTC AEC3 on the mic path; reference tapped at the TTS output stream (all providers) | diff --git a/docs/vi/realtime-voice_vi.md b/docs/vi/realtime-voice_vi.md index e48cf8ae6..459f902cf 100644 --- a/docs/vi/realtime-voice_vi.md +++ b/docs/vi/realtime-voice_vi.md @@ -141,6 +141,46 @@ với reconnect send/receive của provider, vì các loop đó chưa tồn tạ `connect()` thành công. Không cần restart HAL hay chờ audio mới; các lượt voice vẫn fallback xuống agent chính cho tới khi kết nối hồi phục. +## Khử vọng âm (AEC) + +`hal/drivers/voice/aec.py` đưa audio mic qua APM của WebRTC (AEC3), lấy audio +đang phát làm tín hiệu tham chiếu. Nó **độc lập với provider**: tham chiếu được +lấy tại `_WatchedStream.write` (`tts/service.py`) — điểm duy nhất mà mọi đường +phát ra loa đều đi qua: giọng tổng hợp, phần drain của `speak_queue`, và **native +audio** của realtime. Lấy ở đó thay vì tại lúc tổng hợp là có chủ ý: TTS render +một câu nhanh hơn thời gian thực rất nhiều, còn output stream ghi đúng tốc độ +phát — đúng nhịp mà mic nghe thấy. + +**Mặc định tắt** (`HAL_AEC_ENABLED=false`). Nó cần binding +`aec-audio-processing`, vốn **không** phải dependency của hal — PyPI không có +wheel Linux nào, nên thiết bị cần wheel tự build. Khi import thất bại, +`configure()` log một lần và mọi entry point trở thành no-op; đường voice hoạt +động y như trước. + +| Env | Mặc định | Ý nghĩa | +|-----|----------|---------| +| `HAL_AEC_ENABLED` | `false` | Công tắc chính | +| `HAL_AEC_DELAY_MS` | `150` | Gợi ý độ trễ loa→mic. Độ trễ đo trên lamp ~154 ms; sửa 80→150 đưa ERLE từ 10.9 lên 18.6 dB | +| `HAL_AEC_NS` | `true` | Bật thêm khử nhiễu của APM | +| `HAL_AEC_TAIL_S` | `0.5` | Tiếp tục khử trong khoảng này sau lần ghi loa cuối, rồi bypass APM | +| `HAL_AEC_DUMP_DIR` | — | Ghi `aec_mic/ref/out.wav` để phân tích ERLE offline | + +Chỉ **barge-in monitor** và vòng VAD chính được bọc. Vì `_wait_for_tts()` đóng +mic khi thiết bị đang nói, barge-in monitor hiện là đường duy nhất mic còn mở +trong lúc phát — nên đó cũng là chỗ duy nhất đo được AEC, cho tới khi mic được +giữ mở suốt lúc phát (full duplex, chưa làm). Cổng reverb cố ý không khử để giữ +nguyên timing của nó. + +`process()` gom audio về khung cố định 10 ms của APM và trả về đúng số mẫu mà +caller yêu cầu (mồi một lần bằng tối đa 10 ms im lặng), nên khung 64 ms của hal +không đổi. ERLE được log định kỳ khi loa đang hoạt động — **0 dB nghĩa là bộ khử +không làm gì cả**. + +> Image đã load sẵn `module-echo-cancel` của PulseAudio (`setup.sh`), nhưng +> không có gì đi tới nó: một udev rule đặt `PULSE_IGNORE=1` cho card loa để hal +> tự sở hữu, còn capture đi thẳng qua `arecord -D plughw:`. Module đó không có +> tham chiếu lẫn client; nó không phải thứ đang khử vọng âm ở đây. + ## Biểu cảm cảm xúc (fire-and-forget) Nếu thiết bị khai báo capability `expression` @@ -770,3 +810,4 @@ trong `config.json`: | `models/`, `enums/` | Kiểu input/output/event, enum provider + gateway | | `resources/` | System prompt (chung + theo provider) | | `../voice/voice_service.py` | Tích hợp: stream audio mic, tiêu thụ output, route delegate/handled | +| `../voice/aec.py` | WebRTC AEC3 trên đường mic; tham chiếu lấy tại TTS output stream (mọi provider) | diff --git a/hal/drivers/audio_route.py b/hal/drivers/audio_route.py index 09fcd8fb2..04e3c5109 100644 --- a/hal/drivers/audio_route.py +++ b/hal/drivers/audio_route.py @@ -142,6 +142,14 @@ def bt_active() -> bool: def _swap_tts(output_idx: Optional[int]) -> None: + # A new sink is a new acoustic path — the canceller's buffered reference + # belongs to the old one. + try: + from hal.drivers.voice import aec + + aec.reset() + except Exception: + pass tts = state.tts_service if tts is None: return diff --git a/hal/drivers/voice/_internal/config.py b/hal/drivers/voice/_internal/config.py index 52204c7a0..72399bcac 100644 --- a/hal/drivers/voice/_internal/config.py +++ b/hal/drivers/voice/_internal/config.py @@ -100,6 +100,25 @@ WARM_MIC_ECHO_SKIP_MAX_S = float(os.environ.get("HAL_WARM_MIC_ECHO_SKIP_MAX_S", "0.3")) +# --------------------------------------------------------------------------- +# Acoustic echo cancellation (WebRTC AEC3) — see drivers/voice/aec.py. +# Off by default: it needs the `aec-audio-processing` native binding, which is +# not a hal dependency. Absent, every AEC entry point is a no-op. +# --------------------------------------------------------------------------- +AEC_ENABLED = os.environ.get("HAL_AEC_ENABLED", "false").lower() == "true" +# Speaker→mic delay hint. AEC3 estimates the real delay itself, but the hint +# decides how fast it converges. Measured on a lamp (USB mic + USB speaker) the +# true lag is ~154ms; correcting the hint from 80 to 150 took ERLE from 10.9 to +# 18.6 dB overall and 6.5 to 14.2 dB during the convergence phase. +AEC_DELAY_MS = int(os.environ.get("HAL_AEC_DELAY_MS", "150")) +AEC_NOISE_SUPPRESSION = os.environ.get("HAL_AEC_NS", "true").lower() == "true" +# Keep cancelling for this long after the last speaker write, then bypass the +# APM until playback resumes. +AEC_TAIL_S = float(os.environ.get("HAL_AEC_TAIL_S", "0.5")) +# Set to a directory to write aec_mic/ref/out.wav for offline ERLE analysis. +AEC_DUMP_DIR = os.environ.get("HAL_AEC_DUMP_DIR", "") + + # --------------------------------------------------------------------------- # STT keepalive — pre-connect WS before speech is detected to cut latency # --------------------------------------------------------------------------- diff --git a/hal/drivers/voice/aec.py b/hal/drivers/voice/aec.py new file mode 100644 index 000000000..7f5f258ad --- /dev/null +++ b/hal/drivers/voice/aec.py @@ -0,0 +1,324 @@ +"""Acoustic echo cancellation for the mic path (WebRTC AEC3). + +Provider-independent: the reference is tapped at the TTS output stream, so it +covers synthesized speech, queued speech and realtime native audio alike. + +Requires `aec-audio-processing` (SWIG binding over libwebrtc-audio-processing-2). +It is NOT a hal dependency — absent or unloadable, every entry point here +degrades to a no-op and the voice path behaves exactly as before. + +Wiring: + tts/service.py _WatchedStream.write → reference_write() + voice_service.py mic open → wrap_mic() +""" + +import logging +import os +import threading +import time +from math import gcd + +logger = logging.getLogger("hal.voice.aec") + +FRAME_MS = 10 # APM's fixed frame size +SUPPORTED_RATES = (8000, 16000, 32000, 48000) + +_lock = threading.Lock() +_canceller = None +_reference = None +_unavailable_logged = False + + +class EchoReference: + """FIFO of audio handed to the speaker, drained by the canceller. + + Bounded to `max_ms`: older audio is past any useful alignment and letting it + pile up drifts the reference out of step with the mic. + """ + + def __init__(self, rate: int, max_ms: int = 500): + self._buffer = bytearray() + self._max_bytes = int(rate * 2 * max_ms / 1000) + self._lock = threading.Lock() + self._last_write = 0.0 + + def write(self, pcm: bytes) -> None: + with self._lock: + self._buffer.extend(pcm) + if len(self._buffer) > self._max_bytes: + del self._buffer[: len(self._buffer) - self._max_bytes] + self._last_write = time.monotonic() + + def read(self, nbytes: int) -> bytes: + """Take the next `nbytes` of played audio, zero-padded if it ran dry.""" + with self._lock: + if len(self._buffer) >= nbytes: + out = bytes(self._buffer[:nbytes]) + del self._buffer[:nbytes] + return out + out = bytes(self._buffer) + b"\x00" * (nbytes - len(self._buffer)) + self._buffer.clear() + return out + + def clear(self) -> None: + with self._lock: + self._buffer.clear() + + def idle_for(self) -> float: + """Seconds since the last speaker write (inf if nothing was ever played).""" + with self._lock: + if self._last_write <= 0.0: + return float("inf") + return time.monotonic() - self._last_write + + +class EchoCanceller: + """Runs mic audio through WebRTC's APM with played audio as the reference. + + `process()` is the only hot path: it buffers to APM's fixed 10 ms frames and + returns exactly as many samples as the caller asked for, so it drops into an + existing read loop without changing framing. + """ + + def __init__(self, rate: int, reference: EchoReference, delay_ms: int, + noise_suppression: bool, dump_dir: str = ""): + self._rate = rate + self._reference = reference + self._frame_bytes = int(rate * FRAME_MS / 1000) * 2 + self._pending = bytearray() + self._out = bytearray() + self._apm = None + self._dump = None + self._erle_log_at = 0.0 + self._erle_mic = 0.0 + self._erle_out = 0.0 + self._erle_frames = 0 + + from aec_audio_processing import AudioProcessor + + self._apm = AudioProcessor( + enable_aec=True, + enable_ns=noise_suppression, + enable_agc=False, # AGC rides the gain up under the bot's own voice + enable_vad=False, # hal runs webrtcvad/silero/ten-vad for that + ) + self._apm.set_stream_format(rate, 1) + self._apm.set_reverse_stream_format(rate, 1) + self._apm.set_stream_delay(delay_ms) + logger.info( + "AEC3 active: %dHz, %dms frames, delay hint %dms, ns=%s", + rate, FRAME_MS, delay_ms, noise_suppression, + ) + if dump_dir: + self._open_dump(dump_dir, rate) + + def reset(self) -> None: + """Drop buffered audio after a route change or stream reopen.""" + self._pending.clear() + self._out.clear() + self._reference.clear() + + def process(self, pcm: bytes) -> bytes: + """Cancel the speaker signal out of `pcm`, returning the same byte count. + + Primes with up to one 10 ms frame of silence on the first call; from then + on input and output stay length-for-length. + """ + want = len(pcm) + self._pending.extend(pcm) + while len(self._pending) >= self._frame_bytes: + mic = bytes(self._pending[: self._frame_bytes]) + del self._pending[: self._frame_bytes] + played = self._reference.read(self._frame_bytes) + self._apm.process_reverse_stream(played) + cleaned = self._apm.process_stream(mic) + self._out.extend(cleaned) + self._accumulate_erle(mic, cleaned, played) + if self._dump: + self._dump["mic"].writeframes(mic) + self._dump["ref"].writeframes(played) + self._dump["out"].writeframes(cleaned) + if len(self._out) < want: + self._out[:0] = b"\x00" * (want - len(self._out)) + out = bytes(self._out[:want]) + del self._out[:want] + return out + + def close(self) -> None: + self._apm = None + if self._dump: + for handle in self._dump.values(): + handle.close() + self._dump = None + + def _accumulate_erle(self, mic: bytes, cleaned: bytes, played: bytes) -> None: + """Log echo return loss enhancement while the speaker is actually active. + + This is the number that says whether the canceller is doing anything: + 0 dB means it is not. + """ + if not any(played): + return + import numpy as np + + m = np.frombuffer(mic, dtype=np.int16).astype(np.float32) + c = np.frombuffer(cleaned, dtype=np.int16).astype(np.float32) + self._erle_mic += float(np.mean(m ** 2)) + self._erle_out += float(np.mean(c ** 2)) + self._erle_frames += 1 + now = time.monotonic() + if self._erle_frames >= 100 and now - self._erle_log_at > 2.0: + if self._erle_out > 0 and self._erle_mic > 0: + import math + + erle = 10 * math.log10(self._erle_mic / self._erle_out) + logger.info("AEC ERLE %.1f dB over %d frames", erle, self._erle_frames) + self._erle_log_at = now + self._erle_mic = self._erle_out = 0.0 + self._erle_frames = 0 + + def _open_dump(self, dump_dir: str, rate: int) -> None: + import wave + + os.makedirs(dump_dir, exist_ok=True) + self._dump = {} + for name in ("mic", "ref", "out"): + handle = wave.open(os.path.join(dump_dir, f"aec_{name}.wav"), "wb") + handle.setnchannels(1) + handle.setsampwidth(2) + handle.setframerate(rate) + self._dump[name] = handle + logger.info("AEC dumping mic/ref/out wavs to %s", dump_dir) + + +class AecStream: + """Mic stream wrapper that cancels echo out of every read. + + Wraps any object with the `sd.InputStream` read contract (also satisfied by + ArecordStream): `read(frames) -> (int16 ndarray, overflowed)`. + """ + + def __init__(self, inner, canceller: EchoCanceller, tail_s: float, np): + self._inner = inner + self._canceller = canceller + self._tail_s = tail_s + self._np = np + self._bypassed = True + + def __enter__(self): + self._inner = self._inner.__enter__() + self._canceller.reset() + return self + + def __exit__(self, *args): + return self._inner.__exit__(*args) + + def read(self, frames): + data, overflowed = self._inner.read(frames) + if overflowed: + return data, overflowed + # Skip the APM entirely when nothing has played recently — echo only + # exists near playback, and this is a 24/7 loop on an 8-core A55. + if self._canceller._reference.idle_for() > self._tail_s: + if not self._bypassed: + self._canceller.reset() + self._bypassed = True + return data, overflowed + self._bypassed = False + try: + cleaned = self._canceller.process(data.tobytes()) + except Exception as e: + logger.warning("AEC process failed, passing mic through: %s", e) + return data, overflowed + return ( + self._np.frombuffer(cleaned, dtype=self._np.int16).reshape(frames, -1), + overflowed, + ) + + def __getattr__(self, name): + return getattr(self._inner, name) + + +def configure(rate: int) -> bool: + """Build the canceller for a mic sample rate. Safe to call repeatedly. + + Returns False when AEC is off, the rate is unsupported, or the native + binding is missing — callers then use the raw mic. + """ + global _canceller, _reference, _unavailable_logged + + from hal.drivers.voice._internal import config as voice_cfg + + if not voice_cfg.AEC_ENABLED: + return False + if rate not in SUPPORTED_RATES: + logger.warning("AEC disabled: %dHz not supported by the APM %s", rate, SUPPORTED_RATES) + return False + with _lock: + if _canceller is not None and _canceller._rate == rate: + return True + if _canceller is not None: + _canceller.close() + _canceller = None + try: + _reference = EchoReference(rate) + _canceller = EchoCanceller( + rate, + _reference, + voice_cfg.AEC_DELAY_MS, + voice_cfg.AEC_NOISE_SUPPRESSION, + voice_cfg.AEC_DUMP_DIR, + ) + except Exception as e: + _reference = None + _canceller = None + if not _unavailable_logged: + _unavailable_logged = True + logger.warning( + "AEC unavailable (%s) — mic runs uncancelled. Install with: " + "uv pip install aec-audio-processing", + e, + ) + return False + return True + + +def wrap_mic(mic_ctx, rate: int, np): + """Return `mic_ctx` wrapped with echo cancellation, or unchanged when off.""" + if not configure(rate): + return mic_ctx + from hal.drivers.voice._internal import config as voice_cfg + + return AecStream(mic_ctx, _canceller, voice_cfg.AEC_TAIL_S, np) + + +def reference_write(samples, rate: int) -> None: + """Record float32 audio on its way to the speaker. Never raises. + + Called from the TTS output stream write, i.e. at playback rate — which is + the timing the mic sees. Tapping at synthesis instead would be wrong: TTS + renders a sentence far faster than real time. + """ + ref = _reference + if ref is None or _canceller is None: + return + try: + import numpy as np + + mono = np.asarray(samples, dtype=np.float32).reshape(-1) + dst = _canceller._rate + if rate != dst: + import scipy.signal + + g = gcd(dst, rate) + mono = scipy.signal.resample_poly(mono, dst // g, rate // g) + pcm16 = (np.clip(mono, -1.0, 1.0) * 32767).astype(np.int16) + ref.write(pcm16.tobytes()) + except Exception as e: + logger.debug("AEC reference write skipped: %s", e) + + +def reset() -> None: + """Drop canceller state — call after an audio route change.""" + if _canceller is not None: + _canceller.reset() diff --git a/hal/drivers/voice/tts/service.py b/hal/drivers/voice/tts/service.py index 7b4098d34..0a71c251b 100644 --- a/hal/drivers/voice/tts/service.py +++ b/hal/drivers/voice/tts/service.py @@ -20,6 +20,7 @@ import numpy as np +from hal.drivers.voice import aec from hal.drivers.voice.tts.backend import ( TTSBackend, TTS_SAMPLE_RATE, @@ -79,6 +80,10 @@ def __init__(self, stream, owner): def write(self, data): self._owner._write_started_ts = time.monotonic() + # Echo reference, tapped at playback rate — the timing the mic sees. + # Covers synthesized speech, the queue drain and realtime native audio, + # since all three reach the device through this one stream. + aec.reference_write(data, self._owner._stream_rate) try: return self._stream.write(data) finally: diff --git a/hal/drivers/voice/voice_service.py b/hal/drivers/voice/voice_service.py index 0eca57bce..26f828a8d 100644 --- a/hal/drivers/voice/voice_service.py +++ b/hal/drivers/voice/voice_service.py @@ -51,6 +51,7 @@ from hal.drivers.voice._internal.turn_dispatch import dispatch_turn from hal.drivers.voice._internal.vad_filters import SileroVADFilter, WebRTCVADFilter from hal.drivers.voice._internal.wakeword_focus import WakeWordFocus +from hal.drivers.voice import aec from hal.drivers.voice.backchannel import Backchannel from hal.drivers.voice.stt import STTProvider @@ -657,6 +658,9 @@ def _monitor_barge_in(self) -> bool: blocksize=frame_size, device=self._input_device, ) + # The one place the mic is already open while the speaker plays, so + # the only place AEC can be measured before full duplex exists. + mic_ctx = aec.wrap_mic(mic_ctx, device_rate, np) with mic_ctx as mic: while self._running and self._tts_is_speaking(): data, overflowed = mic.read(frame_size) @@ -752,6 +756,7 @@ def _loop(self): blocksize=frame_size, device=self._input_device, ) + mic_ctx = aec.wrap_mic(mic_ctx, device_rate, self._np) with mic_ctx as mic: logger.info( "Listening for speech (RMS=%d, rate=%dHz, backend=%s)...", diff --git a/hal/pipecat_rt.py b/hal/pipecat_rt.py new file mode 100644 index 000000000..8cf189632 --- /dev/null +++ b/hal/pipecat_rt.py @@ -0,0 +1,662 @@ +"""Pipecat-based realtime voice agent — self-contained, cascaded (text in / text out). + +Standalone by design: it shares nothing with `hal/realtime/` and imports none of +its provider machinery. ASR and TTS stay with the existing hal stack, so this +module only owns the middle of the turn: + + transcript ──▶ LLMContext ──▶ LLM (OpenAI-compatible) ──▶ text chunks + └─ tools ──▶ handled here, or raised to the host + +Pipecat runs on a private asyncio loop in a daemon thread; the public API is +sync and queue-based so a caller on hal's voice thread never touches the loop. + + agent = PipecatAgent(PipecatConfig(), instructions=..., host_tools=[...]) + agent.start() + for event in agent.run_turn("what time is it"): + ... + agent.stop() + +Requires `pipecat-ai[openai]` and `httpx[http2]`; neither is a hal dependency +yet. `start()` returns False when they are missing and `available` stays False. +""" + +import asyncio +import json +import logging +import os +import queue +import threading +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Callable, Generator, Iterable + +logger = logging.getLogger("hal.pipecat") + +DONE = object() # sentinel on the event queue: this turn produced its last frame + + +# --- Configuration --------------------------------------------------------- + + +@dataclass +class PipecatConfig: + base_url: str = os.environ.get("HAL_PIPECAT_BASE_URL", "") + api_key: str = os.environ.get("HAL_PIPECAT_API_KEY", "not-needed") + model: str = os.environ.get("HAL_PIPECAT_MODEL", "") + # Caps generated tokens, which INCLUDES tool-call arguments — a small budget + # truncates the JSON and the turn is lost with a parse error, not a retry. + max_tokens: int = int(os.environ.get("HAL_PIPECAT_MAX_TOKENS", "512")) + temperature: float = float(os.environ.get("HAL_PIPECAT_TEMPERATURE", "0.6")) + http2: bool = os.environ.get("HAL_PIPECAT_HTTP2", "true").lower() == "true" + max_retries: int = int(os.environ.get("HAL_PIPECAT_MAX_RETRIES", "2")) + warm_timeout_s: float = float(os.environ.get("HAL_PIPECAT_WARM_TIMEOUT_S", "5")) + turn_timeout_s: float = float(os.environ.get("HAL_PIPECAT_TURN_TIMEOUT_S", "20")) + tool_result_timeout_s: float = float(os.environ.get("HAL_PIPECAT_TOOL_TIMEOUT_S", "4")) + # In-session history kept in the message list. The durable memory lives in + # the system prompt, which is rebuilt only on recycle — appending here keeps + # the cached prefix intact, re-folding memory into the prompt would not. + max_history_messages: int = int(os.environ.get("HAL_PIPECAT_MAX_HISTORY", "128")) + search_budget_per_turn: int = int(os.environ.get("HAL_PIPECAT_SEARCH_BUDGET", "2")) + gemini_api_key: str = os.environ.get("HAL_PIPECAT_GEMINI_KEY", "") + gemini_search_model: str = os.environ.get( + "HAL_PIPECAT_GEMINI_SEARCH_MODEL", "gemini-3.5-flash-lite" + ) + search_timeout_s: float = float(os.environ.get("HAL_PIPECAT_SEARCH_TIMEOUT_S", "6")) + + +# --- Turn events ----------------------------------------------------------- + + +@dataclass +class TextChunk: + text: str + + +@dataclass +class ToolCall: + """A tool the host must run (device control, delegation, camera).""" + + name: str + arguments: str + call_id: str + + +@dataclass +class TurnMetrics: + """Where a turn's latency went. `server` bundles network RTT + prefill.""" + + wait: float = 0.0 + server: float = 0.0 + stream: float = 0.0 + ttft: float = 0.0 + prompt_chars: int = 0 + + def __str__(self) -> str: + return ( + f"ttft {self.ttft:.2f}s | wait {self.wait:.2f}s | server {self.server:.2f}s " + f"| stream {self.stream:.2f}s | prompt {self.prompt_chars / 1000:.1f}k chars" + ) + + +@dataclass +class _TurnState: + """Per-turn scratch space. + + Reset in place, never replaced: the LLM service and the collector capture + this object when the pipeline is built, so a fresh instance would strand + their writes on an orphan and the turn would never see its own end frame. + """ + + started: float = 0.0 + request_sent: float = 0.0 + headers: float = 0.0 + first_token: float = 0.0 + prompt_chars: int = 0 + searches: int = 0 + pending: dict = field(default_factory=dict) + calls_started: int = 0 + response_ended: bool = False + + def reset(self, started: float) -> None: + self.started = started + self.request_sent = self.headers = self.first_token = 0.0 + self.prompt_chars = self.searches = self.calls_started = 0 + self.response_ended = False + self.pending.clear() + + +# --- Agent ----------------------------------------------------------------- + + +class PipecatAgent: + """One cascaded voice-agent session. + + Args: + config: endpoint, budgets and timeouts. + instructions: the system prompt — the stable, cache-friendly prefix. + host_tools: tools the HOST executes. Each is + `(name, description, parameters_json_schema)`; calls are yielded as + `ToolCall` and must be answered with `tool_result()`. + """ + + def __init__( + self, + config: PipecatConfig, + instructions: str = "", + host_tools: Iterable[tuple[str, str, dict]] = (), + ) -> None: + self._config = config + self._instructions = instructions + self._host_tools = list(host_tools) + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._worker = None + self._context = None + self._llm = None + self._events: queue.Queue = queue.Queue() + self._turn = _TurnState() + self._turn_lock = threading.Lock() + self._ready = threading.Event() + self._stopping = threading.Event() + self._session_id = str(uuid.uuid4()) + self._search_client = None + self.last_metrics: TurnMetrics | None = None + + # --- Lifecycle --- + + @property + def available(self) -> bool: + return self._ready.is_set() and not self._stopping.is_set() + + def start(self) -> bool: + """Build the pipeline and warm the prompt cache. False when unusable.""" + if not self._config.base_url or not self._config.model: + logger.warning("[pipecat] base_url/model not configured — disabled") + return False + try: + import pipecat # noqa: F401 + except ImportError as e: + logger.warning("[pipecat] not installed (%s) — disabled", e) + return False + + self._stopping.clear() + started = threading.Event() + self._thread = threading.Thread( + target=self._run_loop, args=(started,), daemon=True, name="pipecat-loop" + ) + self._thread.start() + started.wait(timeout=30) + return self.available + + def stop(self) -> None: + self._stopping.set() + self._ready.clear() + loop = self._loop + if loop is None: + return + try: + asyncio.run_coroutine_threadsafe(self._shutdown(), loop).result(timeout=5) + except Exception: + logger.exception("[pipecat] shutdown failed") + # Cancel pipecat's own tasks and let them unwind before stopping the + # loop; tearing it down under them logs a wall of "Task was destroyed" + # and leaks whatever they were holding. + try: + asyncio.run_coroutine_threadsafe(_cancel_tasks(), loop).result(timeout=5) + except Exception: + logger.debug("[pipecat] task cancellation incomplete") + loop.call_soon_threadsafe(loop.stop) + if self._thread is not None: + self._thread.join(timeout=5) + self._thread = None + + def _run_loop(self, started: threading.Event) -> None: + loop = asyncio.new_event_loop() + self._loop = loop + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(self._build()) + self._ready.set() + except Exception: + logger.exception("[pipecat] pipeline build failed") + finally: + started.set() + try: + loop.run_forever() + finally: + loop.close() + asyncio.set_event_loop(None) + + async def _build(self) -> None: + from pipecat.pipeline.pipeline import Pipeline + from pipecat.pipeline.worker import PipelineParams, PipelineWorker + from pipecat.processors.aggregators.llm_context import LLMContext + from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, + ) + from pipecat.workers.runner import WorkerRunner + + self._llm = _build_llm(self._config, self._turn) + self._context = LLMContext( + [{"role": "system", "content": self._instructions}], + tools=self._build_tools(), + ) + aggregators = LLMContextAggregatorPair(self._context) + self._worker = PipelineWorker( + Pipeline( + [ + aggregators.user(), + self._llm, + _Collector(self._events, self._turn), + aggregators.assistant(), + ] + ), + params=PipelineParams(enable_metrics=True), + idle_timeout_secs=None, + conversation_id=self._session_id, + ) + asyncio.create_task(WorkerRunner(handle_sigint=False).run(self._worker)) + await self._warm() + + async def _shutdown(self) -> None: + from pipecat.frames.frames import EndFrame + + if self._worker is not None: + await self._worker.queue_frame(EndFrame()) + if self._search_client is not None: + await self._search_client.aclose() + self._search_client = None + + # --- Turns --- + + def run_turn(self, transcript: str, image=None) -> Generator[Any, None, None]: + """Run one turn, yielding TextChunk / ToolCall until the model is done. + + Args: + transcript: the user's final STT text — this agent never sees audio. + image: optional BGR frame added to this turn only. + """ + if not self.available: + return + with self._turn_lock: + self._drain() + self._turn.reset(time.perf_counter()) + deadline = time.monotonic() + self._config.turn_timeout_s + self._submit(self._queue_turn(transcript, image)) + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + logger.warning("[pipecat] turn timed out with no completion") + break + try: + event = self._events.get(timeout=min(remaining, 1.0)) + except queue.Empty: + continue + if event is DONE: + break + yield event + self.last_metrics = self._metrics() + logger.info("[pipecat] %s", self.last_metrics) + self._trim_history() + + def tool_result(self, call_id: str, output: str, run_llm: bool = True) -> None: + """Answer a ToolCall. `run_llm=False` records it without a new inference.""" + future = self._turn.pending.pop(call_id, None) + if future is None or self._loop is None: + return + self._loop.call_soon_threadsafe( + lambda: future.done() or future.set_result((output, run_llm)) + ) + + def _settle_call(self, run_llm: bool) -> None: + """Close the turn once a tool call has been answered, if nothing follows. + + The response's end frame arrives while the handler is still waiting on + the host, so the decision can only be made here: a result that triggers + a new inference means more output is coming; one that does not means + the turn ended with that call. Runs on the loop thread, same as the + collector, so the counters need no lock. + """ + turn = self._turn + turn.calls_started = max(0, turn.calls_started - 1) + if run_llm: + turn.response_ended = False + elif turn.calls_started == 0 and turn.response_ended: + turn.response_ended = False + self._events.put(DONE) + + def abort_turn(self) -> None: + """Release anything awaiting a tool result and end the turn now.""" + for call_id in list(self._turn.pending): + self.tool_result(call_id, '{"result": "aborted"}', run_llm=False) + self._events.put(DONE) + + def add_context(self, text: str) -> None: + """Append a one-off context line (time, speaker) ahead of the next turn.""" + if self._context is not None and text: + self._context.add_message({"role": "user", "content": text}) + + def rebuild_context(self, instructions: str) -> None: + """Recycle: swap the system prompt and drop in-session history. + + Cheap here — context is client-side, so there is no reconnect. Durable + memory rides in `instructions`. + """ + self._instructions = instructions + if self._context is not None: + self._context.set_messages([{"role": "system", "content": instructions}]) + + async def _queue_turn(self, transcript: str, image) -> None: + from pipecat.frames.frames import TranscriptionFrame + + if image is not None: + self._add_image(image) + await self._worker.queue_frame( + TranscriptionFrame(transcript, "user", str(time.time())) + ) + + def _add_image(self, image) -> None: + """Write the frame straight into the context. + + Deliberately not a UserImageRawFrame: the aggregator answers that with + its own inference, which makes every turn reply twice. + """ + try: + import cv2 + + ok, buf = cv2.imencode(".jpg", image) + if ok: + self._context.add_image_frame_message( + format="JPEG", size=(image.shape[1], image.shape[0]), + image=buf.tobytes(), + ) + except Exception as e: + logger.warning("[pipecat] image attach failed: %s", e) + + # --- Tools --- + + def _build_tools(self): + from pipecat.adapters.schemas.function_schema import FunctionSchema + from pipecat.adapters.schemas.tools_schema import ToolsSchema + + schemas = [ + FunctionSchema( + name=name, + description=description, + properties=params.get("properties", {}), + required=params.get("required", []), + handler=self._host_handler(name), + ) + for name, description, params in self._host_tools + ] + if self._config.gemini_api_key: + schemas.append( + FunctionSchema( + name="web_search", + description=( + "Real-time access to current information on the web. Use it for " + "anything after your training cutoff, anything happening now, and " + "any fact you are not certain of — news, results, prices, weather, " + "schedules. You DO have this access: never reply that you cannot " + "provide real-time information." + ), + properties={"query": {"type": "string", "description": "What to search for."}}, + required=["query"], + handler=self._search_handler, + ) + ) + return ToolsSchema(standard_tools=schemas) if schemas else None + + def _host_handler(self, name: str) -> Callable: + """Raise the call to the host and wait for `tool_result()`.""" + + async def handler(params) -> None: + from pipecat.frames.frames import FunctionCallResultProperties + + future = asyncio.get_running_loop().create_future() + self._turn.pending[params.tool_call_id] = future + self._events.put( + ToolCall( + name=name, + arguments=json.dumps(params.arguments or {}), + call_id=params.tool_call_id, + ) + ) + try: + output, run_llm = await asyncio.wait_for( + future, timeout=self._config.tool_result_timeout_s + ) + except asyncio.TimeoutError: + self._turn.pending.pop(params.tool_call_id, None) + output, run_llm = '{"error": "host did not answer"}', False + await params.result_callback( + output, properties=FunctionCallResultProperties(run_llm=run_llm) + ) + self._settle_call(run_llm) + + return handler + + async def _search_handler(self, params) -> None: + """Gemini grounding search, capped per turn. + + The cap is not optional: uncapped, a model that cannot find a stated + answer re-queries until it gives up — seconds of dead air per turn. + """ + query = (params.arguments or {}).get("query", "") + if self._turn.searches >= self._config.search_budget_per_turn: + await params.result_callback( + {"error": "search budget for this turn is spent; answer with what you have"} + ) + self._settle_call(True) + return + self._turn.searches += 1 + started = time.perf_counter() + try: + answer = await self._gemini_search(query) + except Exception as e: + logger.warning("[pipecat] web_search failed: %s", e) + await params.result_callback({"error": "search unavailable"}) + self._settle_call(True) + return + logger.info( + "[pipecat] web_search %.2fs query=%r -> %r", + time.perf_counter() - started, query[:60], answer[:80], + ) + await params.result_callback({"answer": answer}) + self._settle_call(True) + + async def _gemini_search(self, query: str) -> str: + import httpx + + if self._search_client is None: + self._search_client = httpx.AsyncClient(timeout=self._config.search_timeout_s) + url = ( + "https://generativelanguage.googleapis.com/v1beta/models/" + f"{self._config.gemini_search_model}:generateContent" + ) + payload = { + "contents": [{"role": "user", "parts": [{"text": query}]}], + "tools": [{"google_search": {}}], + "systemInstruction": { + "parts": [{ + "text": "Answer in ONE short factual sentence. No markdown, no lists, " + "no citations, no preamble. If unknown, say so in one sentence.", + }] + }, + "generationConfig": {"maxOutputTokens": 120, "thinkingLevel": "minimal"}, + } + response = await self._search_client.post( + url, json=payload, headers={"x-goog-api-key": self._config.gemini_api_key} + ) + response.raise_for_status() + parts = response.json()["candidates"][0]["content"]["parts"] + return " ".join(p.get("text", "") for p in parts).strip() + + # --- Internals --- + + def _submit(self, coro) -> None: + if self._loop is not None: + asyncio.run_coroutine_threadsafe(coro, self._loop) + + def _drain(self) -> None: + """Clear anything a previous turn left behind, so a turn reads only its own.""" + dropped = 0 + while True: + try: + self._events.get_nowait() + except queue.Empty: + break + dropped += 1 + if dropped: + logger.info("[pipecat] dropped %d stale event(s)", dropped) + + def _metrics(self) -> TurnMetrics: + t = self._turn + return TurnMetrics( + wait=(t.request_sent - t.started) if t.request_sent else 0.0, + server=(t.headers - t.request_sent) if t.headers and t.request_sent else 0.0, + stream=(t.first_token - t.headers) if t.first_token and t.headers else 0.0, + ttft=(t.first_token - t.started) if t.first_token else 0.0, + prompt_chars=t.prompt_chars, + ) + + def _trim_history(self) -> None: + """Keep the system prompt, drop the oldest turns past the window.""" + if self._context is None: + return + messages = self._context.get_messages() + limit = self._config.max_history_messages + if len(messages) <= limit + 1: + return + self._context.set_messages([messages[0]] + messages[-limit:]) + + async def _warm(self) -> None: + """Prefill the shared prefix once so the first real turn does not. + + Worth it here because the prompt is large (identity + skills + memory); + on a small prefix it measures as noise. Time-boxed — a busy endpoint has + blocked this for 13 s. + """ + try: + invocation = self._llm.get_llm_adapter().get_llm_invocation_params( + self._context, convert_developer_to_user=True + ) + params = { + "model": self._config.model, + "messages": self._context.get_messages() + [{"role": "user", "content": "hi"}], + "max_tokens": 2, + "stream": False, + } + if invocation.get("tools"): + params["tools"] = invocation["tools"] + started = time.perf_counter() + await asyncio.wait_for( + self._llm._client.chat.completions.create(**params), + timeout=self._config.warm_timeout_s, + ) + logger.info("[pipecat] context warmed in %.2fs", time.perf_counter() - started) + except asyncio.TimeoutError: + logger.warning("[pipecat] context warm-up timed out — starting anyway") + except Exception as e: + logger.warning("[pipecat] context warm-up skipped: %s", e) + + +# --- Pipeline pieces ------------------------------------------------------- + + +async def _cancel_tasks() -> None: + current = asyncio.current_task() + pending = [t for t in asyncio.all_tasks() if t is not current] + for task in pending: + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + +def _build_llm(config: PipecatConfig, turn: _TurnState): + from pipecat.services.openai.llm import OpenAILLMService + + class _TimedLLM(OpenAILLMService): + """Adds HTTP/2, retries and request timing to pipecat's OpenAI client. + + Pipecat builds the client HTTP/1.1-only with retries off; h2 measured + 413 ms median TTFT against 642 ms, and the endpoint returns transient + 500s that a retry absorbs. + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._turn = turn + + def create_client(self, api_key=None, base_url=None, **kwargs): + import httpx + from openai import AsyncOpenAI, DefaultAsyncHttpxClient + + limits = httpx.Limits( + max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None + ) + try: + http_client = DefaultAsyncHttpxClient(http2=config.http2, limits=limits) + except ImportError: + logger.warning('[pipecat] HTTP/2 needs httpx[http2]; using HTTP/1.1') + http_client = DefaultAsyncHttpxClient(limits=limits) + client = AsyncOpenAI( + api_key=api_key, base_url=base_url, http_client=http_client + ) + return client.with_options(max_retries=config.max_retries) + + async def get_chat_completions(self, context): + self._turn.prompt_chars = sum( + len(str(m.get("content", ""))) for m in context.get_messages() + ) + self._turn.request_sent = time.perf_counter() + stream = await super().get_chat_completions(context) + self._turn.headers = time.perf_counter() + return stream + + return _TimedLLM( + api_key=config.api_key, + base_url=config.base_url, + settings=OpenAILLMService.Settings( + model=config.model, + max_tokens=config.max_tokens, + temperature=config.temperature, + ), + ) + + +class _Collector: + """Bridges pipeline frames onto the sync event queue.""" + + def __new__(cls, events: queue.Queue, turn: _TurnState): + from pipecat.frames.frames import ( + ErrorFrame, + FunctionCallsStartedFrame, + LLMFullResponseEndFrame, + TextFrame, + ) + from pipecat.processors.frame_processor import FrameProcessor + + class Collector(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame) and frame.text: + if not turn.first_token: + turn.first_token = time.perf_counter() + events.put(TextChunk(frame.text)) + elif isinstance(frame, FunctionCallsStartedFrame): + turn.calls_started += len(frame.function_calls or [None]) + elif isinstance(frame, ErrorFrame): + logger.warning("[pipecat] %s", frame) + events.put(DONE) + elif isinstance(frame, LLMFullResponseEndFrame): + # A response that raised tool calls is not the end of the + # turn — the results decide. See PipecatAgent._settle_call. + if turn.calls_started: + turn.response_ended = True + else: + events.put(DONE) + await self.push_frame(frame, direction) + + return Collector() From e6a3fb26ebd808f0138dd5a1c858bc3062b757b3 Mon Sep 17 00:00:00 2001 From: 20e4e8cf72f6d639 <26b5f3883ed5f10f> Date: Fri, 21 Aug 2026 17:01:58 +0700 Subject: [PATCH 2/2] wip --- docs/realtime-voice.md | 122 ++++++++++- docs/vi/realtime-voice_vi.md | 122 ++++++++++- hal/config.py | 28 ++- hal/drivers/voice/_internal/pipecat_turn.py | 201 ++++++++++++++++++ hal/drivers/voice/voice_service.py | 32 ++- hal/pipecat_rt.py | 173 ++++++++++++++- hal/realtime/pipecat_session.py | 178 ++++++++++++++++ system/device/realtime.go | 33 ++- system/domain/device.go | 2 +- system/server/config/realtime.go | 75 +++++-- .../src/pages/settings/RealtimeSection.tsx | 27 ++- .../web/src/pages/settings/SettingsPanel.tsx | 10 +- 12 files changed, 942 insertions(+), 61 deletions(-) create mode 100644 hal/drivers/voice/_internal/pipecat_turn.py create mode 100644 hal/realtime/pipecat_session.py diff --git a/docs/realtime-voice.md b/docs/realtime-voice.md index c2b343b5c..cb547fde1 100644 --- a/docs/realtime-voice.md +++ b/docs/realtime-voice.md @@ -405,14 +405,19 @@ snapshots normally. ## Providers -Three interchangeable backends, selected by `HAL_REALTIME_PROVIDER` / -`realtime.provider` (`none` | `gemini` | `openai` | `qwen`): +Four backends, selected by `HAL_REALTIME_PROVIDER` / `realtime.provider` +(`none` | `gemini` | `openai` | `qwen` | `pipecat`). The first three are +**audio-native** — they take microphone frames and return speech. `pipecat` is +**cascaded**: HAL keeps its own STT and TTS and pipecat drives only the middle of +the turn. It shares nothing with `voice_agent/` and is not a `VoiceAgentBase`; +see [Pipecat](#pipecat--the-cascaded-provider) below. | Provider | Class | Threading model | Default model | Sample rate | |----------|-------|-----------------|---------------|-------------| | Gemini Live | `voice_agent/gemini_live.py` `GeminiLiveAgent` | private asyncio loop on a `gemini-io` thread; send/recv threads submit coroutines via `run_coroutine_threadsafe` | `gemini-2.5-flash-native-audio-preview-12-2025` | 16000 Hz | | OpenAI Realtime | `voice_agent/openai_realtime.py` `OpenAIRealtimeAgent` | fully synchronous; one `RealtimeConnection` shared by send/recv threads, serialized by a reentrant lock | `gpt-realtime-2` | 24000 Hz | | Qwen Omni Realtime | `voice_agent/qwen_realtime.py` `QwenRealtimeAgent` | fully synchronous; raw `websockets.sync.client` socket shared by send/recv threads, reusing the openai_realtime thread/queue skeleton | `qwen3.5-omni-plus-realtime` | 16000 Hz in / 24000 Hz out | +| Pipecat (cascaded) | `pipecat_session.py` `PipecatSession` over `hal/pipecat_rt.py` `PipecatAgent` | private asyncio loop on a `pipecat-loop` thread; the public API is sync and queue-based | `llm_model` (no default of its own) | n/a — text in, text out | Gemini Live uses `google-genai` and keeps its private asyncio loop owned by its `gemini-io` thread. Teardown first closes/cancels the provider receive task, @@ -491,6 +496,89 @@ queue-based contract: so a silent/no-response turn falls back to the main agent without long dead-air). - `available` ⇔ the websocket/session is connected (`_connected`). +### Pipecat — the cascaded provider + +Selected with `realtime.provider = "pipecat"`. The pipeline reduces to + +``` +transcript ──▶ LLMContext ──▶ LLM (OpenAI-compatible) ──▶ text chunks + └─ tools ──▶ handled here, or raised to the host +``` + +so the audio half of the turn never changes: the same STT produces the +transcript and the same TTS speaks the reply, sentence by sentence. The turn +driver is `drivers/voice/_internal/pipecat_turn.py` (`run_pipecat_turn`), the +cascaded twin of `realtime_turn.py`. It returns the same `RealtimeTurnResult`, +and it shares the wait filler, the thinking cue and the CoT leak filter with the +audio-native driver, so both brains feel identical to the user. + +What it does **not** have, by construction: + +- **No native audio** — the model's voice is never played; the device speaks. +- **No speech emotion or tone recognition** — the model sees text, not audio. + This is the deliberate trade for latency; `express_emotion` still works, + driven by the reply's content rather than the speaker's voice. +- **No `look` / in-session vision** and no look-replay — visual questions + delegate to the main agent. +- **No WS-recovery retry** — there is no long-lived session to lose. +- **No voice or reasoning knob** — the device TTS owns the voice and the model + exposes no reasoning tier, so `ValidateRealtimeKnobs` rejects both. + +`PipecatSession` duck-types the slice of `RealtimeOrchestrator` that +`voice_service` touches during capture (`available`, `prepare_turn`, +`append_audio`, `send_text`, `sample_rate`, `rebuilding`, +`wait_until_available`, `save_turn`), so only the turn call branches. +`append_audio` is a documented no-op: this brain is fed the finished transcript. + +Persona and memory come from the **same** gateway-keyed `ContextManagerBase` the +audio-native providers use, so identity, `summary.md` and the skills catalog are +identical. Instructions are rebuilt every `HAL_REALTIME_SESSION_MAX_TURNS` turns +so a rename or new memory lands; between rebuilds the prompt is byte-stable, +which is what keeps the gateway's prefix cache warm. + +Tools: `delegate_to_main` (answered with `run_llm=False` — the turn is over, the +main agent takes it), `express_emotion` (fire-and-forget on a daemon thread, +answered with `run_llm=True` so the model still owes the user words), and an +optional Gemini-grounded `web_search` when `realtime.pipecat.search_api_key` is +set, budgeted per turn by `HAL_PIPECAT_SEARCH_BUDGET`. + +**Per-turn metrics.** Each turn logs one line at INFO: + +``` +[pipecat] ttft 0.62s | wait 0.00s | server 0.61s | stream 0.01s | + prompt 38.3k chars/10866 tok | out 11 tok | payload 40.8 KB +``` + +| Field | Meaning | +|-------|---------| +| `ttft` | transcript queued → first text frame out | +| `wait` | time inside the pipeline before the request went out (aggregation) | +| `server` | request sent → response headers: network RTT + prefill | +| `stream` | headers → first token | +| `prompt` | context text size, and prompt tokens the server billed | +| `out` | completion tokens, with `(N reasoning)` when reported | +| `images` | image parts in the context (omitted when none) | +| `payload` | bytes actually written on the wire, measured by an httpx request hook | +| `N requests` | shown only when a turn made more than one call | + +Token counts and payload bytes are **turn totals, not per-request**: answering a +tool call costs a second request that re-sends the whole prompt, so an +`express_emotion` turn roughly doubles both. That is the visible price of the +tool, and the reason the counters accumulate rather than overwrite. + +`cached` (prefix-cache hits) appears only when the endpoint returns +`prompt_tokens_details.cached_tokens`, and `reasoning` only with +`completion_tokens_details.reasoning_tokens`. Both are mapped by pipecat; a +gateway that omits them simply leaves the field out of the line. + +**Dependency note.** `pipecat-ai` is not a HAL dependency and pins +`onnxruntime~=1.24.3`, which would downgrade the 1.27.x the device uses for +TEN-VAD and insightface. Install it with `--no-deps` plus the modules pipecat +imports at package load (`loguru`, `pyloudnorm`, `soxr`, `resampy`, `markdown`, +`nltk`, `num2words`, `defusedxml`, `docopt`) so onnxruntime is left alone. +`PipecatAgent.start()` returns `False` when the package is missing, and the +provider then falls back to the main agent instead of failing the turn. + ### OpenAI connection safety The OpenAI agent shares a single `RealtimeConnection` between its send and recv @@ -763,8 +851,8 @@ spurious HAL restart on the next boot after an os-server-only field changes. Modelled in Go at `system/server/config/realtime.go`; read in HAL at `hal/config.py`. Shared fields sit at the top; per-provider knobs live in -`gemini` / `openai` / `qwen` sub-objects, with `provider` selecting the active -one (`none` or absent → realtime off). Empty `api_key` / `base_url` fall back to +`gemini` / `openai` / `qwen` / `pipecat` sub-objects, with `provider` selecting +the active one (`none` or absent → realtime off). Empty `api_key` / `base_url` fall back to `llm_api_key` / `llm_base_url` — **except qwen**: its credentials are its own (`realtime.qwen.api_key` / `realtime.qwen.base_url`, Go struct `QwenRealtime`), with deliberately **no fallback** to the shared `realtime.api_key`/`base_url` or @@ -773,6 +861,14 @@ through the `campaign-api` proxy. Set them via `realtime.qwen.*` in config.json or via env on the device (`DASHSCOPE_API_KEY`, `HAL_QWEN_REALTIME_BASE_URL` in `/opt/hal/.env`); with neither set the WS handshake fails loudly in the hal log. +`pipecat` likewise keeps its own credentials in `realtime.pipecat.*` (Go struct +`PipecatRealtime`) and ignores the shared `realtime.api_key`/`base_url`, which +carry the WS-suffixed proxy values. Its endpoint is a plain OpenAI-compatible +`/v1` host, so when a field is empty it falls back to the **AI brain's** +(`llm_base_url` / `llm_api_key` / `llm_model`) — already such a host. With no +base_url or model resolvable, `PipecatSession.start()` logs and stays +unavailable, and turns fall through to the main agent. + > **Leave `base_url` blank unless you have a non-proxy endpoint.** (Applies to > gemini/openai; qwen never derives from `llm_base_url` — it uses its own > `realtime.qwen.base_url`.) When empty, HAL @@ -791,7 +887,8 @@ or via env on the device (`DASHSCOPE_API_KEY`, `HAL_QWEN_REALTIME_BASE_URL` in "provider": "gemini", "gemini": { "model": "gemini-3.1-flash-live-preview", "voice": "Kore", "thinking_level": "MINIMAL" }, "openai": { "model": "gpt-realtime-2", "voice": "alloy", "reasoning_effort": "minimal" }, - "qwen": { "model": "qwen3.5-omni-plus-realtime", "voice": "Ethan", "api_key": "sk-…", "base_url": "wss://…" } + "qwen": { "model": "qwen3.5-omni-plus-realtime", "voice": "Ethan", "api_key": "sk-…", "base_url": "wss://…" }, + "pipecat": { "model": "qwen/qwen3.6-35b-a3b", "base_url": "https://…/v1", "api_key": "…", "search_api_key": "AIza…" } } } ``` @@ -847,7 +944,7 @@ is a top-level `config.json` flag: | `HAL_WAKEWORD_FOLLOWUP_TIMEOUT_S` | `20` | Idle seconds for the short post-command focus window. Each accepted `voice_command` or `voice_followup` refreshes it. `0` disables follow-ups and requires a wake phrase for every mic session. Ignored when `wakeword` is false. | | `HAL_SILENCE_VAD_ENABLED` | `true` | Require Silero to confirm speech before the end-of-turn silence clock is refreshed. RMS remains the cheap pre-gate; set `false` to fall back to pure-RMS silence detection. | | `HAL_SILENCE_VAD_WINDOW_FRAMES` | `3` | Number of frames batched per Silero run for that check — Silero costs ~20 ms/frame on ARM and its LSTM needs more than one 64 ms frame to settle. | -| `HAL_REALTIME_PROVIDER` | `gemini` | `none` \| `gemini` \| `openai` \| `qwen` | +| `HAL_REALTIME_PROVIDER` | `gemini` | `none` \| `gemini` \| `openai` \| `qwen` \| `pipecat` | | `HAL_REALTIME_TURN_DETECTION` | `off` | `server_vad` \| `semantic_vad` \| `off` (Gemini: off = manual activity detection) | | `HAL_REALTIME_RECV_QUEUE_TIMEOUT_S` | `8.0` | Max seconds `receive()` waits for the next output event before ending a silent turn (fallback to main agent) | | `HAL_REALTIME_LOOK_RECV_TIMEOUT_S` | `20.0` | Silent-turn watchdog used instead of the default for turns where a `look` fired (per-turn, via `extend_recv_timeout()`). Gemini's forced thinking over a text-dense frame can stay silent >8 s right before the answer — the default watchdog was killing those turns | @@ -876,6 +973,16 @@ is a top-level `config.json` flag: | `HAL_QWEN_REALTIME_BASE_URL` | — | DashScope workspace WS host; overrides `realtime.qwen.base_url`. Never derived from `llm_base_url` | | `HAL_QWEN_REALTIME_MODEL` | `qwen3.5-omni-plus-realtime` | turbo is legacy: no function calls, ignores turn context | | `HAL_QWEN_REALTIME_VOICE` | `Ethan` | 3.5-plus: also `Serena`; turbo-only: `Cherry` \| `Chelsie` | +| `HAL_PIPECAT_BASE_URL` | `llm_base_url` | OpenAI-compatible `/v1` host; overrides `realtime.pipecat.base_url` | +| `HAL_PIPECAT_API_KEY` | `llm_api_key` | Overrides `realtime.pipecat.api_key` | +| `HAL_PIPECAT_MODEL` | `llm_model` | Overrides `realtime.pipecat.model` | +| `HAL_PIPECAT_GEMINI_KEY` | — | Enables the `web_search` tool (Gemini grounding). Unset → no search tool is offered | +| `HAL_PIPECAT_SEARCH_BUDGET` | `2` | Max `web_search` calls per turn | +| `HAL_PIPECAT_MAX_TOKENS` | `512` | Caps generated tokens — **includes tool-call arguments**, so a small budget truncates the JSON and loses the turn | +| `HAL_PIPECAT_MAX_HISTORY` | `128` | In-session messages kept before trimming; durable memory rides in the system prompt | +| `HAL_PIPECAT_HTTP2` | `true` | HTTP/2 to the gateway | +| `HAL_PIPECAT_TURN_TIMEOUT_S` | `20` | Per-turn watchdog | +| `HAL_PIPECAT_LOG_LEVEL` | `WARNING` | pipecat logs through loguru at DEBUG by default, which floods the device journal with per-frame lines | | `HAL_REALTIME_MEMORY_PATH` | `/realtime/memory.jsonl` | | | `HAL_REALTIME_MAX_MEMORY_ENTRIES` / `_TRIM_KEEP` | `1000` / `500` | | | `HAL_REALTIME_SUMMARIZER_ENABLED` | `true` | | @@ -896,4 +1003,7 @@ is a top-level `config.json` flag: | `models/`, `enums/` | Input/output/event types, provider + gateway enums | | `resources/` | System prompts (shared + per-provider) | | `../voice/voice_service.py` | Integration: streams mic audio, consumes output, routes delegate/handled | +| `pipecat_session.py` | Cascaded brain: owns the `PipecatAgent`, reuses the context manager, duck-types the capture surface | +| `../../pipecat_rt.py` | Standalone pipecat engine (`PipecatAgent`): pipeline build, tool bridging, per-turn metrics. Runnable on its own: `python -m hal.pipecat_rt` | +| `../voice/_internal/pipecat_turn.py` | Cascaded turn driver (`run_pipecat_turn`) — the text twin of `realtime_turn.py` | | `../voice/aec.py` | WebRTC AEC3 on the mic path; reference tapped at the TTS output stream (all providers) | diff --git a/docs/vi/realtime-voice_vi.md b/docs/vi/realtime-voice_vi.md index 40dac0ef2..feacc33ce 100644 --- a/docs/vi/realtime-voice_vi.md +++ b/docs/vi/realtime-voice_vi.md @@ -375,14 +375,19 @@ chụp thì không có gì để bàn giao, agent chụp như bình thường. ## Các provider -Ba backend thay thế cho nhau, chọn bằng `HAL_REALTIME_PROVIDER` -(`none` | `gemini` | `openai` | `qwen`): +Bốn backend, chọn bằng `HAL_REALTIME_PROVIDER` / `realtime.provider` +(`none` | `gemini` | `openai` | `qwen` | `pipecat`). Ba cái đầu là +**audio-native** — nhận frame mic và trả về tiếng nói. `pipecat` là **cascaded**: +HAL giữ nguyên STT và TTS của mình, pipecat chỉ lo phần giữa của lượt. Nó không +dùng chung gì với `voice_agent/` và không phải `VoiceAgentBase`; xem +[Pipecat](#pipecat--provider-cascaded) bên dưới. | Provider | Class | Mô hình threading | Model mặc định | Sample rate | |----------|-------|-------------------|----------------|-------------| | Gemini Live | `voice_agent/gemini_live.py` `GeminiLiveAgent` | event loop asyncio riêng trên thread `gemini-io`; thread send/recv submit coroutine qua `run_coroutine_threadsafe` | `gemini-2.5-flash-native-audio-preview-12-2025` | 16000 Hz | | OpenAI Realtime | `voice_agent/openai_realtime.py` `OpenAIRealtimeAgent` | thuần đồng bộ; 1 `RealtimeConnection` dùng chung bởi thread send/recv, serialize bằng reentrant lock | `gpt-realtime-2` | 24000 Hz | | Qwen Omni Realtime | `voice_agent/qwen_realtime.py` `QwenRealtimeAgent` | thuần đồng bộ; client `websockets.sync.client` thô | `qwen3.5-omni-plus-realtime` | 16000 Hz | +| Pipecat (cascaded) | `pipecat_session.py` `PipecatSession` trên `hal/pipecat_rt.py` `PipecatAgent` | event loop asyncio riêng trên thread `pipecat-loop`; API public là sync, dựa trên queue | `llm_model` (không có default riêng) | không có — text vào, text ra | Gemini Live dùng `google-genai` và private asyncio loop của nó do thread `gemini-io` sở hữu. Teardown đóng/hủy provider receive task trước, rồi mới join @@ -456,6 +461,89 @@ trên queue: và fallback sang main agent mà không bị dead-air dài). - `available` ⇔ websocket/session đã connect (`_connected`). +### Pipecat — provider cascaded + +Chọn bằng `realtime.provider = "pipecat"`. Pipeline rút gọn còn + +``` +transcript ──▶ LLMContext ──▶ LLM (tương thích OpenAI) ──▶ text chunk + └─ tool ──▶ xử lý tại chỗ, hoặc đẩy lên host +``` + +nên nửa audio của lượt không đổi: vẫn STT đó tạo transcript và vẫn TTS đó đọc +câu trả lời, từng câu một. Turn driver là +`drivers/voice/_internal/pipecat_turn.py` (`run_pipecat_turn`), bản song sinh +cascaded của `realtime_turn.py`. Nó trả về cùng `RealtimeTurnResult`, và dùng +chung wait filler, thinking cue, CoT leak filter với driver audio-native — nên +với người dùng, hai bộ não cảm giác như nhau. + +Những thứ nó **không** có, theo thiết kế: + +- **Không native audio** — giọng của model không bao giờ được phát; device tự nói. +- **Không nhận diện cảm xúc/ngữ điệu giọng** — model thấy text, không thấy audio. + Đây là đánh đổi có chủ đích để lấy latency; `express_emotion` vẫn chạy, dựa + trên nội dung câu trả lời thay vì giọng người nói. +- **Không có `look` / thị giác trong phiên**, không look-replay — câu hỏi thị + giác delegate cho main agent. +- **Không retry khôi phục WS** — không có session dài hạn để mất. +- **Không có knob voice hay reasoning** — TTS của device quyết định giọng và + model không expose mức reasoning, nên `ValidateRealtimeKnobs` từ chối cả hai. + +`PipecatSession` duck-type đúng phần bề mặt của `RealtimeOrchestrator` mà +`voice_service` chạm tới lúc capture (`available`, `prepare_turn`, +`append_audio`, `send_text`, `sample_rate`, `rebuilding`, +`wait_until_available`, `save_turn`), nên chỉ lời gọi turn là rẽ nhánh. +`append_audio` là no-op có chủ đích: bộ não này nhận transcript đã hoàn chỉnh. + +Persona và memory đến từ **cùng** `ContextManagerBase` (khóa theo gateway) mà +provider audio-native dùng, nên identity, `summary.md` và catalog skills giống +hệt. Instructions được dựng lại mỗi `HAL_REALTIME_SESSION_MAX_TURNS` lượt để một +lần đổi tên hay memory mới kịp vào; giữa hai lần dựng, prompt ổn định từng byte — +đó chính là thứ giữ ấm prefix cache của gateway. + +Tool: `delegate_to_main` (trả kết quả với `run_llm=False` — lượt kết thúc, main +agent tiếp quản), `express_emotion` (fire-and-forget trên daemon thread, trả với +`run_llm=True` để model vẫn còn nợ người dùng lời nói), và `web_search` tùy chọn +dùng Gemini grounding khi có `realtime.pipecat.search_api_key`, giới hạn theo +lượt bằng `HAL_PIPECAT_SEARCH_BUDGET`. + +**Metric mỗi lượt.** Mỗi lượt ghi một dòng log ở mức INFO: + +``` +[pipecat] ttft 0.62s | wait 0.00s | server 0.61s | stream 0.01s | + prompt 38.3k chars/10866 tok | out 11 tok | payload 40.8 KB +``` + +| Field | Ý nghĩa | +|-------|---------| +| `ttft` | từ lúc transcript vào hàng đợi → text frame đầu tiên đi ra | +| `wait` | thời gian trong pipeline trước khi request được gửi (aggregation) | +| `server` | gửi request → nhận header: RTT mạng + prefill | +| `stream` | header → token đầu tiên | +| `prompt` | kích thước text của context, và số prompt token server tính | +| `out` | completion token, kèm `(N reasoning)` khi server báo | +| `images` | số image part trong context (bỏ qua khi không có) | +| `payload` | số byte thực sự ghi lên dây, đo bằng httpx request hook | +| `N requests` | chỉ hiện khi một lượt gọi nhiều hơn một request | + +Số token và payload là **tổng của cả lượt, không phải từng request**: trả lời một +tool call tốn thêm một request gửi lại toàn bộ prompt, nên một lượt có +`express_emotion` gần như nhân đôi cả hai. Đó là cái giá nhìn thấy được của tool, +và cũng là lý do các bộ đếm cộng dồn thay vì ghi đè. + +`cached` (prefix-cache hit) chỉ xuất hiện khi endpoint trả +`prompt_tokens_details.cached_tokens`, và `reasoning` chỉ khi có +`completion_tokens_details.reasoning_tokens`. Pipecat map sẵn cả hai; gateway nào +không trả thì field đó đơn giản là không nằm trong dòng log. + +**Lưu ý dependency.** `pipecat-ai` không phải dependency của HAL và pin +`onnxruntime~=1.24.3`, sẽ hạ cấp bản 1.27.x mà device đang dùng cho TEN-VAD và +insightface. Hãy cài với `--no-deps` cộng thêm các module pipecat import lúc load +package (`loguru`, `pyloudnorm`, `soxr`, `resampy`, `markdown`, `nltk`, +`num2words`, `defusedxml`, `docopt`) để onnxruntime được giữ nguyên. +`PipecatAgent.start()` trả `False` khi thiếu package, và provider fallback về +main agent thay vì làm hỏng lượt. + ### An toàn connection của OpenAI Agent OpenAI dùng chung 1 `RealtimeConnection` giữa thread send và recv. Mọi @@ -718,8 +806,8 @@ chỉ-thuộc-os-server. Model ở Go tại `system/server/config/realtime.go`; đọc ở HAL tại `hal/config.py`. Field chung ở trên; knob theo provider nằm trong sub-object -`gemini` / `openai` / `qwen`, `provider` chọn cái đang active (`none` hoặc vắng → -tắt realtime). `api_key` / `base_url` rỗng → fallback `llm_api_key` / +`gemini` / `openai` / `qwen` / `pipecat`, `provider` chọn cái đang active (`none` +hoặc vắng → tắt realtime). `api_key` / `base_url` rỗng → fallback `llm_api_key` / `llm_base_url` — **trừ qwen**: credential của qwen là của riêng nó (`realtime.qwen.api_key` / `realtime.qwen.base_url`, Go struct `QwenRealtime` còn có `model`/`voice`), **cố tình không** fallback về `realtime.api_key`/ @@ -728,6 +816,14 @@ không đi qua proxy `campaign-api`. Set qua `realtime.qwen.*` trong config.json hoặc qua env trên device (`DASHSCOPE_API_KEY`, `HAL_QWEN_REALTIME_BASE_URL` trong `/opt/hal/.env`); thiếu cả hai thì WS handshake fail rõ ràng trong log hal. +`pipecat` cũng giữ credential riêng trong `realtime.pipecat.*` (Go struct +`PipecatRealtime`) và bỏ qua `realtime.api_key`/`base_url` chung — vốn mang giá +trị proxy có suffix WS. Endpoint của nó là host tương thích OpenAI `/v1` thuần, +nên field nào trống sẽ fallback về **AI brain** (`llm_base_url` / `llm_api_key` / +`llm_model`) — vốn đã là host như vậy. Nếu không resolve được base_url hoặc +model, `PipecatSession.start()` ghi log và ở trạng thái không khả dụng, các lượt +rơi về main agent. + > **Để `base_url` trống trừ khi có endpoint riêng (không qua proxy).** Khi trống, > HAL tự suy ra `/ws/gemini` (hoặc `/ws/openai`) — đúng suffix WS mà > proxy `campaign-api` route. Nếu `base_url` bị set bằng `llm_base_url` trần (thiếu @@ -746,7 +842,8 @@ trong `/opt/hal/.env`); thiếu cả hai thì WS handshake fail rõ ràng trong "provider": "gemini", "gemini": { "model": "gemini-3.1-flash-live-preview", "voice": "Kore", "thinking_level": "MINIMAL" }, "openai": { "model": "gpt-realtime-2", "voice": "alloy", "reasoning_effort": "minimal" }, - "qwen": { "api_key": "sk-…", "base_url": "wss://…", "model": "qwen3.5-omni-plus-realtime", "voice": "Ethan" } + "qwen": { "api_key": "sk-…", "base_url": "wss://…", "model": "qwen3.5-omni-plus-realtime", "voice": "Ethan" }, + "pipecat": { "model": "qwen/qwen3.6-35b-a3b", "base_url": "https://…/v1", "api_key": "…", "search_api_key": "AIza…" } } } ``` @@ -800,7 +897,7 @@ trong `config.json`: | `HAL_WAKEWORD_FOLLOWUP_TIMEOUT_S` | `20` | Số giây idle của cửa sổ focus sau lệnh. Mỗi `voice_command` hoặc `voice_followup` được nhận sẽ refresh cửa sổ. `0` tắt follow-up và buộc mỗi phiên mic phải có wake phrase. Bị bỏ qua khi `wakeword` là false. | | `HAL_SILENCE_VAD_ENABLED` | `true` | Yêu cầu Silero xác nhận có tiếng nói trước khi refresh đồng hồ im lặng kết thúc lượt. RMS vẫn là cổng chặn rẻ chạy trước; đặt `false` để quay về phát hiện im lặng thuần RMS. | | `HAL_SILENCE_VAD_WINDOW_FRAMES` | `3` | Số frame gom lại cho mỗi lần chạy Silero ở bước kiểm đó — Silero tốn ~20 ms/frame trên ARM và LSTM của nó cần hơn một frame 64 ms mới ổn định. | -| `HAL_REALTIME_PROVIDER` | `gemini` | `none` \| `gemini` \| `openai` \| `qwen` | +| `HAL_REALTIME_PROVIDER` | `gemini` | `none` \| `gemini` \| `openai` \| `qwen` \| `pipecat` | | `HAL_REALTIME_TURN_DETECTION` | `off` | `server_vad` \| `semantic_vad` \| `off` (Gemini: off = activity detection thủ công) | | `HAL_REALTIME_RECV_QUEUE_TIMEOUT_S` | `8.0` | Số giây tối đa `receive()` chờ output event kế tiếp trước khi kết thúc lượt im lặng (fallback sang main agent) | | `HAL_REALTIME_LOOK_RECV_TIMEOUT_S` | `20.0` | Watchdog im-lặng dùng thay mặc định cho turn có `look` (theo từng turn, qua `extend_recv_timeout()`). Gemini bị ép thinking trên frame dày chữ có thể im >8 s ngay trước khi trả lời — watchdog mặc định giết nhầm mấy turn đó | @@ -829,6 +926,16 @@ trong `config.json`: | `HAL_QWEN_REALTIME_BASE_URL` | — | WS host DashScope (`wss:///api-ws/v1/realtime`); **không** fallback về `llm_base_url` — chỉ đọc `realtime.qwen.base_url` khi env trống | | `HAL_QWEN_REALTIME_MODEL` | `qwen3.5-omni-plus-realtime` | turbo legacy: không gọi function call, lờ turn context | | `HAL_QWEN_REALTIME_VOICE` | `Ethan` | 3.5-plus: thêm Serena; chỉ-turbo: Cherry \| Chelsie | +| `HAL_PIPECAT_BASE_URL` | `llm_base_url` | Host tương thích OpenAI `/v1`; override `realtime.pipecat.base_url` | +| `HAL_PIPECAT_API_KEY` | `llm_api_key` | Override `realtime.pipecat.api_key` | +| `HAL_PIPECAT_MODEL` | `llm_model` | Override `realtime.pipecat.model` | +| `HAL_PIPECAT_GEMINI_KEY` | — | Bật tool `web_search` (Gemini grounding). Không set → không đăng ký tool search | +| `HAL_PIPECAT_SEARCH_BUDGET` | `2` | Số lần `web_search` tối đa mỗi lượt | +| `HAL_PIPECAT_MAX_TOKENS` | `512` | Giới hạn token sinh ra — **tính cả argument của tool call**, nên budget nhỏ sẽ cắt cụt JSON và mất lượt | +| `HAL_PIPECAT_MAX_HISTORY` | `128` | Số message giữ trong phiên trước khi trim; memory dài hạn nằm trong system prompt | +| `HAL_PIPECAT_HTTP2` | `true` | Dùng HTTP/2 tới gateway | +| `HAL_PIPECAT_TURN_TIMEOUT_S` | `20` | Watchdog mỗi lượt | +| `HAL_PIPECAT_LOG_LEVEL` | `WARNING` | pipecat log qua loguru ở mức DEBUG mặc định, làm ngập journal của device với log từng frame | | `HAL_REALTIME_MEMORY_PATH` | `/realtime/memory.jsonl` | | | `HAL_REALTIME_MAX_MEMORY_ENTRIES` / `_TRIM_KEEP` | `1000` / `500` | | | `HAL_REALTIME_SUMMARIZER_ENABLED` | `true` | | @@ -849,4 +956,7 @@ trong `config.json`: | `models/`, `enums/` | Kiểu input/output/event, enum provider + gateway | | `resources/` | System prompt (chung + theo provider) | | `../voice/voice_service.py` | Tích hợp: stream audio mic, tiêu thụ output, route delegate/handled | +| `pipecat_session.py` | Bộ não cascaded: sở hữu `PipecatAgent`, dùng lại context manager, duck-type bề mặt capture | +| `../../pipecat_rt.py` | Engine pipecat độc lập (`PipecatAgent`): dựng pipeline, bắc cầu tool, metric mỗi lượt. Chạy độc lập được: `python -m hal.pipecat_rt` | +| `../voice/_internal/pipecat_turn.py` | Turn driver cascaded (`run_pipecat_turn`) — bản text của `realtime_turn.py` | | `../voice/aec.py` | WebRTC AEC3 trên đường mic; tham chiếu lấy tại TTS output stream (mọi provider) | diff --git a/hal/config.py b/hal/config.py index 882bb16e7..d79546101 100644 --- a/hal/config.py +++ b/hal/config.py @@ -661,6 +661,7 @@ def _os_cfg_realtime() -> dict: _RT_GEMINI: dict = _RT.get("gemini") if isinstance(_RT.get("gemini"), dict) else {} _RT_OPENAI: dict = _RT.get("openai") if isinstance(_RT.get("openai"), dict) else {} _RT_QWEN: dict = _RT.get("qwen") if isinstance(_RT.get("qwen"), dict) else {} +_RT_PIPECAT: dict = _RT.get("pipecat") if isinstance(_RT.get("pipecat"), dict) else {} def _rt_str(env_key: str, cfg_val, default: str) -> str: @@ -683,7 +684,7 @@ def _rt_enabled() -> bool: REALTIME_ENABLED: bool = _rt_enabled() -REALTIME_PROVIDER: str = _rt_str("HAL_REALTIME_PROVIDER", _RT.get("provider"), "gemini") # none | gemini | openai | qwen +REALTIME_PROVIDER: str = _rt_str("HAL_REALTIME_PROVIDER", _RT.get("provider"), "gemini") # none | gemini | openai | qwen | pipecat # When enabled, do not send a voice turn to the realtime agent until an STT # interim transcript starts with one of the configured wake phrases. This is a # top-level config.json setting because it also gates the non-realtime Go path. @@ -1199,6 +1200,31 @@ def _rt_enabled() -> bool: ) REALTIME_QWEN_SAMPLE_RATE: int = 16000 +# --- Realtime: Pipecat (cascaded) --- +# Not an audio-native brain: HAL keeps its own STT and TTS and pipecat drives +# only the middle of the turn (text in -> LLM + tools -> text out), so there is +# no voice or reasoning knob here. The endpoint is any OpenAI-compatible /v1 +# host; blank values fall back to the AI brain's, which is already such a host. +REALTIME_PIPECAT_API_KEY: str = ( + os.environ.get("HAL_PIPECAT_API_KEY", "") + or _RT_PIPECAT.get("api_key", "") + or _os_cfg_get("llm_api_key", "") +) +REALTIME_PIPECAT_BASE_URL: str = ( + os.environ.get("HAL_PIPECAT_BASE_URL", "") + or _RT_PIPECAT.get("base_url", "") + or _os_cfg_get("llm_base_url", "") +) +REALTIME_PIPECAT_MODEL: str = _rt_str( + "HAL_PIPECAT_MODEL", _RT_PIPECAT.get("model"), _os_cfg_get("llm_model", "") +) +# Gemini grounding search, offered to the model as a `web_search` tool. Separate +# key because the gateway above is typically self-hosted and has no search. +REALTIME_PIPECAT_SEARCH_KEY: str = ( + os.environ.get("HAL_PIPECAT_GEMINI_KEY", "") + or _RT_PIPECAT.get("search_api_key", "") +) + # --- Realtime: Context manager --- OPENCLAW_WORKSPACE_DIR: str = os.environ.get("HAL_OPENCLAW_WORKSPACE_DIR", "/root/.openclaw/workspace") HERMES_WORKSPACE_DIR: str = os.environ.get("HAL_HERMES_WORKSPACE_DIR", "/root/.hermes") diff --git a/hal/drivers/voice/_internal/pipecat_turn.py b/hal/drivers/voice/_internal/pipecat_turn.py new file mode 100644 index 000000000..a085e34d3 --- /dev/null +++ b/hal/drivers/voice/_internal/pipecat_turn.py @@ -0,0 +1,201 @@ +"""Pipecat turn handling — the cascaded twin of realtime_turn.py. + +Same contract (drive one turn, return a RealtimeTurnResult) over a different +input: the finished STT transcript instead of committed audio. There is no +native-audio path, no look-replay and no WS-recovery retry — a cascaded brain +has no session to lose, and an empty transcript means there is simply nothing +to send. + +The wait filler, thinking cue and CoT leak filter are shared with realtime_turn +so both brains feel identical to the user. +""" + +import json +import logging +import threading +from typing import Callable + +from hal import config as hal_config +from hal.pipecat_rt import TextChunk, ToolCall +from hal.realtime.orchestrator import ( + DEFAULT_EMOTION_INTENSITY, + DELEGATE_TOOL_NAME, + EMOTION_TOOL_NAME, + RealtimeOrchestrator, +) +from hal.drivers.voice._internal.cot_leak_filter import CoTLeakFilter, clean_transcript +from hal.drivers.voice._internal.realtime_turn import ( + SENTENCE_ENDS, + RealtimeTurnResult, + _reply_language_name, + _thinking_cue_clear, + _thinking_cue_start, + _WaitFiller, +) + +logger = logging.getLogger("hal.voice") + + +def _fire_emotion(arguments: str) -> None: + """Run express_emotion off-thread so the reply never waits on the face.""" + try: + args: dict = json.loads(arguments) if arguments else {} + except (ValueError, TypeError): + return + emotion: str = str(args.get("emotion", "")).strip().lower() + if not emotion: + return + intensity: float = DEFAULT_EMOTION_INTENSITY + try: + intensity = max(0.0, min(1.0, float(args.get("intensity", intensity)))) + except (ValueError, TypeError): + pass + threading.Thread( + target=RealtimeOrchestrator._fire_emotion, + args=(emotion, intensity), + daemon=True, + ).start() + + +def run_pipecat_turn( + session, + tts, + strip_markers: Callable[[str], str], + combined: str, +) -> RealtimeTurnResult: + """Send the transcript to the pipecat brain and speak its reply. + + Returns how the turn resolved so the caller can forward (delegate), + suppress (handled), or fall back to the main agent. + """ + if not hal_config.REALTIME_ENABLED: + return RealtimeTurnResult() + if not session.available: + logger.warning("[pipecat] enabled but not available — falling back to OS server") + return RealtimeTurnResult() + + text: str = (combined or "").strip() + if not text: + logger.info("[pipecat] empty transcript — nothing to send (no audio path)") + return RealtimeTurnResult() + + delegated = False + handled = False + delegate_msg = "" + text_parts: list[str] = [] + sentence_buf = "" + first_sentence_sent = False + reply_lang: str = _reply_language_name() + leak_filter = CoTLeakFilter(reply_lang) + wait_filler = _WaitFiller() + + _thinking_cue_start() + wait_filler.arm() + try: + for event in session.run_turn(text): + if isinstance(event, ToolCall): + if event.name == DELEGATE_TOOL_NAME: + delegated = True + try: + delegate_msg = str( + json.loads(event.arguments or "{}").get("message", "") + ) + except (ValueError, TypeError): + delegate_msg = "" + # The main-agent hop that follows fires its own filler. + wait_filler.cancel() + session.tool_result( + event.call_id, '{"status": "delegated"}', run_llm=False + ) + continue + if event.name == EMOTION_TOOL_NAME: + _fire_emotion(event.arguments) + # run_llm=True: the face is a side effect, the reply still owes + # the user words. + session.tool_result(event.call_id, '{"status": "ok"}', run_llm=True) + continue + logger.warning("[pipecat] unknown tool %r", event.name) + session.tool_result( + event.call_id, '{"error": "unknown tool"}', run_llm=True + ) + continue + + if delegated or not isinstance(event, TextChunk): + continue + + text_parts.append(event.text) + sentence_buf += event.text + if tts is not None and sentence_buf.rstrip().endswith(SENTENCE_ENDS): + sentence: str = leak_filter.filter_text(strip_markers(sentence_buf)) + if sentence: + if not first_sentence_sent: + logger.info("[pipecat] First sentence → speak: %r", sentence[:80]) + wait_filler.cancel() + if not tts.speak(sentence): + tts.speak_queue(sentence) + first_sentence_sent = True + _thinking_cue_clear() + else: + logger.info( + "[pipecat] Next sentence → speak_queue: %r", sentence[:80] + ) + tts.speak_queue(sentence) + sentence_buf = "" + + transcript: str = clean_transcript( + strip_markers("".join(text_parts)), reply_lang + ) + + if delegated: + logger.info("[pipecat] Delegated → will forward to OS server") + else: + remaining: str = leak_filter.filter_text(strip_markers(sentence_buf)) + if remaining and tts is not None: + if not first_sentence_sent: + logger.info("[pipecat] Final fragment → speak: %r", remaining[:80]) + wait_filler.cancel() + if not tts.speak(remaining): + tts.speak_queue(remaining) + first_sentence_sent = True + _thinking_cue_clear() + else: + logger.info( + "[pipecat] Final fragment → speak_queue: %r", remaining[:80] + ) + tts.speak_queue(remaining) + # Same rule as the audio-native path: only claim the turn when the + # device actually spoke, so an empty reply still reaches the main agent. + if first_sentence_sent or transcript: + handled = True + logger.info( + "[pipecat] Chit-chat complete — agent_reply=%r", + transcript[:200] if transcript else "(empty)", + ) + session.save_turn( + user_text=text, agent_text=transcript or "(empty)" + ) + else: + logger.info("[pipecat] No output (empty / timeout) — falling back") + _thinking_cue_clear() + try: + from hal.routes.led import restore_led + + restore_led() + except Exception: + pass + except Exception as e: + logger.warning("[pipecat] Processing failed: %s — will forward to OS server", e) + _thinking_cue_clear() + try: + session.abort_turn() + except Exception: + pass + return RealtimeTurnResult(delegated=True) + finally: + wait_filler.cancel() + try: + session.turn_finished() + except Exception: + logger.exception("[pipecat] turn bookkeeping failed") + + return RealtimeTurnResult(delegated, handled, transcript, delegate_msg) diff --git a/hal/drivers/voice/voice_service.py b/hal/drivers/voice/voice_service.py index 26f828a8d..2b04edb2f 100644 --- a/hal/drivers/voice/voice_service.py +++ b/hal/drivers/voice/voice_service.py @@ -41,6 +41,7 @@ run_realtime_turn, should_dispatch_to_main, ) +from hal.drivers.voice._internal.pipecat_turn import run_pipecat_turn from hal.drivers.voice._internal.sensing_sender import SensingSender from hal.drivers.voice._internal.session_finalize import finalize_session from hal.drivers.voice._internal.speaker_decorate import ( @@ -218,11 +219,25 @@ def __init__( # OS server event sender (with echo similarity filter) self._sensing_sender = SensingSender(tts_service=tts_service) - # Realtime voice agent — parallel audio pipeline (Gemini Live / OpenAI Realtime). - self._realtime = RealtimeOrchestrator( - gateway=AgentGateway(hal_config.AGENT_GATEWAY), - enable_expression=enable_expression, + # Realtime voice agent. Two shapes behind one handle: the audio-native + # providers (Gemini Live / OpenAI Realtime / Qwen Omni) stream mic frames, + # while pipecat is cascaded and takes the finished STT transcript. Both + # answer the same capture-path surface, so only the turn driver branches. + self._pipecat: bool = ( + hal_config.REALTIME_PROVIDER.strip().lower() == "pipecat" ) + if self._pipecat: + from hal.realtime.pipecat_session import PipecatSession + + self._realtime = PipecatSession( + gateway=AgentGateway(hal_config.AGENT_GATEWAY), + enable_expression=enable_expression, + ) + else: + self._realtime = RealtimeOrchestrator( + gateway=AgentGateway(hal_config.AGENT_GATEWAY), + enable_expression=enable_expression, + ) # Hook into TTS on_speak_end to feed spoken text back to the realtime agent. # With turn_complete=False on text inputs, this won't trigger a standalone response. @@ -1722,7 +1737,14 @@ def _send_pre_roll(): except Exception as e: logger.warning("[realtime] speaker correction send failed: %s", e) - if realtime_turn_started: + if realtime_turn_started and self._pipecat: + rt = run_pipecat_turn( + self._realtime, + self._tts, + self.strip_rt_markers, + combined, + ) + elif realtime_turn_started: rt = run_realtime_turn( self._realtime, self._tts, diff --git a/hal/pipecat_rt.py b/hal/pipecat_rt.py index 8cf189632..4a787a766 100644 --- a/hal/pipecat_rt.py +++ b/hal/pipecat_rt.py @@ -25,6 +25,7 @@ import logging import os import queue +import sys import threading import time import uuid @@ -63,6 +64,9 @@ class PipecatConfig: "HAL_PIPECAT_GEMINI_SEARCH_MODEL", "gemini-3.5-flash-lite" ) search_timeout_s: float = float(os.environ.get("HAL_PIPECAT_SEARCH_TIMEOUT_S", "6")) + # pipecat logs through loguru at DEBUG, which floods the device journal with + # per-frame lines. Raise to DEBUG only when tracing the pipeline. + log_level: str = os.environ.get("HAL_PIPECAT_LOG_LEVEL", "WARNING") # --- Turn events ----------------------------------------------------------- @@ -84,19 +88,51 @@ class ToolCall: @dataclass class TurnMetrics: - """Where a turn's latency went. `server` bundles network RTT + prefill.""" + """Where a turn's latency went and what it cost. + + A turn that calls a tool makes more than one request, so token counts and + payload bytes are TURN TOTALS across `requests`, not per-request values. + """ wait: float = 0.0 server: float = 0.0 stream: float = 0.0 ttft: float = 0.0 prompt_chars: int = 0 + prompt_tokens: int = 0 + completion_tokens: int = 0 + cached_tokens: int = 0 + reasoning_tokens: int = 0 + images: int = 0 + payload_bytes: int = 0 + requests: int = 0 def __str__(self) -> str: - return ( - f"ttft {self.ttft:.2f}s | wait {self.wait:.2f}s | server {self.server:.2f}s " - f"| stream {self.stream:.2f}s | prompt {self.prompt_chars / 1000:.1f}k chars" - ) + parts = [ + f"ttft {self.ttft:.2f}s", + f"wait {self.wait:.2f}s", + f"server {self.server:.2f}s", + f"stream {self.stream:.2f}s", + ] + prompt = f"prompt {self.prompt_chars / 1000:.1f}k chars" + if self.prompt_tokens: + prompt += f"/{self.prompt_tokens} tok" + if self.cached_tokens: + pct = 100.0 * self.cached_tokens / self.prompt_tokens + prompt += f" ({self.cached_tokens} cached, {pct:.0f}%)" + parts.append(prompt) + if self.completion_tokens or self.reasoning_tokens: + out = f"out {self.completion_tokens} tok" + if self.reasoning_tokens: + out += f" ({self.reasoning_tokens} reasoning)" + parts.append(out) + if self.images: + parts.append(f"images {self.images}") + if self.payload_bytes: + parts.append(f"payload {self.payload_bytes / 1024:.1f} KB") + if self.requests > 1: + parts.append(f"{self.requests} requests") + return " | ".join(parts) @dataclass @@ -113,6 +149,13 @@ class _TurnState: headers: float = 0.0 first_token: float = 0.0 prompt_chars: int = 0 + prompt_tokens: int = 0 + completion_tokens: int = 0 + cached_tokens: int = 0 + reasoning_tokens: int = 0 + images: int = 0 + payload_bytes: int = 0 + requests: int = 0 searches: int = 0 pending: dict = field(default_factory=dict) calls_started: int = 0 @@ -122,6 +165,8 @@ def reset(self, started: float) -> None: self.started = started self.request_sent = self.headers = self.first_token = 0.0 self.prompt_chars = self.searches = self.calls_started = 0 + self.prompt_tokens = self.completion_tokens = self.cached_tokens = 0 + self.reasoning_tokens = self.images = self.payload_bytes = self.requests = 0 self.response_ended = False self.pending.clear() @@ -179,6 +224,13 @@ def start(self) -> bool: except ImportError as e: logger.warning("[pipecat] not installed (%s) — disabled", e) return False + try: + from loguru import logger as _loguru + + _loguru.remove() + _loguru.add(sys.stderr, level=self._config.log_level) + except Exception: + pass self._stopping.clear() started = threading.Event() @@ -252,7 +304,7 @@ async def _build(self) -> None: aggregators.assistant(), ] ), - params=PipelineParams(enable_metrics=True), + params=PipelineParams(enable_metrics=True, enable_usage_metrics=True), idle_timeout_secs=None, conversation_id=self._session_id, ) @@ -519,6 +571,13 @@ def _metrics(self) -> TurnMetrics: stream=(t.first_token - t.headers) if t.first_token and t.headers else 0.0, ttft=(t.first_token - t.started) if t.first_token else 0.0, prompt_chars=t.prompt_chars, + prompt_tokens=t.prompt_tokens, + completion_tokens=t.completion_tokens, + cached_tokens=t.cached_tokens, + reasoning_tokens=t.reasoning_tokens, + images=t.images, + payload_bytes=t.payload_bytes, + requests=t.requests, ) def _trim_history(self) -> None: @@ -574,6 +633,28 @@ async def _cancel_tasks() -> None: await asyncio.gather(*pending, return_exceptions=True) +def _measure_messages(messages) -> tuple[int, int]: + """(text chars, image parts) over a message list. + + Counts text separately from images: a base64 image part is ~100k characters + and would otherwise swamp the prompt-size number it shares a field with. + """ + chars = images = 0 + for message in messages: + content = message.get("content", "") + if isinstance(content, str): + chars += len(content) + continue + for part in content or (): + if not isinstance(part, dict): + chars += len(str(part)) + elif part.get("type") == "text": + chars += len(part.get("text", "")) + else: + images += 1 + return chars, images + + def _build_llm(config: PipecatConfig, turn: _TurnState): from pipecat.services.openai.llm import OpenAILLMService @@ -596,25 +677,45 @@ def create_client(self, api_key=None, base_url=None, **kwargs): limits = httpx.Limits( max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None ) + async def on_request(request) -> None: + # The body as it actually goes on the wire — the only honest + # payload number, since the SDK reshapes what the context holds. + try: + turn.payload_bytes += len(request.content or b"") + turn.requests += 1 + except Exception: + pass + + hooks = {"request": [on_request]} try: - http_client = DefaultAsyncHttpxClient(http2=config.http2, limits=limits) + http_client = DefaultAsyncHttpxClient( + http2=config.http2, limits=limits, event_hooks=hooks + ) except ImportError: logger.warning('[pipecat] HTTP/2 needs httpx[http2]; using HTTP/1.1') - http_client = DefaultAsyncHttpxClient(limits=limits) + http_client = DefaultAsyncHttpxClient(limits=limits, event_hooks=hooks) client = AsyncOpenAI( api_key=api_key, base_url=base_url, http_client=http_client ) return client.with_options(max_retries=config.max_retries) async def get_chat_completions(self, context): - self._turn.prompt_chars = sum( - len(str(m.get("content", ""))) for m in context.get_messages() - ) + chars, images = _measure_messages(context.get_messages()) + self._turn.prompt_chars = chars + self._turn.images = images self._turn.request_sent = time.perf_counter() stream = await super().get_chat_completions(context) self._turn.headers = time.perf_counter() return stream + async def start_llm_usage_metrics(self, tokens): + """Accumulate: a turn with a tool call bills more than one request.""" + self._turn.prompt_tokens += tokens.prompt_tokens or 0 + self._turn.completion_tokens += tokens.completion_tokens or 0 + self._turn.cached_tokens += getattr(tokens, "cache_read_input_tokens", 0) or 0 + self._turn.reasoning_tokens += getattr(tokens, "reasoning_tokens", 0) or 0 + await super().start_llm_usage_metrics(tokens) + return _TimedLLM( api_key=config.api_key, base_url=config.base_url, @@ -660,3 +761,53 @@ async def process_frame(self, frame, direction): await self.push_frame(frame, direction) return Collector() + + +# --- Smoke test ------------------------------------------------------------ +# python -m hal.pipecat_rt ["prompt" ...] — needs HAL_PIPECAT_BASE_URL/_MODEL. + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(message)s") + + prompts = sys.argv[1:] or [ + "hello, who are you?", + "what is the capital of Vietnam?", + "play some jazz music please", + ] + delegate = ( + "delegate_to_main", + "Hand the request to the main system: device control, music, scheduling, " + "memory, skills. Pass a short summary of what the user wants.", + { + "properties": { + "message": {"type": "string", "description": "What the user actually asked for."} + }, + "required": ["message"], + }, + ) + agent = PipecatAgent( + PipecatConfig(), + instructions=( + "You are Lamp, a small desk robot. Keep replies to one or two short spoken " + "sentences. When the user asks for device control, music, scheduling or " + "memory, call delegate_to_main instead of answering." + ), + host_tools=[delegate], + ) + started = time.perf_counter() + if not agent.start(): + sys.exit("start failed — missing pipecat-ai/httpx[http2], or base_url/model unset") + print(f"ready in {time.perf_counter() - started:.2f}s") + + for prompt in prompts: + print(f"\nuser: {prompt}") + reply = "" + for event in agent.run_turn(prompt): + if isinstance(event, TextChunk): + reply += event.text + elif isinstance(event, ToolCall): + print(f" tool {event.name}({event.arguments})") + agent.tool_result(event.call_id, '{"result": "delegated"}', run_llm=False) + print(f"lamp: {reply.strip()!r}\n {agent.last_metrics}") + + agent.stop() diff --git a/hal/realtime/pipecat_session.py b/hal/realtime/pipecat_session.py new file mode 100644 index 000000000..8c3216b42 --- /dev/null +++ b/hal/realtime/pipecat_session.py @@ -0,0 +1,178 @@ +"""Cascaded realtime brain — pipecat driving the LLM half of a voice turn. + +The audio-native providers (gemini/openai/qwen) take microphone frames and +return speech. This one does not: HAL keeps its own STT and TTS, and pipecat +owns only transcript -> LLM + tools -> text. It therefore shares nothing with +`voice_agent/` and never becomes a `VoiceAgentBase`. + +It does duck-type the slice of `RealtimeOrchestrator` that `voice_service` +touches during capture, so only the turn driver has to branch. The audio +methods are deliberate no-ops — this brain is fed the finished transcript. +""" + +import logging +from typing import Any, Generator + +from hal import config +from hal.pipecat_rt import PipecatAgent, PipecatConfig +from hal.realtime.config import _load_language +from hal.realtime.context_manager import ContextManagerBase +from hal.realtime.enums import AgentGateway +from hal.realtime.orchestrator import ( + DEFAULT_SAMPLE_RATE, + DELEGATE_TOOL, + EMOTION_TOOL, + RealtimeOrchestrator, +) +from hal.realtime.summarizer import RealtimeSummarizer + +logger = logging.getLogger("hal.realtime") + + +def _as_host_tool(tool: dict[str, Any]) -> tuple[str, str, dict]: + """Convert a realtime tool schema to pipecat's (name, description, params).""" + return tool["name"], tool["description"], tool.get("parameters", {}) + + +class PipecatSession: + """One pipecat conversation, with the same persona/memory context the + audio-native providers get. + + Instructions are rebuilt every REALTIME_SESSION_MAX_TURNS turns so a rename + or new memory lands; between rebuilds the prompt is byte-stable, which is + what keeps the gateway's prefix cache warm. + """ + + def __init__( + self, + gateway: AgentGateway = AgentGateway.OPENCLAW, + enable_expression: bool = False, + ) -> None: + self._expression_enabled = enable_expression + self._agent: PipecatAgent | None = None + self._turns_since_rebuild = 0 + + summarizer: RealtimeSummarizer | None = None + if config.REALTIME_SUMMARIZER_ENABLED: + try: + summarizer = RealtimeSummarizer() + except Exception as e: + logger.warning("[pipecat] summarizer unavailable: %s", e) + context_cls = RealtimeOrchestrator.CONTEXT_MANAGERS.get(gateway) + if context_cls is None: + from hal.realtime.context_manager import OpenClawContextManager + + context_cls = OpenClawContextManager + self._context: ContextManagerBase = context_cls( + workspace_dir=RealtimeOrchestrator.WORKSPACE_DIRS.get( + gateway, config.OPENCLAW_WORKSPACE_DIR + ), + language=_load_language() or "English", + provider="pipecat", + summarizer=summarizer, + ) + + # --- lifecycle --------------------------------------------------------- + + def start(self) -> None: + if not config.REALTIME_PIPECAT_BASE_URL or not config.REALTIME_PIPECAT_MODEL: + logger.warning( + "[pipecat] disabled — base_url=%r model=%r (set realtime.pipecat.*)", + config.REALTIME_PIPECAT_BASE_URL, + config.REALTIME_PIPECAT_MODEL, + ) + return + tools = [_as_host_tool(DELEGATE_TOOL)] + if self._expression_enabled: + tools.append(_as_host_tool(EMOTION_TOOL)) + agent = PipecatAgent( + PipecatConfig( + base_url=config.REALTIME_PIPECAT_BASE_URL, + api_key=config.REALTIME_PIPECAT_API_KEY or "not-needed", + model=config.REALTIME_PIPECAT_MODEL, + gemini_api_key=config.REALTIME_PIPECAT_SEARCH_KEY, + ), + instructions=self._context.build_instructions(), + host_tools=tools, + ) + if not agent.start(): + logger.warning("[pipecat] start failed — falling back to the main agent") + return + self._agent = agent + logger.info( + "[pipecat] ready — model=%s base_url=%s", + config.REALTIME_PIPECAT_MODEL, + config.REALTIME_PIPECAT_BASE_URL, + ) + + def stop(self) -> None: + agent, self._agent = self._agent, None + if agent is not None: + agent.stop() + try: + self._context.summarize_device_memory() + self._context.summarize_realtime_memory() + except Exception: + logger.exception("[pipecat] memory summarization failed on shutdown") + + # --- orchestrator-shaped surface used by voice_service ----------------- + + @property + def available(self) -> bool: + return self._agent is not None and self._agent.available + + @property + def sample_rate(self) -> int: + return DEFAULT_SAMPLE_RATE + + @property + def rebuilding(self) -> bool: + return False + + def wait_until_available(self, timeout_s: float = 2.0) -> bool: + return self.available + + def prepare_turn(self) -> None: + """No session to warm: the HTTP client is pooled and the prompt cached.""" + + def append_audio(self, frame: bytes) -> None: + """No-op — the transcript, not the audio, reaches this brain.""" + + def send_text(self, text: str) -> None: + if self._agent is not None: + self._agent.add_context(text) + + def save_turn(self, user_text: str, agent_text: str) -> None: + self._context.add_turn(user_text, agent_text) + + # --- turn -------------------------------------------------------------- + + def run_turn(self, transcript: str) -> Generator[Any, None, None]: + if self._agent is None: + return iter(()) + return self._agent.run_turn(transcript) + + def tool_result(self, call_id: str, output: str, run_llm: bool = True) -> None: + if self._agent is not None: + self._agent.tool_result(call_id, output, run_llm=run_llm) + + def abort_turn(self) -> None: + if self._agent is not None: + self._agent.abort_turn() + + @property + def last_metrics(self): + return self._agent.last_metrics if self._agent is not None else None + + def turn_finished(self) -> None: + """Count the turn and rebuild instructions once the cap is reached.""" + self._turns_since_rebuild += 1 + cap: int = config.REALTIME_SESSION_MAX_TURNS + if cap <= 0 or self._turns_since_rebuild < cap or self._agent is None: + return + self._turns_since_rebuild = 0 + try: + self._agent.rebuild_context(self._context.build_instructions()) + logger.info("[pipecat] context rebuilt after %d turns", cap) + except Exception as e: + logger.warning("[pipecat] context rebuild failed: %s", e) diff --git a/system/device/realtime.go b/system/device/realtime.go index 38ea054af..3de10695c 100644 --- a/system/device/realtime.go +++ b/system/device/realtime.go @@ -44,10 +44,11 @@ func applyRealtimeSet(c *config.Config, d domain.RealtimeSetData) { if d.Provider != "" { rt.Provider = strings.ToLower(strings.TrimSpace(d.Provider)) } - // Credentials are provider-routed: qwen keeps its own api_key/base_url in - // the qwen sub-object (HAL deliberately ignores the shared fields for qwen - // — they hold the campaign-api credentials used by gemini/openai). - if strings.ToLower(strings.TrimSpace(rt.Provider)) == "qwen" { + // Credentials are provider-routed: qwen and pipecat keep their own + // api_key/base_url in their sub-object (HAL deliberately ignores the shared + // fields for both — those hold the campaign-api credentials gemini/openai use). + switch strings.ToLower(strings.TrimSpace(rt.Provider)) { + case "qwen": if d.APIKey != "" || d.BaseURL != "" { if rt.Qwen == nil { rt.Qwen = &config.QwenRealtime{} @@ -59,7 +60,19 @@ func applyRealtimeSet(c *config.Config, d domain.RealtimeSetData) { rt.Qwen.BaseURL = d.BaseURL } } - } else { + case "pipecat": + if d.APIKey != "" || d.BaseURL != "" { + if rt.Pipecat == nil { + rt.Pipecat = &config.PipecatRealtime{} + } + if d.APIKey != "" { + rt.Pipecat.APIKey = d.APIKey + } + if d.BaseURL != "" { + rt.Pipecat.BaseURL = d.BaseURL + } + } + default: if d.APIKey != "" { rt.APIKey = d.APIKey } @@ -107,7 +120,15 @@ func applyRealtimeSet(c *config.Config, d domain.RealtimeSetData) { if d.Voice != "" { rt.Qwen.Voice = d.Voice } - // no reasoning knob — validateRealtimeSet already rejected it + // no reasoning knob — validateRealtimeSet already rejected it + case "pipecat": + if rt.Pipecat == nil { + rt.Pipecat = &config.PipecatRealtime{} + } + if d.Model != "" { + rt.Pipecat.Model = d.Model + } + // no voice/reasoning knobs — validateRealtimeSet already rejected them } } diff --git a/system/domain/device.go b/system/domain/device.go index 5cfd5183b..098e0bfa4 100644 --- a/system/domain/device.go +++ b/system/domain/device.go @@ -1156,7 +1156,7 @@ type MQTTTTSSetAck struct { // `realtime.set` downlink (data block) and the HTTP UpdateConfig `realtime` field. type RealtimeSetData struct { Enabled *bool `json:"enabled,omitempty"` // nil = leave unchanged - Provider string `json:"provider,omitempty"` // gemini | openai | none + Provider string `json:"provider,omitempty"` // gemini | openai | qwen | pipecat | none Model string `json:"model,omitempty"` // active provider's model Voice string `json:"voice,omitempty"` // active provider's voice Reasoning string `json:"reasoning,omitempty"` // gemini thinking_level OR openai reasoning_effort diff --git a/system/server/config/realtime.go b/system/server/config/realtime.go index 798fca88e..954e1b753 100644 --- a/system/server/config/realtime.go +++ b/system/server/config/realtime.go @@ -25,13 +25,14 @@ import ( type RealtimeConfig struct { // Enabled toggles the realtime brain. Unset → true (mirrors HAL's // HAL_REALTIME_ENABLED default); set false to disable. - Enabled *bool `json:"enabled,omitempty" yaml:"enabled"` - Provider string `json:"provider,omitempty" yaml:"provider"` // none|gemini|openai|qwen ("" == none) - APIKey string `json:"api_key,omitempty" yaml:"apiKey"` // empty → falls back to LLMAPIKey - BaseURL string `json:"base_url,omitempty" yaml:"baseURL"` // empty → falls back to LLMBaseURL - Gemini *GeminiRealtime `json:"gemini,omitempty" yaml:"gemini"` - OpenAI *OpenAIRealtime `json:"openai,omitempty" yaml:"openai"` - Qwen *QwenRealtime `json:"qwen,omitempty" yaml:"qwen"` + Enabled *bool `json:"enabled,omitempty" yaml:"enabled"` + Provider string `json:"provider,omitempty" yaml:"provider"` // none|gemini|openai|qwen|pipecat ("" == none) + APIKey string `json:"api_key,omitempty" yaml:"apiKey"` // empty → falls back to LLMAPIKey + BaseURL string `json:"base_url,omitempty" yaml:"baseURL"` // empty → falls back to LLMBaseURL + Gemini *GeminiRealtime `json:"gemini,omitempty" yaml:"gemini"` + OpenAI *OpenAIRealtime `json:"openai,omitempty" yaml:"openai"` + Qwen *QwenRealtime `json:"qwen,omitempty" yaml:"qwen"` + Pipecat *PipecatRealtime `json:"pipecat,omitempty" yaml:"pipecat"` } // GeminiRealtime holds Gemini Live's provider-specific knobs. Empty fields → HAL @@ -76,6 +77,23 @@ type QwenRealtime struct { Search *bool `json:"search,omitempty" yaml:"search"` } +// PipecatRealtime holds the cascaded pipecat brain's knobs. It is NOT an +// audio-native provider: HAL keeps its own STT/TTS and pipecat drives only the +// LLM half of the turn, so there is no voice and no reasoning field. The +// endpoint is any OpenAI-compatible /v1 host; empty api_key/base_url/model fall +// back to the AI brain's (llm_api_key / llm_base_url / llm_model), which is +// already such a host. The shared realtime.api_key/base_url are deliberately +// NOT consulted — those carry the WS-suffixed campaign-api values gemini and +// openai use. +type PipecatRealtime struct { + APIKey string `json:"api_key,omitempty" yaml:"apiKey"` + BaseURL string `json:"base_url,omitempty" yaml:"baseURL"` // e.g. https://host/v1 + Model string `json:"model,omitempty" yaml:"model"` + // SearchAPIKey enables the `web_search` tool (Gemini grounding). Separate + // key because the gateway above is typically self-hosted with no search. + SearchAPIKey string `json:"search_api_key,omitempty" yaml:"searchAPIKey"` +} + // Realtime per-provider defaults — what os-server resolves (and pushes) when the // operator hasn't overridden a knob. Model/voice match HAL's defaults // (hal/config.py): the Gemini model is the flash (cheapest) live variant. The @@ -196,12 +214,19 @@ func (c *Config) RealtimeBaseURLOverride() string { if c.Realtime == nil { return "" } - // qwen keeps its own base_url in the sub-object (never the shared field). - if c.RealtimeProvider() == "qwen" { + // qwen and pipecat keep their own base_url in the sub-object (never the + // shared field, which carries the WS-suffixed campaign-api value). + switch c.RealtimeProvider() { + case "qwen": if c.Realtime.Qwen != nil { return c.Realtime.Qwen.BaseURL } return "" + case "pipecat": + if c.Realtime.Pipecat != nil { + return c.Realtime.Pipecat.BaseURL + } + return "" } return c.Realtime.BaseURL } @@ -212,8 +237,11 @@ func (c *Config) RealtimeHasAPIKey() bool { if c.Realtime == nil { return false } - if c.RealtimeProvider() == "qwen" { + switch c.RealtimeProvider() { + case "qwen": return c.Realtime.Qwen != nil && c.Realtime.Qwen.APIKey != "" + case "pipecat": + return c.Realtime.Pipecat != nil && c.Realtime.Pipecat.APIKey != "" } return c.Realtime.APIKey != "" } @@ -237,6 +265,12 @@ func (c *Config) RealtimeModel() string { return c.Realtime.Qwen.Model } return defaultRealtimeQwenModel + case "pipecat": + // No default of its own: unset means "reuse the AI brain's model", which + // HAL resolves from llm_model. + if c.Realtime != nil && c.Realtime.Pipecat != nil { + return c.Realtime.Pipecat.Model + } } return "" } @@ -301,7 +335,7 @@ var ( // cheapest (the default). Voices match the maps below; KEEP IN SYNC with the HAL // enums (hal/realtime/enums). var ( - RealtimeProviders = []string{"gemini", "openai", "qwen", "none"} + RealtimeProviders = []string{"gemini", "openai", "qwen", "pipecat", "none"} RealtimeGeminiVoiceList = []string{"Puck", "Charon", "Kore", "Fenrir", "Aoede"} RealtimeOpenAIVoiceList = []string{"alloy", "ash", "coral", "echo", "fable", "onyx", "nova", "sage", "shimmer"} RealtimeQwenVoiceList = []string{"Cherry", "Serena", "Ethan", "Chelsie"} @@ -321,8 +355,8 @@ type RealtimeOptions struct { func GetRealtimeOptions() RealtimeOptions { return RealtimeOptions{ Providers: RealtimeProviders, - Voices: map[string][]string{"gemini": RealtimeGeminiVoiceList, "openai": RealtimeOpenAIVoiceList, "qwen": RealtimeQwenVoiceList}, - Reasoning: map[string][]string{"gemini": RealtimeGeminiThinkingList, "openai": RealtimeOpenAIReasoningList, "qwen": {}}, + Voices: map[string][]string{"gemini": RealtimeGeminiVoiceList, "openai": RealtimeOpenAIVoiceList, "qwen": RealtimeQwenVoiceList, "pipecat": {}}, + Reasoning: map[string][]string{"gemini": RealtimeGeminiThinkingList, "openai": RealtimeOpenAIReasoningList, "qwen": {}, "pipecat": {}}, } } @@ -330,10 +364,10 @@ func GetRealtimeOptions() RealtimeOptions { // the off-synonyms / empty). Anything else is rejected. func ValidateRealtimeProvider(provider string) error { switch strings.ToLower(strings.TrimSpace(provider)) { - case "gemini", "openai", "qwen", "none", "off", "disabled", "": + case "gemini", "openai", "qwen", "pipecat", "none", "off", "disabled", "": return nil default: - return fmt.Errorf("invalid realtime provider %q (want gemini|openai|qwen|none)", provider) + return fmt.Errorf("invalid realtime provider %q (want gemini|openai|qwen|pipecat|none)", provider) } } @@ -364,8 +398,17 @@ func ValidateRealtimeKnobs(provider, voice, reasoning string) error { if reasoning != "" { return fmt.Errorf("qwen realtime has no reasoning knob, got %q", reasoning) } + case "pipecat": + // Cascaded: the device's own TTS owns the voice and the model exposes no + // reasoning tier, so both knobs are meaningless here. + if voice != "" { + return fmt.Errorf("pipecat has no voice knob (device TTS owns it), got %q", voice) + } + if reasoning != "" { + return fmt.Errorf("pipecat has no reasoning knob, got %q", reasoning) + } default: - return fmt.Errorf("realtime model/voice/reasoning require a concrete provider (gemini|openai|qwen), got %q", provider) + return fmt.Errorf("realtime model/voice/reasoning require a concrete provider (gemini|openai|qwen|pipecat), got %q", provider) } return nil } diff --git a/system/web/src/pages/settings/RealtimeSection.tsx b/system/web/src/pages/settings/RealtimeSection.tsx index ccc58999d..262389d1f 100644 --- a/system/web/src/pages/settings/RealtimeSection.tsx +++ b/system/web/src/pages/settings/RealtimeSection.tsx @@ -3,11 +3,13 @@ import { C, LockedField, LockedPasswordField, SectionCard } from "@/components/s import { getRealtimeOptions } from "@/lib/api"; import type { LlmLoadedState } from "@/hooks/setup/types"; -// Realtime voice-agent (Gemini Live / OpenAI Realtime) config. Values map 1:1 to -// the config.json `realtime` block (HAL reads it; os-server restarts HAL on save). -// Voice + reasoning are provider-specific — keep these lists in sync with +// Realtime voice-agent config. Values map 1:1 to the config.json `realtime` +// block (HAL reads it; os-server restarts HAL on save). Voice + reasoning are +// provider-specific — keep these lists in sync with // system/server/config/realtime.go (ValidateRealtimeKnobs) and the HAL enums. -const PROVIDERS = ["gemini", "openai", "qwen", "none"]; +// pipecat is the odd one out: cascaded rather than audio-native, so it has +// neither knob and its endpoint is a plain OpenAI-compatible /v1 host. +const PROVIDERS = ["gemini", "openai", "qwen", "pipecat", "none"]; // Display labels for the Provider dropdown. Values on the wire stay lowercase // (server-side switch keys off "gemini" / "openai" / …); only the human-facing @@ -17,6 +19,7 @@ const PROVIDER_LABEL: Record = { gemini: "Gemini", openai: "OpenAI", qwen: "Qwen", + pipecat: "Pipecat", none: "None", }; const displayProvider = (v: string): string => @@ -25,15 +28,25 @@ const VOICES: Record = { gemini: ["Puck", "Charon", "Kore", "Fenrir", "Aoede"], openai: ["alloy", "ash", "coral", "echo", "fable", "onyx", "nova", "sage", "shimmer"], qwen: ["Cherry", "Serena", "Ethan", "Chelsie"], + pipecat: [], }; // Reasoning depth = cost knob. First entry (cheapest) is the default. -// qwen realtime has no reasoning knob → empty list hides the selector. +// qwen realtime and pipecat have no reasoning knob → empty list hides the selector. const REASONING: Record = { gemini: ["MINIMAL", "LOW", "MEDIUM", "HIGH"], openai: ["minimal", "low", "medium", "high", "xhigh"], qwen: [], + pipecat: [], }; +// A provider only accepts the knobs it actually has: pipecat has neither (the +// device TTS owns the voice, and the model exposes no reasoning tier) and qwen +// has no reasoning. The server REJECTS an unsupported knob, so the save payload +// must be filtered by these — hiding the selector is not enough, since the +// state keeps whatever the previous provider left behind. +export const providerHasVoice = (p: string): boolean => (VOICES[p] ?? []).length > 0; +export const providerHasReasoning = (p: string): boolean => (REASONING[p] ?? []).length > 0; + export interface RealtimeLoadedState { apiKey: boolean; } @@ -88,7 +101,7 @@ export function RealtimeSection({
@@ -118,7 +131,7 @@ export function RealtimeSection({ )} - + )} diff --git a/system/web/src/pages/settings/SettingsPanel.tsx b/system/web/src/pages/settings/SettingsPanel.tsx index 50bc3f4f4..af940cb73 100644 --- a/system/web/src/pages/settings/SettingsPanel.tsx +++ b/system/web/src/pages/settings/SettingsPanel.tsx @@ -11,7 +11,7 @@ import { WifiSection } from "@/pages/settings/WifiSection"; import { VoiceSection as EditVoiceSection } from "@/pages/settings/VoiceSection"; import { FaceSection as EditFaceSection } from "@/pages/settings/FaceSection"; import { TTSSection } from "@/pages/settings/TTSSection"; -import { RealtimeSection } from "@/pages/settings/RealtimeSection"; +import { RealtimeSection, providerHasReasoning, providerHasVoice } from "@/pages/settings/RealtimeSection"; import { AgentRuntimeSection } from "@/pages/settings/AgentRuntimeSection"; import { TimezoneSection } from "@/pages/settings/TimezoneSection"; import { STTSection, type SttProvider } from "@/pages/settings/STTSection"; @@ -440,7 +440,13 @@ export function SettingsPanel({ activeSection }: { activeSection: SettingsSectio if (adminPassword) body.admin_password = adminPassword; // Realtime block — server applies + restarts hal. api_key only when typed. const realtime: Record = { enabled: realtimeEnabled, provider: realtimeProvider }; - if (realtimeProvider !== "none") { realtime.voice = realtimeVoice; realtime.reasoning = realtimeReasoning; } + // Only the knobs this provider actually has: the server rejects the rest, + // and the state still holds the previous provider's values (a provider whose + // knob the server reports empty leaves the useState default in place). + if (realtimeProvider !== "none") { + if (providerHasVoice(realtimeProvider)) realtime.voice = realtimeVoice; + if (providerHasReasoning(realtimeProvider)) realtime.reasoning = realtimeReasoning; + } if (realtimeBaseUrl) realtime.base_url = realtimeBaseUrl; if (realtimeApiKey) realtime.api_key = realtimeApiKey; body.realtime = realtime;