Add Utils for Multi-turn Evaluation (#46325)

* Add multi-turn eval utils

* run black

* update utils to use constants

* move import

* rename trace to turn

* update messages validation and utils

* run black

Author: salma-elshafey

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/constants.py

@@ -61,6 +61,17 @@ class EvaluatorScoringPattern(Enum):
     SCALE_1_5 = "scale_1_5"  # 1-5 scale (quality evaluators)
 
 
+class EvaluationLevel(str, Enum):
+    """Supported evaluation levels for multi-turn evaluators.
+
+    - ``CONVERSATION``: Force conversation-level evaluation using the multi-turn path.
+    - ``TURN``: Force turn-level evaluation using the single-turn query/response path.
+    """
+
+    CONVERSATION = "conversation"
+    TURN = "turn"
+
+
 class Tasks:
     """Defines types of annotation tasks supported by RAI Service."""
 

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py

@@ -6,7 +6,7 @@
 import re
 import math
 import threading
-from typing import Any, List, Literal, Mapping, Optional, Type, TypeVar, Tuple, Union, cast, get_args, get_origin
+from typing import Any, Dict, List, Literal, Mapping, Optional, Type, TypeVar, Tuple, Union, cast, get_args, get_origin
 
 import nltk
 from azure.storage.blob import ContainerClient
@@ -962,3 +962,305 @@ def upload(path: str, container_client: ContainerClient, logger=None):
             category=ErrorCategory.UPLOAD_ERROR,
             blame=ErrorBlame.SYSTEM_ERROR,
         ) from e
