wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add LLMService.append_system_instruction()

Browse Source 
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.
This commit is contained in:
Mark Backman
2026-05-21 16:51:13 -04:00
parent e8ec7c585f
commit ee3d1128ec
4 changed files with 126 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/xxxx.added.2.md Normal file
View File

@@ -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.

73
src/pipecat/services/llm_service.py
View File

@@ -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

76
tests/test_llm_service.py
View File

@@ -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")

5
tests/test_user_turn_completion_mixin.py
View File

@@ -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."""