AgapHost/openai/pipecat/bot.py

import asyncio
import os
import re
import uuid
import logging

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from livekit import api as lkapi

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.openai.stt import OpenAISTTService
from pipecat.services.openai.tts import OpenAITTSService
from pipecat.transports.livekit.transport import LiveKitTransport, LiveKitParams


# ── TTS text normalizer ──────────────────────────────────────────────────────
# Replaces symbols and abbreviations with spoken Russian words so Silero TTS
# doesn't truncate on unknown characters.

_NORM_RULES: list[tuple[re.Pattern, str]] = [
    # Temperature: +12°C / -5°С / 12 °C → плюс двенадцать градусов цельсия
    (re.compile(r"([+-]?\d+)\s*°\s*[CСcс]", re.IGNORECASE), r"\1 градусов цельсия"),
    # Bare degree sign: 90° → 90 градусов
    (re.compile(r"(\d+)\s*°"), r"\1 градусов"),
    # Percent
    (re.compile(r"(\d+)\s*%"), r"\1 процентов"),
    # Speed: m/s, м/с, km/h, км/ч
    (re.compile(r"\bm/s\b", re.IGNORECASE), "метров в секунду"),
    (re.compile(r"\bм/с\b"), "метров в секунду"),
    (re.compile(r"\bkm/h\b", re.IGNORECASE), "километров в час"),
    (re.compile(r"\bкм/ч\b"), "километров в час"),
    # Currency
    (re.compile(r"\$\s*(\d+)"), r"\1 долларов"),
    (re.compile(r"(\d+)\s*\$"), r"\1 долларов"),
    (re.compile(r"€\s*(\d+)"), r"\1 евро"),
    (re.compile(r"(\d+)\s*€"), r"\1 евро"),
    (re.compile(r"(\d+)\s*₽"), r"\1 рублей"),
    # Plus/minus signs before numbers
    (re.compile(r"\+(\d)"), r"плюс \1"),
    (re.compile(r"-(\d)"), r"минус \1"),
    # Common abbreviations
    (re.compile(r"\bкг\b"), "килограмм"),
    (re.compile(r"\bг\b(?=\s|$)"), "грамм"),
    (re.compile(r"\bмм\b"), "миллиметров"),
    (re.compile(r"\bсм\b"), "сантиметров"),
    (re.compile(r"\bкм\b"), "километров"),
    # Strip remaining special chars that TTS can't handle
    (re.compile(r"[°•·†‡§¶©®™«»<>{}[\]|\\~^`]"), ""),
]


def normalize_for_tts(text: str) -> str:
    """Replace symbols with spoken Russian equivalents."""
    for pattern, replacement in _NORM_RULES:
        text = pattern.sub(replacement, text)
    return text


class TTSTextNormalizer(FrameProcessor):
    """Intercepts TextFrames between LLM and TTS, normalizing symbols to words."""

    async def process_frame(self, frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
        await super().process_frame(frame, direction)
        if isinstance(frame, TextFrame):
            original = frame.text
            normalized = normalize_for_tts(original)
            if normalized != original:
                logger.debug(f"TTSTextNormalizer: {original!r} → {normalized!r}")
            await self.push_frame(TextFrame(text=normalized), direction)
        else:
            await self.push_frame(frame, direction)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ── Config ────────────────────────────────────────────────────────────────────
LIVEKIT_URL        = os.getenv("LIVEKIT_URL",        "ws://host.docker.internal:7880")  # bot connects here
LIVEKIT_PUBLIC_URL = os.getenv("LIVEKIT_PUBLIC_URL", "wss://lk.alogins.net")            # browser connects here
LIVEKIT_API_KEY    = os.getenv("LIVEKIT_API_KEY",    "devkey")
LIVEKIT_SECRET     = os.getenv("LIVEKIT_SECRET",     "")
ADOLF_URL       = os.getenv("ADOLF_URL",        "http://host.docker.internal:8000/v1")
STT_URL         = os.getenv("STT_URL",          "http://host.docker.internal:8880/v1")
TTS_URL         = os.getenv("TTS_URL",          "http://host.docker.internal:8881/v1")
STT_MODEL       = os.getenv("STT_MODEL",        "deepdml/faster-whisper-large-v3-turbo-ct2")
TTS_VOICE       = os.getenv("TTS_VOICE",        "onyx")

SYSTEM_PROMPT = "You are Adolf, a helpful voice assistant. Keep replies concise — 1-3 sentences. No markdown."

app = FastAPI(title="Pipecat Voice Bot")
app.mount("/static", StaticFiles(directory="static"), name="static")


# ── LiveKit helpers ───────────────────────────────────────────────────────────
def _lk_token(room: str, identity: str, is_bot: bool = False) -> str:
    grants = lkapi.VideoGrants(
        room_join=True,
        room=room,
        can_publish=True,
        can_subscribe=True,
        can_publish_data=True,
    )
    token = (
        lkapi.AccessToken(LIVEKIT_API_KEY, LIVEKIT_SECRET)
        .with_identity(identity)
        .with_name("Adolf Bot" if is_bot else identity)
        .with_grants(grants)
    )
    return token.to_jwt()


async def _create_room(room_name: str) -> None:
    lk = lkapi.LiveKitAPI(LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_SECRET)
    try:
        await lk.room.create_room(
            lkapi.CreateRoomRequest(name=room_name, empty_timeout=300, max_participants=5)
        )
    finally:
        await lk.aclose()


# ── Pipecat pipeline ──────────────────────────────────────────────────────────
async def _run_bot(room_name: str) -> None:
    bot_token = _lk_token(room_name, "pipecat-bot", is_bot=True)

    transport = LiveKitTransport(
        url=LIVEKIT_URL,
        token=bot_token,
        room_name=room_name,
        params=LiveKitParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(params=VADParams(
                stop_secs=0.8,       # wait 0.8s of silence before end-of-speech
                start_secs=0.2,      # start speech detection after 0.2s
                confidence=0.85,     # high confidence to avoid triggering on ambient noise
            )),
        ),
    )

    stt = OpenAISTTService(
        api_key="dummy",
        base_url=STT_URL,
        model=STT_MODEL,
        language="ru",
    )

    llm = OpenAILLMService(
        api_key="dummy",
        base_url=ADOLF_URL,
        model="adolf-light",
    )

    tts = OpenAITTSService(
        api_key="dummy",
        base_url=TTS_URL,
        model="silero",
        voice=TTS_VOICE,
    )

    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    normalizer = TTSTextNormalizer()

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        normalizer,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=False))

    @transport.event_handler("on_participant_disconnected")
    async def on_disconnect(transport, participant):
        identity = participant if isinstance(participant, str) else getattr(participant, "identity", str(participant))
        logger.info(f"Participant {identity} left — stopping bot")
        await task.cancel()

    runner = PipelineRunner()
    logger.info(f"Bot starting in room={room_name}")
    await runner.run(task)
    logger.info(f"Bot done in room={room_name}")


# ── API ───────────────────────────────────────────────────────────────────────
class ConnectResponse(BaseModel):
    room: str
    token: str
    url: str


@app.post("/connect", response_model=ConnectResponse)
async def connect():
    room_name = f"voice-{uuid.uuid4().hex[:6]}"
    await _create_room(room_name)
    user_token = _lk_token(room_name, "user")
    asyncio.create_task(_run_bot(room_name))
    return ConnectResponse(room=room_name, token=user_token, url=LIVEKIT_PUBLIC_URL)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/", response_class=HTMLResponse)
async def index():
    with open("static/index.html") as f:
        return f.read()