Merge pull request #1083 from team-telnyx/creating_telnyx_chatbot

Creating telnyx chatbot

examples/telnyx-chatbot/.gitignore

@@ -0,0 +1,161 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+runpod.toml

examples/telnyx-chatbot/Dockerfile

@@ -0,0 +1,20 @@
+# Use an official Python runtime as a parent image
+FROM python:3.10-bullseye
+
+# Set the working directory in the container
+WORKDIR /telnyx-chatbot
+
+# Copy the requirements file into the container
+COPY requirements.txt .
+
+# Install any needed packages specified in requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the current directory contents into the container
+COPY . .
+
+# Expose the desired port
+EXPOSE 8765
+
+# Run the application
+CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8765"]

examples/telnyx-chatbot/README.md

@@ -0,0 +1,112 @@
+# Telnyx Chatbot
+
+This project is a FastAPI-based chatbot that integrates with Telnyx to handle WebSocket connections and provide real-time communication. The project includes endpoints for starting a call and handling WebSocket connections.
+
+## Table of Contents
+
+- [Telnyx Chatbot](#telnyx-chatbot)
+  - [Table of Contents](#table-of-contents)
+  - [Features](#features)
+  - [Requirements](#requirements)
+  - [Installation](#installation)
+  - [Configure Telnyx TeXML application](#configure-telnyx-texml-application)
+  - [Running the Application](#running-the-application)
+    - [Using Python (Option 1)](#using-python-option-1)
+    - [Using Docker (Option 2)](#using-docker-option-2)
+  - [Usage](#usage)
+
+## Features
+
+- **FastAPI**: A modern, fast (high-performance), web framework for building APIs with Python 3.6+.
+- **WebSocket Support**: Real-time communication using WebSockets.
+- **CORS Middleware**: Allowing cross-origin requests for testing.
+- **Dockerized**: Easily deployable using Docker.
+
+## Requirements
+
+- Python 3.10
+- Docker (for containerized deployment)
+- ngrok (for tunneling)
+- Telnyx Account
+
+## Installation
+
+1. **Set up a virtual environment** (optional but recommended):
+
+   ```sh
+   python -m venv venv
+   source venv/bin/activate  # On Windows, use `venv\Scripts\activate`
+   ```
+
+2. **Install dependencies**:
+
+   ```sh
+   pip install -r requirements.txt
+   ```
+
+3. **Create .env**:
+   Copy the example environment file and update with your settings:
+
+   ```sh
+   cp env.example .env
+   ```
+
+4. **Install ngrok**:
+   Follow the instructions on the [ngrok website](https://ngrok.com/download) to download and install ngrok.
+
+## Configure Telnyx TeXML application
+
+1. **Start ngrok**:
+   In a new terminal, start ngrok to tunnel the local server:
+
+   ```sh
+   ngrok http 8765
+   ```
+
+2. **Update the Telnyx TeXML applications Webhook**:
+
+   - Go to your TeXML configuration page
+   - Provide the ngrok URL to the Webhook URL field and ensure the POST method is selected
+   - Click Save at the bottom of the page
+
+3. **Configure streams.xml**:
+   - Copy the template file to create your local version:
+     ```sh
+     cp templates/streams.xml.template templates/streams.xml
+     ```
+   - In `templates/streams.xml`, replace `<your server url>` with your ngrok URL (without `https://`)
+   - The final URL should look like: `wss://abc123.ngrok.io/ws`
+   - The encoding (`bidirectionalCodec`) should be `PCMU` or `PCMA` depending on your needs. Based on selected encoding, set the outbound_encoding in `server.py` when the bot is initialized.
+   - The inbound encoding can be controlled from the application configuration for inbound calls and dial/transfer commands for outbound calls.
+
+## Running the Application
+
+Choose one of these two methods to run the application:
+
+### Using Python (Option 1)
+
+**Run the FastAPI application**:
+
+```sh
+# Make sure you’re in the project directory and your virtual environment is activated
+python server.py
+```
+
+### Using Docker (Option 2)
+
+1. **Build the Docker image**:
+
+   ```sh
+   docker build -t telnyx-chatbot .
+   ```
+
+2. **Run the Docker container**:
+   ```sh
+   docker run -it --rm -p 8765:8765 telnyx-chatbot
+   ```
+
+The server will start on port 8765. Keep this running while you test with Telnyx.
+
+## Usage
+
+To start a call, simply make a call to your configured Telnyx phone number. The webhook URL will direct the call to your FastAPI application, which will handle it accordingly.

examples/telnyx-chatbot/bot.py

@@ -0,0 +1,97 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import EndFrame
+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.serializers.telnyx import TelnyxFrameSerializer
+
+from pipecat.services.elevenlabs import ElevenLabsTTSService, Language
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.network.fastapi_websocket import (
+    FastAPIWebsocketParams,
+    FastAPIWebsocketTransport,
+)
+
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def run_bot(websocket_client, stream_id, outbound_encoding, inbound_encoding):
+    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=TelnyxFrameSerializer(stream_id, outbound_encoding, inbound_encoding),
+        ),
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = ElevenLabsTTSService(
+        api_key=os.getenv("ELEVENLABS_API_KEY"),
+        voice_id="CwhRBWXzGAHq8TQ4Fs17",
+        output_format="pcm_24000",
+        params=ElevenLabsTTSService.InputParams(language=Language.EN),
+    )
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in an audio call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Websocket input from client
+            stt,  # Speech-To-Text
+            context_aggregator.user(),
+            llm,  # LLM
+            tts,  # Text-To-Speech
+            transport.output(),  # Websocket output to client
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        await task.queue_frames([EndFrame()])
+
+    runner = PipelineRunner(handle_sigint=False)
+
+    await runner.run(task)

examples/telnyx-chatbot/env.example

@@ -0,0 +1,3 @@
+OPENAI_API_KEY=
+DEEPGRAM_API_KEY=
+ELEVENLABS_API_KEY=

examples/telnyx-chatbot/requirements.txt

@@ -0,0 +1,5 @@
+pipecat-ai[openai,silero,deepgram,elevenlabs]
+fastapi
+uvicorn
+python-dotenv
+loguru

examples/telnyx-chatbot/server.py

@@ -0,0 +1,37 @@
+import json
+
+import uvicorn
+from bot import run_bot
+from fastapi import FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.responses import HTMLResponse
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allow all origins for testing
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.post("/")
+async def start_call():
+    print("POST TeXML")
+    return HTMLResponse(content=open("templates/streams.xml").read(), media_type="application/xml")
+
+@app.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    start_data = websocket.iter_text()
+    await start_data.__anext__()
+    call_data = json.loads(await start_data.__anext__())
+    print(call_data, flush=True)
+    stream_id = call_data["stream_id"]
+    outbound_encoding = call_data["start"]["media_format"]["encoding"]
+    print("WebSocket connection accepted")
+    await run_bot(websocket, stream_id, outbound_encoding, "PCMU")
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8765)

examples/telnyx-chatbot/templates/streams.xml.template

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Response>
+  <Connect>
+    <Stream url="wss://<your server url>/ws" bidirectionalMode="rtp"></Stream>
+  </Connect>
+  <Pause length="40"/>
+</Response>

src/pipecat/audio/utils.py

@@ -81,7 +81,7 @@ def ulaw_to_pcm(ulaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
 
     # Resample
     out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate)
-
+    
     return out_pcm_bytes
 
 
@@ -93,3 +93,23 @@ def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
     ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2)
 
     return ulaw_bytes
+
+
+def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes:     
+    # Convert a-law to PCM
+    in_pcm_bytes = audioop.alaw2lin(alaw_bytes, 2)
+
+    # Resample
+    out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate)
+    
+    return out_pcm_bytes
+
+
+def pcm_to_alaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
+    # Resample
+    in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate)
+
+    # Convert PCM to μ-law
+    alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2)
+
+    return alaw_bytes

src/pipecat/serializers/telnyx.py

@@ -0,0 +1,99 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import base64
+import json
+
+from pydantic import BaseModel
+
+from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm, pcm_to_alaw, alaw_to_pcm
+from pipecat.frames.frames import (
+    AudioRawFrame,
+    Frame,
+    InputAudioRawFrame,
+    InputDTMFFrame,
+    KeypadEntry,
+    StartInterruptionFrame,
+)
+from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
+
+
+class TelnyxFrameSerializer(FrameSerializer):
+    class InputParams(BaseModel):
+        telnyx_sample_rate: int = 8000
+        sample_rate: int = 16000
+        inbound_encoding: str = "PCMU"
+        outbound_encoding: str = "PCMU"
+
+    def __init__(self, stream_id: str, outbound_encoding: str, inbound_encoding: str, params: InputParams = InputParams()):
+        self._stream_id = stream_id
+        params.outbound_encoding = outbound_encoding
+        params.inbound_encoding = inbound_encoding  
+        self._params = params
+
+    @property
+    def type(self) -> FrameSerializerType:
+        return FrameSerializerType.TEXT
+
+    def serialize(self, frame: Frame) -> str | bytes | None:
+        if isinstance(frame, AudioRawFrame):
+            data = frame.audio
+            
+            if self._params.inbound_encoding == "PCMU":
+                serialized_data = pcm_to_ulaw(
+                    data, frame.sample_rate, self._params.telnyx_sample_rate
+                    )
+            elif self._params.inbound_encoding == "PCMA":
+                serialized_data = pcm_to_alaw(
+                    data, frame.sample_rate, self._params.telnyx_sample_rate
+                    )
+            else:
+                raise ValueError(f"Unsupported encoding: {self._params.encoding}")
+            
+            payload = base64.b64encode(serialized_data).decode("utf-8")
+            answer = {
+                "event": "media",
+                "media": {"payload": payload},
+            }
+
+            return json.dumps(answer)
+
+        if isinstance(frame, StartInterruptionFrame):
+            answer = {"event": "clear"}
+            return json.dumps(answer)
+
+    def deserialize(self, data: str | bytes) -> Frame | None:
+        message = json.loads(data)
+        
+        if message["event"] == "media":
+            payload_base64 = message["media"]["payload"]
+            payload = base64.b64decode(payload_base64)
+
+            if self._params.outbound_encoding == "PCMU":
+                deserialized_data = ulaw_to_pcm(
+                    payload, self._params.telnyx_sample_rate, self._params.sample_rate
+                )
+            elif self._params.outbound_encoding == "PCMA":
+                deserialized_data = alaw_to_pcm(
+                    payload, self._params.telnyx_sample_rate, self._params.sample_rate
+                )
+            else:
+                raise ValueError(f"Unsupported encoding: {self._params.encoding}")
+
+            audio_frame = InputAudioRawFrame(
+                audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate
+            )
+            return audio_frame
+        elif message["event"] == "dtmf":
+            digit = message.get("dtmf", {}).get("digit")
+
+            try:
+                return InputDTMFFrame(KeypadEntry(digit))
+            except ValueError as e:
+                # Handle case where string doesn't match any enum value
+                return None
+        else:
+            return None