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

Merge pull request #3957 from pipecat-ai/mb/user-turn-completion-system-instruction

Browse Source 
Move turn completion instructions to system_instruction
This commit is contained in:
Mark Backman
2026-03-09 09:17:06 -04:00
committed by GitHub
parent 8a02e6fbc5 64155e8f06
commit 9eaa99c8e2
5 changed files with 155 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3957.fixed.md Normal file
View File

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

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

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

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