diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 41979cb5f2..a7ea6e1676 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/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] diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index be3c82e424..c8461fc7c6 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/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 diff --git a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py b/python/packages/openai/tests/openai/test_openai_chat_completion_client.py index 9c1c103987..fb5e7c471e 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py +++ b/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()