From 7377d88cf5d6660f9b952043a5fc865b3023723a Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 20 Mar 2026 14:36:23 -0400 Subject: [PATCH] Move system_instruction tests into test_get_llm_invocation_params.py --- tests/test_adapter_system_instruction.py | 587 ----------------------- tests/test_get_llm_invocation_params.py | 485 +++++++++++++++++++ 2 files changed, 485 insertions(+), 587 deletions(-) delete mode 100644 tests/test_adapter_system_instruction.py diff --git a/tests/test_adapter_system_instruction.py b/tests/test_adapter_system_instruction.py deleted file mode 100644 index 125702dd6..000000000 --- a/tests/test_adapter_system_instruction.py +++ /dev/null @@ -1,587 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Unit tests for system_instruction and developer message handling in LLM adapters. - -Tests cover: - -1. system_instruction only (no system/developer in context) -2. Initial "system" message only (no system_instruction) -3. Initial "developer" message only (no system_instruction) -> promoted to system instruction -4. Both system_instruction and initial "system" message -> warns -5. Both system_instruction and initial "developer" message -> does NOT warn; developer becomes "user" -6. Non-OpenAI adapters: subsequent "developer" messages converted to "user" -7. Non-OpenAI adapters: initial "system" discarded when system_instruction provided -8. Gemini: non-initial "system" message is converted to "user" (not extracted) -9. Single system-only message: converted to "user" instead of extracting (empty list prevention) -""" - -import unittest -from unittest.mock import patch - -from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter -from pipecat.processors.aggregators.llm_context import LLMContext, LLMStandardMessage - - -class TestOpenAIAdapterSystemInstruction(unittest.TestCase): - """Tests for the OpenAI ChatCompletion adapter.""" - - def setUp(self): - self.adapter = OpenAILLMAdapter() - - def test_system_instruction_only(self): - """system_instruction alone is prepended as a system message.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") - - self.assertEqual(params["messages"][0]["role"], "system") - self.assertEqual(params["messages"][0]["content"], "Be helpful.") - self.assertEqual(params["messages"][1]["role"], "user") - - def test_initial_system_message_only(self): - """Initial system message without system_instruction passes through.""" - 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(len(params["messages"]), 2) - self.assertEqual(params["messages"][0]["role"], "system") - self.assertEqual(params["messages"][0]["content"], "You are helpful.") - - def test_both_system_instruction_and_system_message_warns(self): - """system_instruction + initial system message warns but allows both.""" - 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() - warning_msg = mock_logger.warning.call_args[0][0] - self.assertIn("may be unintended", warning_msg) - - # Both are present: prepended system_instruction + original system message - self.assertEqual(params["messages"][0]["content"], "Be concise.") - self.assertEqual(params["messages"][1]["content"], "You are helpful.") - - def test_both_system_instruction_and_developer_message_no_warning(self): - """system_instruction + initial developer message does NOT warn.""" - 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() - - # system_instruction prepended, developer message stays in messages - self.assertEqual(params["messages"][0]["content"], "Be concise.") - self.assertEqual(params["messages"][1]["role"], "developer") - - def test_warning_fires_only_once(self): - """Conflict warning fires only once per adapter instance.""" - 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: - self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") - self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") - mock_logger.warning.assert_called_once() - - -class TestAnthropicAdapterSystemInstruction(unittest.TestCase): - """Tests for the Anthropic adapter.""" - - def setUp(self): - from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter - - self.adapter = AnthropicLLMAdapter() - - def test_system_instruction_only(self): - """system_instruction alone becomes the system parameter.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params( - context, enable_prompt_caching=False, system_instruction="Be helpful." - ) - - self.assertEqual(params["system"], "Be helpful.") - self.assertEqual(len(params["messages"]), 1) - self.assertEqual(params["messages"][0]["role"], "user") - - def test_initial_system_message_only(self): - """Initial system message is extracted as the system parameter.""" - 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, enable_prompt_caching=False) - - self.assertEqual(params["system"], "You are helpful.") - self.assertEqual(len(params["messages"]), 1) - self.assertEqual(params["messages"][0]["role"], "user") - - def test_initial_developer_message_promoted(self): - """Initial developer message without system_instruction is promoted to system.""" - messages: list[LLMStandardMessage] = [ - {"role": "developer", "content": "Extra context."}, - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False) - - self.assertEqual(params["system"], "Extra context.") - self.assertEqual(len(params["messages"]), 1) - - 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, - enable_prompt_caching=False, - system_instruction="Be concise.", - ) - mock_logger.warning.assert_called_once() - warning_msg = mock_logger.warning.call_args[0][0] - self.assertIn("Using system_instruction", warning_msg) - - self.assertEqual(params["system"], "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": "assistant", "content": "Hi"}, - {"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, - enable_prompt_caching=False, - system_instruction="Be concise.", - ) - mock_logger.warning.assert_not_called() - - self.assertEqual(params["system"], "Be concise.") - # Developer message should have been converted to "user" - self.assertEqual(params["messages"][0]["role"], "user") - self.assertEqual(params["messages"][0]["content"], "Extra context.") - - def test_subsequent_developer_messages_converted_to_user(self): - """Subsequent developer messages are converted to user role.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - {"role": "assistant", "content": "Hi"}, - {"role": "developer", "content": "More instructions"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False) - - # Developer message was converted to "user" - self.assertEqual(params["messages"][2]["role"], "user") - self.assertEqual(params["messages"][2]["content"], "More instructions") - - def test_initial_system_discarded_when_system_instruction_provided(self): - """Initial system message is discarded when system_instruction is provided.""" - messages: list[LLMStandardMessage] = [ - {"role": "system", "content": "Old instruction."}, - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - - with patch("pipecat.adapters.base_llm_adapter.logger"): - params = self.adapter.get_llm_invocation_params( - context, - enable_prompt_caching=False, - system_instruction="New instruction.", - ) - - self.assertEqual(params["system"], "New instruction.") - # Only the user message should remain - self.assertEqual(len(params["messages"]), 1) - self.assertEqual(params["messages"][0]["role"], "user") - - def test_single_system_message_becomes_user(self): - """A lone system message is converted to user (not extracted) to prevent empty history.""" - messages: list[LLMStandardMessage] = [ - {"role": "system", "content": "You are helpful."}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False) - - from anthropic import NOT_GIVEN - - self.assertEqual(params["system"], NOT_GIVEN) - self.assertEqual(len(params["messages"]), 1) - self.assertEqual(params["messages"][0]["role"], "user") - - -class TestBedrockAdapterSystemInstruction(unittest.TestCase): - """Tests for the AWS Bedrock adapter.""" - - def setUp(self): - from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMAdapter - - self.adapter = AWSBedrockLLMAdapter() - - def test_system_instruction_only(self): - """system_instruction alone becomes the system parameter.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") - - self.assertEqual(params["system"], [{"text": "Be helpful."}]) - - def test_initial_system_message_only(self): - """Initial system message is extracted as the system parameter.""" - 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"], [{"text": "You are helpful."}]) - self.assertEqual(len(params["messages"]), 1) - - def test_initial_developer_message_promoted(self): - """Initial developer message without system_instruction is promoted.""" - 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.assertEqual(params["system"], [{"text": "Extra context."}]) - - 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"], [{"text": "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"], [{"text": "Be concise."}]) - self.assertEqual(params["messages"][0]["role"], "user") - - def test_subsequent_developer_messages_converted_to_user(self): - """Subsequent developer messages are converted to user role.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - {"role": "assistant", "content": "Hi"}, - {"role": "developer", "content": "More instructions"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context) - - self.assertEqual(params["messages"][2]["role"], "user") - - def test_single_system_message_becomes_user(self): - """A lone system message is converted to user to prevent empty history.""" - messages: list[LLMStandardMessage] = [ - {"role": "system", "content": "You are helpful."}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context) - - self.assertIsNone(params["system"]) - self.assertEqual(len(params["messages"]), 1) - self.assertEqual(params["messages"][0]["role"], "user") - - -class TestGeminiAdapterSystemInstruction(unittest.TestCase): - """Tests for the Gemini adapter.""" - - def setUp(self): - from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter - - self.adapter = GeminiLLMAdapter() - - def test_system_instruction_only(self): - """system_instruction alone becomes the system_instruction parameter.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") - - self.assertEqual(params["system_instruction"], "Be helpful.") - - def test_initial_system_message_only(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_initial_developer_message_promoted(self): - """Initial developer message without system_instruction is promoted.""" - 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.assertEqual(params["system_instruction"], "Extra context.") - - 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_non_initial_system_message_not_extracted(self): - """Non-initial system message is converted to user, not extracted as system instruction.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - {"role": "system", "content": "Late system message"}, - {"role": "user", "content": "How are you?"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context) - - # No system instruction should be extracted from non-initial position - self.assertIsNone(params["system_instruction"]) - # The system message should have been converted to user role in the Gemini Content - # (we check that 3 messages are present, meaning no extraction happened) - self.assertEqual(len(params["messages"]), 3) - - def test_subsequent_developer_messages_converted_to_user(self): - """Subsequent developer messages are converted to user role.""" - messages: list[LLMStandardMessage] = [ - {"role": "user", "content": "Hello"}, - {"role": "developer", "content": "More instructions"}, - ] - context = LLMContext(messages=messages) - params = self.adapter.get_llm_invocation_params(context) - - self.assertEqual(len(params["messages"]), 2) - # Second message (developer) should be converted to user in Google format - self.assertEqual(params["messages"][1].role, "user") - - -class TestBaseLLMAdapterHelpers(unittest.TestCase): - """Tests for the shared helper methods on BaseLLMAdapter.""" - - def setUp(self): - # Use OpenAILLMAdapter as a concrete implementation for testing the base helpers - self.adapter = OpenAILLMAdapter() - - def test_extract_system_message(self): - """System message is extracted from messages[0].""" - messages = [ - {"role": "system", "content": "Be helpful."}, - {"role": "user", "content": "Hello"}, - ] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction=None - ) - - self.assertEqual(content, "Be helpful.") - self.assertEqual(role, "system") - self.assertEqual(len(messages), 1) # popped - - def test_extract_developer_without_system_instruction(self): - """Developer message is extracted when no system_instruction.""" - messages = [ - {"role": "developer", "content": "Context."}, - {"role": "user", "content": "Hello"}, - ] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction=None - ) - - self.assertEqual(content, "Context.") - self.assertEqual(role, "developer") - self.assertEqual(len(messages), 1) - - def test_developer_with_system_instruction_converts_to_user(self): - """Developer message with system_instruction is converted to user, not extracted.""" - messages = [ - {"role": "developer", "content": "Context."}, - {"role": "user", "content": "Hello"}, - ] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction="Be helpful." - ) - - self.assertIsNone(content) - self.assertIsNone(role) - self.assertEqual(len(messages), 2) # not popped - self.assertEqual(messages[0]["role"], "user") # converted to user - - def test_single_system_message_becomes_user(self): - """Single system message is converted to user instead of extracting (empty prevention).""" - messages = [ - {"role": "system", "content": "Be helpful."}, - ] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction=None - ) - - self.assertIsNone(content) - self.assertIsNone(role) - self.assertEqual(len(messages), 1) # not popped - self.assertEqual(messages[0]["role"], "user") - - def test_non_system_message_ignored(self): - """Non-system/developer first message is ignored.""" - messages = [ - {"role": "user", "content": "Hello"}, - ] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction=None - ) - - self.assertIsNone(content) - self.assertIsNone(role) - self.assertEqual(len(messages), 1) - - def test_empty_messages(self): - """Empty messages list returns None.""" - messages = [] - content, role = self.adapter._extract_initial_system_or_developer( - messages, system_instruction=None - ) - - self.assertIsNone(content) - self.assertIsNone(role) - - def test_resolve_both_system_discard(self): - """Resolve with discard=True: system_instruction wins, warns.""" - with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: - result = self.adapter._resolve_system_instruction( - "from context", "system", "from settings", discard_context_system=True - ) - mock_logger.warning.assert_called_once() - - self.assertEqual(result, "from settings") - - def test_resolve_both_system_keep(self): - """Resolve with discard=False: warns but returns system_instruction.""" - with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: - result = self.adapter._resolve_system_instruction( - "from context", "system", "from settings", discard_context_system=False - ) - mock_logger.warning.assert_called_once() - - self.assertEqual(result, "from settings") - - def test_resolve_only_system_instruction(self): - """Only system_instruction: returns it, no warning.""" - with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: - result = self.adapter._resolve_system_instruction( - None, None, "from settings", discard_context_system=True - ) - mock_logger.warning.assert_not_called() - - self.assertEqual(result, "from settings") - - def test_resolve_only_context_system_discard(self): - """Only context system (discard=True): returns it.""" - result = self.adapter._resolve_system_instruction( - "from context", "system", None, discard_context_system=True - ) - - self.assertEqual(result, "from context") - - def test_resolve_only_context_system_keep(self): - """Only context system (discard=False): returns None (already in messages).""" - result = self.adapter._resolve_system_instruction( - "from context", "system", None, discard_context_system=False - ) - - self.assertIsNone(result) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_get_llm_invocation_params.py b/tests/test_get_llm_invocation_params.py index 279288661..d88d6193d 100644 --- a/tests/test_get_llm_invocation_params.py +++ b/tests/test_get_llm_invocation_params.py @@ -14,6 +14,8 @@ For OpenAI adapter: 2. LLMSpecificMessage objects with llm='openai' are included and others are filtered out 3. Complex message structures (like multi-part content) are preserved 4. System instructions are preserved throughout messages at any position +5. system_instruction is prepended as a system message, with conflict warnings +6. Developer messages pass through without triggering warnings For Gemini adapter: 1. LLMStandardMessage objects are converted to Gemini Content format @@ -22,6 +24,8 @@ For Gemini adapter: 4. System messages are extracted as system_instruction (without duplication) 5. Single system instruction is converted to user message when no other messages exist 6. Multiple system instructions: first extracted, later ones converted to user messages +7. system_instruction overrides context system message, with conflict warnings +8. Developer messages are promoted to system instruction or converted to user For Anthropic adapter: 1. LLMStandardMessage objects are converted to Anthropic MessageParam format @@ -30,6 +34,8 @@ For Anthropic adapter: 4. System messages: first extracted as system parameter, later ones converted to user messages 5. Consecutive messages with same role are merged into multi-content-block messages 6. Empty text content is converted to "(empty)" +7. system_instruction overrides context system message, with conflict warnings +8. Developer messages are promoted to system instruction or converted to user For AWS Bedrock adapter: 1. LLMStandardMessage objects are converted to AWS Bedrock format @@ -38,9 +44,16 @@ For AWS Bedrock adapter: 4. System messages: first extracted as system parameter, later ones converted to user messages 5. Consecutive messages with same role are merged into multi-content-block messages 6. Empty text content is converted to "(empty)" +7. system_instruction overrides context system message, with conflict warnings +8. Developer messages are promoted to system instruction or converted to user + +For BaseLLMAdapter helpers: +1. _extract_initial_system_or_developer: system/developer extraction and conversion logic +2. _resolve_system_instruction: conflict resolution between context and settings """ import unittest +from unittest.mock import patch from google.genai.types import Content, Part @@ -214,6 +227,82 @@ class TestOpenAIGetLLMInvocationParams(unittest.TestCase): self.assertEqual(params["messages"][4]["role"], "user") self.assertEqual(params["messages"][6]["role"], "assistant") + def test_system_instruction_only(self): + """system_instruction alone is prepended as a system message.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") + + self.assertEqual(params["messages"][0]["role"], "system") + self.assertEqual(params["messages"][0]["content"], "Be helpful.") + self.assertEqual(params["messages"][1]["role"], "user") + + def test_initial_system_message_only(self): + """Initial system message without system_instruction passes through.""" + 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(len(params["messages"]), 2) + self.assertEqual(params["messages"][0]["role"], "system") + self.assertEqual(params["messages"][0]["content"], "You are helpful.") + + def test_both_system_instruction_and_system_message_warns(self): + """system_instruction + initial system message warns but allows both.""" + 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() + warning_msg = mock_logger.warning.call_args[0][0] + self.assertIn("may be unintended", warning_msg) + + # Both are present: prepended system_instruction + original system message + self.assertEqual(params["messages"][0]["content"], "Be concise.") + self.assertEqual(params["messages"][1]["content"], "You are helpful.") + + def test_both_system_instruction_and_developer_message_no_warning(self): + """system_instruction + initial developer message does NOT warn.""" + 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() + + # system_instruction prepended, developer message stays in messages + self.assertEqual(params["messages"][0]["content"], "Be concise.") + self.assertEqual(params["messages"][1]["role"], "developer") + + def test_warning_fires_only_once(self): + """Conflict warning fires only once per adapter instance.""" + 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: + self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") + self.adapter.get_llm_invocation_params(context, system_instruction="Be concise.") + mock_logger.warning.assert_called_once() + class TestGeminiGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -478,6 +567,100 @@ class TestGeminiGetLLMInvocationParams(unittest.TestCase): # Should have 2 model messages (converted from assistant) self.assertEqual(len(model_messages), 2) + def test_system_instruction_only(self): + """system_instruction alone becomes the system_instruction parameter.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") + + self.assertEqual(params["system_instruction"], "Be helpful.") + + def test_initial_system_message_only(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_initial_developer_message_promoted(self): + """Initial developer message without system_instruction is promoted.""" + 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.assertEqual(params["system_instruction"], "Extra context.") + + 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_non_initial_system_message_not_extracted(self): + """Non-initial system message is converted to user, not extracted as system instruction.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + {"role": "system", "content": "Late system message"}, + {"role": "user", "content": "How are you?"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + # No system instruction should be extracted from non-initial position + self.assertIsNone(params["system_instruction"]) + # The system message should have been converted to user role in the Gemini Content + # (we check that 3 messages are present, meaning no extraction happened) + self.assertEqual(len(params["messages"]), 3) + + def test_subsequent_developer_messages_converted_to_user(self): + """Subsequent developer messages are converted to user role.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + {"role": "developer", "content": "More instructions"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(len(params["messages"]), 2) + # Second message (developer) should be converted to user in Google format + self.assertEqual(params["messages"][1].role, "user") + class TestAnthropicGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -737,6 +920,108 @@ class TestAnthropicGetLLMInvocationParams(unittest.TestCase): self.assertEqual(params["messages"][0]["role"], "user") self.assertEqual(params["messages"][0]["content"], "You are a helpful assistant.") + def test_system_instruction_only(self): + """system_instruction alone becomes the system parameter.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params( + context, enable_prompt_caching=False, system_instruction="Be helpful." + ) + + self.assertEqual(params["system"], "Be helpful.") + self.assertEqual(len(params["messages"]), 1) + self.assertEqual(params["messages"][0]["role"], "user") + + def test_initial_developer_message_promoted(self): + """Initial developer message without system_instruction is promoted to system.""" + messages: list[LLMStandardMessage] = [ + {"role": "developer", "content": "Extra context."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False) + + self.assertEqual(params["system"], "Extra context.") + self.assertEqual(len(params["messages"]), 1) + + 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, + enable_prompt_caching=False, + system_instruction="Be concise.", + ) + mock_logger.warning.assert_called_once() + warning_msg = mock_logger.warning.call_args[0][0] + self.assertIn("Using system_instruction", warning_msg) + + self.assertEqual(params["system"], "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": "assistant", "content": "Hi"}, + {"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, + enable_prompt_caching=False, + system_instruction="Be concise.", + ) + mock_logger.warning.assert_not_called() + + self.assertEqual(params["system"], "Be concise.") + # Developer message should have been converted to "user" + self.assertEqual(params["messages"][0]["role"], "user") + self.assertEqual(params["messages"][0]["content"], "Extra context.") + + def test_subsequent_developer_messages_converted_to_user(self): + """Subsequent developer messages are converted to user role.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi"}, + {"role": "developer", "content": "More instructions"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, enable_prompt_caching=False) + + # Developer message was converted to "user" + self.assertEqual(params["messages"][2]["role"], "user") + self.assertEqual(params["messages"][2]["content"], "More instructions") + + def test_initial_system_discarded_when_system_instruction_provided(self): + """Initial system message is discarded when system_instruction is provided.""" + messages: list[LLMStandardMessage] = [ + {"role": "system", "content": "Old instruction."}, + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + + with patch("pipecat.adapters.base_llm_adapter.logger"): + params = self.adapter.get_llm_invocation_params( + context, + enable_prompt_caching=False, + system_instruction="New instruction.", + ) + + self.assertEqual(params["system"], "New instruction.") + # Only the user message should remain + self.assertEqual(len(params["messages"]), 1) + self.assertEqual(params["messages"][0]["role"], "user") + class TestAWSBedrockGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -996,6 +1281,72 @@ class TestAWSBedrockGetLLMInvocationParams(unittest.TestCase): params["messages"][0]["content"][0]["text"], "You are a helpful assistant." ) + def test_system_instruction_only(self): + """system_instruction alone becomes the system parameter.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.") + + self.assertEqual(params["system"], [{"text": "Be helpful."}]) + + def test_initial_developer_message_promoted(self): + """Initial developer message without system_instruction is promoted.""" + 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.assertEqual(params["system"], [{"text": "Extra context."}]) + + 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"], [{"text": "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"], [{"text": "Be concise."}]) + self.assertEqual(params["messages"][0]["role"], "user") + + def test_subsequent_developer_messages_converted_to_user(self): + """Subsequent developer messages are converted to user role.""" + messages: list[LLMStandardMessage] = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi"}, + {"role": "developer", "content": "More instructions"}, + ] + context = LLMContext(messages=messages) + params = self.adapter.get_llm_invocation_params(context) + + self.assertEqual(params["messages"][2]["role"], "user") + class TestPerplexityGetLLMInvocationParams(unittest.TestCase): def setUp(self) -> None: @@ -1214,5 +1565,139 @@ class TestPerplexityGetLLMInvocationParams(unittest.TestCase): self.assertEqual(params["messages"], []) +class TestBaseLLMAdapterHelpers(unittest.TestCase): + """Tests for the shared helper methods on BaseLLMAdapter.""" + + def setUp(self): + # Use OpenAILLMAdapter as a concrete implementation for testing the base helpers + self.adapter = OpenAILLMAdapter() + + def test_extract_system_message(self): + """System message is extracted from messages[0].""" + messages = [ + {"role": "system", "content": "Be helpful."}, + {"role": "user", "content": "Hello"}, + ] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction=None + ) + + self.assertEqual(content, "Be helpful.") + self.assertEqual(role, "system") + self.assertEqual(len(messages), 1) # popped + + def test_extract_developer_without_system_instruction(self): + """Developer message is extracted when no system_instruction.""" + messages = [ + {"role": "developer", "content": "Context."}, + {"role": "user", "content": "Hello"}, + ] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction=None + ) + + self.assertEqual(content, "Context.") + self.assertEqual(role, "developer") + self.assertEqual(len(messages), 1) + + def test_developer_with_system_instruction_converts_to_user(self): + """Developer message with system_instruction is converted to user, not extracted.""" + messages = [ + {"role": "developer", "content": "Context."}, + {"role": "user", "content": "Hello"}, + ] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction="Be helpful." + ) + + self.assertIsNone(content) + self.assertIsNone(role) + self.assertEqual(len(messages), 2) # not popped + self.assertEqual(messages[0]["role"], "user") # converted to user + + def test_single_system_message_becomes_user(self): + """Single system message is converted to user instead of extracting (empty prevention).""" + messages = [ + {"role": "system", "content": "Be helpful."}, + ] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction=None + ) + + self.assertIsNone(content) + self.assertIsNone(role) + self.assertEqual(len(messages), 1) # not popped + self.assertEqual(messages[0]["role"], "user") + + def test_non_system_message_ignored(self): + """Non-system/developer first message is ignored.""" + messages = [ + {"role": "user", "content": "Hello"}, + ] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction=None + ) + + self.assertIsNone(content) + self.assertIsNone(role) + self.assertEqual(len(messages), 1) + + def test_empty_messages(self): + """Empty messages list returns None.""" + messages = [] + content, role = self.adapter._extract_initial_system_or_developer( + messages, system_instruction=None + ) + + self.assertIsNone(content) + self.assertIsNone(role) + + def test_resolve_both_system_discard(self): + """Resolve with discard=True: system_instruction wins, warns.""" + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + result = self.adapter._resolve_system_instruction( + "from context", "system", "from settings", discard_context_system=True + ) + mock_logger.warning.assert_called_once() + + self.assertEqual(result, "from settings") + + def test_resolve_both_system_keep(self): + """Resolve with discard=False: warns but returns system_instruction.""" + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + result = self.adapter._resolve_system_instruction( + "from context", "system", "from settings", discard_context_system=False + ) + mock_logger.warning.assert_called_once() + + self.assertEqual(result, "from settings") + + def test_resolve_only_system_instruction(self): + """Only system_instruction: returns it, no warning.""" + with patch("pipecat.adapters.base_llm_adapter.logger") as mock_logger: + result = self.adapter._resolve_system_instruction( + None, None, "from settings", discard_context_system=True + ) + mock_logger.warning.assert_not_called() + + self.assertEqual(result, "from settings") + + def test_resolve_only_context_system_discard(self): + """Only context system (discard=True): returns it.""" + result = self.adapter._resolve_system_instruction( + "from context", "system", None, discard_context_system=True + ) + + self.assertEqual(result, "from context") + + def test_resolve_only_context_system_keep(self): + """Only context system (discard=False): returns None (already in messages).""" + result = self.adapter._resolve_system_instruction( + "from context", "system", None, discard_context_system=False + ) + + self.assertIsNone(result) + + if __name__ == "__main__": unittest.main()