Integracion de Proveedores

VOCALS utiliza un patron de abstraccion de proveedores que te permite agregar implementaciones personalizadas de STT, LLM y TTS sin modificar el pipeline principal. Esta guia recorre las interfaces base, los metodos requeridos y el proceso de registro.

Interfaces Base

Todos los proveedores heredan de BaseProvider, que maneja el almacenamiento de API keys y requiere un metodo validate(). Cada etapa del pipeline tiene su propia clase abstracta.

BaseProvider

# backend/app/providers/base.py

class BaseProvider(ABC):
    def __init__(self, api_key: str, config: Optional[Dict[str, Any]] = None):
        self.api_key = api_key
        self.config = config or {}

    @abstractmethod
    async def validate(self) -> bool:
        """Validate the provider configuration and API key.
        Returns True if valid, raises an exception with details otherwise."""
        ...

    async def list_models(self) -> List[str]:
        """Return available model IDs. Override in subclasses.
        Returns an empty list by default."""
        return []

STTProvider

class STTProvider(BaseProvider):
    @abstractmethod
    async def transcribe(self, audio_stream: AsyncIterator[bytes]) -> AsyncIterator[str]:
        """Transcribe streaming audio to text.

        Args:
            audio_stream: Async iterator of PCM audio chunks (16kHz, 16-bit mono).

        Yields:
            Transcribed text fragments (partial or final results).
        """
        ...

LLMProvider

class LLMProvider(BaseProvider):
    @abstractmethod
    async def generate(
        self,
        messages: list,
        system_prompt: Optional[str] = None,
    ) -> AsyncIterator[str]:
        """Generate a streaming response from the LLM.

        Args:
            messages: Conversation history as list of dicts with 'role' and 'content'.
            system_prompt: Optional system prompt to prepend.

        Yields:
            Text tokens/chunks as they are generated.
        """
        ...

TTSProvider

class TTSProvider(BaseProvider):
    @abstractmethod
    async def synthesize(self, text: str) -> AsyncIterator[bytes]:
        """Synthesize text to streaming audio.

        Args:
            text: Text to convert to speech.

        Yields:
            Audio data chunks (PCM 16kHz, 16-bit mono).
        """
        ...

Implementacion de un Proveedor Personalizado

Aqui tienes un ejemplo completo de como agregar un proveedor TTS personalizado.

Paso 1: Crear la Clase del Proveedor

Crea un nuevo archivo en el subdirectorio correspondiente:

# backend/app/providers/tts/my_tts.py

from typing import Any, AsyncIterator, Dict, List, Optional
import httpx

from app.providers.base import TTSProvider


