Praxis CI
bba99418df
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---
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""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()) |