+
+
+# region Multi-turn utilities
+
+
+def _merge_query_response_messages(query, response):
+    """Merge query and response message lists into a single conversation.
+
+    :param query: The query messages.
+    :type query: List[dict]
+    :param response: The response messages.
+    :type response: List[dict]
+    :return: The merged conversation messages.
+    :rtype: List[dict]
+    """
+    return [*query, *response]
+
+
+def _split_messages_at_latest_user(messages):
+    """Split messages into query/response slices at the latest user turn.
+
+    :param messages: The conversation messages.
+    :type messages: List[dict]
+    :return: A tuple of (query_messages, response_messages).
+    :rtype: Tuple[List[dict], List[dict]]
+    """
+    latest_user_index = max(i for i, message in enumerate(messages) if message["role"] == "user")
+    return messages[: latest_user_index + 1], messages[latest_user_index + 1 :]
+
+
+def _wrap_string_messages(query, response):
+    """Wrap string query/response into separate message lists.
+
+    :param query: The query string.
+    :type query: str
+    :param response: The response string.
+    :type response: str
+    :return: A tuple of (query_messages, response_messages).
+    :rtype: Tuple[List[dict], List[dict]]
+    """
+    return (
+        [{"role": "user", "content": [{"type": "text", "text": query}]}],
+        [{"role": "assistant", "content": [{"type": "text", "text": response}]}],
+    )
+
+
+def serialize_messages(messages):
+    """Serialize a list of chat messages into a labeled text transcript for multi-turn prompts.
+
+    **Input format:** List of message dicts, each with ``"role"`` (``user``, ``assistant``, ``tool``,
+    ``system``, ``developer``) and ``"content"`` (string or list of content-block dicts like
+    ``{"type": "text", "text": "..."}``). Tool messages may include ``tool_call_id`` and content
+    blocks of type ``tool_result``/``tool_call``.
+
+    **Output format:** Plain-text transcript with labeled turns::
+
+        User turn 1:
+          <user text>
+
+        Agent turn 1:
+          <assistant text>
+          [TOOL_CALL] func_name({"arg": "val"})
+          [TOOL_RESULT] <result>
+
+        User turn 2:
+          <user text>
+        ...
+
+    System/developer messages are included as a system preamble. Consecutive messages of the same
+    role are grouped into a single turn. Assistant string content is auto-normalized to content-block
+    format for consistent formatting.
+
+    :param messages: Chat messages with role and content.
+    :type messages: List[dict]
+    :return: Formatted text transcript.
+    :rtype: str
+    """
+    if not messages:
+        return ""
+
+    from azure.ai.evaluation._evaluators._common._validators._validation_constants import MessageRole
+
+    all_user_queries = []
+    all_agent_responses = []
+    cur_user_query = []
+    cur_agent_response = []
+    system_message = None
+
+    for msg in messages:
+        if not isinstance(msg, dict):
+            continue
+        role = msg.get("role")
+        if not role:
+            continue
+
+        # _get_agent_response expects content as list of dicts, not a plain string
+        normalized = msg
+        if role == MessageRole.ASSISTANT and isinstance(msg.get("content"), str):
+            normalized = {**msg, "content": [{"type": "text", "text": msg["content"]}]}
+
+        if role in (MessageRole.SYSTEM, MessageRole.DEVELOPER):
+            content = msg.get("content", "")
+            if isinstance(content, list):
+                system_message = "\n".join(_extract_text_from_content(content))
+            else:
+                system_message = content
+
+        elif role == MessageRole.USER and "content" in msg:
+            if cur_agent_response:
+                formatted = _get_agent_response(cur_agent_response, include_tool_messages=True)
+                all_agent_responses.append([formatted])
+                cur_agent_response = []
+            content = msg["content"]
+            if isinstance(content, str):
+                text_in_msg = [content]
+            else:
+                text_in_msg = _extract_text_from_content(content)
+            if text_in_msg:
+                cur_user_query.append(text_in_msg)
+
+        elif role in (MessageRole.ASSISTANT, MessageRole.TOOL):
+            if cur_user_query:
+                all_user_queries.append(cur_user_query)
+                cur_user_query = []
+            cur_agent_response.append(normalized)
+
+    # Flush any remaining buffered turn
+    if cur_user_query:
+        all_user_queries.append(cur_user_query)
+    if cur_agent_response:
+        formatted = _get_agent_response(cur_agent_response, include_tool_messages=True)
+        all_agent_responses.append([formatted])
+
+    conversation_history: Dict = {
+        "user_queries": all_user_queries,
+        "agent_responses": all_agent_responses[: len(all_user_queries) - 1] if len(all_user_queries) > 0 else [],
+    }
+    if system_message:
+        conversation_history["system_message"] = system_message
+
+    result = _pretty_format_conversation_history(conversation_history)
+
+    # Append any trailing agent turn (the final response after the last user query)
+    start = max(len(all_user_queries) - 1, 0)
+    for i, agent_response in enumerate(all_agent_responses[start:], start=start):
+        result += f"Agent turn {i + 1}:\n"
+        for msg_text in agent_response:
+            if isinstance(msg_text, list):
+                for submsg in msg_text:
+                    result += "  " + "\n  ".join(submsg.split("\n")) + "\n"
+            else:
+                result += "  " + "\n  ".join(msg_text.split("\n")) + "\n"
+        result += "\n"
+
+    return result.rstrip("\n")
+
+
+def _resolve_evaluation_level(evaluation_level, error_target):
+    """Validate and normalize the evaluation_level parameter.
+
+    :param evaluation_level: The evaluation level to resolve.
+    :type evaluation_level: Optional[Union[EvaluationLevel, str]]
+    :param error_target: The error target for exceptions.
+    :type error_target: ErrorTarget
+    :return: The resolved EvaluationLevel or None for auto-detect.
+    :rtype: Optional[EvaluationLevel]
+    """
+    from .constants import EvaluationLevel
+
+    valid = [level.value for level in EvaluationLevel]
+    if evaluation_level is None or evaluation_level == "":
+        return None
+    if isinstance(evaluation_level, EvaluationLevel):
+        return evaluation_level
+    if isinstance(evaluation_level, str):
+        try:
+            return EvaluationLevel(evaluation_level)
+        except ValueError:
+            raise EvaluationException(
+                message=(f"Invalid evaluation_level '{evaluation_level}'. " f"Must be one of: {valid}."),
+                blame=ErrorBlame.USER_ERROR,
+                category=ErrorCategory.INVALID_VALUE,
+                target=error_target,
+            )
+    raise EvaluationException(
+        message=(f"Invalid evaluation_level '{evaluation_level}'. " f"Must be one of: {valid}."),
+        blame=ErrorBlame.USER_ERROR,
+        category=ErrorCategory.INVALID_VALUE,
+        target=error_target,
+    )
+
+
+def _is_intermediate_response(response):
+    """Check if response is intermediate (last content item is function_call or mcp_approval_request).
+
+    An intermediate response is one where the assistant's last message ends with a
+    function_call or mcp_approval_request content type, meaning the conversation is
+    still in progress and not yet ready for evaluation.
+
+    :param response: The response messages.
+    :type response: List[dict]
+    :return: True if the response is intermediate, False otherwise.
+    :rtype: bool
+    """
+    if isinstance(response, list) and len(response) > 0:
+        last_msg = response[-1]
+        if isinstance(last_msg, dict) and last_msg.get("role") == "assistant":
+            content = last_msg.get("content", [])
+            if isinstance(content, list) and len(content) > 0:
+                last_content = content[-1]
+                if isinstance(last_content, dict) and last_content.get("type") in (
+                    "function_call",
+                    "mcp_approval_request",
+                ):
+                    return True
+    return False
+
+
+def _drop_mcp_approval_messages(messages):
+    """Remove MCP approval request/response messages from a conversation.
+
+    MCP approval messages are protocol-level messages that should not be included
+    in the evaluation input.
+
+    :param messages: The conversation messages.
+    :type messages: List[dict]
+    :return: The filtered messages without MCP approval request/response messages.
+    :rtype: List[dict]
+    """
+    if not isinstance(messages, list):
+        return messages
+    return [
+        msg
+        for msg in messages
+        if not (
+            isinstance(msg, dict)
+            and isinstance(msg.get("content"), list)
+            and (
+                (
+                    msg.get("role") == "assistant"
+                    and any(isinstance(c, dict) and c.get("type") == "mcp_approval_request" for c in msg["content"])
+                )
+                or (
+                    msg.get("role") == "tool"
+                    and any(isinstance(c, dict) and c.get("type") == "mcp_approval_response" for c in msg["content"])
+                )
+            )
+        )
+    ]
+
+
+def _normalize_function_call_types(messages):
+    """Normalize function_call/function_call_output/openapi_call/openapi_call_output types to tool_call/tool_result.
+
+    This ensures a consistent content type vocabulary for downstream evaluators
+    regardless of how the original messages were authored.
+
+    :param messages: The conversation messages.
+    :type messages: List[dict]
+    :return: The messages with normalized content types.
+    :rtype: List[dict]
+    """
+    if not isinstance(messages, list):
+        return messages
+    for msg in messages:
+        if not isinstance(msg, dict) or not isinstance(msg.get("content"), list):
+            continue
+        for item in msg["content"]:
+            if not isinstance(item, dict):
+                continue
+            t = item.get("type")
+            if t == "function_call":
+                item["type"] = "tool_call"
+            elif t == "function_call_output":
+                item["type"] = "tool_result"
+                if "function_call_output" in item:
+                    item["tool_result"] = item.pop("function_call_output")
+            elif t == "openapi_call":
+                item["type"] = "tool_call"
+            elif t == "openapi_call_output":
+                item["type"] = "tool_result"
+                if "openapi_call_output" in item:
+                    item["tool_result"] = item.pop("openapi_call_output")
+    return messages
+
+
+def _preprocess_messages(messages):
+    """Preprocess conversation messages by dropping MCP approval messages and normalizing function call types.
+
+    This should be called before passing messages to serialization or evaluation functions.
+
+    :param messages: The conversation messages.
+    :type messages: List[dict]
+    :return: The preprocessed messages.
+    :rtype: List[dict]
+    """
+    messages = _drop_mcp_approval_messages(messages)
+    messages = _normalize_function_call_types(messages)
+    return messages
+
+
+# endregion Multi-turn utilities

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_validators/__init__.py

@@ -8,11 +8,13 @@
 from ._tool_definitions_validator import ToolDefinitionsValidator
 from ._tool_calls_validator import ToolCallsValidator
 from ._task_navigation_efficiency_validator import TaskNavigationEfficiencyValidator
+from ._messages_validator import MessagesOrQueryResponseInputValidator
 
 __all__ = [
     "ValidatorInterface",
     "ConversationValidator",
     "ToolDefinitionsValidator",
     "ToolCallsValidator",
     "TaskNavigationEfficiencyValidator",
+    "MessagesOrQueryResponseInputValidator",
 ]