Add Novita AI LLM service provider

pyproject.toml

@@ -95,6 +95,7 @@ moondream = [ "accelerate~=1.10.0", "einops~=0.8.0", "pyvips[binary]~=3.0.0", "t
 neuphonic = [ "pipecat-ai[websockets-base]" ]
 noisereduce = [ "noisereduce~=3.0.3" ]
 nvidia = [ "nvidia-riva-client>=2.21.1,<3" ]
+novita = []
 openai = [ "pipecat-ai[websockets-base]" ]
 rnnoise = [ "pyrnnoise~=0.4.1" ]
 openpipe = [ "openpipe>=4.50.0,<6" ]

src/pipecat/services/novita/__init__.py

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import sys
+
+from pipecat.services import DeprecatedModuleProxy
+
+from .llm import *
+
+sys.modules[__name__] = DeprecatedModuleProxy(globals(), "novita", "novita.llm")

src/pipecat/services/novita/llm.py

@@ -0,0 +1,81 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Novita AI LLM service implementation using OpenAI-compatible interface."""
+
+from dataclasses import dataclass
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.services.openai.base_llm import BaseOpenAILLMService
+from pipecat.services.openai.llm import OpenAILLMService
+
+
+@dataclass
+class NovitaLLMSettings(BaseOpenAILLMService.Settings):
+    """Settings for NovitaLLMService."""
+
+    pass
+
+
+class NovitaLLMService(OpenAILLMService):
+    """A service for interacting with Novita AI's API using the OpenAI-compatible interface.
+
+    This service extends OpenAILLMService to connect to Novita AI's API endpoint while
+    maintaining full compatibility with OpenAI's interface and functionality.
+    """
+
+    Settings = NovitaLLMSettings
+    _settings: Settings
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "https://api.novita.ai/openai",
+        model: Optional[str] = None,
+        settings: Optional[Settings] = None,
+        **kwargs,
+    ):
+        """Initialize Novita AI LLM service.
+
+        Args:
+            api_key: The API key for accessing Novita AI's API.
+            base_url: The base URL for Novita AI API. Defaults to "https://api.novita.ai/openai".
+            model: The model identifier to use. Defaults to "moonshotai/kimi-k2.5".
+
+                .. deprecated:: 0.0.105
+                    Use ``settings=NovitaLLMService.Settings(model=...)`` instead.
+
+            settings: Runtime-updatable settings. When provided alongside deprecated
+                parameters, ``settings`` values take precedence.
+            **kwargs: Additional keyword arguments passed to OpenAILLMService.
+        """
+        default_settings = self.Settings(model="moonshotai/kimi-k2.5")
+
+        if model is not None:
+            self._warn_init_param_moved_to_settings("model", "model")
+            default_settings.model = model
+
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(api_key=api_key, base_url=base_url, settings=default_settings, **kwargs)
+
+    def create_client(self, api_key=None, base_url=None, **kwargs):
+        """Create OpenAI-compatible client for Novita AI API endpoint.
+
+        Args:
+            api_key: The API key to use for the client. If None, uses instance api_key.
+            base_url: The base URL for the API. If None, uses instance base_url.
+            **kwargs: Additional keyword arguments passed to the parent create_client method.
+
+        Returns:
+            An OpenAI-compatible client configured for Novita AI's API.
+        """
+        logger.debug(f"Creating Novita AI client with api {base_url}")
+        return super().create_client(api_key, base_url, **kwargs)

tests/test_novita_llm.py

@@ -0,0 +1,67 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Unit tests for Novita LLM service."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.services.novita.llm import NovitaLLMService
+
+
+@pytest.mark.asyncio
+async def test_novita_llm_stream_closed_on_cancellation():
+    """Test that the stream is closed when CancelledError occurs during iteration.
+
+    This prevents socket leaks when the pipeline is interrupted (e.g., user interruption).
+    """
+    with patch.object(NovitaLLMService, "create_client"):
+        service = NovitaLLMService(api_key="test-key", model="test-model")
+        service._client = AsyncMock()
+
+        stream_closed = False
+
+        class MockAsyncStream:
+            """Mock AsyncStream that tracks close() calls and raises CancelledError."""
+
+            def __init__(self):
+                self.iteration_count = 0
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                self.iteration_count += 1
+                if self.iteration_count > 1:
+                    raise asyncio.CancelledError()
+                mock_chunk = AsyncMock()
+                mock_chunk.usage = None
+                mock_chunk.choices = []
+                return mock_chunk
+
+            async def close(self):
+                nonlocal stream_closed
+                stream_closed = True
+
+        mock_stream = MockAsyncStream()
+
+        service._stream_chat_completions_specific_context = AsyncMock(return_value=mock_stream)
+        service._stream_chat_completions_universal_context = AsyncMock(return_value=mock_stream)
+        service.start_ttfb_metrics = AsyncMock()
+        service.stop_ttfb_metrics = AsyncMock()
+        service.start_llm_usage_metrics = AsyncMock()
+
+        context = LLMContext(
+            messages=[{"role": "user", "content": "Hello"}],
+        )
+
+        with pytest.raises(asyncio.CancelledError):
+            await service._process_context(context)
+
+        assert stream_closed, "Stream should be closed even when CancelledError occurs"