diff --git a/pyproject.toml b/pyproject.toml index 03fe4ebd3..f929b0f59 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/services/novita/__init__.py b/src/pipecat/services/novita/__init__.py new file mode 100644 index 000000000..624adbcd9 --- /dev/null +++ b/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") diff --git a/src/pipecat/services/novita/llm.py b/src/pipecat/services/novita/llm.py new file mode 100644 index 000000000..2cbb447a3 --- /dev/null +++ b/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) diff --git a/tests/test_novita_llm.py b/tests/test_novita_llm.py new file mode 100644 index 000000000..e0f2c71b4 --- /dev/null +++ b/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"