feat: support models requiring strict alternating user-assistant roles

python/packages/autogen-agentchat/src/autogen_agentchat/agents/_assistant_agent.py

@@ -57,7 +57,7 @@
     ToolCallSummaryMessage,
 )
 from ..state import AssistantAgentState
-from ..utils import remove_images
+from ..utils import ensure_alternating_roles, remove_images
 from ._base_chat_agent import BaseChatAgent
 
 event_logger = logging.getLogger(EVENT_LOGGER_NAME)
@@ -1640,11 +1640,24 @@ async def load_state(self, state: Mapping[str, Any]) -> None:
 
     @staticmethod
     def _get_compatible_context(model_client: ChatCompletionClient, messages: List[LLMMessage]) -> Sequence[LLMMessage]:
-        """Ensure that the messages are compatible with the underlying client, by removing images if needed."""
-        if model_client.model_info["vision"]:
-            return messages
-        else:
-            return remove_images(messages)
+        """Ensure that the messages are compatible with the underlying client.
+
+        This method handles two compatibility concerns:
+        1. Removing images for non-vision models.
+        2. Ensuring strict alternating user-assistant roles for models that require it
+           (e.g., DeepSeek R1, Mistral). See https://github.com/microsoft/autogen/issues/5965
+        """
+        from autogen_core.models import ModelFamily
+
+        result: List[LLMMessage] = messages if model_client.model_info["vision"] else remove_images(messages)
+        # Enforce alternating roles if the model requires it (explicit flag or family-based).
+        needs_alternating = model_client.model_info.get("requires_alternating_roles", False)
+        if not needs_alternating:
+            family = model_client.model_info.get("family", "")
+            needs_alternating = ModelFamily.requires_alternating_roles(family)
+        if needs_alternating:
+            result = ensure_alternating_roles(result)
+        return result
 
     def _to_config(self) -> AssistantAgentConfig:
         """Convert the assistant agent to a declarative config."""

python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py

@@ -23,6 +23,7 @@
 
 from ... import TRACE_LOGGER_NAME
 from ...base import ChatAgent, Team, TerminationCondition
