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

Move turn completion instructions to system_instruction

Browse Source 
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.
This commit is contained in:
Mark Backman
2026-03-08 10:41:40 -04:00
parent 764c3c4f32
commit efda57de5c
4 changed files with 154 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/pipecat/processors/aggregators/llm_response_universal.py
View File

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

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

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

11
tests/test_context_aggregators_universal.py
View File

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

117
tests/test_user_turn_completion_mixin.py
View File

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