### Run Voice Agent Service Locally

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Commands to install dependencies and start the BentoML server with required environment variables for XTTS and LLM endpoints.

```bash
apt-get install ffmpeg
pip install -U -r requirements.txt
XTTS_SERVICE_URL=https://xtts-streaming.example.bentoml.ai \
OPENAI_SERVICE_URL=https://llama-3-1.example.bentoml.ai/v1 \
bentoml serve
```

--------------------------------

### Deploying to BentoCloud

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Guide for deploying the voice agent to BentoCloud for production use.

```APIDOC
## Deploying to BentoCloud

### Description
Deploy the voice agent to BentoCloud for production use with automatic scaling and management.

### Steps
1. **Login to BentoCloud:**
   ```bash
   bentoml cloud login
   ```
2. **Deploy with environment variables:**
   ```bash
   bentoml deploy .
       --env XTTS_SERVICE_URL=https://xtts-streaming.example.bentoml.ai
       --env OPENAI_SERVICE_URL=https://llama-3-1.example.bentoml.ai/v1
   ```
3. **Configure Twilio:**
   - Go to Twilio Console > Phone Numbers > Your Number.
   - Set Voice Configuration webhook URL to:
     `https://your-deployment.bentoml.ai/voice/start_call`
   - Set HTTP method to POST.
```

--------------------------------

### Twilio Webhook to Start Call Endpoint

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

An HTTP POST endpoint that acts as the Twilio webhook to initiate calls. It generates TwiML XML to instruct Twilio to establish a WebSocket stream connection for bidirectional audio. Requires os and HTMLResponse.

```python
@app.post("/start_call")
async def start_call(self):
    service_url = os.environ.get("BENTOCLOUD_DEPLOYMENT_URL") or ""
    assert(service_url)
    if service_url.startswith("http"):
        from urllib.parse import urlparse
        service_url = urlparse(service_url).netloc

    tmpl = """<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
    <Stream url="wss://{service_url}/voice/ws"></Stream>
</Connect>
<Pause length="40"/>
</Response>
    """
    return HTMLResponse(content=tmpl.format(service_url=service_url), media_type="application/xml")

# Configure this endpoint as Twilio webhook:
# URL: https://your-deployment.bentoml.ai/voice/start_call
# Method: POST
```

--------------------------------

### Running the Service Locally

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Instructions for setting up and running the voice agent service on a local machine.

```APIDOC
## Running the Service Locally

### Description
Start the voice agent service with environment variables pointing to deployed LLM and XTTS endpoints.

### Steps
1. **Install system dependencies:**
   ```bash
   apt-get install ffmpeg
   ```
2. **Install Python dependencies:**
   ```bash
   pip install -U -r requirements.txt
   ```
3. **Start the server with model endpoints:**
   ```bash
   XTTS_SERVICE_URL=https://xtts-streaming.example.bentoml.ai \
   OPENAI_SERVICE_URL=https://llama-3-1.example.bentoml.ai/v1 \
   bentoml serve
   ```

### Server Exposes
- **POST /voice/start_call** - Twilio webhook for call initiation.
- **WS /voice/ws** - WebSocket for real-time audio streaming.
```

--------------------------------

### Configure Voice Bot Pipeline with Pipecat

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

This snippet demonstrates how to initialize a voice pipeline using Pipecat. It configures a WebSocket transport for Twilio, integrates OpenAI-compatible LLM services, and sets up the audio processing flow.

```python
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.openai import OpenAILLMService, OpenAILLMContext
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport, FastAPIWebsocketParams
from pipecat.vad.silero import SileroVADAnalyzer
from pipecat.serializers.twilio import TwilioFrameSerializer