+from ...utils import ensure_alternating_roles
 from ...messages import (
     BaseAgentEvent,
     BaseChatMessage,
@@ -245,11 +246,21 @@ async def _select_speaker(self, roles: str, participants: List[str], max_attempt
             select_speaker_messages = [UserMessage(content=select_speaker_prompt, source="user")]
 
         num_attempts = 0
+        # Check if the model requires strict alternating roles (explicit flag or family-based).
+        _requires_alternating = self._model_client.model_info.get("requires_alternating_roles", False)
+        if not _requires_alternating:
+            _family = self._model_client.model_info.get("family", "")
+            _requires_alternating = ModelFamily.requires_alternating_roles(_family)
+
         while num_attempts < max_attempts:
             num_attempts += 1
+            # Apply alternating role enforcement if needed.
+            send_messages: List[SystemMessage | UserMessage | AssistantMessage] = select_speaker_messages
+            if _requires_alternating:
+                send_messages = ensure_alternating_roles(select_speaker_messages)  # type: ignore[assignment]
             if self._model_client_streaming:
                 chunk: CreateResult | str = ""
-                async for _chunk in self._model_client.create_stream(messages=select_speaker_messages):
+                async for _chunk in self._model_client.create_stream(messages=send_messages):
                     chunk = _chunk
                     if self._emit_team_events:
                         if isinstance(chunk, str):
@@ -266,7 +277,7 @@ async def _select_speaker(self, roles: str, participants: List[str], max_attempt
                 assert isinstance(chunk, CreateResult)
                 response = chunk
             else:
-                response = await self._model_client.create(messages=select_speaker_messages)
+                response = await self._model_client.create(messages=send_messages)
             assert isinstance(response.content, str)
             select_speaker_messages.append(AssistantMessage(content=response.content, source="selector"))
             # NOTE: we use all participant names to check for mentions, even if the previous speaker is not allowed.

python/packages/autogen-agentchat/src/autogen_agentchat/utils/__init__.py

@@ -2,6 +2,6 @@
 This module implements various utilities common to AgentChat agents and teams.
 """
 
-from ._utils import content_to_str, remove_images
+from ._utils import content_to_str, ensure_alternating_roles, remove_images
 
-__all__ = ["content_to_str", "remove_images"]
+__all__ = ["content_to_str", "ensure_alternating_roles", "remove_images"]

python/packages/autogen-agentchat/src/autogen_agentchat/utils/_utils.py

@@ -1,7 +1,14 @@
-from typing import List, Union
+from typing import List, Sequence, Union
 
 from autogen_core import FunctionCall, Image
-from autogen_core.models import FunctionExecutionResult, LLMMessage, UserMessage
+from autogen_core.models import (
+    AssistantMessage,
+    FunctionExecutionResult,
+    FunctionExecutionResultMessage,
+    LLMMessage,
+    SystemMessage,
+    UserMessage,
+)
 from pydantic import BaseModel
 
 # Type aliases for convenience
@@ -42,3 +49,146 @@ def remove_images(messages: List[LLMMessage]) -> List[LLMMessage]:
         else:
             str_messages.append(message)
     return str_messages
+
+
+def _get_role(message: LLMMessage) -> str:
+    """Get the effective role of a message for alternation checking."""
+    if isinstance(message, SystemMessage):
+        return "system"
+    elif isinstance(message, UserMessage):
+        return "user"
+    elif isinstance(message, AssistantMessage):
+        return "assistant"
+    elif isinstance(message, FunctionExecutionResultMessage):
+        # Function results are treated as user-role messages by most APIs
+        return "user"
+    return "unknown"
+
+
+def _merge_user_contents(contents: List[str]) -> str:
+    """Merge multiple user message contents into a single string."""
+    return "\n".join(c for c in contents if c)
+
+
+def _merge_assistant_contents(contents: List[str]) -> str:
+    """Merge multiple assistant message contents into a single string."""
+    return "\n".join(c for c in contents if c)
+
+
+def ensure_alternating_roles(messages: Sequence[LLMMessage]) -> List[LLMMessage]:
+    """Ensure messages follow strict alternating user-assistant roles.
+
+    This is required by certain model APIs (e.g., DeepSeek R1, Mistral) that reject
+    requests with consecutive messages of the same role.
+
+    The strategy is:
+    1. Preserve leading system messages as-is.
+    2. For consecutive user-role messages (UserMessage or FunctionExecutionResultMessage),
+       merge their text content into a single UserMessage.
+    3. For consecutive assistant messages, merge their text content into a single
+       AssistantMessage.
+    4. If FunctionExecutionResultMessage contains tool results (non-text), insert an
+       empty assistant message before it to maintain alternation when needed.
+    5. Between two assistant messages, inject an empty user message.
+    6. Between two user messages, inject an empty assistant message.
+
+    Args:
+        messages: The original message sequence.
+
+    Returns:
+        A new list of messages with strict alternating user-assistant roles.
+    """
+    if not messages:
+        return []
+
+    result: List[LLMMessage] = []
+
+    # Preserve leading system messages
+    idx = 0
+    while idx < len(messages) and isinstance(messages[idx], SystemMessage):
+        result.append(messages[idx])
+        idx += 1
+
+    if idx >= len(messages):
+        return result
+
+    # Process remaining messages, enforcing alternation
+    for i in range(idx, len(messages)):
+        msg = messages[i]
+        role = _get_role(msg)
+
+        if role == "system":
+            # Convert non-leading system messages to user messages
+            result.append(UserMessage(content=msg.content, source="system"))  # type: ignore
+            continue
+
+        if not result or _get_role(result[-1]) == "system":
+            # First non-system message, just add it
+            result.append(msg)
+            continue
+
+        prev_role = _get_role(result[-1])
+
+        if role == prev_role:
+            # Same role as previous -- merge content
+            if role == "user":
+                prev_msg = result[-1]
+                if isinstance(prev_msg, UserMessage) and isinstance(msg, UserMessage):
+                    prev_content = content_to_str(prev_msg.content)
+                    curr_content = content_to_str(msg.content)
+                    merged = _merge_user_contents([prev_content, curr_content])
+                    result[-1] = UserMessage(content=merged, source=prev_msg.source)
+                elif isinstance(prev_msg, UserMessage) and isinstance(msg, FunctionExecutionResultMessage):
+                    # Merge function result text into the previous user message
+                    prev_content = content_to_str(prev_msg.content)
+                    func_content = "\n".join(r.content for r in msg.content)
+                    merged = _merge_user_contents([prev_content, func_content])
+                    result[-1] = UserMessage(content=merged, source=prev_msg.source)
+                elif isinstance(prev_msg, FunctionExecutionResultMessage) and isinstance(msg, UserMessage):
+                    # Convert previous to UserMessage and merge
+                    prev_content = "\n".join(r.content for r in prev_msg.content)
+                    curr_content = content_to_str(msg.content)
+                    merged = _merge_user_contents([prev_content, curr_content])
+                    result[-1] = UserMessage(content=merged, source=msg.source)
+                elif isinstance(prev_msg, FunctionExecutionResultMessage) and isinstance(
+                    msg, FunctionExecutionResultMessage
+                ):
+                    # Merge two function result messages
+                    result[-1] = FunctionExecutionResultMessage(
+                        content=list(prev_msg.content) + list(msg.content)
+                    )
+                else:
+                    # Fallback: inject an assistant message to break the sequence
+                    result.append(AssistantMessage(content="", source="system"))
+                    result.append(msg)
+            elif role == "assistant":
+                prev_msg = result[-1]
+                if isinstance(prev_msg, AssistantMessage) and isinstance(msg, AssistantMessage):
+                    prev_content = content_to_str(prev_msg.content)
+                    curr_content = content_to_str(msg.content)
+                    merged = _merge_assistant_contents([prev_content, curr_content])
+                    # Merge thoughts if present
+                    thoughts = [t for t in [prev_msg.thought, msg.thought] if t]
+                    merged_thought = "\n".join(thoughts) if thoughts else None
+                    result[-1] = AssistantMessage(
+                        content=merged, source=prev_msg.source, thought=merged_thought
+                    )
+                else:
+                    # Fallback: inject a user message to break the sequence
+                    result.append(UserMessage(content="", source="system"))
+                    result.append(msg)
+            else:
+                # Unknown same-role situation, inject opposite role
+                result.append(UserMessage(content="", source="system"))
+                result.append(msg)
+        elif role == "user" and prev_role == "assistant":
+            # Correct alternation
+            result.append(msg)
+        elif role == "assistant" and prev_role == "user":
+            # Correct alternation
+            result.append(msg)
+        else:
+            # Any other transition, just append
+            result.append(msg)
+
+    return result