class MyTTSProvider(TTSProvider):
    """Custom TTS provider implementation."""

    # Hardcoded model list (or fetch from API in list_models)
    MODELS = ["model-standard", "model-hd"]

    def __init__(self, api_key: str, config: Optional[Dict[str, Any]] = None):
        super().__init__(api_key, config)
        self.model = config.get("model", "model-standard")
        self.voice = config.get("voice_id", "default")
        self.base_url = "https://api.my-tts-service.com/v1"

    async def validate(self) -> bool:
        """Test the API key by making a lightweight API call."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}/voices",
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=10.0,
            )
            if resp.status_code == 401:
                raise ValueError("Invalid API key")
            resp.raise_for_status()
        return True

    async def list_models(self) -> List[str]:
        """Return available models. Can query the API or return a static list."""
        return self.MODELS

    async def synthesize(self, text: str) -> AsyncIterator[bytes]:
        """Stream synthesized audio as PCM 16kHz 16-bit mono chunks."""
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{self.base_url}/synthesize",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "text": text,
                    "model": self.model,
                    "voice": self.voice,
                    "output_format": "pcm_16000",
                },
                timeout=30.0,
            ) as resp:
                resp.raise_for_status()
                async for chunk in resp.aiter_bytes(chunk_size=4096):
                    yield chunk

Paso 2: Registrar el Proveedor

Agrega la importacion y la llamada de registro al registro de proveedores:

# backend/app/providers/registry.py

def _register_all() -> None:
    # ... existing registrations ...

    from app.providers.tts.my_tts import MyTTSProvider
    register_provider("tts", "my_tts", MyTTSProvider)

La funcion register_provider recibe tres argumentos:

Argumento	Tipo	Descripcion
`provider_type`	string	`stt`, `llm`, o `tts`
`name`	string	Identificador unico usado en la API (ej. `my_tts`)
`cls`	Type[BaseProvider]	La clase del proveedor

Paso 3: Usar el Proveedor

Despues del registro, el proveedor esta disponible inmediatamente a traves de la API:

# Crear una configuracion de proveedor
curl -X POST \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "tts",
    "name": "my_tts",
    "api_key": "your-api-key-here",
    "model_id": "model-hd",
    "extra_config": { "voice_id": "custom-voice" }
  }' \
  https://api.usevocals.com/api/v1/providers

# Probar el proveedor
curl -X POST \
  -H "Authorization: Bearer $JWT" \
  https://api.usevocals.com/api/v1/providers/{provider_id}/test

# Listar modelos disponibles
curl -H "Authorization: Bearer $JWT" \
  https://api.usevocals.com/api/v1/providers/{provider_id}/models

Luego asigna el proveedor a un agente:

curl -X PUT \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{"active_tts_provider_id": "provider-uuid-here"}' \
  https://api.usevocals.com/api/v1/agents/{agent_id}

Registro de Proveedores

El registro (backend/app/providers/registry.py) mantiene tres diccionarios -- uno por tipo de proveedor -- que mapean nombres a clases:

_stt_providers: Dict[str, Type[STTProvider]] = {}
_llm_providers: Dict[str, Type[LLMProvider]] = {}
_tts_providers: Dict[str, Type[TTSProvider]] = {}

Funciones principales:

Funcion	Descripcion
`register_provider(type, name, cls)`	Registrar una clase de proveedor
`get_provider(type, name, api_key, config)`	Instanciar un proveedor registrado
`list_providers(type)`	Listar nombres de proveedores registrados para un tipo

La funcion _register_all() se ejecuta en tiempo de importacion y registra todos los proveedores incluidos.

Proveedores Registrados Actualmente

Proveedores STT

Nombre	Clase	Descripcion
`deepgram`	`DeepgramSTTProvider`	STT en tiempo real via Deepgram Nova
`openai`	`OpenAISTTProvider`	API de OpenAI Whisper
`whisper`	`WhisperSTTProvider`	OpenAI Whisper (local)
`elevenlabs`	`ElevenLabsSTTProvider`	ElevenLabs STT
`qwen`	`QwenSTTProvider`	Alibaba Qwen STT
`fish`	`FishSTTProvider`	Fish Audio STT

Proveedores LLM

Nombre	Clase	Descripcion
`openai`	`OpenAILLMProvider`	GPT-4o, GPT-4, GPT-3.5
`claude`	`ClaudeLLMProvider`	Modelos Anthropic Claude
`google`	`GoogleLLMProvider`	Modelos Google Gemini
`kimi`	`KimiLLMProvider`	Modelos Moonshot Kimi

Proveedores TTS

Nombre	Clase	Descripcion
`deepgram`	`DeepgramTTSProvider`	Deepgram Aura TTS
`openai`	`OpenAITTSProvider`	OpenAI TTS
`elevenlabs`	`ElevenLabsTTSProvider`	ElevenLabs TTS
`qwen`	`QwenTTSProvider`	Alibaba Qwen TTS
`resemble`	`ResembleTTSProvider`	Resemble AI TTS
`fish`	`FishTTSProvider`	Fish Audio TTS

Requisitos de Formato de Audio

Todos los proveedores STT reciben audio como un AsyncIterator[bytes] de fragmentos PCM 16kHz, 16-bit mono. El orquestador se encarga de la conversion desde el formato mulaw/8kHz de Twilio antes de pasar el audio al proveedor STT.

Todos los proveedores TTS deben producir audio como fragmentos PCM 16kHz, 16-bit mono. El orquestador convierte la salida de vuelta a mulaw/8kHz para Twilio.

Si la API de tu proveedor utiliza un formato diferente, realiza la conversion dentro de tu implementacion del proveedor.

Buenas Practicas

Streaming: Usa APIs de streaming cuando esten disponibles. El pipeline procesa audio en tiempo real, por lo que las APIs por lotes agregan una latencia significativa.
Manejo de timeouts: Establece timeouts razonables en las llamadas HTTP. El orquestador reintentara o degradara elegantemente ante fallos del proveedor.
Mensajes de error: Lanza ValueError desde validate() con un mensaje claro (ej. "Invalid API key", "Model not found"). Estos mensajes se devuelven al usuario a traves de la API.
Listado de modelos: Implementa list_models() para consultar la API del proveedor cuando sea posible. Recurre a una lista predefinida si la API no soporta la enumeracion de modelos. Devuelve una lista vacia solo como ultimo recurso.
Acceso a configuracion: Usa self.config para leer configuraciones especificas del proveedor pasadas a traves de extra_config al crear o actualizar un proveedor mediante la API.

Interfaces Base​

BaseProvider​

STTProvider​

LLMProvider​

TTSProvider​

Implementacion de un Proveedor Personalizado​

Paso 1: Crear la Clase del Proveedor​

Paso 2: Registrar el Proveedor​

Paso 3: Usar el Proveedor​

Registro de Proveedores​

Proveedores Registrados Actualmente​

Proveedores STT​

Proveedores LLM​

Proveedores TTS​

Requisitos de Formato de Audio​

Buenas Practicas​