praxis/server/assist/pipeline.py

"""build_assist_pipeline — the assist-mode Pipecat pipeline (D-061, D-065, D-066, TASK-05-01).

Reuses the v0.1 voice services (_build_transport, _build_stt, _build_llm from
server/pipeline.py — FIXED, not rewritten). Swaps the system prompt for the
≤150-token assist prompt (AssistContextBinder). Defaults to Piper TTS for
assist (D-065 — ~80ms first audio vs Cartesia ~120ms). Inserts the
LiveAssistGuardrailProcessor between llm and tts (D-060 layer 2).

Pipeline structure:
  transport.input() → stt → latency_observer → user_aggregator → llm →
  latency_observer → LiveAssistGuardrailProcessor → tts → latency_observer →
  transport.output() → assistant_aggregator

No opening line (assist is invoked mid-shift — no scripted opener, unlike
practice which plays the scenario opening line).
"""

from __future__ import annotations

import os
from typing import Any

from loguru import logger

from server.assist.context import AssistContext
from server.assist.guardrail_processor import LiveAssistGuardrailProcessor
from server.guardrails.live_assist import LiveAssistGuardrail


def _env(key: str, default: str = "") -> str:
    return os.environ.get(key, default).strip()


def _build_tts_piper() -> Any:
    """Build the Piper TTS service (D-065 — default for assist, ~80ms first audio)."""
    from pipecat.services.piper.tts import PiperTTSService

    voice_model = _env("PIPER_VOICE_MODEL")
    if not voice_model:
        logger.warning("PIPER_VOICE_MODEL not set — Piper TTS will not speak (pipeline still starts).")
    return PiperTTSService(voice_id=voice_model or "missing")


def _build_tts_assist() -> Any:
    """Build the TTS service for assist mode (D-065).

    Default: Piper (self-hosted, ~80ms). Fallback: Cartesia if
    PRAXIS_ASSIST_TTS=cartesia (for testing without Piper).
    """
    choice = _env("PRAXIS_ASSIST_TTS", "piper").lower()
    if choice == "cartesia":
        from server.pipeline import _build_tts

        return _build_tts()  # Cartesia (practice path)
    return _build_tts_piper()


def build_assist_pipeline(
    webrtc_connection,
    *,
    context: AssistContext,
    guardrail: LiveAssistGuardrail | None = None,
    session: Any | None = None,
) -> tuple:
    """Assemble the assist-mode Pipecat pipeline (TASK-05-01, D-061, D-065, D-066).

    Reuses _build_transport, _build_stt, _build_llm from server/pipeline.py.
    Uses Piper TTS by default (D-065). Inserts the LiveAssistGuardrailProcessor
    between llm and tts. No opening line (assist is invoked mid-shift).

    Returns (pipeline, task, runner, transport) — no scenario_runtime (assist
    has an AssistContext, not a ScenarioRuntime).
    """
    from pipecat.pipeline.pipeline import Pipeline
    from pipecat.pipeline.runner import PipelineRunner
    from pipecat.pipeline.task import PipelineParams, PipelineTask
    from pipecat.processors.aggregators.llm_context import LLMContext
    from pipecat.processors.aggregators.llm_response_universal import (
        LLMContextAggregator,
    )

    from server.latency import LatencyObserver
    from server.pipeline import _build_llm, _build_stt, _build_transport

    transport = _build_transport(webrtc_connection)
    stt = _build_stt()
    llm = _build_llm()
    tts = _build_tts_assist()

    latency_observer = LatencyObserver()

    # Build the LLM context from the ≤150-token assist prompt (D-066).
    llm_context = LLMContext(messages=[{"role": "system", "content": context.system_prompt}])
    user_aggregator = LLMContextAggregator(context=llm_context, role="user")
    assistant_aggregator = LLMContextAggregator(context=llm_context, role="assistant")

    # In-loop guardrail processor (D-060 layer 2, REQ-IDEATE-02).
    if guardrail is None:
        guardrail = LiveAssistGuardrail()
    guardrail_processor = LiveAssistGuardrailProcessor(
        guardrail=guardrail, session=session, llm_context=llm_context
    )

    pipeline = Pipeline(
        [
            transport.input(),          # WebRTC audio in
            stt,                        # Deepgram Nova-3
            latency_observer,           # timestamp ASR-ready
            user_aggregator,            # collect user transcript into context
            llm,                        # Ollama gemma4:cloud (assist prompt)
            latency_observer,           # timestamp LLM-first-token
            guardrail_processor,        # LiveAssistGuardrail (post-LLM, pre-TTS)
            tts,                        # Piper (default) or Cartesia
            latency_observer,           # timestamp TTS-first-audio
            transport.output(),         # WebRTC audio out
            assistant_aggregator,       # collect assistant text into context
        ]
    )

    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            allow_interruptions=True,   # D-008 abort-and-yield
            enable_metrics=True,        # latency measurement
            metrics_request_timeout=10.0,
        ),
    )

    runner = PipelineRunner(handle_sigint=False)
    # No opening line — assist is invoked mid-shift (no scripted opener).
    return pipeline, task, runner, transport


__all__ = ["build_assist_pipeline"]