Add system_instruction parameter to run_inference (#3968)

* Add system_instruction parameter to run_inference

Allow callers to provide a custom system instruction directly when calling
run_inference, without having to construct provider-specific context objects.

For OpenAI, the instruction is prepended as a system message (preserving
existing messages). For Anthropic, Google, and AWS Bedrock, it overrides the
single system field with a warning when an existing system instruction is
present in the context.

* Use system_instruction parameter in _generate_summary

Pass the summarization prompt via run_inference's system_instruction
parameter instead of embedding it as a system message in the context.

* Add changelog for #3968

changelog/3968.added.md

@@ -0,0 +1 @@
+- Added `system_instruction` parameter to `run_inference` across all LLM services, allowing callers to override the system prompt for one-shot inference calls. Used by `_generate_summary` to pass the summarization prompt cleanly.

src/pipecat/pipeline/llm_switcher.py

@@ -52,17 +52,19 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
         """
         return self.strategy.active_service
 
-    async def run_inference(self, context: LLMContext) -> Optional[str]:
+    async def run_inference(self, context: LLMContext, **kwargs) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.
 
         Args:
             context: The LLM context containing conversation history.
+            **kwargs: Additional arguments forwarded to the active LLM's run_inference
+                (e.g. max_tokens, system_instruction).
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
         """
         if self.active_llm:
-            return await self.active_llm.run_inference(context=context)
+            return await self.active_llm.run_inference(context=context, **kwargs)
         return None
 
     def register_function(

src/pipecat/services/anthropic/llm.py

@@ -346,7 +346,10 @@ class AnthropicLLMService(LLMService):
             return response
 
     async def run_inference(
-        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+        self,
+        context: LLMContext | OpenAILLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
     ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
@@ -354,6 +357,8 @@ class AnthropicLLMService(LLMService):
             context: The LLM context containing conversation history.
             max_tokens: Optional maximum number of tokens to generate. If provided,
                 overrides the service's default max_tokens setting.
+            system_instruction: Optional system instruction to use for this inference.
+                If provided, overrides any system instruction in the context.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -375,6 +380,15 @@ class AnthropicLLMService(LLMService):
             system = getattr(context, "system", NOT_GIVEN)
             tools = context.tools or []
 
+        # Override system instruction if provided
+        if system_instruction is not None:
+            if system and system is not NOT_GIVEN:
+                logger.warning(
+                    f"{self}: Both system_instruction and a system message in context are set."
+                    " Using system_instruction."
+                )
+            system = system_instruction
+
         # Build params using the same method as streaming completions
         params = {
             "model": self._settings.model,

src/pipecat/services/aws/llm.py

@@ -923,7 +923,10 @@ class AWSBedrockLLMService(LLMService):
         return inference_config
 
     async def run_inference(
-        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+        self,
+        context: LLMContext | OpenAILLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
     ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
@@ -931,6 +934,8 @@ class AWSBedrockLLMService(LLMService):
             context: The LLM context containing conversation history.
             max_tokens: Optional maximum number of tokens to generate. If provided,
                 overrides the service's default max_tokens setting.
+            system_instruction: Optional system instruction to use for this inference.
+                If provided, overrides any system instruction in the context.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -947,6 +952,15 @@ class AWSBedrockLLMService(LLMService):
             messages = context.messages
             system = getattr(context, "system", None)  # [{"text": "system message"}]
 
+        # Override system instruction if provided
+        if system_instruction is not None:
+            if system:
+                logger.warning(
+                    f"{self}: Both system_instruction and a system message in context are set."
+                    " Using system_instruction."
+                )
+            system = [{"text": system_instruction}]
+
         # Prepare request parameters using the same method as streaming
         inference_config = self._build_inference_config()
 

src/pipecat/services/google/llm.py

@@ -881,7 +881,10 @@ class GoogleLLMService(LLMService):
         self._client = genai.Client(api_key=self._api_key, http_options=self._http_options)
 
     async def run_inference(
-        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+        self,
+        context: LLMContext | OpenAILLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
     ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
@@ -889,6 +892,8 @@ class GoogleLLMService(LLMService):
             context: The LLM context containing conversation history.
             max_tokens: Optional maximum number of tokens to generate. If provided,
                 overrides the service's default max_tokens setting.
+            system_instruction: Optional system instruction to use for this inference.
+                If provided, overrides any system instruction in the context.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -908,6 +913,15 @@ class GoogleLLMService(LLMService):
             system = getattr(context, "system_message", None)
             tools = context.tools or []
 
+        # Override system instruction if provided
+        if system_instruction is not None:
+            if system:
+                logger.warning(
+                    f"{self}: Both system_instruction and a system message in context are set."
+                    " Using system_instruction."
+                )
+            system = system_instruction
+
         # Build generation config using the same method as streaming
         generation_params = self._build_generation_params(
             system_instruction=system, tools=tools if tools else None

src/pipecat/services/llm_service.py

@@ -244,7 +244,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
         return self.get_llm_adapter().create_llm_specific_message(message)
 
     async def run_inference(
-        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+        self,
+        context: LLMContext | OpenAILLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
     ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
@@ -254,6 +257,8 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             context: The LLM context containing conversation history.
             max_tokens: Optional maximum number of tokens to generate. If provided,
                 overrides the service's default max_tokens/max_completion_tokens setting.
+            system_instruction: Optional system instruction to use for this inference.
+                If provided, overrides any system instruction in the context.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -535,23 +540,17 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
 
         # Create summary context
         transcript = LLMContextSummarizationUtil.format_messages_for_summary(result.messages)
-        prompt_messages = [
-            {
-                "role": "system",
-                "content": frame.summarization_prompt,
-            },
-            {
-                "role": "user",
-                "content": f"Conversation history:\n{transcript}",
-            },
-        ]
-        summary_context = LLMContext(messages=prompt_messages)
+        summary_context = LLMContext(
+            messages=[{"role": "user", "content": f"Conversation history:\n{transcript}"}]
+        )
 
         # Generate summary using run_inference
         # This will be overridden by each LLM service implementation
         try:
             summary_text = await self.run_inference(
-                summary_context, max_tokens=frame.target_context_tokens
+                summary_context,
+                max_tokens=frame.target_context_tokens,
+                system_instruction=frame.summarization_prompt,
             )
         except NotImplementedError:
             raise RuntimeError(

src/pipecat/services/openai/base_llm.py

@@ -342,7 +342,10 @@ class BaseOpenAILLMService(LLMService):
         return params
 
     async def run_inference(
-        self, context: LLMContext | OpenAILLMContext, max_tokens: Optional[int] = None
+        self,
+        context: LLMContext | OpenAILLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
     ) -> Optional[str]:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context.
 
@@ -350,6 +353,8 @@ class BaseOpenAILLMService(LLMService):
             context: The LLM context containing conversation history.
             max_tokens: Optional maximum number of tokens to generate. If provided,
                 overrides the service's default max_tokens/max_completion_tokens setting.
+            system_instruction: Optional system instruction to use for this inference.
+                If provided, overrides any system instruction in the context.
 
         Returns:
             The LLM's response as a string, or None if no response is generated.
@@ -371,6 +376,16 @@ class BaseOpenAILLMService(LLMService):
         params["stream"] = False
         params.pop("stream_options", None)
 
+        # Prepend system instruction if provided
+        if system_instruction is not None:
+            messages = params.get("messages", [])
+            if messages and messages[0].get("role") == "system":
+                logger.warning(
+                    f"{self}: Both system_instruction and a system message in context are set."
+                    " Using system_instruction."
+                )
+            params["messages"] = [{"role": "system", "content": system_instruction}] + messages
+
         # Override max_tokens if provided
         if max_tokens is not None:
             # Use max_completion_tokens for newer models, fallback to max_tokens

tests/test_run_inference.py

@@ -511,5 +511,257 @@ async def test_aws_bedrock_run_inference_client_exception():
             await service.run_inference(mock_context)
 
 
-if __name__ == "__main__":
-    unittest.main()
+# --- system_instruction parameter tests ---
+
+
+@pytest.mark.asyncio
+async def test_openai_run_inference_system_instruction_overrides_context():
+    """Test that system_instruction overrides the system message from context."""
+    with patch.object(OpenAILLMService, "create_client"):
+        service = OpenAILLMService(model="gpt-4")
+        service._client = AsyncMock()
+
+        mock_context = MagicMock(spec=LLMContext)
+        mock_adapter = MagicMock()
+        test_messages = [
+            {"role": "system", "content": "Original system message"},
+            {"role": "user", "content": "Hello"},
+        ]
+        mock_adapter.get_llm_invocation_params.return_value = OpenAILLMInvocationParams(
+            messages=test_messages, tools=OPENAI_NOT_GIVEN, tool_choice=OPENAI_NOT_GIVEN
+        )
+        service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = "Response"
+        service._client.chat.completions.create.return_value = mock_response
+
+        result = await service.run_inference(
+            mock_context, system_instruction="New system instruction"
+        )
+
+        assert result == "Response"
+        call_kwargs = service._client.chat.completions.create.call_args.kwargs
+        messages = call_kwargs["messages"]
+        # system_instruction should be prepended as the first message
+        assert messages[0] == {"role": "system", "content": "New system instruction"}
+        # Original system message should still be present
+        assert messages[1] == {"role": "system", "content": "Original system message"}
+        # User message should still be present
+        assert messages[2] == {"role": "user", "content": "Hello"}
+        assert len(messages) == 3
+
+
+@pytest.mark.asyncio
+async def test_openai_run_inference_system_instruction_none_unchanged():
+    """Test that when system_instruction is None, behavior is unchanged."""
+    with patch.object(OpenAILLMService, "create_client"):
+        service = OpenAILLMService(model="gpt-4")
+        service._client = AsyncMock()
+
+        mock_context = MagicMock(spec=LLMContext)
+        mock_adapter = MagicMock()
+        test_messages = [
+            {"role": "system", "content": "Original system message"},
+            {"role": "user", "content": "Hello"},
+        ]
+        mock_adapter.get_llm_invocation_params.return_value = OpenAILLMInvocationParams(
+            messages=test_messages, tools=OPENAI_NOT_GIVEN, tool_choice=OPENAI_NOT_GIVEN
+        )
+        service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = "Response"
+        service._client.chat.completions.create.return_value = mock_response
+
+        result = await service.run_inference(mock_context)
+
+        assert result == "Response"
+        call_kwargs = service._client.chat.completions.create.call_args.kwargs
+        messages = call_kwargs["messages"]
+        assert messages[0] == {"role": "system", "content": "Original system message"}
+        assert messages[1] == {"role": "user", "content": "Hello"}
+
+
+@pytest.mark.asyncio
+async def test_anthropic_run_inference_system_instruction_overrides_context():
+    """Test that system_instruction overrides the system message for Anthropic."""
+    service = AnthropicLLMService(api_key="test-key", model="claude-3-sonnet-20240229")
+    service._client = AsyncMock()
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": "Hello"}]
+    mock_adapter.get_llm_invocation_params.return_value = AnthropicLLMInvocationParams(
+        messages=test_messages, system="Original system", tools=[]
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_response = MagicMock()
+    mock_response.content = [MagicMock()]
+    mock_response.content[0].text = "Response"
+    service._client.beta.messages.create.return_value = mock_response
+
+    result = await service.run_inference(mock_context, system_instruction="New system instruction")
+
+    assert result == "Response"
+    call_kwargs = service._client.beta.messages.create.call_args.kwargs
+    assert call_kwargs["system"] == "New system instruction"
+    assert call_kwargs["messages"] == test_messages
+
+
+@pytest.mark.asyncio
+async def test_anthropic_run_inference_system_instruction_none_unchanged():
+    """Test that when system_instruction is None, Anthropic behavior is unchanged."""
+    service = AnthropicLLMService(api_key="test-key", model="claude-3-sonnet-20240229")
+    service._client = AsyncMock()
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": "Hello"}]
+    mock_adapter.get_llm_invocation_params.return_value = AnthropicLLMInvocationParams(
+        messages=test_messages, system="Original system", tools=[]
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_response = MagicMock()
+    mock_response.content = [MagicMock()]
+    mock_response.content[0].text = "Response"
+    service._client.beta.messages.create.return_value = mock_response
+
+    result = await service.run_inference(mock_context)
+
+    assert result == "Response"
+    call_kwargs = service._client.beta.messages.create.call_args.kwargs
+    assert call_kwargs["system"] == "Original system"
+
+
+@pytest.mark.asyncio
+async def test_google_run_inference_system_instruction_overrides_context():
+    """Test that system_instruction overrides the system message for Google."""
+    service = GoogleLLMService(api_key="test-key", model="gemini-2.0-flash")
+    service._client = AsyncMock()
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": "Hello"}]
+    mock_adapter.get_llm_invocation_params.return_value = GeminiLLMInvocationParams(
+        messages=test_messages, system_instruction="Original system", tools=NotGiven()
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_response = MagicMock()
+    mock_response.candidates = [MagicMock()]
+    mock_response.candidates[0].content = MagicMock()
+    mock_response.candidates[0].content.parts = [MagicMock()]
+    mock_response.candidates[0].content.parts[0].text = "Response"
+    service._client.aio = AsyncMock()
+    service._client.aio.models = AsyncMock()
+    service._client.aio.models.generate_content = AsyncMock(return_value=mock_response)
+
+    result = await service.run_inference(mock_context, system_instruction="New system instruction")
+
+    assert result == "Response"
+    call_kwargs = service._client.aio.models.generate_content.call_args.kwargs
+    config = call_kwargs["config"]
+    assert config.system_instruction == "New system instruction"
+
+
+@pytest.mark.asyncio
+async def test_google_run_inference_system_instruction_none_unchanged():
+    """Test that when system_instruction is None, Google behavior is unchanged."""
+    service = GoogleLLMService(api_key="test-key", model="gemini-2.0-flash")
+    service._client = AsyncMock()
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": "Hello"}]
+    mock_adapter.get_llm_invocation_params.return_value = GeminiLLMInvocationParams(
+        messages=test_messages, system_instruction="Original system", tools=NotGiven()
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_response = MagicMock()
+    mock_response.candidates = [MagicMock()]
+    mock_response.candidates[0].content = MagicMock()
+    mock_response.candidates[0].content.parts = [MagicMock()]
+    mock_response.candidates[0].content.parts[0].text = "Response"
+    service._client.aio = AsyncMock()
+    service._client.aio.models = AsyncMock()
+    service._client.aio.models.generate_content = AsyncMock(return_value=mock_response)
+
+    result = await service.run_inference(mock_context)
+
+    assert result == "Response"
+    call_kwargs = service._client.aio.models.generate_content.call_args.kwargs
+    config = call_kwargs["config"]
+    assert config.system_instruction == "Original system"
+
+
+@pytest.mark.asyncio
+async def test_aws_bedrock_run_inference_system_instruction_overrides_context():
+    """Test that system_instruction overrides the system message for AWS Bedrock."""
+    service = AWSBedrockLLMService(model="anthropic.claude-3-sonnet-20240229-v1:0")
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": [{"text": "Hello"}]}]
+    mock_adapter.get_llm_invocation_params.return_value = AWSBedrockLLMInvocationParams(
+        messages=test_messages,
+        system=[{"text": "Original system"}],
+        tools=[],
+        tool_choice=None,
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_client = AsyncMock()
+    mock_response = {"output": {"message": {"content": [{"text": "Response"}]}}}
+    mock_client.converse.return_value = mock_response
+
+    mock_context_manager = AsyncMock()
+    mock_context_manager.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_context_manager.__aexit__ = AsyncMock(return_value=None)
+
+    with patch.object(service._aws_session, "client", return_value=mock_context_manager):
+        result = await service.run_inference(
+            mock_context, system_instruction="New system instruction"
+        )
+
+        assert result == "Response"
+        call_kwargs = mock_client.converse.call_args.kwargs
+        assert call_kwargs["system"] == [{"text": "New system instruction"}]
+        assert call_kwargs["messages"] == test_messages
+
+
+@pytest.mark.asyncio
+async def test_aws_bedrock_run_inference_system_instruction_none_unchanged():
+    """Test that when system_instruction is None, AWS Bedrock behavior is unchanged."""
+    service = AWSBedrockLLMService(model="anthropic.claude-3-sonnet-20240229-v1:0")
+
+    mock_context = MagicMock(spec=LLMContext)
+    mock_adapter = MagicMock()
+    test_messages = [{"role": "user", "content": [{"text": "Hello"}]}]
+    mock_adapter.get_llm_invocation_params.return_value = AWSBedrockLLMInvocationParams(
+        messages=test_messages,
+        system=[{"text": "Original system"}],
+        tools=[],
+        tool_choice=None,
+    )
+    service.get_llm_adapter = MagicMock(return_value=mock_adapter)
+
+    mock_client = AsyncMock()
+    mock_response = {"output": {"message": {"content": [{"text": "Response"}]}}}
+    mock_client.converse.return_value = mock_response
+
+    mock_context_manager = AsyncMock()
+    mock_context_manager.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_context_manager.__aexit__ = AsyncMock(return_value=None)
+
+    with patch.object(service._aws_session, "client", return_value=mock_context_manager):
+        result = await service.run_inference(mock_context)
+
+        assert result == "Response"
+        call_kwargs = mock_client.converse.call_args.kwargs
+        assert call_kwargs["system"] == [{"text": "Original system"}]