From efda57de5c041e0912cd649869049f094d37d8b6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Mar 2026 10:41:40 -0400 Subject: [PATCH 1/2] Move turn completion instructions to system_instruction Turn completion instructions were being injected as a system message in the LLM context, which caused warning spam when system_instruction was also set, did not persist across full context updates, and broke LLMs that do not support consecutive system messages. Instead, compose the turn completion instructions into the LLM service system_instruction field. This is managed via _base_system_instruction which stores the original value for restoration when turn completion is disabled. --- .../aggregators/llm_response_universal.py | 6 - src/pipecat/services/llm_service.py | 33 +++++ tests/test_context_aggregators_universal.py | 11 +- tests/test_user_turn_completion_mixin.py | 117 ++++++++++++++++++ 4 files changed, 154 insertions(+), 13 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index cf6c81e5f..db8ed4a3d 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -565,9 +565,6 @@ class LLMUserAggregator(LLMContextAggregator): ) ) - # Auto-inject turn completion instructions into context - self._context.add_message({"role": "system", "content": config.completion_instructions}) - async def _stop(self, frame: EndFrame): await self._maybe_emit_user_turn_stopped(on_session_end=True) await self._cleanup() @@ -636,9 +633,6 @@ class LLMUserAggregator(LLMContextAggregator): async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame): self.set_messages(frame.messages) - if self._params.filter_incomplete_user_turns: - config = self._params.user_turn_completion_config or UserTurnCompletionConfig() - self._context.add_message({"role": "system", "content": config.completion_instructions}) if frame.run_llm: await self.push_context_frame() diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 909b555a2..44858038d 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -212,6 +212,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): self._run_in_parallel = run_in_parallel self._function_call_timeout_secs = function_call_timeout_secs self._filter_incomplete_user_turns: bool = False + self._base_system_instruction: Optional[str] = None self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} @@ -326,6 +327,19 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await self._cancel_sequential_runner_task() await self._cancel_summary_task() + def _compose_system_instruction(self): + """Compose system_instruction by appending turn completion instructions. + + Combines the base system instruction with turn completion instructions + and writes the result to ``self._settings.system_instruction``. + """ + base = self._base_system_instruction + completion_instructions = self._user_turn_completion_config.completion_instructions + if base: + self._settings.system_instruction = f"{base}\n\n{completion_instructions}" + else: + self._settings.system_instruction = completion_instructions + async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]: """Apply a settings delta, handling turn-completion fields. @@ -345,9 +359,28 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): f"{self}: Incomplete turn filtering " f"{'enabled' if self._filter_incomplete_user_turns else 'disabled'}" ) + if self._filter_incomplete_user_turns: + # Save the current system_instruction before composing + self._base_system_instruction = self._settings.system_instruction + self._compose_system_instruction() + else: + # Restore original system_instruction + self._settings.system_instruction = self._base_system_instruction + self._base_system_instruction = None if "user_turn_completion_config" in changed and self._filter_incomplete_user_turns: self.set_user_turn_completion_config(self._settings.user_turn_completion_config) + self._compose_system_instruction() + + if ( + "system_instruction" in changed + and self._filter_incomplete_user_turns + and "filter_incomplete_user_turns" not in changed + ): + # system_instruction changed while turn completion is active. + # Treat the new value as the new base and recompose. + self._base_system_instruction = self._settings.system_instruction + self._compose_system_instruction() return changed diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index b22abf6c6..7d6f73f2d 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -50,7 +50,6 @@ from pipecat.turns.user_mute import ( MuteUntilFirstBotCompleteUserMuteStrategy, ) from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy -from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig from pipecat.turns.user_turn_strategies import UserTurnStrategies USER_TURN_STOP_TIMEOUT = 0.2 @@ -156,7 +155,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): ) assert context.messages[0]["content"] == "Hi there!" - async def test_llm_messages_update_reinjects_turn_completion_instructions(self): + async def test_llm_messages_update_does_not_inject_turn_completion_into_context(self): context = LLMContext() params = LLMUserAggregatorParams(filter_incomplete_user_turns=True) pipeline = Pipeline([LLMUserAggregator(context, params=params)]) @@ -170,13 +169,11 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): pipeline, frames_to_send=frames_to_send, ) - config = UserTurnCompletionConfig() - # The context should contain the new messages plus the re-injected instructions - assert len(context.messages) == 3 + # Turn completion instructions are now set via system_instruction on the + # LLM service, not injected into context messages. + assert len(context.messages) == 2 assert context.messages[0]["content"] == "You are a helpful assistant." assert context.messages[1]["content"] == "Hello!" - assert context.messages[2]["role"] == "system" - assert context.messages[2]["content"] == config.completion_instructions async def test_default_user_turn_strategies(self): context = LLMContext() diff --git a/tests/test_user_turn_completion_mixin.py b/tests/test_user_turn_completion_mixin.py index 273d8cc90..e503ecbf2 100644 --- a/tests/test_user_turn_completion_mixin.py +++ b/tests/test_user_turn_completion_mixin.py @@ -10,10 +10,14 @@ from unittest.mock import AsyncMock from pipecat.frames.frames import LLMFullResponseEndFrame, LLMTextFrame from pipecat.processors.frame_processor import FrameProcessor +from pipecat.services.llm_service import LLMService +from pipecat.services.settings import LLMSettings from pipecat.turns.user_turn_completion_mixin import ( USER_TURN_COMPLETE_MARKER, + USER_TURN_COMPLETION_INSTRUCTIONS, USER_TURN_INCOMPLETE_LONG_MARKER, USER_TURN_INCOMPLETE_SHORT_MARKER, + UserTurnCompletionConfig, UserTurnCompletionLLMServiceMixin, ) @@ -140,5 +144,118 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase self.assertFalse(processor._turn_suppressed) +class MockLLMService(LLMService): + """Minimal LLM service for testing system_instruction composition.""" + + def __init__(self, **kwargs): + settings = LLMSettings( + model="test-model", + system_instruction=kwargs.pop("system_instruction", None), + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=None, + user_turn_completion_config=None, + ) + super().__init__(settings=settings, **kwargs) + + +class TestSystemInstructionComposition(unittest.IsolatedAsyncioTestCase): + """Tests for turn completion system_instruction composition in LLMService.""" + + async def test_enable_turn_completion_sets_system_instruction(self): + """Enabling turn completion should set system_instruction to completion instructions.""" + service = MockLLMService() + self.assertIsNone(service._settings.system_instruction) + + delta = LLMSettings(filter_incomplete_user_turns=True) + await service._update_settings(delta) + + self.assertEqual(service._settings.system_instruction, USER_TURN_COMPLETION_INSTRUCTIONS) + self.assertIsNone(service._base_system_instruction) + + async def test_enable_turn_completion_appends_to_existing_system_instruction(self): + """Enabling turn completion should append instructions to existing system_instruction.""" + service = MockLLMService(system_instruction="You are a helpful assistant.") + + delta = LLMSettings(filter_incomplete_user_turns=True) + await service._update_settings(delta) + + expected = f"You are a helpful assistant.\n\n{USER_TURN_COMPLETION_INSTRUCTIONS}" + self.assertEqual(service._settings.system_instruction, expected) + self.assertEqual(service._base_system_instruction, "You are a helpful assistant.") + + async def test_disable_turn_completion_restores_system_instruction(self): + """Disabling turn completion should restore the original system_instruction.""" + service = MockLLMService(system_instruction="You are a helpful assistant.") + + # Enable + await service._update_settings(LLMSettings(filter_incomplete_user_turns=True)) + self.assertIn(USER_TURN_COMPLETION_INSTRUCTIONS, service._settings.system_instruction) + + # Disable + await service._update_settings(LLMSettings(filter_incomplete_user_turns=False)) + self.assertEqual(service._settings.system_instruction, "You are a helpful assistant.") + self.assertIsNone(service._base_system_instruction) + + async def test_disable_turn_completion_restores_none(self): + """Disabling turn completion when original was None should restore None.""" + service = MockLLMService() + + await service._update_settings(LLMSettings(filter_incomplete_user_turns=True)) + self.assertEqual(service._settings.system_instruction, USER_TURN_COMPLETION_INSTRUCTIONS) + + await service._update_settings(LLMSettings(filter_incomplete_user_turns=False)) + self.assertIsNone(service._settings.system_instruction) + + async def test_update_system_instruction_while_turn_completion_active(self): + """Changing system_instruction while turn completion is active should recompose.""" + service = MockLLMService(system_instruction="Original prompt.") + + await service._update_settings(LLMSettings(filter_incomplete_user_turns=True)) + expected = f"Original prompt.\n\n{USER_TURN_COMPLETION_INSTRUCTIONS}" + self.assertEqual(service._settings.system_instruction, expected) + + # Now update system_instruction + await service._update_settings(LLMSettings(system_instruction="New prompt.")) + expected = f"New prompt.\n\n{USER_TURN_COMPLETION_INSTRUCTIONS}" + self.assertEqual(service._settings.system_instruction, expected) + self.assertEqual(service._base_system_instruction, "New prompt.") + + async def test_update_config_recomposes_with_custom_instructions(self): + """Updating turn completion config should recompose with new instructions.""" + service = MockLLMService(system_instruction="Base prompt.") + + await service._update_settings(LLMSettings(filter_incomplete_user_turns=True)) + + custom_config = UserTurnCompletionConfig(instructions="Custom turn instructions.") + await service._update_settings(LLMSettings(user_turn_completion_config=custom_config)) + + expected = "Base prompt.\n\nCustom turn instructions." + self.assertEqual(service._settings.system_instruction, expected) + + async def test_simultaneous_enable_and_system_instruction_change(self): + """Enabling turn completion and changing system_instruction in the same delta + should use the new system_instruction as the base.""" + service = MockLLMService(system_instruction="Original prompt.") + + await service._update_settings( + LLMSettings( + filter_incomplete_user_turns=True, + system_instruction="New prompt.", + ) + ) + + # apply_update sets system_instruction to "New prompt." before _update_settings + # runs, so the base should be the new value the user explicitly set. + self.assertEqual(service._base_system_instruction, "New prompt.") + expected = f"New prompt.\n\n{USER_TURN_COMPLETION_INSTRUCTIONS}" + self.assertEqual(service._settings.system_instruction, expected) + + if __name__ == "__main__": unittest.main() From 64155e8f0638721df82ebaf56582750d020bf071 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Mar 2026 10:44:45 -0400 Subject: [PATCH 2/2] Add changelog for #3957 --- changelog/3957.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/3957.fixed.md diff --git a/changelog/3957.fixed.md b/changelog/3957.fixed.md new file mode 100644 index 000000000..a501aecb1 --- /dev/null +++ b/changelog/3957.fixed.md @@ -0,0 +1 @@ +- Fixed turn completion instructions being injected as a context system message instead of using `system_instruction`. This caused warning spam when `system_instruction` was also set and didn't persist across full context updates.