From bb33045389fea76d47afaca9730cd09a8f43604e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 24 Mar 2026 17:30:35 -0400 Subject: [PATCH] Add system instruction conflict resolution tests for realtime adapters Test that OpenAI Realtime, Grok Realtime, and Nova Sonic adapters prefer init-provided system_instruction over context-provided, warn on conflicts, and don't warn for developer messages. --- tests/test_get_llm_invocation_params.py | 126 ++++++++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/tests/test_get_llm_invocation_params.py b/tests/test_get_llm_invocation_params.py index 883c60631..6f0015ada 100644 --- a/tests/test_get_llm_invocation_params.py +++ b/tests/test_get_llm_invocation_params.py @@ -2085,6 +2085,48 @@ class TestOpenAIRealtimeGetLLMInvocationParams(unittest.TestCase): self.assertEqual(params["messages"], []) self.assertIsNone(params["system_instruction"]) + def test_both_system_instruction_and_system_message_warns(self): + """system_instruction + initial system message warns and uses system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_called_once() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_both_system_instruction_and_developer_message_no_warning(self): + """system_instruction + initial developer message: no warning, developer becomes user.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_not_called() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_system_instruction_only(self): + """system_instruction without context system message returns system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") + + self.assertEqual(params["system_instruction"], "Be concise.") + class TestGrokRealtimeGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -2135,6 +2177,48 @@ class TestGrokRealtimeGetLLMInvocationParams(unittest.TestCase): self.assertEqual(params["messages"], []) self.assertIsNone(params["system_instruction"]) + def test_both_system_instruction_and_system_message_warns(self): + """system_instruction + initial system message warns and uses system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_called_once() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_both_system_instruction_and_developer_message_no_warning(self): + """system_instruction + initial developer message: no warning, developer becomes user.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_not_called() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_system_instruction_only(self): + """system_instruction without context system message returns system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") + + self.assertEqual(params["system_instruction"], "Be concise.") + class TestAWSNovaSonicGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -2179,6 +2263,48 @@ class TestAWSNovaSonicGetLLMInvocationParams(unittest.TestCase): # Developer becomes user, plus assistant self.assertEqual(len(params["messages"]), 2) + def test_both_system_instruction_and_system_message_warns(self): + """system_instruction + initial system message warns and uses system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_called_once() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_both_system_instruction_and_developer_message_no_warning(self): + """system_instruction + initial developer message: no warning, developer becomes user.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + params = self.adapter.get_llm_invocation_params( + context, system_instruction="Be concise." + ) + mock_logger.warning.assert_not_called() + + self.assertEqual(params["system_instruction"], "Be concise.") + + def test_system_instruction_only(self): + """system_instruction without context system message returns system_instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") + + self.assertEqual(params["system_instruction"], "Be concise.") + class TestBaseLLMAdapterHelpers(unittest.TestCase): """Tests for the shared helper methods on BaseLLMAdapter."""