From ee3d1128ec5a8e67aa8bc864ec02ff99cff6803f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 21 May 2026 16:51:13 -0400 Subject: [PATCH] Add LLMService.append_system_instruction() Composes durable text onto a user-provided system instruction (alongside the turn-completion and async-tool-cancellation addons) so it is prepended on every inference and survives context-message resets. The user's base prompt is now snapshotted once and the effective instruction is always rebuilt from it, replacing the prior lazy capture/restore logic with a single invariant. --- changelog/xxxx.added.2.md | 1 + src/pipecat/services/llm_service.py | 73 ++++++++++++++--------- tests/test_llm_service.py | 76 +++++++++++++++++++++++- tests/test_user_turn_completion_mixin.py | 5 +- 4 files changed, 126 insertions(+), 29 deletions(-) create mode 100644 changelog/xxxx.added.2.md diff --git a/changelog/xxxx.added.2.md b/changelog/xxxx.added.2.md new file mode 100644 index 000000000..af44f9bf8 --- /dev/null +++ b/changelog/xxxx.added.2.md @@ -0,0 +1 @@ +- Added `LLMService.append_system_instruction(...)`, which composes durable text onto a user-provided system instruction (alongside the turn-completion and async-tool-cancellation instructions) so it is prepended on every inference and survives context-message resets. diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index bea25878e..219f52f36 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -300,7 +300,13 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] self._enable_async_tool_cancellation: bool = enable_async_tool_cancellation self._filter_incomplete_user_turns: bool = False self._async_tool_cancellation_enabled: bool = False - self._base_system_instruction: str | None = None + # The user's base system instruction, without composed addons. Captured + # here and refreshed when the user changes ``system_instruction`` at + # runtime; ``_compose_system_instruction`` always rebuilds the effective + # instruction from this plus any appended / addon instructions. + base_si = self._settings.system_instruction + self._base_system_instruction: str | None = base_si if isinstance(base_si, str) else None + self._appended_system_instructions: list[str] = [] # `adapter_class` is typed as `type[BaseLLMAdapter]` so subclasses # don't need to spell out the generic parameter just to subclass # (backward compatibility for 3rd-party providers outside this repo). @@ -392,15 +398,37 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] await self._cancel_sequential_runner_task() await self._cancel_summary_task() - def _compose_system_instruction(self): - """Compose system_instruction from the base and all active addon instructions. + def append_system_instruction(self, instruction: str) -> None: + """Append durable text to the system instruction, preserving the user's prompt. - Combines the base system instruction with turn completion instructions - (when enabled) and async tool cancellation instructions (when enabled), - writing the result to ``self._settings.system_instruction``. + The text is composed onto the end of the system instruction (joined + with a blank line) and re-applied on every inference, so it survives + context-message resets (e.g. ``LLMMessagesUpdateFrame(messages=[])``). + Intended for framework components that own an LLM and need to add + standard guidance to a user-provided prompt — for example, ``UIWorker`` + appends the UI wire-format guide. Appended instructions compose after + the user's base prompt and alongside the turn-completion and + async-tool-cancellation instructions. + + Args: + instruction: The instruction text to append. + """ + self._appended_system_instructions.append(instruction) + self._compose_system_instruction() + + def _compose_system_instruction(self): + """Rebuild ``system_instruction`` from the base prompt and all active addons. + + Joins the user's base system instruction (the single source of truth, + captured at construction and refreshed on runtime ``system_instruction`` + updates) with any appended instructions (e.g. the ``UIWorker`` prompt + guide), turn completion instructions (when enabled), and async tool + cancellation instructions (when enabled). Safe to call repeatedly — it + always rebuilds from the base, so it never compounds. """ base = self._base_system_instruction parts = [base] if base else [] + parts.extend(self._appended_system_instructions) if self._filter_incomplete_user_turns: parts.append(self._user_turn_completion_config.completion_instructions) if self._async_tool_cancellation_enabled: @@ -428,29 +456,24 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] 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 "system_instruction" in changed: + # The user replaced the base prompt; re-snapshot it so composition + # rebuilds the effective instruction from the new value. + base_si = self._settings.system_instruction + self._base_system_instruction = base_si if isinstance(base_si, str) else None if "user_turn_completion_config" in changed and self._filter_incomplete_user_turns: self.set_user_turn_completion_config( assert_given(self._settings.user_turn_completion_config) ) - self._compose_system_instruction() - if ( - "system_instruction" in changed - and (self._filter_incomplete_user_turns or self._async_tool_cancellation_enabled) - and "filter_incomplete_user_turns" not in changed - ): - # system_instruction changed while composition is active. - # Treat the new value as the new base and recompose. - self._base_system_instruction = self._settings.system_instruction + # Any of these fields changes the composed instruction; rebuild it. + if changed.keys() & { + "filter_incomplete_user_turns", + "system_instruction", + "user_turn_completion_config", + }: self._compose_system_instruction() return changed @@ -1060,10 +1083,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter] logger.debug(f"{self}: Enabling async tool cancellation") self._async_tool_cancellation_enabled = True - - if self._base_system_instruction is None: - self._base_system_instruction = self._settings.system_instruction - self._compose_system_instruction() self._adapter.builtin_tools[CANCEL_ASYNC_TOOL_NAME] = CANCEL_ASYNC_TOOL_SCHEMA diff --git a/tests/test_llm_service.py b/tests/test_llm_service.py index f084b0753..20fe2212d 100644 --- a/tests/test_llm_service.py +++ b/tests/test_llm_service.py @@ -34,7 +34,7 @@ class MockLLMService(LLMService): def __init__(self, **kwargs): settings = LLMSettings( model="test-model", - system_instruction=None, + system_instruction=kwargs.pop("system_instruction", None), temperature=None, max_tokens=None, top_p=None, @@ -317,3 +317,77 @@ class TestLLMService(unittest.IsolatedAsyncioTestCase): muted = await strategy.process_frame(frame) self.assertFalse(muted) + + +class TestAppendSystemInstruction(unittest.IsolatedAsyncioTestCase): + """Coverage for `LLMService.append_system_instruction`.""" + + def _service(self, system_instruction: str | None = None) -> MockLLMService: + # Construct with the prompt so the base snapshot happens the real way + # (in __init__), rather than poking _base_system_instruction directly. + return MockLLMService(system_instruction=system_instruction) + + def test_append_preserves_existing_prompt(self): + service = self._service("APP") + service.append_system_instruction("GUIDE") + self.assertEqual(service._settings.system_instruction, "APP\n\nGUIDE") + + def test_append_with_no_base_uses_text_alone(self): + service = self._service(None) + service.append_system_instruction("GUIDE") + self.assertEqual(service._settings.system_instruction, "GUIDE") + + def test_multiple_appends_join_in_order(self): + service = self._service("APP") + service.append_system_instruction("G1") + service.append_system_instruction("G2") + self.assertEqual(service._settings.system_instruction, "APP\n\nG1\n\nG2") + + async def test_appended_guide_survives_turn_completion_toggle(self): + service = self._service("APP") + service.append_system_instruction("GUIDE") + + # Enabling turn completion composes after the appended guide, once. + await service._update_settings(LLMSettings(filter_incomplete_user_turns=True)) + composed = service._settings.system_instruction + self.assertTrue(composed.startswith("APP\n\nGUIDE\n\n")) + self.assertEqual(composed.count("GUIDE"), 1) + + # Disabling restores base + guide (without the turn instructions). + await service._update_settings(LLMSettings(filter_incomplete_user_turns=False)) + self.assertEqual(service._settings.system_instruction, "APP\n\nGUIDE") + + async def test_runtime_system_instruction_update_preserves_appended(self): + service = self._service("APP") + service.append_system_instruction("GUIDE") + + # A runtime system_instruction change replaces the base but keeps the + # appended guide composed onto the end. + await service._update_settings(LLMSettings(system_instruction="NEW")) + self.assertEqual(service._settings.system_instruction, "NEW\n\nGUIDE") + + async def test_base_set_after_append_composes(self): + # No base at construction; the guide is appended first, then the user + # sets a system_instruction at runtime. The guide is retained. + service = self._service(None) + service.append_system_instruction("GUIDE") + self.assertEqual(service._settings.system_instruction, "GUIDE") + + await service._update_settings(LLMSettings(system_instruction="APP")) + self.assertEqual(service._settings.system_instruction, "APP\n\nGUIDE") + + async def test_appended_guide_survives_async_tool_cancellation_toggle(self): + service = self._service("APP") + service.append_system_instruction("GUIDE") + + # Enabling async tool cancellation composes after the appended guide, + # without duplicating it. + service._setup_async_tool_cancellation() + composed = service._settings.system_instruction + self.assertTrue(composed.startswith("APP\n\nGUIDE\n\n")) + self.assertEqual(composed.count("GUIDE"), 1) + self.assertNotEqual(composed, "APP\n\nGUIDE") # async instructions appended + + # Disabling recomposes back to base + guide. + service._teardown_async_tool_cancellation() + self.assertEqual(service._settings.system_instruction, "APP\n\nGUIDE") diff --git a/tests/test_user_turn_completion_mixin.py b/tests/test_user_turn_completion_mixin.py index 5a4f0c640..bb6a8e5db 100644 --- a/tests/test_user_turn_completion_mixin.py +++ b/tests/test_user_turn_completion_mixin.py @@ -224,7 +224,10 @@ class TestSystemInstructionComposition(unittest.IsolatedAsyncioTestCase): # 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) + # The base prompt is retained — it's the single source of truth that + # composition rebuilds from; disabling just recomposes without the + # turn-completion addon. + self.assertEqual(service._base_system_instruction, "You are a helpful assistant.") async def test_disable_turn_completion_restores_none(self): """Disabling turn completion when original was None should restore None."""