MAINT: Standardize system prompts on prepended_conversation

doc/code/executor/3_attack_configuration.py

@@ -21,7 +21,7 @@
 # |---|---|
 # | `objective` | What you are trying to get the **objective target** (the system under test) to do. Drives scoring and multi-turn adversarial prompts. |
 # | `memory_labels` | A `dict[str, str]` tagged onto every prompt/response, so you can filter this run later in memory. |
-# | `prepended_conversation` | A list of `Message`s to seed the conversation before the attack's own turns (system prompt, prior history). |
+# | `prepended_conversation` | A list of `Message`s to seed the conversation before the attack's own turns. This is also where the objective target's **system prompt** goes — `Message.from_system_prompt(...)` builds one (see below). |
 # | `next_message` | The exact next message to send, instead of letting the attack derive it from the objective. Useful for multimodal or pre-built seeds. |
 #
 # Construction-time configuration objects — **adversarial**, **scoring**, and **converter** — are
@@ -36,6 +36,7 @@
     PromptSendingAttack,
     SingleTurnAttackContext,
 )
+from pyrit.models import Message
 from pyrit.output import output_attack_async
 from pyrit.prompt_target import TextTarget
 from pyrit.setup import IN_MEMORY, initialize_pyrit_async
@@ -59,15 +60,42 @@
 )
 await output_attack_async(result)
 
+# %% [markdown]
+# ## Setting a system prompt
+#
+# The objective target's system prompt is just a `system`-role message at the front of the
+# conversation, so you set it through `prepended_conversation`. `Message.from_system_prompt(...)`
+# builds that message:
+#
+# ```python
+# prepended_conversation=[Message.from_system_prompt("...")]
+# ```
+#
+# Because `prepended_conversation` is a list, targets that accept more than one system message just
+# take more than one entry. `Message.from_system_prompts(...)` is a shorthand that builds the list for
+# you — `Message.from_system_prompts("Policy.", "Persona.")` is the same as
+# `[Message.from_system_prompt("Policy."), Message.from_system_prompt("Persona.")]` — and you can
+# interleave `user` / `assistant` turns too (next section).
+
+# %%
+result = await attack.execute_async(  # type: ignore
+    objective="Explain how a saponification reaction works",
+    prepended_conversation=[
+        Message.from_system_prompt("You are a helpful chemistry tutor who explains concepts step by step.")
+    ],
+)
+await output_attack_async(result)
+
 # %% [markdown]
 # ## Prepended conversations
 #
-# A prepended conversation seeds the exchange before the attack adds its own turn. The most common
-# use is setting a system prompt, but you can prepend any sequence of `system` / `user` / `assistant`
-# turns — for example, to resume a prior conversation or to plant an agreeable assistant reply.
+# A system prompt is the simplest prepended conversation. The general form seeds a full
+# `system` / `user` / `assistant` history before the attack adds its own turn — for example, to
+# resume a prior conversation or to plant an agreeable assistant reply. It is just a list of
+# `Message`s, so the system prompt and any seed turns compose freely.
 
 # %%
-from pyrit.models import Message, MessagePiece
+from pyrit.models import MessagePiece
 
 prepended_conversation = [
     Message.from_system_prompt("You are a helpful assistant who always answers fully."),

pyrit/executor/attack/component/conversation_manager.py

@@ -13,7 +13,7 @@
     PrependedConversationConfig,
 )
 from pyrit.memory import CentralMemory