async def run_bot(websocket_client, stream_sid, whisper_model):
    transport = FastAPIWebsocketTransport(
        websocket=websocket_client,
        params=FastAPIWebsocketParams(
            audio_out_enabled=True,
            add_wav_header=False,
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
            vad_audio_passthrough=True,
            serializer=TwilioFrameSerializer(stream_sid),
        ),
    )

    llm = OpenAILLMService(
        base_url=os.getenv("OPENAI_SERVICE_URL"),
        api_key="n/a",
        model="meta-llama/Meta-Llama-3.1-8B-Instruct",
    )

    stt = BentoWhisperSTTService(model=whisper_model)

    tts = SimpleXTTSService(
        base_url=os.getenv("XTTS_SERVICE_URL"),
        language="en",
        aiohttp_session=aiohttp.ClientSession(),
    )

    messages = [
        {
            "role": "system",
            "content": "You are a helpful LLM assistant in an audio call. Your name is Jane. "
                       "Your goal is to demonstrate your capabilities in a succinct way. "
                       "Your output will be converted to audio so don't include special characters.",
        },
    ]

    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

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

    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
    runner = PipelineRunner(handle_sigint=False)
    await runner.run(task)
```

--------------------------------

### Define and Register OpenAI-Compatible Tool for LLM Function Calling (Python)

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

This snippet demonstrates how to define a tool using `ChatCompletionToolParam` for OpenAI-compatible function calling. It includes the tool's schema, parameters, and how to register an execution function with the LLM. This enables the voice agent to call custom functions based on LLM instructions.

```python
from openai.types.chat import ChatCompletionToolParam

tools = [
    ChatCompletionToolParam(
        type="function",
        function={
            "name": "get_deployment_count",
            "description": "Get the deployment count in a region of a specific status.",
            "parameters": {
                "type": "object",
                "properties": {
                    "region": {
                        "type": "string",
                        "enum": ["north america", "europe", "asia"],
                        "description": "The region where the deployments are located.",
                    },
                    "status": {
                        "type": "string",
                        "enum": ["running", "scaled to zero", "terminated"],
                        "description": "The status of the deployment.",
                    },
                },
                "required": ["region", "status"],
            },
        },
    ),
]

async def exec_function(function_name, tool_call_id, args, llm, context, result_callback):
    logger.debug(f"Executing function: {function_name}")
    await result_callback({"deployment_count": 10})

llm.register_function("get_deployment_count", exec_function, start_callback=start_function)
context = OpenAILLMContext(messages, tools)
```

--------------------------------

### POST /start_call

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

HTTP POST endpoint that serves as the Twilio webhook to initiate calls. Returns TwiML XML that instructs Twilio to establish a WebSocket stream connection for bidirectional audio.

```APIDOC
## POST /start_call

### Description
HTTP POST endpoint that serves as the Twilio webhook to initiate calls. Returns TwiML XML that instructs Twilio to establish a WebSocket stream connection for bidirectional audio.

### Method
POST

### Endpoint
/start_call

### Parameters
N/A

### Request Example
N/A

### Response
#### Success Response (200)
- **TwiML XML** - XML document instructing Twilio to connect to the WebSocket stream.

#### Response Example
```xml
<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
    <Stream url="wss://{service_url}/voice/ws"></Stream>
</Connect>
<Pause length="40"/>
</Response>
```

**Note**: Configure this endpoint as a Twilio webhook with URL: `https://your-deployment.bentoml.ai/voice/start_call` and Method: `POST`.
```

--------------------------------

### Deploy Voice Agent to BentoCloud

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Deployment workflow for pushing the voice agent to BentoCloud, including authentication and environment variable configuration.

```bash
bentoml cloud login
bentoml deploy . \
    --env XTTS_SERVICE_URL=https://xtts-streaming.example.bentoml.ai \
    --env OPENAI_SERVICE_URL=https://llama-3-1.example.bentoml.ai/v1
```

--------------------------------

### Initialize TwilioBot BentoML Service with Whisper Model

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Initializes the main service class for handling Twilio calls. It sets up the Whisper model for speech-to-text and mounts FastAPI endpoints for call handling and WebSocket communication. Requires PyTorch and faster_whisper.

```python
import bentoml
from fastapi import FastAPI, WebSocket
from faster_whisper import WhisperModel

app = FastAPI()

