Merge pull request #3595 from pipecat-ai/aleix/add-kokoro-tts-support

services(tss): add new KokoroTTSService

changelog/3595.added.md

@@ -0,0 +1 @@
+- Added `KokoroTTSService` for local text-to-speech synthesis using the Kokoro-82M model.

examples/foundational/07zj-interruptible-kokoro.py

@@ -0,0 +1,132 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.kokoro.tts import KokoroTTSService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
+from pipecat.turns.user_turn_strategies import UserTurnStrategies
+
+load_dotenv(override=True)
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = KokoroTTSService(voice_id="af_heart")
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = LLMContext(messages)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(
+            user_turn_strategies=UserTurnStrategies(
+                stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
+            ),
+        ),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            user_aggregator,  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            assistant_aggregator,  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

pyproject.toml

@@ -75,6 +75,7 @@ heygen = [ "livekit>=1.0.13", "pipecat-ai[websockets-base]" ]
 hume = [ "hume>=0.11.2" ]
 inworld = []
 koala = [ "pvkoala~=2.0.3" ]
+kokoro = [ "kokoro>=0.9.4,<1", "requests>=2.32.5,<3" ]
 krisp = [ "pipecat-ai-krisp~=0.4.0" ]
 langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ]
 livekit = [ "livekit~=1.0.13", "livekit-api~=1.0.5", "tenacity>=8.2.3,<10.0.0", "pyjwt>=2.10.1" ]
@@ -95,7 +96,7 @@ rnnoise = [ "pyrnnoise~=0.4.1" ]
 openpipe = [ "openpipe>=4.50.0,<6" ]
 openrouter = []
 perplexity = []
-piper = [ "piper-tts>=1.3.0,<2" ]
+piper = [ "piper-tts>=1.3.0,<2", "requests>=2.32.5,<3" ]
 playht = [ "pipecat-ai[websockets-base]" ]
 qwen = []
 remote-smart-turn = []

scripts/evals/run-release-evals.py

@@ -139,6 +139,7 @@ TESTS_07 = [
     ("07zg-interruptible-camb.py", EVAL_SIMPLE_MATH),
     ("07zh-interruptible-hathora.py", EVAL_SIMPLE_MATH),
     ("07zi-interruptible-piper.py", EVAL_SIMPLE_MATH),
+    ("07zj-interruptible-kokoro.py", EVAL_SIMPLE_MATH),
     # Needs a local XTTS docker instance running.
     # ("07i-interruptible-xtts.py", EVAL_SIMPLE_MATH),
     # Needs a Krisp license.

src/pipecat/services/kokoro/tts.py

@@ -0,0 +1,155 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Kokoro TTS service implementation."""
+
+import asyncio
+from typing import AsyncGenerator, AsyncIterator, Optional
+
+import numpy as np
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.audio.utils import create_stream_resampler
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.tts_service import TTSService
+from pipecat.transcriptions.language import Language, resolve_language
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+try:
+    from kokoro import KPipeline
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Kokoro, you need to `pip install pipecat-ai[kokoro]`.")
+    raise Exception(f"Missing module: {e}")
+
+
+def language_to_kokoro_language(language: Language) -> str:
+    """Convert a Language enum to Kokoro language code.
+
+    Args:
+        language: The Language enum value to convert.
+
+    Returns:
+        The corresponding Kokoro language code, or None if not supported.
+    """
+    LANGUAGE_MAP = {
+        Language.EN: "a",
+        Language.EN_US: "a",
+        Language.EN_GB: "b",
+        Language.ES: "e",
+        Language.FR: "f",
+        Language.HI: "h",
+        Language.IT: "i",
+        Language.JA: "j",
+        Language.PT: "p",
+        Language.ZH: "z",
+    }
+
+    return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+
+
+class KokoroTTSService(TTSService):
+    """Kokoro TTS service implementation.
+
+    Provides local text-to-speech synthesis using the Kokoro-82M model.
+    Automatically downloads the model on first use.
+    """
+
+    class InputParams(BaseModel):
+        """Input parameters for Kokoro TTS configuration.
+
+        Parameters:
+            language: Language to use for synthesis.
+        """
+
+        language: Language = Language.EN
+
+    def __init__(
+        self,
+        *,
+        voice_id: str,
+        repo_id="hexgrad/Kokoro-82M",
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Kokoro TTS service.
+
+        Args:
+            voice_id: Voice identifier to use for synthesis.
+            repo_id: Hugging Face repository ID for the Kokoro model.
+                Defaults to "hexgrad/Kokoro-82M".
+            params: Configuration parameters for synthesis.
+            **kwargs: Additional arguments passed to parent `TTSService`.
+        """
+        super().__init__(**kwargs)
+
+        params = params or KokoroTTSService.InputParams()
+
+        self._voice_id = voice_id
+        self._lang_code = language_to_kokoro_language(params.language)
+        self._pipeline = KPipeline(lang_code=self._lang_code, repo_id=repo_id)
+
+        self._resampler = create_stream_resampler()
+
+    def can_generate_metrics(self) -> bool:
+        """Indicate that this service supports TTFB and usage metrics."""
+        return True
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Synthesize speech from text using the Kokoro pipeline.
+
+        Runs the Kokoro pipeline in a background thread and streams audio
+        frames as they become available.
+
+        Args:
+            text: The text to synthesize.
+        """
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        def async_next(it):
+            try:
+                return next(it)
+            except StopIteration:
+                return None
+
+        async def async_iterator(iterator) -> AsyncIterator[bytes]:
+            while True:
+                item = await asyncio.to_thread(async_next, iterator)
+                if item is None:
+                    return
+
+                (_, _, audio) = item
+
+                # Kokoro outputs a PyTorch tensor at 24kHz, convert to int16 bytes
+                audio_np = np.array(audio)
+                audio_int16 = (audio_np * 32767).astype(np.int16).tobytes()
+
+                yield audio_int16
+
+        try:
+            await self.start_ttfb_metrics()
+            await self.start_tts_usage_metrics(text)
+            yield TTSStartedFrame()
+
+            async for frame in self._stream_audio_frames_from_iterator(
+                async_iterator(self._pipeline(text, voice=self._voice_id)),
+                in_sample_rate=24000,
+            ):
+                await self.stop_ttfb_metrics()
+                yield frame
+        except Exception as e:
+            yield ErrorFrame(error=f"Unknown error occurred: {e}")
+        finally:
+            logger.debug(f"{self}: Finished TTS [{text}]")
+            await self.stop_ttfb_metrics()
+            yield TTSStoppedFrame()

src/pipecat/services/piper/tts.py

@@ -225,7 +225,6 @@ class PiperHttpTTSService(TTSService):
                     await self.stop_ttfb_metrics()
                     yield frame
         except Exception as e:
-            logger.error(f"{self} exception: {e}")
             yield ErrorFrame(error=f"Unknown error occurred: {e}")
         finally:
             logger.debug(f"{self}: Finished TTS [{text}]")