diff --git a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py index 1b62f5edc..657c83bf9 100644 --- a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py +++ b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py @@ -125,7 +125,8 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]): universal_context_messages = copy.deepcopy(universal_context_messages) - # If we have a "system" message as our first message, let's pull that out into "instruction" + # If we have a "system" message as our first message, + # pull that out into "instruction" if universal_context_messages[0].get("role") == "system": system = universal_context_messages.pop(0) content = system.get("content") @@ -136,8 +137,13 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]): if system_instruction: self._system_instruction = system_instruction + # Convert any remaining "system"/"developer" messages to "user", + # as Nova Sonic only supports "user" and "assistant" in history. + for msg in universal_context_messages: + if msg.get("role") in ("system", "developer"): + msg["role"] = "user" + # Process remaining messages to fill out conversation history. - # Nova Sonic supports "user" and "assistant" messages in history. for universal_context_message in universal_context_messages: message = self._from_universal_context_message(universal_context_message) if message: diff --git a/src/pipecat/adapters/services/grok_realtime_adapter.py b/src/pipecat/adapters/services/grok_realtime_adapter.py index 03a3ed475..07b1de843 100644 --- a/src/pipecat/adapters/services/grok_realtime_adapter.py +++ b/src/pipecat/adapters/services/grok_realtime_adapter.py @@ -128,6 +128,11 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter): if not messages: return self.ConvertedMessages(messages=[], system_instruction=system_instruction) + # Convert any remaining "system"/"developer" messages to "user" + for msg in messages: + if msg.get("role") in ("system", "developer"): + msg["role"] = "user" + # Single user message can be sent normally if len(messages) == 1 and messages[0].get("role") == "user": return self.ConvertedMessages( diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 136dc1a2f..6bea0e2ed 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -116,8 +116,8 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): messages = copy.deepcopy(universal_context_messages) system_instruction = None - # If we have a "system" message as our first message, let's pull that out into session - # "instructions" + # If we have a "system" message as our first message, + # pull that out into session "instructions" if messages[0].get("role") == "system": system = messages.pop(0) content = system.get("content") @@ -128,6 +128,11 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): if not messages: return self.ConvertedMessages(messages=[], system_instruction=system_instruction) + # Convert any remaining "system"/"developer" messages to "user" + for msg in messages: + if msg.get("role") in ("system", "developer"): + msg["role"] = "user" + # If we have just a single "user" item, we can just send it normally if len(messages) == 1 and messages[0].get("role") == "user": return self.ConvertedMessages( diff --git a/tests/test_get_llm_invocation_params.py b/tests/test_get_llm_invocation_params.py index 524d409c5..e409e2cd3 100644 --- a/tests/test_get_llm_invocation_params.py +++ b/tests/test_get_llm_invocation_params.py @@ -70,9 +70,12 @@ from google.genai.types import Content, Part from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter +from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter +from pipecat.adapters.services.grok_realtime_adapter import GrokRealtimeLLMAdapter from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter +from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMAdapter from pipecat.adapters.services.perplexity_adapter import PerplexityLLMAdapter from pipecat.processors.aggregators.llm_context import ( @@ -1950,6 +1953,151 @@ class TestOpenAIResponsesGetLLMInvocationParams(unittest.TestCase): self.assertNotIn("instructions", params) +class TestOpenAIRealtimeGetLLMInvocationParams(unittest.TestCase): + def setUp(self) -> None: + self.adapter = OpenAIRealtimeLLMAdapter() + + def test_system_message_extracted_as_instruction(self): + """Initial system message is extracted as system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + self.assertEqual(len(params["messages"]), 1) + + def test_developer_message_becomes_user(self): + """Developer message is converted to user, not extracted as system instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertIsNone(params["system_instruction"]) + # Developer converted to user, then packed with the other user message + self.assertEqual(len(params["messages"]), 1) + + def test_subsequent_developer_message_becomes_user(self): + """Non-initial developer message is converted to user.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "developer", "content": "Extra context."}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + # Developer message converted to user + self.assertEqual(len(params["messages"]), 1) + + def test_empty_messages(self): + """Empty messages list returns empty.""" + context = LLMContext(messages=[]) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["messages"], []) + self.assertIsNone(params["system_instruction"]) + + +class TestGrokRealtimeGetLLMInvocationParams(unittest.TestCase): + def setUp(self) -> None: + self.adapter = GrokRealtimeLLMAdapter() + + def test_system_message_extracted_as_instruction(self): + """Initial system message is extracted as system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + self.assertEqual(len(params["messages"]), 1) + + def test_developer_message_becomes_user(self): + """Developer message is converted to user, not extracted as system instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertIsNone(params["system_instruction"]) + # Developer converted to user, then packed with the other user message + self.assertEqual(len(params["messages"]), 1) + + def test_subsequent_developer_message_becomes_user(self): + """Non-initial developer message is converted to user.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "developer", "content": "Extra context."}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + self.assertEqual(len(params["messages"]), 1) + + def test_empty_messages(self): + """Empty messages list returns empty.""" + context = LLMContext(messages=[]) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["messages"], []) + self.assertIsNone(params["system_instruction"]) + + +class TestAWSNovaSonicGetLLMInvocationParams(unittest.TestCase): + def setUp(self) -> None: + self.adapter = AWSNovaSonicLLMAdapter() + + def test_system_message_extracted_as_instruction(self): + """Initial system message is extracted as system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + self.assertEqual(len(params["messages"]), 1) + + def test_developer_message_becomes_user(self): + """Developer message is converted to user, not extracted as system instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertIsNone(params["system_instruction"]) + # Both messages should be present (developer as user, plus the real user) + self.assertEqual(len(params["messages"]), 2) + + def test_subsequent_developer_message_becomes_user(self): + """Non-initial developer message is converted to user.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "developer", "content": "Extra context."}, + {"role": "assistant", "content": "Hi"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["system_instruction"], "You are helpful.") + # Developer becomes user, plus assistant + self.assertEqual(len(params["messages"]), 2) + + class TestBaseLLMAdapterHelpers(unittest.TestCase): """Tests for the shared helper methods on BaseLLMAdapter."""