Python: Fix duplicate system message from instructions (#5049) (#5051)

Add deduplication to `prepend_instructions_to_messages()` to skip
instructions that are already present as leading messages with the
same role and text. This prevents duplicate system messages when
instructions are injected by multiple layers (e.g. Agent + chat client).

Fixes #5049

python/packages/core/agent_framework/_types.py

@@ -1796,7 +1796,19 @@ def prepend_instructions_to_messages(
     if isinstance(instructions, str):
         instructions = [instructions]
 
-    instruction_messages = [Message(role, [instr]) for instr in instructions]
+    # Skip instructions that are already present as leading messages with the
+    # same role and text.  This prevents duplicate system messages when
+    # instructions are injected by multiple layers (e.g. Agent + chat client).
+    deduplicated: list[str] = []
+    for idx, instr in enumerate(instructions):
+        if idx < len(messages) and messages[idx].role == role and messages[idx].text == instr:
+            continue
+        deduplicated.append(instr)
+
+    if not deduplicated:
+        return messages
+
+    instruction_messages = [Message(role, [instr]) for instr in deduplicated]
     return [*instruction_messages, *messages]
 
 

python/packages/core/tests/core/test_types.py

@@ -4068,3 +4068,87 @@ def test_oauth_consent_request_serialization_roundtrip():
 
 
 # endregion
+
+
+# region prepend_instructions_to_messages tests
+
+
+def test_prepend_instructions_basic():
+    """Test that instructions are prepended as system message."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [Message("user", ["Hello"])]
+    result = prepend_instructions_to_messages(messages, "You are helpful.")
+    assert len(result) == 2
+    assert result[0].role == "system"
+    assert result[0].text == "You are helpful."
+    assert result[1].role == "user"
+
+
+def test_prepend_instructions_none():
+    """Test that None instructions returns messages unchanged."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [Message("user", ["Hello"])]
+    result = prepend_instructions_to_messages(messages, None)
+    assert result is messages
+
+
+def test_prepend_instructions_skips_duplicate():
+    """Test that duplicate system instructions are not prepended again."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [
+        Message("system", ["You are helpful."]),
+        Message("user", ["Hello"]),
+    ]
+    result = prepend_instructions_to_messages(messages, "You are helpful.")
+    assert len(result) == 2
+    assert result[0].role == "system"
+    assert result[0].text == "You are helpful."
+    assert result[1].role == "user"
+
+
+def test_prepend_instructions_skips_duplicate_list():
+    """Test deduplication with a list of instructions."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [
+        Message("system", ["First instruction"]),
+        Message("system", ["Second instruction"]),
+        Message("user", ["Hello"]),
+    ]
+    result = prepend_instructions_to_messages(messages, ["First instruction", "Second instruction"])
+    assert len(result) == 3
+    assert result[0].text == "First instruction"
+    assert result[1].text == "Second instruction"
+    assert result[2].text == "Hello"
+
+
+def test_prepend_instructions_adds_when_different():
+    """Test that different instructions are still prepended."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [
+        Message("system", ["Old instruction"]),
+        Message("user", ["Hello"]),
+    ]
+    result = prepend_instructions_to_messages(messages, "New instruction")
+    assert len(result) == 3
+    assert result[0].role == "system"
+    assert result[0].text == "New instruction"
+    assert result[1].text == "Old instruction"
+    assert result[2].text == "Hello"
+
+
+def test_prepend_instructions_custom_role():
+    """Test prepending with a custom role."""
+    from agent_framework._types import prepend_instructions_to_messages
+
+    messages = [Message("user", ["Hello"])]
+    result = prepend_instructions_to_messages(messages, "Be concise.", role="developer")
+    assert len(result) == 2
+    assert result[0].role == "developer"
+
+
+# endregion

python/packages/openai/tests/openai/test_openai_chat_completion_client.py

@@ -1233,6 +1233,32 @@ def test_prepare_options_with_instructions(
     assert prepared_options["messages"][0]["content"] == "You are a helpful assistant."
 
 
+def test_prepare_options_with_instructions_no_duplicate(
+    openai_unit_test_env: dict[str, str],
+) -> None:
+    """Test that duplicate system message from instructions is not added again.
+
+    Regression test for https://github.com/microsoft/agent-framework/issues/5049
+    """
+    client = OpenAIChatCompletionClient()
+
+    # Simulate messages that already contain the system instruction
+    messages = [
+        Message(role="system", text="You are a helpful assistant."),
+        Message(role="user", text="Hello"),
+    ]
+    options = {"instructions": "You are a helpful assistant."}
+
+    prepared_options = client._prepare_options(messages, options)
+
+    # Should NOT duplicate the system message
+    assert "messages" in prepared_options
+    assert len(prepared_options["messages"]) == 2
+    assert prepared_options["messages"][0]["role"] == "system"
+    assert prepared_options["messages"][0]["content"] == "You are a helpful assistant."
+    assert prepared_options["messages"][1]["role"] == "user"
+
+
 def test_prepare_message_with_author_name(openai_unit_test_env: dict[str, str]) -> None:
     """Test that author_name is included in prepared message."""
     client = OpenAIChatCompletionClient()