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. diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index ca25650b4..e1e356007 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -568,9 +568,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() @@ -639,9 +636,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()