From 2b2b0f81215cf56bc711c7fb38897e6aa83ec255 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 09:23:26 -0400 Subject: [PATCH 1/4] Add MistralLLMService --- src/pipecat/services/mistral/__init__.py | 0 src/pipecat/services/mistral/llm.py | 188 +++++++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 src/pipecat/services/mistral/__init__.py create mode 100644 src/pipecat/services/mistral/llm.py diff --git a/src/pipecat/services/mistral/__init__.py b/src/pipecat/services/mistral/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/mistral/llm.py b/src/pipecat/services/mistral/llm.py new file mode 100644 index 000000000..0c9fee634 --- /dev/null +++ b/src/pipecat/services/mistral/llm.py @@ -0,0 +1,188 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Mistral LLM service implementation using OpenAI-compatible interface.""" + +from typing import List, Sequence + +from loguru import logger +from openai import AsyncStream +from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam + +from pipecat.frames.frames import FunctionCallFromLLM +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.openai.llm import OpenAILLMService + + +class MistralLLMService(OpenAILLMService): + """A service for interacting with Mistral's API using the OpenAI-compatible interface. + + This service extends OpenAILLMService to connect to Mistral's API endpoint while + maintaining full compatibility with OpenAI's interface and functionality. + """ + + def __init__( + self, + *, + api_key: str, + base_url: str = "https://api.mistral.ai/v1", + model: str = "mistral-small-latest", + **kwargs, + ): + """Initialize the Mistral LLM service. + + Args: + api_key: The API key for accessing Mistral's API. + base_url: The base URL for Mistral API. Defaults to "https://api.mistral.ai/v1". + model: The model identifier to use. Defaults to "mistral-small-latest". + **kwargs: Additional keyword arguments passed to 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 Mistral API endpoint. + + Args: + api_key: The API key for authentication. If None, uses instance key. + base_url: The base URL for the API. If None, uses instance URL. + **kwargs: Additional arguments passed to the client constructor. + + Returns: + An OpenAI-compatible client configured for Mistral API. + """ + logger.debug(f"Creating Mistral client with api {base_url}") + return super().create_client(api_key, base_url, **kwargs) + + def _apply_mistral_assistant_prefix( + self, messages: List[ChatCompletionMessageParam] + ) -> List[ChatCompletionMessageParam]: + """Apply Mistral's assistant message prefix requirement. + + Mistral requires assistant messages to have prefix=True when they + are the final message in a conversation. According to Mistral's API: + - Assistant messages with prefix=True MUST be the last message + - Only add prefix=True to the final assistant message when needed + - This allows assistant messages to be accepted as the last message + + Args: + messages: The original list of messages. + + Returns: + Messages with Mistral prefix requirement applied to final assistant message. + """ + if not messages: + return messages + + # Create a copy to avoid modifying the original + fixed_messages = [dict(msg) for msg in messages] + + # Get the last message + last_message = fixed_messages[-1] + + # Only add prefix=True to the last message if it's an assistant message + # and Mistral would otherwise reject it + if last_message.get("role") == "assistant" and "prefix" not in last_message: + last_message["prefix"] = True + + return fixed_messages + + async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]): + """Execute function calls, filtering out already-completed ones. + + Mistral and OpenAI have different function call detection patterns: + + OpenAI (Stream-based detection): + - Detects function calls only from streaming chunks as the LLM generates them + - Second LLM completion doesn't re-detect existing tool_calls in message history + - Function calls execute exactly once + + Mistral (Message-based detection): + - Detects function calls from the complete message history on each completion + - Second LLM completion with the response re-detects the same tool_calls from + previous messages + - Without filtering, function calls would execute twice + + This method prevents duplicate execution by: + 1. Checking message history for existing tool result messages + 2. Filtering out function calls that already have corresponding results + 3. Only executing function calls that haven't been completed yet + + Note: This filtering prevents duplicate function execution, but the + on_function_calls_started event may still fire twice due to the detection + pattern difference. This is expected behavior. + + Args: + function_calls: The function calls to potentially execute. + """ + if not function_calls: + return + + # Filter out function calls that already have results + calls_to_execute = [] + + # Get messages from the first function call's context (they should all have the same context) + messages = function_calls[0].context.get_messages() if function_calls else [] + + # Get all tool_call_ids that already have results + executed_call_ids = set() + for msg in messages: + if msg.get("role") == "tool" and msg.get("tool_call_id"): + executed_call_ids.add(msg.get("tool_call_id")) + + # Only include function calls that haven't been executed yet + for call in function_calls: + if call.tool_call_id not in executed_call_ids: + calls_to_execute.append(call) + else: + logger.trace( + f"Skipping already-executed function call: {call.function_name}:{call.tool_call_id}" + ) + + # Call parent method with filtered list + if calls_to_execute: + await super().run_function_calls(calls_to_execute) + + async def get_chat_completions( + self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] + ) -> AsyncStream[ChatCompletionChunk]: + """Create a streaming chat completion using Mistral's API. + + Args: + context: The context object containing tools configuration + and other settings for the chat completion. + messages: The list of messages comprising + the conversation history and current request. + + Returns: + A streaming response of chat completion + chunks that can be processed asynchronously. + """ + # Apply Mistral's assistant prefix requirement for API compatibility + fixed_messages = self._apply_mistral_assistant_prefix(messages) + + params = { + "model": self.model_name, + "stream": True, + "messages": fixed_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"], + } + + # Handle Mistral-specific parameter mapping + # Mistral uses "random_seed" instead of "seed" + if self._settings["seed"]: + params["random_seed"] = self._settings["seed"] + + # Add any extra parameters + params.update(self._settings["extra"]) + + chunks = await self._client.chat.completions.create(**params) + return chunks From b7ae2989ac9f52b2977105698b1406b8d75ffec4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 09:40:54 -0400 Subject: [PATCH 2/4] Add foundational 14w-function-calling.py --- CHANGELOG.md | 8 + .../14w-function-calling-mistral.py | 165 ++++++++++++++++++ scripts/evals/run-release-evals.py | 1 + 3 files changed, 174 insertions(+) create mode 100644 examples/foundational/14w-function-calling-mistral.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c67321c4d..2bdaeb794 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,11 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `MistralLLMService`, using Mistral's chat completion API. + ### Fixed - Fixed an issue where `AsyncAITTSService` had very high latency in responding by adding `force=true` when sending the flush command. +### Other + +- Added `14w-function-calling-mistal.py` using `MistralLLMService`. + ## [0.0.80] - 2025-08-13 ### Added diff --git a/examples/foundational/14w-function-calling-mistral.py b/examples/foundational/14w-function-calling-mistral.py new file mode 100644 index 000000000..2fb46beb6 --- /dev/null +++ b/examples/foundational/14w-function-calling-mistral.py @@ -0,0 +1,165 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import TTSSpeakFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.mistral.llm import MistralLLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +# 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(), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + llm = MistralLLMService(api_key=os.getenv("MISTRAL_API_KEY")) + + # You can also register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + 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 user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + + 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(), + stt, + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + 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. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @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() diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 1cee7e684..49acc432a 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -113,6 +113,7 @@ TESTS_14 = [ ("14q-function-calling-qwen.py", PROMPT_WEATHER, EVAL_WEATHER), ("14r-function-calling-aws.py", PROMPT_WEATHER, EVAL_WEATHER), ("14v-function-calling-openai.py", PROMPT_WEATHER, EVAL_WEATHER), + ("14w-function-calling-mistral.py", PROMPT_WEATHER, EVAL_WEATHER), # Currently not working. # ("14c-function-calling-together.py", PROMPT_WEATHER, EVAL_WEATHER), # ("14k-function-calling-cerebras.py", PROMPT_WEATHER, EVAL_WEATHER), From 7f0494aa0428fcd4e881ae3c712c5025759ad65a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Aug 2025 10:32:18 -0400 Subject: [PATCH 3/4] Override build_chat_completion_params for Mistral --- src/pipecat/services/mistral/llm.py | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/src/pipecat/services/mistral/llm.py b/src/pipecat/services/mistral/llm.py index 0c9fee634..e316150ac 100644 --- a/src/pipecat/services/mistral/llm.py +++ b/src/pipecat/services/mistral/llm.py @@ -145,20 +145,15 @@ class MistralLLMService(OpenAILLMService): if calls_to_execute: await super().run_function_calls(calls_to_execute) - async def get_chat_completions( + def build_chat_completion_params( self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] - ) -> AsyncStream[ChatCompletionChunk]: - """Create a streaming chat completion using Mistral's API. + ) -> dict: + """Build parameters for Mistral chat completion request. - Args: - context: The context object containing tools configuration - and other settings for the chat completion. - messages: The list of messages comprising - the conversation history and current request. - - Returns: - A streaming response of chat completion - chunks that can be processed asynchronously. + Handles Mistral-specific requirements including: + - Assistant message prefix requirement for API compatibility + - Parameter mapping (random_seed instead of seed) + - Core completion settings """ # Apply Mistral's assistant prefix requirement for API compatibility fixed_messages = self._apply_mistral_assistant_prefix(messages) @@ -184,5 +179,4 @@ class MistralLLMService(OpenAILLMService): # Add any extra parameters params.update(self._settings["extra"]) - chunks = await self._client.chat.completions.create(**params) - return chunks + return params From 42bd1e9d4043db94c1205c0585c2720da6e2ffa0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Aug 2025 11:15:52 -0400 Subject: [PATCH 4/4] Add Mistral to README and pyproject.toml --- README.md | 2 +- pyproject.toml | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 230aafdd6..fb66a4f27 100644 --- a/README.md +++ b/README.md @@ -54,7 +54,7 @@ You can connect to Pipecat from any platform using our official SDKs: | Category | Services | | ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [NVIDIA Riva](https://docs.pipecat.ai/server/services/stt/riva), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [SambaNova (Whisper)](https://docs.pipecat.ai/server/services/stt/sambanova), [Soniox](https://docs.pipecat.ai/server/services/stt/soniox), [Speechmatics](https://docs.pipecat.ai/server/services/stt/speechmatics), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | -| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [SambaNova](https://docs.pipecat.ai/server/services/llm/sambanova) [Together AI](https://docs.pipecat.ai/server/services/llm/together) | +| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [Mistral](https://docs.pipecat.ai/server/services/llm/mistral), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [SambaNova](https://docs.pipecat.ai/server/services/llm/sambanova) [Together AI](https://docs.pipecat.ai/server/services/llm/together) | | Text-to-Speech | [Async](https://docs.pipecat.ai/server/services/tts/asyncai), [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [Groq](https://docs.pipecat.ai/server/services/tts/groq), [Inworld](https://docs.pipecat.ai/server/services/tts/inworld), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [NVIDIA Riva](https://docs.pipecat.ai/server/services/tts/riva), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | | Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | | Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | diff --git a/pyproject.toml b/pyproject.toml index 1f9b9165c..d61c3efe3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -73,6 +73,7 @@ lmnt = [ "websockets>=13.1,<15.0" ] local = [ "pyaudio~=0.2.14" ] mcp = [ "mcp[cli]~=1.9.4" ] mem0 = [ "mem0ai~=0.1.94" ] +mistral = [] mlx-whisper = [ "mlx-whisper~=0.4.2" ] moondream = [ "accelerate~=1.10.0", "einops~=0.8.0", "pyvips[binary]~=3.0.0", "timm~=1.0.13", "transformers>=4.48.0" ] nim = []