EchoRoute - Universal Voice Agent Router

A production-quality async Python SDK for routing voice agent requests across multiple providers with a clean plugin architecture.

🚀 Key Features

• Universal Interface: Single API for multiple voice providers
• Plugin Architecture: Providers register themselves using their own APIs
• Production Ready: Comprehensive error handling, retries, and monitoring
• Async First: Built with asyncio for high performance
• Fallback Support: Automatic failover between providers
• Load Balancing: Distribute requests across available providers
• Health Monitoring: Real-time provider health checks
• Extensible: Easy to add new providers

📦 Installation

From PyPI (Recommended)

# Basic installation
pip install echoroute

# With OpenAI support
pip install echoroute[openai]

# With Sarvam AI support
pip install echoroute[sarvam]

# With all providers
pip install echoroute[all]

# For development
pip install echoroute[dev]

From Source

git clone https://github.com/gaurav98095/echoroute.git
cd echoroute
pip install -e .[dev]

🏗️ Architecture

The key architectural principle:

👉 Providers register themselves to the router using their own APIs 👉 After registration, the user interacts with a single common interface 👉 The caller must never care which provider is used

🚀 Quick Start

Basic Usage

import asyncio
from echoroute import (
    VoiceRouter,
    VoiceRequest,
    OpenAIProvider,
    ProviderConfig,
    AudioFormat,
    Language
)

async def main():
    # Create and initialize router
    router = VoiceRouter()
    await router.initialize()

    # Register OpenAI provider
    openai_config = ProviderConfig(
        name="openai",
        api_key="your-openai-api-key"
    )
    openai_provider = OpenAIProvider(openai_config)
    await router.register_provider(openai_provider)

    # Create a voice synthesis request
    request = VoiceRequest(
        text="Hello, world! This is a test of the voice router.",
        language=Language.EN_US,
        output_format=AudioFormat.MP3
    )

    # Process the request - router automatically selects best provider
    response = await router.process_request(request)

    # Access synthesized audio
    audio_data = response.synthesis.audio.data
    with open("output.mp3", "wb") as f:
        f.write(audio_data)

    await router.shutdown()

asyncio.run(main())

Multiple Providers with Fallback

import asyncio
from echoroute import (
    VoiceRouter,
    VoiceRequest,
    OpenAIProvider,
    MockProvider,  # For testing
    ProviderConfig,
    Priority
)

async def main():
    router = VoiceRouter()
    await router.initialize()

    # Register primary provider (OpenAI)
    primary_config = ProviderConfig(
        name="openai-primary",
        api_key="your-openai-key",
        priority=Priority.HIGH
    )
    await router.register_provider(OpenAIProvider(primary_config))

    # Register fallback provider
    fallback_config = ProviderConfig(
        name="fallback",
        priority=Priority.NORMAL
    )
    await router.register_provider(MockProvider(fallback_config))

    # Request will try OpenAI first, fallback to Mock if needed
    request = VoiceRequest(text="Hello with fallback support!")
    response = await router.process_request(request, fallback=True)

    print(f"Processed by: {response.provider_name}")

    await router.shutdown()

asyncio.run(main())

Configuration File

# config.yaml
router:
  max_concurrent_requests: 50
  request_timeout: 30.0
  fallback_enabled: true
  load_balancing: false

providers:
  - name: "openai"
    api_key: "${OPENAI_API_KEY}"
    priority: "high"
    timeout: 30.0
    enabled: true

  - name: "backup"
    priority: "normal"
    enabled: true
    custom_settings:
      simulate_latency: false

from echoroute.utils import ConfigManager

# Load configuration
config_manager = ConfigManager.from_file("config.yaml")
router_config = config_manager.get_router_config()
provider_configs = config_manager.get_provider_configs()

# Create router with config
router = VoiceRouter(router_config)
await router.initialize()

# Register providers from config
for provider_config in provider_configs:
    if provider_config.name == "openai":
        provider = OpenAIProvider(provider_config)
    else:
        provider = MockProvider(provider_config)
    await router.register_provider(provider)

🔌 Creating Custom Providers

from echoroute.core import BaseProvider
from echoroute.types import (
    VoiceRequest,
    VoiceResponse,
    ProviderCapabilities,
    AudioFormat,
    Language
)

