Update FireworksLLMService to use OpenAILLMService

CHANGELOG.md

@@ -10,12 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `GroqLLMService` and `GrokLLMService` for Groq and Grok API integration, with
-  OpenAI-compatible interface
+  OpenAI-compatible interface.
-- New examples demonstrating function calling with Groq, Grok, and Azure OpenAI
 
-  - `14f-function-calling-groq.py`
+- New examples demonstrating function calling with Groq, Grok, Azure OpenAI,
-  - `14g-function-calling-grok.py`
+  and Fireworks: `14f-function-calling-groq.py`, `14g-function-calling-grok.py`,
-  - `14h-function-calling-azure.py`
+  `14h-function-calling-azure.py`, and `14i-function-calling-fireworks.py`.
 
 - In order to obtain the audio stored by the `AudioBufferProcessor` you can now
   also register an `on_audio_data` event handler. The `on_audio_data` handler
@@ -47,6 +46,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels):
 - Updated the `AzureLLMService` to use the `OpenAILLMService`. Updated the
   `api_version` to `2024-09-01-preview`.
 
+- Updated the `FireworksLLMService` to use the `OpenAILLMService`. Updated the
+  default model to `accounts/fireworks/models/firefunction-v2`.
+
 ### Removed
 
 - Removed `AppFrame`. This was used as a special user custom frame, but there's
@@ -71,6 +73,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels):
 - Fixed Google Gemini message handling to properly convert appended messages to
   Gemini's required format.
 
+- Fixed an issue with `FireworksLLMService` where chat completions were failing
+  by removing the `stream_options` from the chat completion options.
+
 ## [0.0.49] - 2024-11-17
 
 ### Added

examples/foundational/14i-function-calling-fireworks.py

@@ -0,0 +1,140 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.fireworks import FireworksLLMService
+from pipecat.services.openai import OpenAILLMContext
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def start_fetch_weather(function_name, llm, context):
+    # note: we can't push a frame to the LLM here. the bot
+    # can interrupt itself and/or cause audio overlapping glitches.
+    # possible question for Aleix and Chad about what the right way
+    # to trigger speech is, now, with the new queues/async/sync refactors.
+    # await llm.push_frame(TextFrame("Let me check on that."))
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
+    await result_callback({"conditions": "nice", "temperature": "75"})
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        llm = FireworksLLMService(
+            api_key=os.getenv("FIREWORKS_API_KEY"),
+            model="accounts/fireworks/models/firefunction-v2",
+        )
+        # Register a function_name of None to get all functions
+        # sent to the same callback with an additional function_name parameter.
+        llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
+
+        tools = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "get_current_weather",
+                    "description": "Get the current weather",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "location": {
+                                "type": "string",
+                                "description": "The city and state, e.g. San Francisco, CA",
+                            },
+                            "format": {
+                                "type": "string",
+                                "enum": ["celsius", "fahrenheit"],
+                                "description": "The temperature unit to use. Infer this from the users location.",
+                            },
+                        },
+                        "required": ["location", "format"],
+                    },
+                },
+            )
+        ]
+        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 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, tools)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                context_aggregator.user(),
+                llm,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -52,7 +52,7 @@ google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.17.2" ]
 grok = [ "openai~=1.50.2" ]
 groq = [ "openai~=1.50.2" ]
 gstreamer = [ "pygobject~=3.48.2" ]
-fireworks = [ "openai~=1.37.2" ]
+fireworks = [ "openai~=1.50.2" ]
 krisp = [ "pipecat-ai-krisp~=0.3.0" ]
 langchain = [ "langchain~=0.2.14", "langchain-community~=0.2.12", "langchain-openai~=0.1.20" ]
 livekit = [ "livekit~=0.17.5", "livekit-api~=0.7.1", "tenacity~=8.5.0" ]

src/pipecat/services/fireworks.py

@@ -4,26 +4,73 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from pipecat.services.openai import BaseOpenAILLMService
+
+from typing import List
 
 from loguru import logger
 
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai import OpenAILLMService
+
 try:
-    from openai import AsyncOpenAI
+    from openai.types.chat import ChatCompletionMessageParam
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
-        "In order to use Fireworks, you need to `pip install pipecat-ai[fireworks]`. Also, set the `FIREWORKS_API_KEY` environment variable."
+        "In order to use Fireworks, you need to `pip install pipecat-ai[fireworks]`. Also, set `FIREWORKS_API_KEY` environment variable."
     )
     raise Exception(f"Missing module: {e}")
 
 
-class FireworksLLMService(BaseOpenAILLMService):
+class FireworksLLMService(OpenAILLMService):
+    """A service for interacting with Fireworks AI using the OpenAI-compatible interface.
+
+    This service extends OpenAILLMService to connect to Fireworks' API endpoint while
+    maintaining full compatibility with OpenAI's interface and functionality.
+
+    Args:
+        api_key (str): The API key for accessing Fireworks AI
+        model (str, optional): The model identifier to use. Defaults to "accounts/fireworks/models/firefunction-v2"
+        base_url (str, optional): The base URL for Fireworks API. Defaults to "https://api.fireworks.ai/inference/v1"
+        **kwargs: Additional keyword arguments passed to OpenAILLMService
+    """
+
     def __init__(
         self,
         *,
         api_key: str,
-        model: str = "accounts/fireworks/models/firefunction-v1",
+        model: str = "accounts/fireworks/models/firefunction-v2",
         base_url: str = "https://api.fireworks.ai/inference/v1",
+        **kwargs,
     ):
-        super().__init__(api_key=api_key, model=model, base_url=base_url)
+        super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
+
+    def create_client(self, api_key=None, base_url=None, **kwargs):
+        """Create OpenAI-compatible client for Fireworks API endpoint."""
+        logger.debug(f"Creating Fireworks client with api {base_url}")
+        return super().create_client(api_key, base_url, **kwargs)
+
+    async def get_chat_completions(
+        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+    ):
+        """Get chat completions from Fireworks API.
+
+        Removes OpenAI-specific parameters not supported by Fireworks.
+        """
+        params = {
+            "model": self.model_name,
+            "stream": True,
+            "messages": messages,
+            "tools": context.tools,
+            "tool_choice": context.tool_choice,
+            "frequency_penalty": self._settings["frequency_penalty"],
+            "presence_penalty": self._settings["presence_penalty"],
+            "temperature": self._settings["temperature"],
+            "top_p": self._settings["top_p"],
+            "max_tokens": self._settings["max_tokens"],
+        }
+
+        params.update(self._settings["extra"])
+
+        chunks = await self._client.chat.completions.create(**params)
+        return chunks