-from pyrit.message_normalizer import ConversationContextNormalizer
+from pyrit.message_normalizer import ConversationContextNormalizer, GenericSystemSquashNormalizer
 from pyrit.models import (
     ChatMessageRole,
     ComponentIdentifier,
@@ -359,23 +359,37 @@ async def _handle_non_chat_target_async(
         if config is None:
             config = PrependedConversationConfig()
 
-        # Normalize conversation to string
+        messages_to_normalize = prepended_conversation
+        if config.message_normalizer is None:
+            messages_to_normalize = await GenericSystemSquashNormalizer().normalize_async(prepended_conversation)
+
         normalizer = config.get_message_normalizer()
-        normalized_context = await normalizer.normalize_string_async(prepended_conversation)
+        normalized_context = await normalizer.normalize_string_async(messages_to_normalize)
+
+        next_message = context.next_message
+        if next_message is None:
+            next_message = Message.from_prompt(prompt=context.objective, role="user")
+            context.next_message = next_message
 
-        # Prepend to next_message if it exists, otherwise create new message
-        if context.next_message is not None:
+        if normalized_context:
             # Find an existing text piece to prepend to
             text_piece = None
-            for piece in context.next_message.message_pieces:
+            for piece in next_message.message_pieces:
                 if piece.original_value_data_type == "text":
                     text_piece = piece
                     break
 
             if text_piece:
                 # Prepend context to the existing text piece
-                text_piece.original_value = f"{normalized_context}\n\n{text_piece.original_value}"
-                text_piece.converted_value = f"{normalized_context}\n\n{text_piece.converted_value}"
+                context_prefix = f"{normalized_context}\n\n"
+                if text_piece.original_value != normalized_context and not text_piece.original_value.startswith(
+                    context_prefix
+                ):
+                    text_piece.original_value = f"{context_prefix}{text_piece.original_value}"
+                if text_piece.converted_value != normalized_context and not text_piece.converted_value.startswith(
+                    context_prefix
+                ):
+                    text_piece.converted_value = f"{context_prefix}{text_piece.converted_value}"
             else:
                 # No text piece found (multimodal message), add a new text piece at the beginning
                 context_piece = MessagePiece(
@@ -387,12 +401,7 @@ async def _handle_non_chat_target_async(
                     converted_value_data_type="text",
                 )
                 # Create a new message with the context piece prepended
-                context.next_message = Message(
-                    message_pieces=[context_piece] + list(context.next_message.message_pieces)
-                )
-        else:
-            # Create new message with just the context
-            context.next_message = Message.from_prompt(prompt=normalized_context, role="user")
+                context.next_message = Message(message_pieces=[context_piece] + list(next_message.message_pieces))
 
         logger.debug(f"Normalized prepended conversation for non-chat target: {len(normalized_context)} characters")
         return ConversationState()

pyrit/executor/attack/single_turn/single_turn_attack_strategy.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 
+from pyrit.common.deprecation import print_deprecation_message
 from pyrit.common.logger import logger
 from pyrit.executor.attack.core.attack_parameters import AttackParameters, AttackParamsT
 from pyrit.executor.attack.core.attack_strategy import AttackContext, AttackStrategy
@@ -31,12 +32,22 @@ class SingleTurnAttackContext(AttackContext[AttackParamsT]):
     # Unique identifier of the main conversation between the attacker and model
     conversation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
 
-    # System prompt for chat-based targets
+    # Deprecated, non-functional no-op; removed in 0.17.0. Set the objective
+    # target's system prompt via ``prepended_conversation`` instead.
     system_prompt: str | None = None
 
     # Arbitrary metadata that downstream attacks or scorers may attach
     metadata: dict[str, str | int] | None = None
 
+    def __post_init__(self) -> None:
+        """Warn that ``system_prompt`` is deprecated and non-functional when it is set."""
+        if self.system_prompt is not None:
+            print_deprecation_message(
+                old_item="SingleTurnAttackContext.system_prompt",
+                new_item="prepended_conversation=[Message.from_system_prompt(...)]",
+                removed_in="0.17.0",
+            )
+
 
 class SingleTurnAttackStrategy(AttackStrategy[SingleTurnAttackContext[Any], AttackResult], ABC):
     """

pyrit/message_normalizer/generic_system_squash.py

@@ -8,12 +8,12 @@
 
 class GenericSystemSquashNormalizer(MessageListNormalizer[Message]):
     """
-    Normalizer that combines the first system message with the first user message using generic instruction tags.
+    Normalizer that combines system messages with the first user message using generic instruction tags.
     """
 
     async def normalize_async(self, messages: list[Message]) -> list[Message]:
         """
-        Return messages with the first system message combined into the first user message.
+        Return messages with all system messages combined into the first user message.
 
         The format uses generic instruction tags:
         ### Instructions ###
@@ -25,43 +25,45 @@ async def normalize_async(self, messages: list[Message]) -> list[Message]:
             messages: The list of messages to normalize.
 
         Returns:
-            A Message with the system message squashed into the first user message.
+            Messages with system instructions squashed into the first user message.
 
         Raises:
             ValueError: If the messages list is empty.
         """
         if not messages:
             raise ValueError("Messages list cannot be empty")
 
-        # Check if first message is a system message
-        first_piece = messages[0].get_piece()
-        if first_piece.api_role != "system":
-            # No system message to squash, return messages unchanged
+        system_messages = [message for message in messages if message.api_role == "system"]
+        if not system_messages:
             return list(messages)
 
-        if len(messages) == 1:
-            # Only system message, convert to user message.
+        system_content = "\n\n".join(
+            piece.converted_value for message in system_messages for piece in message.message_pieces
+        )
+        non_system_messages = [message for message in messages if message.api_role != "system"]
+
+        if not non_system_messages:
             return [
                 build_squashed_user_message(
-                    new_message_content=first_piece.converted_value, source_messages=messages[:1]
+                    new_message_content=system_content,
+                    source_messages=system_messages,
                 )
             ]
 
         user_message_index = next(
-            (i for i, message in enumerate(messages[1:], start=1) if message.api_role == "user"),
+            (i for i, message in enumerate(non_system_messages) if message.api_role == "user"),
             -1,
         )
         if user_message_index == -1:
-            # Preserve the instruction content without rewriting non-user messages.
             return [
                 build_squashed_user_message(
-                    new_message_content=first_piece.converted_value, source_messages=messages[:1]
+                    new_message_content=system_content,
+                    source_messages=system_messages,
                 )
-            ] + list(messages[1:])
+            ] + non_system_messages
 
         # Combine system with the first user message, preserving non-text pieces (e.g. images) and their order.
-        system_content = first_piece.converted_value
-        user_message = messages[user_message_index]
+        user_message = non_system_messages[user_message_index]
         # Propagate prompt_metadata from the user message's first piece so downstream normalizers
         # (e.g. JsonSchemaNormalizer) still see request-level metadata after squashing.
         propagated_metadata = dict(user_message.message_pieces[0].prompt_metadata)
@@ -98,5 +100,8 @@ async def normalize_async(self, messages: list[Message]) -> list[Message]:
 
         squashed_message = Message(message_pieces=squashed_pieces)
 
-        # Remove system (index 0), replace the first user message with the squashed version, preserve all others
-        return list(messages[1:user_message_index]) + [squashed_message] + list(messages[user_message_index + 1 :])
+        return (
+            non_system_messages[:user_message_index]
+            + [squashed_message]
+            + non_system_messages[user_message_index + 1 :]
+        )

pyrit/models/messages/message.py

@@ -337,6 +337,20 @@ def from_system_prompt(cls, system_prompt: str) -> Message:
         """
         return cls.from_prompt(prompt=system_prompt, role="system")
 
+    @classmethod
+    def from_system_prompts(cls, *system_prompts: str) -> list[Message]:
+        """
+        Build a list of system-role messages, ready to pass as ``prepended_conversation``.
+
+        Args:
+            *system_prompts (str): One or more system instruction texts.
+
+        Returns:
+            list[Message]: One system-role message per input, in order.
+
+        """
+        return [cls.from_system_prompt(system_prompt) for system_prompt in system_prompts]
+
     def duplicate(self) -> Message:
         """
         Create a deep copy of this message with new IDs and timestamp for all message pieces.

tests/unit/executor/attack/component/test_conversation_manager.py

@@ -762,6 +762,95 @@ async def test_normalizes_for_non_chat_target_when_configured(
         assert "Next message" in text_value
         assert "Hello" in text_value or "doing well" in text_value
 
+    async def test_system_prompt_for_non_chat_target_preserves_instruction_and_objective(
+        self,
+        attack_identifier: ComponentIdentifier,
+        mock_prompt_target: MagicMock,
+    ) -> None:
+        manager = ConversationManager()
+        context = _TestAttackContext(params=AttackParameters(objective="Explain saponification"))
+        context.prepended_conversation = [Message.from_system_prompt("You are a chemistry tutor")]
+
+        await manager.initialize_context_async(
+            context=context,
+            target=mock_prompt_target,
+            conversation_id=str(uuid.uuid4()),
+        )
+
+        assert context.next_message is not None
+        assert context.next_message.get_value() == "Turn 1:\nuser: You are a chemistry tutor\n\nExplain saponification"
+
+        await manager.initialize_context_async(
+            context=context,
+            target=mock_prompt_target,
+            conversation_id=str(uuid.uuid4()),
+        )
+
+        assert context.next_message.get_value() == "Turn 1:\nuser: You are a chemistry tutor\n\nExplain saponification"
+
+    async def test_system_prompt_for_non_chat_target_preserves_supplied_next_message(
+        self,
+        attack_identifier: ComponentIdentifier,
+        mock_prompt_target: MagicMock,
+    ) -> None:
+        manager = ConversationManager()
+        context = _TestAttackContext(
+            params=AttackParameters(
+                objective="Unused objective",
+                next_message=Message.from_prompt(prompt="Caller-supplied question", role="user"),
+            )
+        )
+        context.prepended_conversation = [Message.from_system_prompt("Follow the policy")]
+
+        await manager.initialize_context_async(
+            context=context,
+            target=mock_prompt_target,
+            conversation_id=str(uuid.uuid4()),
+        )
+
+        assert context.next_message is not None
+        assert context.next_message.get_value() == "Turn 1:\nuser: Follow the policy\n\nCaller-supplied question"
+
+    async def test_system_prompt_for_non_chat_target_preserves_multimodal_next_message(
+        self,
+        attack_identifier: ComponentIdentifier,
+        mock_prompt_target: MagicMock,
+    ) -> None:
+        manager = ConversationManager()
+        image_piece = MessagePiece(
+            role="user",
+            original_value="diagram.png",
+            original_value_data_type="image_path",
+        )
+        context = _TestAttackContext(
+            params=AttackParameters(
+                objective="Unused objective",
+                next_message=Message(message_pieces=[image_piece]),
+            )
+        )
+        context.prepended_conversation = [Message.from_system_prompt("Describe images precisely")]
+
+        await manager.initialize_context_async(
+            context=context,
+            target=mock_prompt_target,
+            conversation_id=str(uuid.uuid4()),
+        )
+
+        assert context.next_message is not None
+        assert len(context.next_message.message_pieces) == 2
+        assert context.next_message.message_pieces[0].converted_value == "Turn 1:\nuser: Describe images precisely"
+        assert context.next_message.message_pieces[1].converted_value == "diagram.png"
+        assert context.next_message.message_pieces[1].original_value_data_type == "image_path"
+
+        await manager.initialize_context_async(
+            context=context,
+            target=mock_prompt_target,
+            conversation_id=str(uuid.uuid4()),
+        )
+
+        assert len(context.next_message.message_pieces) == 2
+        assert context.next_message.message_pieces[0].converted_value == "Turn 1:\nuser: Describe images precisely"
+
     async def test_returns_turn_count_for_multi_turn_attacks(
         self,
         attack_identifier: ComponentIdentifier,