class MyCustomProvider(BaseProvider):
    async def initialize(self) -> None:
        # Initialize your provider (API clients, etc.)
        await self._initialize_metadata()

    async def get_capabilities(self) -> ProviderCapabilities:
        return ProviderCapabilities(
            supports_synthesis=True,
            supports_transcription=True,
            supported_languages=[Language.EN_US],
            supported_audio_formats=[AudioFormat.MP3, AudioFormat.WAV]
        )

    async def process_request(self, request: VoiceRequest) -> VoiceResponse:
        # Implement your provider logic here
        response = VoiceResponse(provider_name=self.name, processing_time=0.0)

        if request.text:
            # Synthesize speech
            response.synthesis = await self._synthesize(request.text)

        if request.audio:
            # Transcribe audio
            response.transcription = await self._transcribe(request.audio)

        return response

    async def health_check(self) -> bool:
        # Check if your provider is healthy
        return True

# Register your custom provider
config = ProviderConfig(name="my-provider", api_key="key")
provider = MyCustomProvider(config)
await router.register_provider(provider)

📊 Monitoring and Health Checks

# Check provider health
health_status = await router.health_check()
print(health_status)  # {"openai": True, "backup": True}

# Get detailed provider status
status = await router.get_provider_status()
for name, details in status.items():
    print(f"{name}: {details['status']}, "
          f"Success Rate: {details['success_rate']:.2%}, "
          f"Latency: {details['latency_ms']:.2f}ms")

🌍 Indian Language Support with Sarvam AI

EchoRoute includes comprehensive support for Indian languages through Sarvam AI:

import asyncio
from echoroute import (
    VoiceRouter,
    VoiceRequest,
    SarvamProvider,
    ProviderConfig,
    Language,
    VoiceModel
)

async def indian_language_example():
    router = VoiceRouter()
    await router.initialize()

    # Register Sarvam AI provider
    sarvam_config = ProviderConfig(
        name="sarvam_ai",
        api_key="your-sarvam-api-key"
    )
    await router.register_provider(SarvamProvider(sarvam_config))

    # Hindi transcription
    hindi_request = VoiceRequest(
        audio=hindi_audio_data,
        language=Language.HI_IN,
        voice_model=VoiceModel.SAARIKA  # Transcription model
    )

    # Auto-detect language and translate to English
    translate_request = VoiceRequest(
        audio=mixed_language_audio,
        language=Language.AUTO,
        voice_model=VoiceModel.SAARAS  # Translation model
    )

    # Generate Hindi speech
    synthesis_request = VoiceRequest(
        text="नमस्ते, यह वीआर राउटर का परीक्षण है।",
        language=Language.HI_IN,
        voice_model=VoiceModel.BULBUL  # TTS model
    )

    await router.shutdown()

Supported Indian Languages

• Hindi (hi-IN), Gujarati (gu-IN), Bengali (bn-IN)
• Tamil (ta-IN), Telugu (te-IN), Marathi (mr-IN)
• Kannada (kn-IN), Malayalam (ml-IN), Punjabi (pa-IN)
• Odia (or-IN), Assamese (as-IN), Nepali (ne-IN)
• Urdu (ur-IN), English Indian (en-IN)

Sarvam AI Models

• SAARIKA: Speech recognition in original language
• SAARAS: Speech recognition + translation to English
• BULBUL: Text-to-speech synthesis

🔧 Advanced Configuration

Environment Variables

export ECHOROUTE_MAX_CONCURRENT_REQUESTS=100
export ECHOROUTE_REQUEST_TIMEOUT=45.0
export ECHOROUTE_FALLBACK_ENABLED=true
export ECHOROUTE_LOG_LEVEL=INFO

Retry Configuration

from echoroute.utils import RetryManager

retry_manager = RetryManager(
    max_retries=3,
    base_delay=1.0,
    exponential_base=2.0,
    jitter=True
)

# Providers automatically use retry logic for transient failures

🧪 Testing

Run the test suite:

pytest tests/

Run with coverage:

pytest tests/ --cov=vrouter --cov-report=html

📝 API Reference

Core Classes

• VoiceRouter: Main router class for managing providers
• BaseProvider: Abstract base class for implementing providers
• ProviderRegistry: Manages provider registration and health
• ConfigManager: Handles configuration loading and management

Data Models

• VoiceRequest: Input request with text/audio and parameters
• VoiceResponse: Response with synthesis/transcription results
• AudioData: Audio data container with format information
• ProviderConfig: Configuration for individual providers
• RouterConfig: Configuration for the router itself

Built-in Providers

• OpenAIProvider: Integration with OpenAI's speech APIs
• SarvamProvider: Integration with Sarvam AI's Indian language speech APIs
• MockProvider: Mock provider for testing and development

🤝 Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests
5. Submit a pull request

📄 License

MIT License - see LICENSE file for details.

🔗 Links

• PyPI Package
• GitHub Repository
• Documentation
• Issue Tracker
• Changelog