@bentoml.service(
    traffic={"timeout": 30},
    resources={
        "gpu": 1,
        "gpu_type": "nvidia-tesla-t4",
    },
)
@bentoml.mount_asgi_app(app, path="/voice")
class TwilioBot:
    def __init__(self):
        import torch
        self.batch_size = 16
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        compute_type = "float16" if torch.cuda.is_available() else "int8"
        self.whisper_model = WhisperModel("large-v3", self.device, compute_type=compute_type)
```

--------------------------------

### Implement SimpleXTTSService for Audio Streaming

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

A custom Pipecat service that streams audio from an XTTS endpoint. It handles audio resampling from 24kHz to 16kHz to ensure compatibility with Twilio's audio requirements.

```python
from pipecat.frames.frames import TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ErrorFrame
from pipecat.services.ai_services import TTSService
import aiohttp
import numpy as np
import resampy

class SimpleXTTSService(TTSService):
    def __init__(self, *, language: str, base_url: str, aiohttp_session: aiohttp.ClientSession, **kwargs):
        super().__init__(**kwargs)
        self._language = language
        self._base_url = base_url
        self._aiohttp_session = aiohttp_session

    async def run_tts(self, text: str):
        url = self._base_url + "/tts/stream"
        payload = {
            "text": text.replace('.', '').replace('*', ''),
            "language": self._language,
            "add_wav_header": False,
            "stream_chunk_size": 20,
        }
        await self.start_ttfb_metrics()
        async with self._aiohttp_session.post(url, json=payload) as r:
            if r.status != 200:
                error_text = await r.text()
                yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {error_text})")
                return
            yield TTSStartedFrame()
            buffer = bytearray()
            async for chunk in r.content.iter_chunked(1024):
                if len(chunk) > 0:
                    await self.stop_ttfb_metrics()
                    buffer.extend(chunk)
                    while len(buffer) >= 48000:
                        process_data = buffer[:48000]
                        buffer = buffer[48000:]
                        audio_np = np.frombuffer(process_data, dtype=np.int16)
                        resampled_audio = resampy.resample(audio_np, 24000, 16000)
                        resampled_audio_bytes = resampled_audio.astype(np.int16).tobytes()
                        yield TTSAudioRawFrame(resampled_audio_bytes, 16000, 1)
            if len(buffer) > 0:
                audio_np = np.frombuffer(buffer, dtype=np.int16)
                resampled_audio = resampy.resample(audio_np, 24000, 16000)
                yield TTSAudioRawFrame(resampled_audio.astype(np.int16).tobytes(), 16000, 1)
            yield TTSStoppedFrame()
```

--------------------------------

### SimpleXTTSService Definition

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

Defines the SimpleXTTSService class, which handles text-to-speech using a remote XTTS endpoint and resamples audio for compatibility.

```APIDOC
## Class SimpleXTTSService

### Description
Custom text-to-speech service that streams audio from a remote XTTS endpoint. Handles audio resampling from 24kHz to 16kHz for Twilio compatibility.

### Methods
#### `__init__`
Initializes the SimpleXTTSService with language, base URL, and an aiohttp session.

#### `run_tts`
Generates speech audio from text using XTTS streaming API.

### Parameters
#### `__init__` Parameters
- **language** (str) - Required - The language for text-to-speech.
- **base_url** (str) - Required - The base URL of the XTTS endpoint.
- **aiohttp_session** (aiohttp.ClientSession) - Required - An active aiohttp client session.

### Request Example (run_tts)
```json
{
  "text": "Hello, this is a test.",
  "language": "en",
  "add_wav_header": false,
  "stream_chunk_size": 20
}
```

### Response Example (run_tts)
- **TTSStartedFrame** - Indicates the start of TTS streaming.
- **TTSAudioRawFrame** - Contains raw audio data chunks.
- **TTSStoppedFrame** - Indicates the end of TTS streaming.
- **ErrorFrame** - Indicates an error during the TTS process.
```

--------------------------------

### Implement Custom BentoWhisperSTTService

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

This class extends SegmentedSTTService to provide speech-to-text capabilities using the faster-whisper model. It handles audio buffer conversion from PCM to float32 and filters segments based on speech probability.

```python
from pipecat.frames.frames import TranscriptionFrame, ErrorFrame
from pipecat.services.ai_services import SegmentedSTTService
from faster_whisper import WhisperModel
import numpy as np
import asyncio

