From 912f1be31c78f63e3b1777ec12f7e9b3b809dbb2 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 10 Mar 2026 12:57:23 -0400 Subject: [PATCH] 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 | 1 + src/pipecat/pipeline/llm_switcher.py | 6 +- src/pipecat/services/anthropic/llm.py | 16 +- src/pipecat/services/aws/llm.py | 16 +- src/pipecat/services/google/llm.py | 16 +- src/pipecat/services/llm_service.py | 25 ++- src/pipecat/services/openai/base_llm.py | 17 +- tests/test_run_inference.py | 256 +++++++++++++++++++++++- 8 files changed, 332 insertions(+), 21 deletions(-) create mode 100644 changelog/3968.added.md diff --git a/changelog/3968.added.md b/changelog/3968.added.md new file mode 100644 index 000000000..cd9be5996 --- /dev/null +++ b/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. diff --git a/src/pipecat/pipeline/llm_switcher.py b/src/pipecat/pipeline/llm_switcher.py index f9f53c066..281bedc65 100644 --- a/src/pipecat/pipeline/llm_switcher.py +++ b/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( diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 7abcec3e0..6bf7bf180 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/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, diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 434011f1f..c77321ead 100644 --- a/src/pipecat/services/aws/llm.py +++ b/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() diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 7f4f8caae..85f8c0ce3 100644 --- a/src/pipecat/services/google/llm.py +++ b/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 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 44858038d..7944f413a 100644 --- a/src/pipecat/services/llm_service.py +++ b/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( diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 5466c75a5..c62147c3d 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/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 diff --git a/tests/test_run_inference.py b/tests/test_run_inference.py index eb18a5fda..4f4021b5d 100644 --- a/tests/test_run_inference.py +++ b/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"}]