Merge pull request #4130 from pipecat-ai/pk/realtime-services-init-v-context-system-instructions-cleanup

Prefer init-provided system instructions in realtime services

changelog/4130.changed.md

@@ -0,0 +1 @@
+- ⚠️ Realtime services (Gemini Live, OpenAI Realtime, Grok Realtime, Nova Sonic) now prefer `system_instruction` from service settings over an initial system message in the LLM context, matching the behavior of non-realtime services. Previously, context-provided system instructions took precedence. A warning is now logged when both are set.

examples/foundational/51-grok-realtime.py

@@ -172,23 +172,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     session_properties = SessionProperties(
         # Voice options: Ara, Rex, Sal, Eve, Leo
         voice="Ara",
-        # System instructions
-        instructions="""You are a helpful and friendly AI assistant powered by Grok.
-
-You have access to several tools:
-- Weather information
-- Current time
-- Restaurant recommendations
-- Web search (built-in)
-- X/Twitter search (built-in)
-
-Your voice and personality should be warm and engaging. Keep your responses
-concise and conversational since this is a voice interaction.
-
-If the user asks about current events or news, use web search.
-If they ask about what people are saying on social media, use X search.
-
-Always be helpful and proactive in offering assistance.""",
         # Grok-specific built-in tools can be added here:
         # tools=[
         #     WebSearchTool(),  # Enable web search
@@ -200,6 +183,22 @@ Always be helpful and proactive in offering assistance.""",
     llm = GrokRealtimeLLMService(
         api_key=os.getenv("GROK_API_KEY"),
         settings=GrokRealtimeLLMService.Settings(
+            system_instruction="""You are a helpful and friendly AI assistant powered by Grok.
+
+    You have access to several tools:
+    - Weather information
+    - Current time
+    - Restaurant recommendations
+    - Web search (built-in)
+    - X/Twitter search (built-in)
+
+    Your voice and personality should be warm and engaging. Keep your responses
+    concise and conversational since this is a voice interaction.
+
+    If the user asks about current events or news, use web search.
+    If they ask about what people are saying on social media, use X search.
+
+    Always be helpful and proactive in offering assistance.""",
             session_properties=session_properties,
         ),
     )

src/pipecat/adapters/services/aws_nova_sonic_adapter.py

@@ -72,20 +72,26 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
         """Get the identifier used in LLMSpecificMessage instances for AWS Nova Sonic."""
         return "aws-nova-sonic"
 
-    def get_llm_invocation_params(self, context: LLMContext) -> AWSNovaSonicLLMInvocationParams:
+    def get_llm_invocation_params(
+        self, context: LLMContext, *, system_instruction: Optional[str] = None
+    ) -> AWSNovaSonicLLMInvocationParams:
         """Get AWS Nova Sonic-specific LLM invocation parameters from a universal LLM context.
 
-        This is a placeholder until support for universal LLMContext machinery is added for AWS Nova Sonic.
-
         Args:
             context: The LLM context containing messages, tools, etc.
+            system_instruction: Optional system instruction from service settings.
 
         Returns:
             Dictionary of parameters for invoking AWS Nova Sonic's LLM API.
         """
         messages = self._from_universal_context_messages(self.get_messages(context))
+        effective_system = self._resolve_system_instruction(
+            messages.system_instruction,
+            system_instruction,
+            discard_context_system=True,
+        )
         return {
-            "system_instruction": messages.system_instruction,
+            "system_instruction": effective_system,
             "messages": messages.messages,
             # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
             "tools": self.from_standard_tools(context.tools) or [],

src/pipecat/adapters/services/grok_realtime_adapter.py

@@ -50,18 +50,26 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
         """Get the identifier used in LLMSpecificMessage instances for Grok Realtime."""
         return "grok-realtime"
 
-    def get_llm_invocation_params(self, context: LLMContext) -> GrokRealtimeLLMInvocationParams:
+    def get_llm_invocation_params(
+        self, context: LLMContext, *, system_instruction: Optional[str] = None
+    ) -> GrokRealtimeLLMInvocationParams:
         """Get Grok Realtime-specific LLM invocation parameters from a universal LLM context.
 
         Args:
             context: The LLM context containing messages, tools, etc.
+            system_instruction: Optional system instruction from service settings.
 
         Returns:
             Dictionary of parameters for invoking Grok's Voice Agent API.
         """
         messages = self._from_universal_context_messages(self.get_messages(context))
+        effective_system = self._resolve_system_instruction(
+            messages.system_instruction,
+            system_instruction,
+            discard_context_system=True,
+        )
         return {
-            "system_instruction": messages.system_instruction,
+            "system_instruction": effective_system,
             "messages": messages.messages,
             "tools": self.from_standard_tools(context.tools) or [],
         }

src/pipecat/adapters/services/open_ai_realtime_adapter.py

@@ -43,20 +43,26 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
         """Get the identifier used in LLMSpecificMessage instances for OpenAI Realtime."""
         return "openai-realtime"
 
-    def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
+    def get_llm_invocation_params(
+        self, context: LLMContext, *, system_instruction: Optional[str] = None
+    ) -> OpenAIRealtimeLLMInvocationParams:
         """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
 
-        This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
-
         Args:
             context: The LLM context containing messages, tools, etc.
+            system_instruction: Optional system instruction from service settings.
 
         Returns:
             Dictionary of parameters for invoking OpenAI Realtime's API.
         """
         messages = self._from_universal_context_messages(self.get_messages(context))
+        effective_system = self._resolve_system_instruction(
+            messages.system_instruction,
+            system_instruction,
+            discard_context_system=True,
+        )
         return {
-            "system_instruction": messages.system_instruction,
+            "system_instruction": effective_system,
             "messages": messages.messages,
             # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
             "tools": self.from_standard_tools(context.tools) or [],

src/pipecat/services/aws/nova_sonic/llm.py

@@ -629,7 +629,9 @@ class AWSNovaSonicLLMService(LLMService):
 
         # Read context
         adapter: AWSNovaSonicLLMAdapter = self.get_llm_adapter()
-        llm_connection_params = adapter.get_llm_invocation_params(self._context)
+        llm_connection_params = adapter.get_llm_invocation_params(
+            self._context, system_instruction=self._settings.system_instruction
+        )
 
         # Send prompt start event, specifying tools.
         # Tools from context take priority over self._tools.
@@ -642,12 +644,9 @@ class AWSNovaSonicLLMService(LLMService):
         await self._send_prompt_start_event(tools)
 
         # Send system instruction.
-        # Instruction from context takes priority over self._settings.system_instruction.
-        system_instruction = (
-            llm_connection_params["system_instruction"]
-            if llm_connection_params["system_instruction"]
-            else self._settings.system_instruction
-        )
+        # The adapter resolves conflicts between init-provided and
+        # context-provided system instructions (preferring init-provided).
+        system_instruction = llm_connection_params["system_instruction"]
         logger.debug(f"Using system instruction: {system_instruction}")
         if system_instruction:
             await self._send_text_event(text=system_instruction, role=Role.SYSTEM)

src/pipecat/services/google/gemini_live/llm.py

@@ -1080,28 +1080,26 @@ class GeminiLiveLLMService(LLMService):
             # We got our initial context
             self._context = context
 
-            # If context contains system instruction or tools, reconnect in
-            # order to apply them.
-            # (Context-provided system instruction and tools take precedence
-            # over the ones provided at initialization time. Note that we could
-            # do more sophisticated comparisons here, but for now this is
-            # sufficient: we'll assume folks won't mean to provide these
-            # settings both in the context and at initialization time. In a
-            # future change, we could/should implement the ability to swap
-            # these settings at any point).
+            # Reconnect if context changes the effective system instruction
+            # or tools compared to the initial connection (which used the
+            # init-provided values). Note that the determination of "effective"
+            # system instruction is delegated to the adapter, which still
+            # chooses the init-provided value if there is one.
             adapter: GeminiLLMAdapter = self.get_llm_adapter()
-            params = adapter.get_llm_invocation_params(self._context)
+            params = adapter.get_llm_invocation_params(
+                self._context, system_instruction=self._system_instruction_from_init
+            )
             system_instruction = params["system_instruction"]
             tools = params["tools"]
-            if system_instruction and self._system_instruction_from_init:
-                logger.warning(
-                    "System instruction provided both at init time and in context; using context-provided value."
-                )
+            system_instruction_changed = system_instruction != self._system_instruction_from_init
             if tools and self._tools_from_init:
                 logger.warning(
                     "Tools provided both at init time and in context; using context-provided value."
                 )
-            if system_instruction or tools:
+            # For tools we simply check presence rather than diffing against
+            # init-provided tools, assuming that if context provides tools
+            # they warrant a reconnect.
+            if system_instruction_changed or tools:
                 await self._reconnect()
 
             # Initialize our bookkeeping of already-completed tool calls in
@@ -1322,10 +1320,12 @@ class GeminiLiveLLMService(LLMService):
             system_instruction = None
             tools = None
             if self._context:
-                params = adapter.get_llm_invocation_params(self._context)
+                params = adapter.get_llm_invocation_params(
+                    self._context, system_instruction=self._system_instruction_from_init
+                )
                 system_instruction = params["system_instruction"]
                 tools = params["tools"]
-            if not system_instruction:
+            else:
                 system_instruction = self._system_instruction_from_init
             if not tools:
                 tools = adapter.from_standard_tools(self._tools_from_init)

src/pipecat/services/grok/realtime/llm.py

@@ -607,11 +607,15 @@ class GrokRealtimeLLMService(LLMService):
         adapter: GrokRealtimeLLMAdapter = self.get_llm_adapter()
 
         if self._context:
-            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+            llm_invocation_params = adapter.get_llm_invocation_params(
+                self._context, system_instruction=self._settings.system_instruction
+            )
 
             if llm_invocation_params["tools"]:
                 settings.tools = llm_invocation_params["tools"]
 
+            # The adapter resolves conflicts between init-provided and
+            # context-provided system instructions (preferring init-provided).
             if llm_invocation_params["system_instruction"]:
                 settings.instructions = llm_invocation_params["system_instruction"]
 

src/pipecat/services/openai/realtime/llm.py

@@ -687,14 +687,16 @@ class OpenAIRealtimeLLMService(LLMService):
         adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
 
         if self._context:
-            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+            llm_invocation_params = adapter.get_llm_invocation_params(
+                self._context, system_instruction=self._settings.system_instruction
+            )
 
             # tools given in the context override the tools in the session properties
             if llm_invocation_params["tools"]:
                 settings.tools = llm_invocation_params["tools"]
 
-            # instructions in the context come from an initial "system" message in the
-            # messages list, and override instructions in the session properties
+            # The adapter resolves conflicts between init-provided and
+            # context-provided system instructions (preferring init-provided).
             if llm_invocation_params["system_instruction"]:
                 settings.instructions = llm_invocation_params["system_instruction"]
 

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."""