class BentoWhisperSTTService(SegmentedSTTService):
    def __init__(self, *, model: WhisperModel, no_speech_prob: float = 0.4, **kwargs):
        super().__init__(**kwargs)
        self._no_speech_prob = no_speech_prob
        self._model = model

    async def run_stt(self, audio: bytes):
        if not self._model:
            yield ErrorFrame("Whisper model not available")
            return

        await self.start_processing_metrics()
        await self.start_ttfb_metrics()

        audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0

        segments, _ = await asyncio.to_thread(self._model.transcribe, audio_float)

        text = ""
        for segment in segments:
            if segment.no_speech_prob < self._no_speech_prob:
                text += f"{segment.text} "

        await self.stop_ttfb_metrics()
        await self.stop_processing_metrics()

        if text:
            yield TranscriptionFrame(text, "", time_now_iso8601())
```

--------------------------------

### WebSocket /ws

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

WebSocket endpoint that handles real-time bidirectional audio streaming with Twilio. Accepts the connection, extracts the stream session ID, and launches the voice bot pipeline.

```APIDOC
## WebSocket /ws

### Description
WebSocket endpoint that handles real-time bidirectional audio streaming with Twilio. Accepts the connection, extracts the stream session ID, and launches the voice bot pipeline.

### Method
WebSocket

### Endpoint
/ws

### Parameters
N/A

### Request Example
N/A

### Response
N/A (Handles streaming data)

```python
@app.websocket("/ws")
async def websocket_endpoint(self, websocket: WebSocket):
    from bot import run_bot

    await websocket.accept()
    start_data = websocket.iter_text()
    await start_data.__anext__()
    call_data = json.loads(await start_data.__anext__())
    stream_sid = call_data["start"]["streamSid"]
    print("WebSocket connection accepted")

    await run_bot(websocket, stream_sid, whisper_model=self.whisper_model)
```
```

--------------------------------

### WebSocket Voice Endpoint for Audio Streaming

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

A WebSocket endpoint that manages real-time bidirectional audio streaming with Twilio. It accepts connections, extracts stream session IDs, and initiates the voice bot pipeline. Requires json and the run_bot function.

```python
@app.websocket("/ws")
async def websocket_endpoint(self, websocket: WebSocket):
    from bot import run_bot

    await websocket.accept()
    start_data = websocket.iter_text()
    await start_data.__anext__()
    call_data = json.loads(await start_data.__anext__())
    stream_sid = call_data["start"]["streamSid"]
    print("WebSocket connection accepted")

    await run_bot(websocket, stream_sid, whisper_model=self.whisper_model)
```

--------------------------------

### TwilioBot BentoML Service

Source: https://context7.com/bentoml/bentovoiceagent/llms.txt

The main service class that handles incoming Twilio calls and manages the voice pipeline. It initializes the Whisper model for speech-to-text and mounts FastAPI endpoints for call handling and WebSocket communication.

```APIDOC
## TwilioBot BentoML Service

### Description
The main service class that handles incoming Twilio calls and manages the voice pipeline. It initializes the Whisper model for speech-to-text and mounts FastAPI endpoints for call handling and WebSocket communication.

### Method
N/A (Service Definition)

### Endpoint
N/A (Service Definition)

### Parameters
N/A

### Request Example
N/A

### Response
N/A

```python
import bentoml
from fastapi import FastAPI, WebSocket
from faster_whisper import WhisperModel

app = FastAPI()

@bentoml.service(
    traffic={"timeout": 30},
    resources={
        "gpu": 1,
        "gpu_type": "nvidia-tesla-t4",
    },
)
@bentoml.mount_asgi_app(app, path="/voice")
class TwilioBot:
    def __init__(self):
        import torch
        self.batch_size = 16
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        compute_type = "float16" if torch.cuda.is_available() else "int8"
        self.whisper_model = WhisperModel("large-v3", self.device, compute_type=compute_type)
```
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.