From bba99418dfddfd8dcfe17ebf57d6b4549257300b Mon Sep 17 00:00:00 2001 From: Praxis CI Date: Sat, 1 Aug 2026 13:04:17 +0000 Subject: [PATCH] =?UTF-8?q?feat(P01-02-04):=20Pipecat=20server=20pipeline?= =?UTF-8?q?=20=E2=80=94=20Silero=20VAD=E2=86=92Deepgram=E2=86=92Ollama?= =?UTF-8?q?=E2=86=92TTS=E2=86=92WebRTC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit server/pipeline.py assembles the Pipecat pipeline (D-017): WebRTC audio in → Deepgram Nova-3 STT → LLMContextAggregator(user) → OLLamaLLMService (gemma4:cloud via https://ollama.com/v1 + bearer, R6) → Cartesia/Piper TTS (selected via PRAXIS_TTS) → WebRTC audio out. Interruptibility via allow_interruptions=True (D-008 abort-and-yield). Hardcoded single-turn system prompt (SLICE-03 replaces with scenario YAML). All keys from env; missing keys log a warning and the pipeline still starts (code structure is the deliverable). server/__main__.py exposes a FastAPI app with /health (reports key-provisioning status) and POST /pipecat/webrtc (accepts an SDP offer, starts a pipeline task, returns the answer). Verified: imports succeed, /health returns 200, routes wired. ---ci--- phase: 1 milestone: v0.1 plan: 02 task: 02-04 status: execute persona: backend-engineer requirements: covered: [REQ-VOICE-01, REQ-VOICE-02, REQ-VOICE-04, REQ-ORCH-01] ---/ci--- --- server/__main__.py | 108 ++++++++++++++++++++++++++ server/pipeline.py | 185 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 293 insertions(+) create mode 100644 server/__main__.py create mode 100644 server/pipeline.py diff --git a/server/__main__.py b/server/__main__.py new file mode 100644 index 0000000..cae260d --- /dev/null +++ b/server/__main__.py @@ -0,0 +1,108 @@ +"""Praxis server entrypoint — starts the Pipecat WebRTC bot server. + +Run: `python -m server` + +Exposes a FastAPI app with: + GET /health — liveness + POST /pipecat/webrtc — accept a WebRTC offer SDP, start a pipeline task + +The server starts and accepts connections even if upstream voice-service keys +are absent (SLICE-02 deliverable = code structure). Missing keys degrade to +no audio/no tokens at runtime, not a crash. +""" + +from __future__ import annotations + +import os +from typing import Any + +from loguru import logger +from pydantic import BaseModel + +# Load .env if present (dev). In production, env is injected directly. +try: + from dotenv import load_dotenv + + load_dotenv() +except ImportError: # pragma: no cover + pass + +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + +from server.pipeline import build_pipeline + + +def _env(key: str, default: str = "") -> str: + return os.environ.get(key, default).strip() + + +HOST = _env("PRAXIS_HOST", "0.0.0.0") +PORT = int(_env("PRAXIS_PORT", "8789")) + + +class WebRTCOffer(BaseModel): + """Client→server WebRTC offer (SDP + type).""" + + sdp: str + type: str = "offer" + + +app = FastAPI(title="Praxis v0.1 voice server", version="0.1.0") +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # dev — the client is a separate Vite origin + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.get("/health") +async def health() -> dict[str, Any]: + """Liveness probe. Reports key-provisioning status for the client.""" + return { + "status": "ok", + "version": "0.1.0", + "keys": { + "deepgram": bool(_env("DEEPGRAM_API_KEY")), + "cartesia": bool(_env("CARTESIA_API_KEY")), + "ollama": bool(_env("OLLAMA_API_KEY")), + }, + "tts": _env("PRAXIS_TTS", "cartesia"), + } + + +@app.post("/pipecat/webrtc") +async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]: + """Accept a WebRTC offer, start a Pipecat pipeline task, return the answer.""" + try: + connection = SmallWebRTCConnection( + ice_servers=[{"urls": "stun:stun.l.google.com:19302"}], + ) + await connection.receive_offer({"sdp": offer.sdp, "type": offer.type}) + await connection.accept() + answer = connection.get_answer() + # Build + run the pipeline for this connection. + pipeline, task, runner, transport = build_pipeline(connection) + # Run the pipeline task in the background; the runner manages its lifecycle. + import asyncio + + asyncio.create_task(runner.run(task)) + return {"sdp": answer["sdp"], "type": answer["type"]} + except Exception as exc: + logger.error(f"WebRTC offer failed: {exc}") + raise HTTPException(status_code=500, detail=str(exc)) + + +def main() -> int: + """Run the server with uvicorn.""" + import uvicorn + + logger.info(f"Praxis v0.1 voice server starting on {HOST}:{PORT}") + uvicorn.run(app, host=HOST, port=PORT, log_level="info") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/server/pipeline.py b/server/pipeline.py new file mode 100644 index 0000000..325e3c7 --- /dev/null +++ b/server/pipeline.py @@ -0,0 +1,185 @@ +"""Praxis Pipecat server pipeline — minimal viable voice loop (SLICE-02 TASK-02-04). + +Pipeline (D-017): + WebRTC audio in → Silero VAD → Deepgram Nova-3 STT → LLMContextAggregator(user) + → OllamaCloudLLM (gemma4:cloud) → LLMContextAggregator(assistant) → Cartesia/Piper TTS + → WebRTC audio out + +Interruptibility (D-008): Pipecat's built-in interrupt handling aborts TTS + yields +the floor when learner VAD fires during AI speech. + +The pipeline starts and accepts connections even if upstream services return auth +errors at runtime — the code structure is the SLICE-02 deliverable. All keys come +from env; missing keys degrade to no audio / no tokens, not crashes. + +Hardcoded single-turn system prompt (no YAML scenario yet — SLICE-03 replaces it). +""" + +from __future__ import annotations + +import os +from typing import Any + +from loguru import logger + + +def _env(key: str, default: str = "") -> str: + return os.environ.get(key, default).strip() + + +# Hardcoded single-turn system prompt (SLICE-02 walking skeleton). +# SLICE-03 TASK-03-07 replaces this with the scenario-driven prompt from YAML. +WALKING_SKELETON_SYSTEM_PROMPT = ( + "You are Jordan, a customer who received a damaged product. " + "You are frustrated but not abusive. You want a refund. " + "Stay in character. Do not break role. " + "Keep responses concise for voice (1-3 sentences)." +) + +WALKING_SKELETON_OPENING_LINE = ( + "Hi, I received my order yesterday and the item is cracked. I want my money back." +) + + +def _build_llm_context(): + """Build the LLMContext with the walking-skeleton system prompt.""" + from pipecat.processors.aggregators.llm_context import LLMContext + + messages = [ + {"role": "system", "content": WALKING_SKELETON_SYSTEM_PROMPT}, + ] + return LLMContext(messages=messages) + + +def _build_transport(webrtc_connection) -> Any: + """Build the SmallWebRTCTransport with audio in/out enabled.""" + from pipecat.transports.base_transport import TransportParams + from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport + + params = TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + audio_out_sample_rate=24000, + ) + return SmallWebRTCTransport(webrtc_connection, params) + + +def _build_stt() -> Any: + """Build the Deepgram Nova-3 STT service (D-013).""" + from pipecat.services.deepgram.stt import DeepgramSTTService + + api_key = _env("DEEPGRAM_API_KEY") + if not api_key: + logger.warning("DEEPGRAM_API_KEY not set — STT will not transcribe (pipeline still starts).") + return DeepgramSTTService( + api_key=api_key or "missing", + live_options=None, # Deepgram defaults are fine for nova-3 + en. + ) + + +def _build_llm() -> Any: + """Build the Pipecat Ollama LLM service pointed at Ollama Cloud (D-020, R6). + + Pipecat's OLLamaLLMService extends OpenAILLMService and accepts a custom + base_url + the OpenAI client api_key (bearer). We point it at + https://ollama.com/v1 with OLLAMA_API_KEY as the bearer. + """ + from pipecat.services.ollama.llm import OLLamaLLMService + + api_key = _env("OLLAMA_API_KEY") + base_url = _env("OLLAMA_BASE_URL", "https://ollama.com/v1") + model = _env("OLLAMA_ROLEPLAY_MODEL", "gemma4:cloud") + if not api_key: + logger.warning("OLLAMA_API_KEY not set — LLM will not respond (pipeline still starts).") + return OLLamaLLMService( + base_url=base_url, + settings=OLLamaLLMService.Settings(model=model, api_key=api_key or "missing"), + ) + + +def _build_tts() -> Any: + """Build the Pipecat TTS service for the selected provider (D-014).""" + choice = _env("PRAXIS_TTS", "cartesia").lower() + + if choice == "piper": + 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", + ) + + # Default: Cartesia + from pipecat.services.cartesia.tts import CartesiaTTSService + + api_key = _env("CARTESIA_API_KEY") + voice_id = _env("CARTESIA_VOICE_ID", "a3536a36-1d18-4efb-a95a-7c44b7b5e384") + if not api_key: + logger.warning("CARTESIA_API_KEY not set — TTS will not speak (pipeline still starts).") + return CartesiaTTSService( + api_key=api_key or "missing", + voice_id=voice_id, + ) + + +def _build_vad_analyzer() -> Any: + """Build the Silero VAD analyzer (D-008 interruptibility).""" + from pipecat.audio.vad.silero import SileroVADAnalyzer + + return SileroVADAnalyzer() + + +def build_pipeline(webrtc_connection): + """Assemble the full Pipecat pipeline + task + runner for one WebRTC session. + + Returns (pipeline, task, runner, transport) so the caller can start the task + on connection and tear it down on disconnect. + """ + from pipecat.pipeline.pipeline import Pipeline + from pipecat.pipeline.runner import PipelineRunner + from pipecat.pipeline.task import PipelineParams, PipelineTask + from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregator, + ) + + transport = _build_transport(webrtc_connection) + stt = _build_stt() + llm = _build_llm() + tts = _build_tts() + + context = _build_llm_context() + user_aggregator = LLMContextAggregator(context=context, role="user") + assistant_aggregator = LLMContextAggregator(context=context, role="assistant") + + pipeline = Pipeline( + [ + transport.input(), # WebRTC audio in + stt, # Deepgram Nova-3 + user_aggregator, # collect user transcript into context + llm, # Ollama gemma4:cloud + tts, # Cartesia/Piper + 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 (TASK-02-06) + metrics_request_timeout=10.0, + ), + ) + + runner = PipelineRunner(handle_sigint=False) + return pipeline, task, runner, transport + + +__all__ = [ + "build_pipeline", + "WALKING_SKELETON_SYSTEM_PROMPT", + "WALKING_SKELETON_OPENING_LINE", +] \ No newline at end of file