diff --git a/CHANGELOG.md b/CHANGELOG.md index c449d450c..009625294 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ 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. + +- New examples demonstrating function calling with Groq, Grok, Azure OpenAI, + and Fireworks: `14f-function-calling-groq.py`, `14g-function-calling-grok.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 will be called every time `buffer_size` (a new constructor argument) is @@ -36,6 +43,12 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): - Updated STT and TTS services with language options that match the supported languages for each service. +- 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 @@ -60,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 diff --git a/README.md b/README.md index cb12f4086..c38b84422 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ Available options include: | Category | Services | Install Command Example | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------- | | Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/api-reference/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/api-reference/services/stt/azure), [Deepgram](https://docs.pipecat.ai/api-reference/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/api-reference/services/stt/gladia), [Whisper](https://docs.pipecat.ai/api-reference/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | -| LLMs | [Anthropic](https://docs.pipecat.ai/api-reference/services/llm/anthropic), [Azure](https://docs.pipecat.ai/api-reference/services/llm/azure), [Fireworks AI](https://docs.pipecat.ai/api-reference/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/api-reference/services/llm/gemini), [Ollama](https://docs.pipecat.ai/api-reference/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/api-reference/services/llm/openai), [Together AI](https://docs.pipecat.ai/api-reference/services/llm/together) | `pip install "pipecat-ai[openai]"` | +| LLMs | [Anthropic](https://docs.pipecat.ai/api-reference/services/llm/anthropic), [Azure](https://docs.pipecat.ai/api-reference/services/llm/azure), [Fireworks AI](https://docs.pipecat.ai/api-reference/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/api-reference/services/llm/gemini), [Grok](https://docs.pipecat.ai/api-reference/services/llm/grok), [Groq](https://docs.pipecat.ai/api-reference/services/llm/groq) [Ollama](https://docs.pipecat.ai/api-reference/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/api-reference/services/llm/openai), [Together AI](https://docs.pipecat.ai/api-reference/services/llm/together) | `pip install "pipecat-ai[openai]"` | | Text-to-Speech | [AWS](https://docs.pipecat.ai/api-reference/services/tts/aws), [Azure](https://docs.pipecat.ai/api-reference/services/tts/azure), [Cartesia](https://docs.pipecat.ai/api-reference/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/api-reference/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/api-reference/services/tts/elevenlabs), [Google](https://docs.pipecat.ai/api-reference/services/tts/google), [LMNT](https://docs.pipecat.ai/api-reference/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/api-reference/services/tts/openai), [PlayHT](https://docs.pipecat.ai/api-reference/services/tts/playht), [Rime](https://docs.pipecat.ai/api-reference/services/tts/rime), [XTTS](https://docs.pipecat.ai/api-reference/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` | | Speech-to-Speech | [OpenAI Realtime](https://docs.pipecat.ai/api-reference/services/s2s/openai) | `pip install "pipecat-ai[openai]"` | | Transport | [Daily (WebRTC)](https://docs.pipecat.ai/api-reference/services/transport/daily), WebSocket, Local | `pip install "pipecat-ai[daily]"` | diff --git a/examples/foundational/14c-function-calling-together.py b/examples/foundational/14c-function-calling-together.py index 283b2fe70..a629753e5 100644 --- a/examples/foundational/14c-function-calling-together.py +++ b/examples/foundational/14c-function-calling-together.py @@ -5,10 +5,15 @@ # import asyncio -import aiohttp 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 @@ -18,14 +23,6 @@ from pipecat.services.openai import OpenAILLMContext from pipecat.services.together import TogetherLLMService from pipecat.transports.services.daily import DailyParams, DailyTransport -from openai.types.chat import ChatCompletionToolParam - -from runner import configure - -from loguru import logger - -from dotenv import load_dotenv - load_dotenv(override=True) logger.remove(0) @@ -125,7 +122,7 @@ async def main(): async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. - # await tts.say("Hi! Ask me about the weather in San Francisco.") + await task.queue_frames([context_aggregator.user().get_context_frame()]) runner = PipelineRunner() diff --git a/examples/foundational/14f-function-calling-groq.py b/examples/foundational/14f-function-calling-groq.py new file mode 100644 index 000000000..bcd26672f --- /dev/null +++ b/examples/foundational/14f-function-calling-groq.py @@ -0,0 +1,139 @@ +# +# 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.groq import GroqLLMService +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 = GroqLLMService( + api_key=os.getenv("GROQ_API_KEY"), model="llama3-groq-70b-8192-tool-use-preview" + ) + # 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", + }, + "unit": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + "required": ["location"], + }, + }, + ) + ] + 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()) diff --git a/examples/foundational/14g-function-calling-grok.py b/examples/foundational/14g-function-calling-grok.py new file mode 100644 index 000000000..0165f1f7e --- /dev/null +++ b/examples/foundational/14g-function-calling-grok.py @@ -0,0 +1,137 @@ +# +# 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.grok import GrokLLMService +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 = GrokLLMService(api_key=os.getenv("GROK_API_KEY")) + # 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()) diff --git a/examples/foundational/14h-function-calling-azure.py b/examples/foundational/14h-function-calling-azure.py new file mode 100644 index 000000000..29b9ef15a --- /dev/null +++ b/examples/foundational/14h-function-calling-azure.py @@ -0,0 +1,141 @@ +# +# 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.azure import AzureLLMService +from pipecat.services.cartesia import CartesiaTTSService +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 = AzureLLMService( + api_key=os.getenv("AZURE_CHATGPT_API_KEY"), + endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), + model=os.getenv("AZURE_CHATGPT_MODEL"), + ) + # 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()) diff --git a/examples/foundational/14i-function-calling-fireworks.py b/examples/foundational/14i-function-calling-fireworks.py new file mode 100644 index 000000000..b3b10df87 --- /dev/null +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index d4ffee4ef..321d3a9bb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,7 @@ Website = "https://pipecat.ai" anthropic = [ "anthropic~=0.34.0" ] assemblyai = [ "assemblyai~=0.34.0" ] aws = [ "boto3~=1.35.27" ] -azure = [ "azure-cognitiveservices-speech~=1.40.0" ] +azure = [ "azure-cognitiveservices-speech~=1.40.0", "openai~=1.50.2" ] canonical = [ "aiofiles~=24.1.0" ] cartesia = [ "cartesia~=1.0.13", "websockets~=13.1" ] daily = [ "daily-python~=0.13.0" ] @@ -49,8 +49,10 @@ examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ] fal = [ "fal-client~=0.4.1" ] gladia = [ "websockets~=13.1" ] 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" ] diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index ce845459d..a95ff7d3c 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -25,13 +25,9 @@ from pipecat.frames.frames import ( TTSStoppedFrame, URLImageRawFrame, ) -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.ai_services import ImageGenService, STTService, TTSService from pipecat.services.openai import ( - BaseOpenAILLMService, - OpenAIAssistantContextAggregator, - OpenAIContextAggregatorPair, - OpenAIUserContextAggregator, + OpenAILLMService, ) from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 @@ -398,33 +394,44 @@ def sample_rate_to_output_format(sample_rate: int) -> SpeechSynthesisOutputForma return sample_rate_map.get(sample_rate, SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm) -class AzureLLMService(BaseOpenAILLMService): +class AzureLLMService(OpenAILLMService): + """A service for interacting with Azure OpenAI using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Azure's OpenAI endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + + Args: + api_key (str): The API key for accessing Azure OpenAI + endpoint (str): The Azure endpoint URL + model (str): The model identifier to use + api_version (str, optional): Azure API version. Defaults to "2024-09-01-preview" + **kwargs: Additional keyword arguments passed to OpenAILLMService + """ + def __init__( - self, *, api_key: str, endpoint: str, model: str, api_version: str = "2023-12-01-preview" + self, + *, + api_key: str, + endpoint: str, + model: str, + api_version: str = "2024-09-01-preview", + **kwargs, ): # Initialize variables before calling parent __init__() because that # will call create_client() and we need those values there. self._endpoint = endpoint self._api_version = api_version - super().__init__(api_key=api_key, model=model) + super().__init__(api_key=api_key, model=model, **kwargs) def create_client(self, api_key=None, base_url=None, **kwargs): + """Create OpenAI-compatible client for Azure OpenAI endpoint.""" + logger.debug(f"Creating Azure OpenAI client with endpoint {self._endpoint}") return AsyncAzureOpenAI( api_key=api_key, azure_endpoint=self._endpoint, api_version=self._api_version, ) - @staticmethod - def create_context_aggregator( - context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True - ) -> OpenAIContextAggregatorPair: - user = OpenAIUserContextAggregator(context) - assistant = OpenAIAssistantContextAggregator( - user, expect_stripped_words=assistant_expect_stripped_words - ) - return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) - class AzureBaseTTSService(TTSService): class InputParams(BaseModel): diff --git a/src/pipecat/services/fireworks.py b/src/pipecat/services/fireworks.py index 632e7ad17..1f7d22b36 100644 --- a/src/pipecat/services/fireworks.py +++ b/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 diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py new file mode 100644 index 000000000..505dfcca5 --- /dev/null +++ b/src/pipecat/services/grok.py @@ -0,0 +1,103 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from loguru import logger + +from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.openai import OpenAILLMService + + +class GrokLLMService(OpenAILLMService): + """A service for interacting with Grok's API using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Grok's API endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + + Args: + api_key (str): The API key for accessing Grok's API + base_url (str, optional): The base URL for Grok API. Defaults to "https://api.x.ai/v1" + model (str, optional): The model identifier to use. Defaults to "grok-beta" + **kwargs: Additional keyword arguments passed to OpenAILLMService + """ + + def __init__( + self, + *, + api_key: str, + base_url: str = "https://api.x.ai/v1", + model: str = "grok-beta", + **kwargs, + ): + super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs) + # Initialize counters for token usage metrics + self._prompt_tokens = 0 + self._completion_tokens = 0 + self._total_tokens = 0 + self._has_reported_prompt_tokens = False + self._is_processing = False + + def create_client(self, api_key=None, base_url=None, **kwargs): + """Create OpenAI-compatible client for Grok API endpoint.""" + logger.debug(f"Creating Grok client with api {base_url}") + return super().create_client(api_key, base_url, **kwargs) + + async def _process_context(self, context: OpenAILLMContext): + """Process a context through the LLM and accumulate token usage metrics. + + This method overrides the parent class implementation to handle Grok's + incremental token reporting style, accumulating the counts and reporting + them once at the end of processing. + + Args: + context (OpenAILLMContext): The context to process, containing messages + and other information needed for the LLM interaction. + """ + # Reset all counters and flags at the start of processing + self._prompt_tokens = 0 + self._completion_tokens = 0 + self._total_tokens = 0 + self._has_reported_prompt_tokens = False + self._is_processing = True + + try: + await super()._process_context(context) + finally: + self._is_processing = False + # Report final accumulated token usage at the end of processing + if self._prompt_tokens > 0 or self._completion_tokens > 0: + self._total_tokens = self._prompt_tokens + self._completion_tokens + tokens = LLMTokenUsage( + prompt_tokens=self._prompt_tokens, + completion_tokens=self._completion_tokens, + total_tokens=self._total_tokens, + ) + await super().start_llm_usage_metrics(tokens) + + async def start_llm_usage_metrics(self, tokens: LLMTokenUsage): + """Accumulate token usage metrics during processing. + + This method intercepts the incremental token updates from Grok's API + and accumulates them instead of passing each update to the metrics system. + The final accumulated totals are reported at the end of processing. + + Args: + tokens (LLMTokenUsage): The token usage metrics for the current chunk + of processing, containing prompt_tokens and completion_tokens counts. + """ + # Only accumulate metrics during active processing + if not self._is_processing: + return + + # Record prompt tokens the first time we see them + if not self._has_reported_prompt_tokens and tokens.prompt_tokens > 0: + self._prompt_tokens = tokens.prompt_tokens + self._has_reported_prompt_tokens = True + + # Update completion tokens count if it has increased + if tokens.completion_tokens > self._completion_tokens: + self._completion_tokens = tokens.completion_tokens diff --git a/src/pipecat/services/groq.py b/src/pipecat/services/groq.py new file mode 100644 index 000000000..c9b49d636 --- /dev/null +++ b/src/pipecat/services/groq.py @@ -0,0 +1,39 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from loguru import logger + +from pipecat.services.openai import OpenAILLMService + + +class GroqLLMService(OpenAILLMService): + """A service for interacting with Groq's API using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Groq's API endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + + Args: + api_key (str): The API key for accessing Groq's API + base_url (str, optional): The base URL for Groq API. Defaults to "https://api.groq.com/openai/v1" + model (str, optional): The model identifier to use. Defaults to "llama-3.1-70b-versatile" + **kwargs: Additional keyword arguments passed to OpenAILLMService + """ + + def __init__( + self, + *, + api_key: str, + base_url: str = "https://api.groq.com/openai/v1", + model: str = "llama-3.1-70b-versatile", + **kwargs, + ): + 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 Groq API endpoint.""" + logger.debug(f"Creating Groq client with api {base_url}") + return super().create_client(api_key, base_url, **kwargs) diff --git a/src/pipecat/services/together.py b/src/pipecat/services/together.py index e2aed4da1..e18e9650d 100644 --- a/src/pipecat/services/together.py +++ b/src/pipecat/services/together.py @@ -9,20 +9,19 @@ from loguru import logger from pipecat.services.openai import OpenAILLMService -try: - # Together.ai is recommending OpenAI-compatible function calling, so we've switched over - # to using the OpenAI client library here rather than the Together Python client library. - from openai import AsyncOpenAI, DefaultAsyncHttpxClient -except ModuleNotFoundError as e: - logger.error(f"Exception: {e}") - logger.error( - "In order to use Together.ai, you need to `pip install pipecat-ai[together]`. Also, set `TOGETHER_API_KEY` environment variable." - ) - raise Exception(f"Missing module: {e}") - class TogetherLLMService(OpenAILLMService): - """This class implements inference with Together's Llama 3.1 models""" + """A service for interacting with Together.ai's API using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Together.ai's API endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + + Args: + api_key (str): The API key for accessing Together.ai's API + base_url (str, optional): The base URL for Together.ai API. Defaults to "https://api.together.xyz/v1" + model (str, optional): The model identifier to use. Defaults to "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo" + **kwargs: Additional keyword arguments passed to OpenAILLMService + """ def __init__( self, @@ -35,5 +34,6 @@ class TogetherLLMService(OpenAILLMService): 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 Together.ai API endpoint.""" logger.debug(f"Creating Together.ai client with api {base_url}") return super().create_client(api_key, base_url, **kwargs)