diff --git a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md index 50defabe0ef2..36ea406fbda2 100644 --- a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md +++ b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md @@ -1,5 +1,17 @@ # Release History +## 1.18.12 (Unreleased) + +### Features Added + +- Added `decision_protocol="openai"` for Decision evaluators, which calls Azure OpenAI's `/openai/v1/decisions` operation on the account's OpenAI endpoint. The Microsoft Decision protocol remains the default, and model names never select a protocol. When an Azure OpenAI Decisions request fails with `model_not_found`, the user error says to set `decision_protocol` to `microsoft`, or remove it, if the deployment is a Microsoft Decision model. Error messages on the Microsoft Decision protocol are unchanged. +- Each protocol is a small adapter that translates the shared Boolean and ordinal questions to its provider contract and binds answers back to the same identifiers, so evaluators keep their scoring, applicability and probability cutoffs. OpenAI score levels are labeled `Level 1` to `Level n`, matching rubric text such as "imply level 2". An OpenAI answer of any other type is rejected as an invalid response, and score probabilities that are not a list of distinct levels fail when the evaluation reads them. The shared transport, authentication, retries, deadlines, cancellation and byte budgets are reused; no OpenAI client upgrade is required. Byte budgets measure the same canonical request on every protocol, so OpenAI admits the same evidence as Microsoft Decision. +- Decision provenance names the protocol in `decision_model.decision_protocol` (`microsoft` or `openai`), and the per-request metrics and execution logs carry it as `protocol`. Provider cache and reasoning counters are kept under `decision_model.usage_details` without being added to token totals, and only when the token count they describe part of is known. Split conversations sum each counter over the turns that report that token count, so the counters never cover requests that the token totals omit. +- An OpenAI question can be refused. A refusal fails an evaluation only when the evaluation reads that answer, so a skipped evaluation never depends on its grade. The failure is a user-classified `EvaluationException` with known usage and `decision_model.error_kind` set to `model_refusal`. +- A provider content filter's rejection (HTTP 400 with code `content_filter`) keeps its user classification and sets `decision_model.error_kind` to `content_filter`. Failed result rows keep both new kinds, as they keep the kinds added in 1.18.11. +- Evidence that is not finite JSON data fails with the same user-classified error on every protocol. +- Each `DecisionEvaluationStrategy.evaluate` call, which requests and interprets one Decision evaluation, records the optional OpenTelemetry metrics `azure.ai.evaluation.decision.evaluations` (count) and `azure.ai.evaluation.decision.evaluation.duration` (ms), with `protocol`, `metric`, `outcome` (`completed`, `skipped`, `error` or `cancelled`, including cancellation by the cancellation callback), `blame` (`none`, `user` or `system`) and `error_kind` (`none` or the failure's `decision_model.error_kind`, such as `context_window_exceeded`, `content_filter` or `model_refusal`). `metric` is the result key of this package's and the Microsoft Foundry catalog's Decision evaluators, and `custom` for any other evaluator, so a customer-defined name never becomes a metric dimension. The metrics complement the per-request metrics: an HTTP 200 response that fails validation or is not applicable is visible here. A conversation scored turn by turn counts each turn. Inputs found not applicable before any request are not counted, and evaluators that interpret a shared request without `evaluate`, such as composite evaluators, are covered by the per-request metrics only. A metric failure, including a broken OpenTelemetry installation, never changes the evaluation result or error. + ## 1.18.11 (2026-11-03) ### Features Added diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py index 63f1d9960485..00d9b23d60dc 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py @@ -22,7 +22,13 @@ from ..._evaluate._utils import _has_aggregator from ..._constants import Prefixes, PF_BATCH_TIMEOUT_SEC from ..._exceptions import EvaluationException -from ..._evaluators._common._model_runners import DECISION_ERROR_KINDS, _is_finite_number, validated_execution +from ..._evaluators._common._decision_protocols import DECISION_PROTOCOLS +from ..._evaluators._common._model_runners import ( + DECISION_ERROR_KINDS, + _is_finite_number, + usage_details, + validated_execution, +) from .._utils import get_int_env_var as get_int @@ -55,6 +61,7 @@ def _decision_error_metadata(exception: EvaluationException) -> Dict[str, Any]: for key, accepted in ( ("backend", ("foundry_decision", "system_one")), ("protocol", ("shared_sdk", "system_one")), + ("decision_protocol", tuple(DECISION_PROTOCOLS)), ("result_usage_scope", ("reported_attempts_only", "final_response_only", "reported_turns_only")), ("error_kind", DECISION_ERROR_KINDS), ): @@ -63,6 +70,9 @@ def _decision_error_metadata(exception: EvaluationException) -> Dict[str, Any]: provenance[key] = value if isinstance(native.get("usage_available"), bool): provenance["usage_available"] = native["usage_available"] + details = usage_details(native.get("usage_details")) + if details: + provenance["usage_details"] = details execution = validated_execution(native.get("execution")) if execution is not None: provenance["execution"] = execution diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_base_model_eval.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_base_model_eval.py index 76730b2ee8b0..d18a70aac3e1 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_base_model_eval.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_base_model_eval.py @@ -4,7 +4,7 @@ import asyncio import math -from typing import Any, Callable, Dict, List, Optional, TypeVar, Union +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TypeVar, Union from azure.core.credentials import TokenCredential from azure.ai.evaluation._constants import EVALUATION_PASS_FAIL_MAPPING @@ -13,8 +13,10 @@ from ._model_runners import ( DecisionModelRunner, PromptyModelRunner, + USAGE_DETAIL_TOKENS, _is_finite_number, is_decision_model_config, + usage_details, ) T = TypeVar("T") @@ -68,7 +70,9 @@ class ModelEvaluatorBase(EvaluatorBase[T]): :keyword cancellation_requested: Optional Decision cancellation callback, polled during authentication, requests and retry waits. Cancelling stops awaiting synchronous credential calls, not their worker threads. :paramtype cancellation_requested: Optional[Callable[[], bool]] - :keyword max_request_bytes: Optional limit on the compact UTF-8 serialized Decision request. + :keyword max_request_bytes: Optional limit on the compact UTF-8 JSON of the canonical Decision request + (deployment, evidence and questions), measured before protocol encoding so every protocol admits the same + evidence. :paramtype max_request_bytes: Optional[int] :keyword max_state_question_bytes: Optional limit on the largest individual state/question pair. :paramtype max_state_question_bytes: Optional[int] @@ -202,10 +206,12 @@ def _decision_model_runner(self) -> DecisionModelRunner: def _aggregate_results(self, per_turn_results: List[Dict[str, T]]) -> Dict[str, Any]: """Aggregate scores normally, but sum independently known Decision usage across turns. - A single shared model is promoted for attribution. Mixed or missing model identities remain per-turn - rather than attributing all usage to the last model. Root properties use the same usage and model, - without a single-request execution summary; other diagnostics retain their final-turn semantics. - Prompty aggregation is unchanged. + Token totals are each summed over the turns that reported them, and cache or reasoning counters over the + turns that reported the token count they describe part of. A single shared model is promoted for + attribution. Mixed or missing model identities remain per-turn rather than attributing all usage to the last + model. When the final turn has properties, root properties use the same usage and model, without a + single-request execution summary; other diagnostics retain their final-turn semantics. Prompty aggregation is + unchanged. """ aggregated: Dict[str, Any] = super()._aggregate_results(per_turn_results) if not self._uses_decision_model: @@ -235,11 +241,15 @@ def _aggregate_results(self, per_turn_results: List[Dict[str, T]]) -> Dict[str, properties_key = f"{self._result_key}_properties" properties = aggregated.get(properties_key) if len(per_turn_results) > 1 and isinstance(properties, dict): - usage = self._known_turn_usage( - aggregated, ("prompt_tokens", "completion_tokens", "total_tokens"), prefix=f"{self._result_key}_" + prefix = f"{self._result_key}_" + fields = ("prompt_tokens", "completion_tokens", "total_tokens") + usage = self._known_turn_usage(aggregated, fields, prefix=prefix) + details = self._summed_usage_details( + (turn.get(properties_key), self._known_turn_usage(turn, fields, prefix=prefix)) + for turn in per_turn_results ) aggregated[properties_key] = self._reconcile_decision_properties( - properties, usage, aggregated.get(f"{self._result_key}_model") + properties, usage, aggregated.get(f"{self._result_key}_model"), details ) return aggregated @@ -248,17 +258,22 @@ def _on_evaluation_error( ) -> None: """Preserve known Decision turn usage without changing the failure or Prompty behavior. - Only the request-owning result's usage is accumulated. A failed turn's execution summary cannot - describe all these turns, so it is omitted and the wider usage scope is explicit. + Only the request-owning result's usage is accumulated, and cache or reasoning counters only from the turns + that reported the token count they describe part of. A failed turn's execution summary cannot describe all + these turns, so it is omitted and the wider usage scope is explicit. When no earlier turn counted usage, the + failure's own properties already describe every counted request and are left unchanged. """ if not self._uses_decision_model or not per_turn_results: return fields = ("prompt_tokens", "completion_tokens", "total_tokens") + properties_key = f"{self._result_key}_properties" contributions = [] + reported = [] for result in per_turn_results: usage = self._known_turn_usage(result, fields, prefix=f"{self._result_key}_") if usage: contributions.append((usage, result.get(f"{self._result_key}_model"))) + reported.append((result.get(properties_key), usage)) if not contributions: return error_properties = getattr(error, "properties", None) @@ -266,6 +281,7 @@ def _on_evaluation_error( failing_usage = self._known_turn_usage(properties, fields) if failing_usage: contributions.append((failing_usage, properties.get("model"))) + reported.append((properties, failing_usage)) usage = {} for field in fields: known = [item[field] for item, _ in contributions if field in item] @@ -275,11 +291,39 @@ def _on_evaluation_error( common_model = models[0] if isinstance(models[0], str) and models[0].strip() else None if not all(model == common_model for model in models): common_model = None - setattr(error, "properties", self._reconcile_decision_properties(properties, usage, common_model)) + details = self._summed_usage_details(reported) + setattr(error, "properties", self._reconcile_decision_properties(properties, usage, common_model, details)) + + @staticmethod + def _summed_usage_details(turns: Iterable[Tuple[Any, Dict[str, int]]]) -> Dict[str, Dict[str, int]]: + """Sum each known cache or reasoning counter over the turns whose matching token count is counted. + + Input counters are summed over the turns that report prompt tokens and output counters over the turns that + report completion tokens, so the counters never cover requests that the token totals omit. + + :param turns: The properties and known token usage of each turn whose Decision response is accounted for. + :type turns: Iterable[Tuple[Any, Dict[str, int]]] + :return: The summed counters grouped like the provider's usage object; empty when no turn reported any. + :rtype: Dict[str, Dict[str, int]] + """ + totals: Dict[str, Dict[str, int]] = {} + for properties, usage in turns: + native = properties.get("decision_model") if isinstance(properties, dict) else None + details = usage_details(native.get("usage_details")) if isinstance(native, dict) else {} + for group, counters in details.items(): + if USAGE_DETAIL_TOKENS[group] not in usage: + continue + summed = totals.setdefault(group, {}) + for counter, value in counters.items(): + summed[counter] = summed.get(counter, 0) + value + return totals @staticmethod def _reconcile_decision_properties( - properties: Dict[str, Any], usage: Dict[str, int], model: Optional[str] + properties: Dict[str, Any], + usage: Dict[str, int], + model: Optional[str], + details: Optional[Dict[str, Dict[str, int]]] = None, ) -> Dict[str, Any]: reconciled = dict(properties) for field in ("prompt_tokens", "completion_tokens", "total_tokens"): @@ -294,6 +338,10 @@ def _reconcile_decision_properties( native = reconciled.get("decision_model") native = dict(native) if isinstance(native, dict) else {} native.pop("execution", None) + if details: + native["usage_details"] = details + else: + native.pop("usage_details", None) if "model" in reconciled: native["model"] = reconciled["model"] else: diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_protocols.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_protocols.py new file mode 100644 index 000000000000..07a184c36695 --- /dev/null +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_protocols.py @@ -0,0 +1,206 @@ +# --------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# --------------------------------------------------------- + +"""Wire adapters for the Decision protocols that the shared runner speaks. + +Evaluators always build one canonical request: evidence (``state``) and ``noul`` or ``score`` questions keyed by +identifier. A protocol translates that request to its provider's wire contract and binds the provider's answers back +to the same identifiers in the canonical answer shape. Protocols never validate answer values, score, apply +thresholds or decide applicability: values are validated only when an evaluator reads them, so an answer that an +evaluation does not use cannot fail it. +""" + +import json +from abc import ABC, abstractmethod +from typing import Any, Dict, Mapping, Optional +from urllib.parse import urlsplit + +REFUSAL = "refusal" + + +class DecisionProtocol(ABC): + """Translates canonical Decision requests and responses for one provider contract.""" + + name: str + supports_refusals = False + # The provider error code for a deployment that exists but serves another protocol, and how to fix that. + mismatch_code: Optional[str] = None + mismatch_hint = "" + + @abstractmethod + def operation_url(self, endpoint: str) -> str: + """Return the provider operation URL for a validated account endpoint. + + :param endpoint: The account endpoint from the model configuration. + :type endpoint: str + :return: The URL that receives the request. + :rtype: str + :raises ValueError: If the endpoint path is not supported by this protocol. + """ + + @abstractmethod + def encode(self, deployment: str, state: Any, questions: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + """Return the wire request body for canonical evidence and questions. + + :param deployment: The deployment name sent as the request's model. + :type deployment: str + :param state: The prepared evidence, as text or data that the runner has already checked is finite JSON. + :type state: Any + :param questions: Canonical ``noul`` and ``score`` questions keyed by identifier. + :type questions: Dict[str, Dict[str, Any]] + :return: The request body. + :rtype: Dict[str, Any] + :raises ValueError: If the provider cannot express the questions. + """ + + @abstractmethod + def bind_answers(self, payload: Dict[str, Any], questions: Mapping[str, Any]) -> Dict[str, Any]: + """Bind the provider's answers to question identifiers in the canonical answer shape. + + :param payload: The provider response. + :type payload: Dict[str, Any] + :param questions: The canonical questions keyed by identifier. + :type questions: Mapping[str, Any] + :return: Answers keyed by question identifier. + :rtype: Dict[str, Any] + :raises ValueError: If the answers cannot be bound to the questions. + """ + + +class MicrosoftDecisionProtocol(DecisionProtocol): + """The Microsoft Decision contract, which is also the canonical shape. A supplied endpoint path is kept. + + Its errors carry no protocol hint, because ``unknown_model`` also answers a mistyped or chat deployment. + """ + + name = "microsoft" + _PATH = "/providers/microsoft/v1/systemone" + + def operation_url(self, endpoint: str) -> str: + return f"{endpoint.rstrip('/')}{self._PATH}" + + def encode(self, deployment: str, state: Any, questions: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + return {"model": deployment, "state": state, "questions": questions} + + def bind_answers(self, payload: Dict[str, Any], questions: Mapping[str, Any]) -> Dict[str, Any]: + answers = payload.get("answers") + if not isinstance(answers, dict): + raise ValueError("answers are missing.") + return answers + + +class OpenAIDecisionProtocol(DecisionProtocol): + """Azure OpenAI Decisions: predicates, ordered score levels and named answers that may be refused. + + A ``noul`` question becomes a ``predicate`` whose instructions also state its true and false criteria, and a + ``score`` question's criteria become levels labeled ``"Level 1"`` to ``"Level n"`` from the lowest, the ordinal + names rubric text such as "imply level 2" uses. Structured evidence is sent as compact JSON text. For example, the + answer ``{"name": "status", "type": "predicate", "probability": 0.9}`` binds to ``{"type": "noul", "noul": 0.9}`` + for question ``status``, and a score's level probabilities become a map keyed by the provider's zero-based level + index. A refused question binds to ``{"type": "refusal"}``. + """ + + name = "openai" + supports_refusals = True + mismatch_code = "model_not_found" + mismatch_hint = ( + "If the deployment is a Microsoft Decision model, set decision_protocol to 'microsoft' or remove it." + ) + MAX_QUESTIONS = 400 + _BASE_PATHS = ("", "/openai", "/openai/v1") + + def operation_url(self, endpoint: str) -> str: + parsed = urlsplit(endpoint) + if parsed.path.rstrip("/") not in self._BASE_PATHS: + raise ValueError("OpenAI Decisions requires an account endpoint or its /openai/v1 base URL.") + return f"{parsed.scheme}://{parsed.netloc}/openai/v1/decisions" + + def encode(self, deployment: str, state: Any, questions: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + if not 1 <= len(questions) <= self.MAX_QUESTIONS: + raise ValueError(f"OpenAI Decisions requires between 1 and {self.MAX_QUESTIONS} questions.") + evidence = ( + state + if isinstance(state, str) + else json.dumps(state, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + ) + encoded = [self._encode_question(name, question) for name, question in questions.items()] + return {"model": deployment, "input": evidence, "questions": encoded} + + @staticmethod + def _encode_question(name: str, question: Dict[str, Any]) -> Dict[str, Any]: + instructions = question.get("instructions") + if not isinstance(instructions, str) or not instructions.strip(): + raise ValueError("Decision questions require nonempty text instructions.") + criteria = question.get("criteria") + if question["type"] == "noul": + if criteria is not None: + if ( + not isinstance(criteria, dict) + or set(criteria) != {"true", "false"} + or not all(isinstance(value, str) and value.strip() for value in criteria.values()) + ): + raise ValueError("Boolean criteria must describe both 'true' and 'false'.") + instructions += ( + f"\n\nThe predicate is true when: {criteria['true']}" + f"\nThe predicate is false when: {criteria['false']}" + ) + return {"name": name, "type": "predicate", "instructions": instructions} + if ( + not isinstance(criteria, list) + or not 2 <= len(criteria) <= 10 + or not all(isinstance(value, str) and value.strip() for value in criteria) + ): + raise ValueError("OpenAI score questions require 2 to 10 ordered text criteria.") + levels = [ + {"label": f"Level {position}", "description": description} + for position, description in enumerate(criteria, start=1) + ] + return {"name": name, "type": "score", "instructions": instructions, "levels": levels} + + def bind_answers(self, payload: Dict[str, Any], questions: Mapping[str, Any]) -> Dict[str, Any]: + answers = payload.get("answers") + if not isinstance(answers, list): + raise ValueError("answers are missing.") + bound: Dict[str, Any] = {} + for answer in answers: + name = answer.get("name") if isinstance(answer, dict) else None + if not isinstance(name, str) or name not in questions or name in bound: + raise ValueError("answer names must match each question exactly once.") + bound[name] = self._canonical_answer(answer) + return bound + + @staticmethod + def _canonical_answer(answer: Dict[str, Any]) -> Dict[str, Any]: + kind = answer.get("type") + if kind == "predicate": + return {"type": "noul", "noul": answer.get("probability")} + if kind == "score": + return { + "type": "score", + "score": answer.get("score"), + "confidence": answer.get("confidence"), + "probabilities": _level_probabilities(answer.get("probabilities")), + } + if kind == REFUSAL: + return {"type": REFUSAL} + # Any other type, including a canonical type such as "noul", matches no question, so the runner rejects it. + return {"type": None} + + +def _level_probabilities(distribution: Any) -> Optional[Dict[str, Any]]: + """Key each level's probability by its index; a malformed distribution is None and fails when it is read.""" + if not isinstance(distribution, list): + return None + probabilities: Dict[str, Any] = {} + for item in distribution: + index = item.get("value") if isinstance(item, dict) else None + if isinstance(index, bool) or not isinstance(index, int) or str(index) in probabilities: + return None + probabilities[str(index)] = item.get("probability") + return probabilities + + +DECISION_PROTOCOLS: Dict[str, DecisionProtocol] = { + protocol.name: protocol for protocol in (MicrosoftDecisionProtocol(), OpenAIDecisionProtocol()) +} diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_strategy.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_strategy.py index 1f75f1d464a8..8186c323525d 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_strategy.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_decision_strategy.py @@ -8,19 +8,137 @@ status gating, score mapping, provenance and result shape. """ +import asyncio import hashlib import json +import time from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException from ._base_model_eval import not_applicable_result -from ._model_runners import DecisionModelRunner, validate_noul_answer, validate_score_answer, validated_execution +from ._decision_protocols import DECISION_PROTOCOLS, REFUSAL +from ._model_runners import ( + _DEFAULT_PROTOCOL, + CANCELLED_BY_CALLBACK, + DECISION_ERROR_KINDS, + MODEL_REFUSAL, + DecisionModelRunner, + _diagnostic, + validate_noul_answer, + validate_score_answer, + validated_execution, +) STATUS_CUTOFF = 0.5 +_METRIC_PREFIX = "azure.ai.evaluation.decision." +# The result keys of this package's and the Microsoft Foundry catalog's Decision evaluators. Evaluation metrics report +# any other result key as "custom", so a customer-defined evaluator name never becomes a metric dimension. +_METRIC_NAMES = frozenset( + { + "coherence", + "customer_satisfaction", + "deflection_rate", + "fluency", + "groundedness", + "intent_resolution", + "output_quality", + "relevance", + "response_completeness", + "retrieval", + "similarity", + "task_adherence", + "task_completion", + "tool_call_accuracy", + "tool_call_success", + "tool_input_accuracy", + "tool_output_utilization", + "tool_selection", + "tool_use_quality", + } +) T = TypeVar("T") +class DecisionRefusalError(EvaluationException): + """The decision model declined to answer a question that the evaluation reads. + + A refusal is not a low probability, a zero score or a skip: the evaluation fails without a score. The model's + content policy decided about the evaluated content, so the error is classified like a chat judge's + content-filter refusal. + """ + + ERROR_KIND = MODEL_REFUSAL + + def __init__(self, question_id: str, error_target: ErrorTarget) -> None: + self.question_id = question_id + message = f"The decision model refused to answer question '{question_id}'; no score is available." + super().__init__( + message=message, + internal_message="The decision model refused to answer an evaluation question.", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.FAILED_EXECUTION, + target=error_target, + ) + + def __reduce__(self): + # Copies and pickles rebuild from the constructor's arguments, then restore attached metadata. + return type(self), (self.question_id, self.target), self.__dict__ + + +def _error_kind(error: EvaluationException) -> str: + """Return the bounded ``decision_model.error_kind`` of a failed evaluation, or ``none`` when it has none.""" + if isinstance(error, DecisionRefusalError): + return MODEL_REFUSAL + properties = getattr(error, "properties", None) + native = properties.get("decision_model") if isinstance(properties, dict) else None + kind = native.get("error_kind") if isinstance(native, dict) else None + return kind if isinstance(kind, str) and kind in DECISION_ERROR_KINDS else "none" + + +def _protocol_name(runner: Any) -> str: + """Return the runner's protocol name: the default for a runner without one, ``unknown`` if unrecognized.""" + name = getattr(runner, "decision_protocol", _DEFAULT_PROTOCOL) + return name if isinstance(name, str) and name in DECISION_PROTOCOLS else "unknown" + + +def _cancelled_by_callback(error: EvaluationException) -> bool: + """Return whether the runner marked the error as a cancellation by the cancellation callback.""" + return getattr(error, CANCELLED_BY_CALLBACK, False) is True + + +def _record_evaluation( + runner: Any, result_key: Any, outcome: str, blame: str, error_kind: str, elapsed_ms: float +) -> None: + """Count one Decision evaluation and record its duration when OpenTelemetry is installed. + + Instruments are looked up per call, like the request metrics, so strategies stay picklable. Attributes are + bounded: the protocol name, the result key of a known evaluator or ``custom``, and fixed outcome, blame and + error kind values. A metric failure, including a broken OpenTelemetry installation, is logged and never changes + the evaluation result or exception. + """ + try: + from opentelemetry.metrics import get_meter # pylint: disable=import-outside-toplevel + + attributes = { + "backend": "decision", + "protocol": _protocol_name(runner), + "metric": result_key if isinstance(result_key, str) and result_key in _METRIC_NAMES else "custom", + "outcome": outcome, + "blame": blame, + "error_kind": error_kind, + } + meter = get_meter("azure.ai.evaluation.decision") + meter.create_counter(_METRIC_PREFIX + "evaluations").add(1, attributes=attributes) + meter.create_histogram(_METRIC_PREFIX + "evaluation.duration", unit="ms").record( + elapsed_ms, attributes=attributes + ) + except ImportError: + return + except Exception: # pylint: disable=broad-except + _diagnostic("Decision evaluation metric sink failed; the evaluation result is unchanged.") + + def _checked_value(validate: Callable[..., T], error_target: ErrorTarget, *args: Any) -> T: try: return validate(*args) @@ -35,18 +153,26 @@ def _checked_value(validate: Callable[..., T], error_target: ErrorTarget, *args: ) from None +def _read_answer(response: Dict[str, Any], question_id: str, error_target: ErrorTarget) -> Dict[str, Any]: + answer = response["answers"][question_id] + if isinstance(answer, dict) and answer.get("type") == REFUSAL: + raise DecisionRefusalError(question_id, error_target) + return answer + + def decision_probability( response: Dict[str, Any], question_id: str, error_target: ErrorTarget = ErrorTarget.EVALUATE ) -> float: """Read and validate the probability of a binary decision answer.""" - return _checked_value(validate_noul_answer, error_target, response["answers"][question_id]) + return _checked_value(validate_noul_answer, error_target, _read_answer(response, question_id, error_target)) def decision_distribution( response: Dict[str, Any], question_id: str, levels: int, error_target: ErrorTarget = ErrorTarget.EVALUATE ) -> List[float]: """Read and validate the distribution of a scored decision answer.""" - return _checked_value(validate_score_answer, error_target, response["answers"][question_id], levels) + answer = _read_answer(response, question_id, error_target) + return _checked_value(validate_score_answer, error_target, answer, levels) def decision_skip_reason( @@ -173,18 +299,44 @@ def __init__( self.extra_properties = extra_properties async def evaluate(self, eval_input: Dict[str, Any]) -> Dict[str, Any]: - response = await self.runner(state=dict(eval_input), questions=self.questions, timeout=self.timeout) - result = self.parse(response, eval_input) - tokens = decision_token_counts(response) - result.update( - { - f"{self.result_key}_prompt_tokens": tokens["input_token_count"], - f"{self.result_key}_completion_tokens": tokens["output_token_count"], - f"{self.result_key}_total_tokens": tokens["total_token_count"], - f"{self.result_key}_model": tokens["model_id"], - } - ) - return result + """Request and interpret one Decision evaluation, recording its outcome and duration as metrics. + + Each call counts once, so a conversation scored turn by turn counts each turn. Outcomes are ``completed``, + ``skipped`` (not applicable), ``error`` with the error's blame and bounded ``error_kind``, or ``cancelled`` + by ``asyncio`` or the cancellation callback. ``metric`` is the evaluator's result key for this package's and + the Microsoft Foundry catalog's evaluators and ``custom`` for any other. + """ + started = time.monotonic() + outcome, blame, error_kind = "error", "system", "none" + try: + response = await self.runner(state=dict(eval_input), questions=self.questions, timeout=self.timeout) + result = self.parse(response, eval_input) + tokens = decision_token_counts(response) + result.update( + { + f"{self.result_key}_prompt_tokens": tokens["input_token_count"], + f"{self.result_key}_completion_tokens": tokens["output_token_count"], + f"{self.result_key}_total_tokens": tokens["total_token_count"], + f"{self.result_key}_model": tokens["model_id"], + } + ) + outcome = "skipped" if result.get(f"{self.result_key}_status") == "skipped" else "completed" + blame = "none" + return result + except asyncio.CancelledError: + outcome, blame = "cancelled", "none" + raise + except EvaluationException as exc: + if _cancelled_by_callback(exc): + outcome, blame = "cancelled", "none" + else: + blame = "user" if exc.blame == ErrorBlame.USER_ERROR else "system" + error_kind = _error_kind(exc) + raise + finally: + _record_evaluation( + self.runner, self.result_key, outcome, blame, error_kind, (time.monotonic() - started) * 1000 + ) def parse(self, response: Dict[str, Any], eval_input: Dict[str, Any]) -> Dict[str, Any]: """Interpret a validated response without making a request. @@ -196,6 +348,7 @@ def parse(self, response: Dict[str, Any], eval_input: Dict[str, Any]) -> Dict[st properties = self.extra_properties(eval_input) if self.extra_properties else {} properties["decision_model"] = { "deployment": self.runner.deployment, + "decision_protocol": _protocol_name(self.runner), "model": response["model"], "rubric_hash": _digest(self.questions[key]), "status_rubric_hash": _digest(self.questions["status"]), @@ -203,6 +356,8 @@ def parse(self, response: Dict[str, Any], eval_input: Dict[str, Any]) -> Dict[st "reason_type": "templated", } execution = validated_execution(response.get("_decision_execution")) + if response.get("usage_details"): + properties["decision_model"]["usage_details"] = response["usage_details"] if execution is not None: properties["decision_model"]["execution"] = execution tokens = decision_token_counts(response) @@ -231,6 +386,8 @@ def parse(self, response: Dict[str, Any], eval_input: Dict[str, Any]) -> Dict[st {name: value for name, value in response["answers"][key].items() if name != "legend"}, ) except EvaluationException as exc: + if isinstance(exc, DecisionRefusalError): + properties["decision_model"]["error_kind"] = DecisionRefusalError.ERROR_KIND exc.properties = properties raise properties["decision_model"].update(details) diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_model_runners.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_model_runners.py index b2035dda5695..4cde411f6d86 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_model_runners.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_model_runners.py @@ -26,6 +26,7 @@ from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException from ..._common.utils import construct_prompty_model_config, validate_model_config from ..._http_utils import get_async_http_client +from ._decision_protocols import DECISION_PROTOCOLS, REFUSAL, DecisionProtocol if os.getenv("AI_EVALS_USE_PF_PROMPTY", "false").lower() == "true": from promptflow.core._flow import AsyncPrompty @@ -43,7 +44,6 @@ def value(self) -> str: DECISION_MODEL_TYPE = "decision" -_DECISION_PATH = "/providers/microsoft/v1/systemone" _DECISION_TOKEN_SCOPE = "https://cognitiveservices.azure.com/.default" LOGGER = logging.getLogger(__name__) _PREFIX = "azure.ai.evaluation.decision." @@ -54,10 +54,27 @@ def value(self) -> str: "api_key", "credential", "probability_threshold", + "decision_protocol", } _DEFAULT_PROBABILITY_THRESHOLD = 0.75 +_DEFAULT_PROTOCOL = "microsoft" +_INVALID_EVIDENCE = "Decision evidence must be finite JSON-serializable data." +# The provider error code, and failed-row error kind, of a content filter rejecting the evaluated content. +CONTENT_FILTER = "content_filter" +# The failed-row error kind of an evaluation whose question the decision model refused to answer. +MODEL_REFUSAL = "model_refusal" +# Set on the EvaluationException raised when the cancellation callback stops a request. Unlike the execution +# summary, which is omitted when it exceeds the hosted metadata bounds, it is always present. +CANCELLED_BY_CALLBACK = "_decision_cancelled_by_callback" # The service adds api_version to every resolved model configuration; the decision endpoint does not use it. _IGNORED_CONFIG_KEYS = {"api_version"} +_USAGE_DETAIL_FIELDS = ( + ("input_tokens_details", ("cached_tokens", "cache_write_tokens")), + ("output_tokens_details", ("reasoning_tokens",)), +) +# The token count that each group of usage details describes part of. A group is kept only when that count is known, +# so the counters never cover requests that the token totals omit. +USAGE_DETAIL_TOKENS = {"input_tokens_details": "prompt_tokens", "output_tokens_details": "completion_tokens"} _ERROR_CODE = re.compile(r"[A-Za-z][A-Za-z0-9_.-]{0,127}") # A 422 states its reason only in a ``detail`` sentence, which is never surfaced. These reasons are recognized by the # provider's fixed sentence openings, so a quoted question identifier later in the sentence cannot change the kind, and @@ -80,7 +97,12 @@ def value(self) -> str: _MAX_PROVIDER_DETAIL_CHARS = 1024 # Reported when a request exceeds the caller's max_request_bytes or max_state_question_bytes and is not sent. _REQUEST_BYTES_EXCEEDED = "request_bytes_exceeded" -DECISION_ERROR_KINDS = tuple(kind for _, kind in _PROVIDER_DETAIL_KINDS) + (_REQUEST_BYTES_EXCEEDED,) +# Every bounded decision_model.error_kind that failed result rows keep and evaluation metrics report. +DECISION_ERROR_KINDS = tuple(kind for _, kind in _PROVIDER_DETAIL_KINDS) + ( + _REQUEST_BYTES_EXCEEDED, + CONTENT_FILTER, + MODEL_REFUSAL, +) def is_decision_model_config(model_config: Any) -> bool: @@ -100,6 +122,33 @@ def _integer(value: Any, maximum: Optional[int] = None) -> bool: ) +def _utf8_json(value: Any) -> bytes: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8") + + +def usage_details(usage: Any) -> Dict[str, Dict[str, int]]: + """Return the known cache and reasoning counters of a provider usage object. + + For example, ``{"input_tokens": 12, "input_tokens_details": {"cached_tokens": 4}}`` yields + ``{"input_tokens_details": {"cached_tokens": 4}}``. The counters describe part of the input or output tokens and + are never added to token totals; counters that are not nonnegative integers are omitted. + + :param usage: The untrusted provider usage object. + :type usage: Any + :return: Known counters grouped like the provider's usage object. + :rtype: Dict[str, Dict[str, int]] + """ + details: Dict[str, Dict[str, int]] = {} + if isinstance(usage, dict): + for group, fields in _USAGE_DETAIL_FIELDS: + values = usage.get(group) + if isinstance(values, dict): + known = {field: values[field] for field in fields if _integer(values.get(field))} + if known: + details[group] = known + return details + + def validated_execution(value: Any) -> Optional[Dict[str, Any]]: """Copy only a complete, strictly typed shared or independent execution summary. @@ -165,10 +214,11 @@ def _diagnostic(message: str) -> None: class DecisionExecution(CustomHookPolicy): """One operation's per-attempt pipeline policy; never shared between concurrent calls.""" - def __init__(self, logger: Any, max_response_bytes: Optional[int]) -> None: + def __init__(self, logger: Any, max_response_bytes: Optional[int], protocol: str = _DEFAULT_PROTOCOL) -> None: super().__init__() self.logger = logger self.max_response_bytes = max_response_bytes + self.protocol = protocol self.started = time.monotonic() self.attempt_started: Optional[float] = None self.attempt_count = 0 @@ -254,6 +304,7 @@ def emit( dimensions = { "backend": "decision", "route": "shared_sdk", + "protocol": self.protocol, "phase": phase, "outcome": outcome, "usage_state": usage, @@ -432,7 +483,7 @@ def validate_score_answer(answer: Dict[str, Any], levels: int) -> List[float]: class DecisionModelRunner: - """Sends prepared evidence and typed questions to a Microsoft-Decision-1 deployment in Foundry. + """Sends prepared evidence and typed questions to a Decision deployment in Foundry. The model configuration must be tagged ``type="decision"`` and contain the Foundry resource endpoint (``azure_endpoint``) and the deployment name (``azure_deployment``). Authentication uses ``api_key`` when @@ -445,24 +496,39 @@ class DecisionModelRunner: only be reused on that loop; cross-loop reuse is not supported. ``probability_threshold`` (default 0.75) is only accepted when ``binary=True``. - ``azure_endpoint`` must be the account resource root, not a project or OpenAI API endpoint. + ``decision_protocol`` selects the provider contract: the default ``"microsoft"`` uses the Microsoft Decision + operation, and ``"openai"`` uses Azure OpenAI's ``/openai/v1/decisions`` operation, which requires a deployment + with the Decision capability in a supported region. Model names never select a protocol. Every protocol returns + the same answers keyed by question identifier, so evaluators score them with the same policies. An OpenAI + question can be refused; the refusal fails an evaluation only when that evaluation reads the refused answer. + + ``azure_endpoint`` must be the account resource root, not a project endpoint. For example, use ``https://account.services.ai.azure.com`` rather than a URL ending in - ``/api/projects/project`` or ``/openai``. Any path supplied is retained when appending the provider operation. + ``/api/projects/project``. OpenAI requires the account's OpenAI endpoint, such as + ``https://account.openai.azure.com``, optionally with its ``/openai`` or ``/openai/v1`` base path; + the Microsoft protocol retains a supplied path when appending the provider operation. Execution controls are keyword-only and separate from model configuration. ``timeout_seconds`` bounds each connect/read; the call's ``timeout`` limits authentication, retries and response body reading. Decision requests use the native async HTTP transport so timeout and cancellation stop body reads. An injected pipeline retains its own transport; that transport must support cancellation for the same guarantee. ``max_retries`` defaults to 3. ``cancellation_requested`` is polled every 50 ms, including retry waits; - it stops awaiting synchronous credentials but cannot stop their worker threads. Byte limits measure - compact unescaped UTF-8 JSON, with ``max_state_question_bytes`` applying to the largest individual pair. - ``max_response_bytes`` limits acceptance before parsing, not transport allocation. + it stops awaiting synchronous credentials but cannot stop their worker threads. Byte limits measure the + compact unescaped UTF-8 JSON of the canonical request ``{"model", "state", "questions"}`` before protocol + encoding, so every protocol admits the same evidence; ``max_state_question_bytes`` applies to the largest + individual ``{"state", "question"}`` pair. ``max_response_bytes`` limits acceptance before parsing, + not transport allocation. A failed request names a recognized reason in ``decision_model.error_kind`` of the error's properties: ``context_window_exceeded`` when the evidence and one question exceed the model's context window, ``request_token_limit_exceeded``, ``question_limit_exceeded`` or ``invalid_question`` for the provider's other - rejections, and ``request_bytes_exceeded`` for a request over the byte limits, which is not sent. + rejections, ``request_bytes_exceeded`` for a request over the byte limits, which is not sent, and + ``content_filter`` when the provider's content filter rejects the evaluated content. Because the limits measure + the canonical request, a host that derives them from a provider's wire limit must allow for the larger OpenAI + encoding, which repeats Boolean criteria in each predicate and sends structured evidence as JSON text. """ + SUPPORTED_PROTOCOLS = frozenset(DECISION_PROTOCOLS) + def __init__( self, model_config: dict, @@ -562,7 +628,14 @@ def __init__( except ValueError: raise _invalid_config("'probability_threshold' must be a number in [0, 1].") from None - self._url = f"{endpoint.rstrip('/')}{_DECISION_PATH}" + protocol = model_config.get("decision_protocol", _DEFAULT_PROTOCOL) + if not isinstance(protocol, str) or protocol not in DECISION_PROTOCOLS: + raise _invalid_config(f"'decision_protocol' must be one of {sorted(DECISION_PROTOCOLS)}.") + self._protocol: DecisionProtocol = DECISION_PROTOCOLS[protocol] + try: + self._url = self._protocol.operation_url(endpoint) + except ValueError as exc: + raise _invalid_config(str(exc)) from None self._deployment = deployment self._api_key = api_key self._user_agent = f"{UserAgentSingleton().value} (type=evaluator subtype={evaluator_name})" @@ -581,6 +654,11 @@ def deployment(self) -> str: """The deployment name sent as the request's model.""" return self._deployment + @property + def decision_protocol(self) -> str: + """The provider protocol, for example ``"microsoft"`` or ``"openai"``.""" + return self._protocol.name + async def __call__(self, *, state: Any, questions: Dict[str, Dict[str, Any]], timeout: float) -> Dict[str, Any]: """Request the decisions for ``questions`` about ``state`` and return the validated response. @@ -590,47 +668,44 @@ async def __call__(self, *, state: Any, questions: Dict[str, Dict[str, Any]], ti :paramtype questions: Dict[str, Dict[str, Any]] :keyword timeout: Overall time limit in seconds. :paramtype timeout: float - :return: ``{"model", "answers", "usage"}`` with the response's structure checked against the questions. + :return: ``{"model", "answers", "usage", "usage_details"}`` with answers keyed by question identifier and + the response's structure checked against the questions. :rtype: Dict[str, Any] """ if not _is_finite_number(timeout) or timeout <= 0: raise _invalid_config("'timeout' must be a positive finite number.") deadline = time.monotonic() + timeout - execution = DecisionExecution(self._execution_logger, self._max_response_bytes) + execution = DecisionExecution(self._execution_logger, self._max_response_bytes, self.decision_protocol) outcome = "error" properties = self._response_properties(None) try: for question in questions.values(): if question.get("type") not in ("noul", "score"): raise _invalid_config(f"Decision question type '{question.get('type')}' is not supported.") - body = {"model": self._deployment, "state": state, "questions": questions} + # Budgets bound the evidence and questions the model reads, so every protocol measures the same canonical + # request and admits the same evidence; wire encoding, such as evidence escaped into JSON text, is not read. + canonical = {"model": self._deployment, "state": state, "questions": questions} try: - content = json.dumps(body, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8") + request = _utf8_json(canonical) state_question_bytes = 0 if self._max_state_question_bytes is not None: state_question_bytes = max( - ( - len( - json.dumps( - {"state": state, "question": question}, - ensure_ascii=False, - separators=(",", ":"), - allow_nan=False, - ).encode("utf-8") - ) - for question in questions.values() - ), + (len(_utf8_json({"state": state, "question": question})) for question in questions.values()), default=0, ) except (ValueError, TypeError, OverflowError, UnicodeError): - raise _failed( - "Decision evidence must be finite JSON-serializable data.", ErrorBlame.USER_ERROR - ) from None - if (self._max_request_bytes is not None and len(content) > self._max_request_bytes) or ( + raise _failed(_INVALID_EVIDENCE, ErrorBlame.USER_ERROR) from None + if (self._max_request_bytes is not None and len(request) > self._max_request_bytes) or ( self._max_state_question_bytes is not None and state_question_bytes > self._max_state_question_bytes ): properties["decision_model"]["error_kind"] = _REQUEST_BYTES_EXCEEDED raise _failed("The decision request exceeds the configured byte budget.", ErrorBlame.USER_ERROR) + try: + body = self._protocol.encode(self._deployment, state, questions) + # A wire body that is the canonical request, as Microsoft's is, is not serialized again. + content = request if body == canonical else _utf8_json(body) + except (ValueError, TypeError, OverflowError, UnicodeError) as exc: + raise _invalid_config(f"Invalid decision request: {exc}") from None try: payload = await asyncio.wait_for( self._await_request(content, deadline, execution), timeout=max(0, deadline - time.monotonic()) @@ -652,6 +727,8 @@ async def __call__(self, *, state: Any, questions: Dict[str, Dict[str, Any]], ti except EvaluationException as exc: if outcome != "deadline": outcome = execution.outcome + if outcome == "cancelled": + setattr(exc, CANCELLED_BY_CALLBACK, True) if exc.properties is None: exc.properties = properties properties = exc.properties @@ -769,8 +846,14 @@ async def _post(self, content: bytes, *, deadline: float, execution: DecisionExe blame = ErrorBlame.USER_ERROR if status in (400, 401, 403, 404, 422) else ErrorBlame.SYSTEM_ERROR code = _provider_error_code(payload) kind = _provider_error_kind(status, payload) + if kind is None and code == CONTENT_FILTER: + # The provider's content filter rejected the evaluated content, not how the request was built. + kind = CONTENT_FILTER diagnostic = f" ({code or kind})" if code or kind else "" - error = _failed(f"The decision model returned HTTP {status}{diagnostic}.", blame) + message = f"The decision model returned HTTP {status}{diagnostic}." + if code is not None and code == self._protocol.mismatch_code: + message = f"{message} {self._protocol.mismatch_hint}" + error = _failed(message, blame) error.properties = self._response_properties(payload) if kind is not None: error.properties["decision_model"]["error_kind"] = kind @@ -779,7 +862,9 @@ async def _post(self, content: bytes, *, deadline: float, execution: DecisionExe def _response_properties(self, payload: Any) -> Dict[str, Any]: """Retain only independently known final-response usage, never raw answers or retry totals.""" - properties: Dict[str, Any] = {"decision_model": {"deployment": self._deployment}} + properties: Dict[str, Any] = { + "decision_model": {"deployment": self._deployment, "decision_protocol": self.decision_protocol} + } usage = payload.get("usage") if isinstance(payload, dict) else None usage = usage if isinstance(usage, dict) else {} for native_key, key in (("input_tokens", "prompt_tokens"), ("output_tokens", "completion_tokens")): @@ -788,24 +873,31 @@ def _response_properties(self, payload: Any) -> Dict[str, Any]: properties[key] = value if "prompt_tokens" in properties and "completion_tokens" in properties: properties["total_tokens"] = properties["prompt_tokens"] + properties["completion_tokens"] + details = { + group: counters + for group, counters in usage_details(usage).items() + if USAGE_DETAIL_TOKENS[group] in properties + } + if details: + properties["decision_model"]["usage_details"] = details model = payload.get("model") if isinstance(payload, dict) else None if isinstance(model, str) and model.strip(): properties["model"] = model return properties - @staticmethod - def _validate_response(payload: Any, questions: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: - # Only the response's structure is checked here; answer values are validated by the evaluator after it - # has read the status answer, so a skipped evaluation never depends on an unusable score. - if not isinstance(payload, dict) or not isinstance(payload.get("answers"), dict): + def _validate_response(self, payload: Any, questions: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + # Only the response's structure is checked here; answer values are validated by the evaluator when it reads + # them, so a skipped evaluation never depends on an unusable or refused score. + if not isinstance(payload, dict): raise ValueError("answers are missing.") - answers = payload["answers"] + answers = self._protocol.bind_answers(payload, questions) if set(answers) != set(questions): raise ValueError("answer identifiers do not match the questions.") for question_id, question in questions.items(): answer = answers[question_id] - question_type = question.get("type") - if not isinstance(answer, dict) or answer.get("type") != question_type: + answer_type = answer.get("type") if isinstance(answer, dict) else None + refused = answer_type == REFUSAL and self._protocol.supports_refusals + if answer_type != question.get("type") and not refused: raise ValueError(f"answer '{question_id}' does not match its question type.") usage = payload.get("usage") @@ -821,4 +913,5 @@ def _validate_response(payload: Any, questions: Dict[str, Dict[str, Any]]) -> Di "model": model, "answers": answers, "usage": {"input_tokens": usage["input_tokens"], "output_tokens": usage["output_tokens"]}, + "usage_details": usage_details(usage), } diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_version.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_version.py index 03d429c451a0..a307b7b58822 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_version.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_version.py @@ -3,4 +3,4 @@ # --------------------------------------------------------- # represents upcoming version -VERSION = "1.18.11" +VERSION = "1.18.12" diff --git a/sdk/evaluation/azure-ai-evaluation/samples/openai_decisions/README.md b/sdk/evaluation/azure-ai-evaluation/samples/openai_decisions/README.md new file mode 100644 index 000000000000..318b2ef81313 --- /dev/null +++ b/sdk/evaluation/azure-ai-evaluation/samples/openai_decisions/README.md @@ -0,0 +1,55 @@ +# Azure OpenAI Decisions evaluators + +Use a GPT-6 Luna deployment with the `decision` capability in a region where Azure OpenAI Decisions is available +(East US 2 is excluded). Use the account's OpenAI endpoint, such as `https://YOUR-ACCOUNT.openai.azure.com`. +This requires `azure-ai-evaluation` 1.18.12 or later; 1.18.11 and earlier reject `decision_protocol`. + +```python +from azure.identity import DefaultAzureCredential +from azure.ai.evaluation import CoherenceEvaluator + +with DefaultAzureCredential() as credential: + evaluator = CoherenceEvaluator( + model_config={ + "type": "decision", + "decision_protocol": "openai", + "azure_endpoint": "https://YOUR-ACCOUNT.openai.azure.com", + "azure_deployment": "YOUR-DECISIONS-DEPLOYMENT", + }, + credential=credential, + ) + result = evaluator(query="When does the museum open?", response="The museum opens at noon.") +``` + +The other built-in evaluators that support Decision models, and the Decision-capable evaluators in the Microsoft +Foundry catalog, accept the same configuration; their rubrics are shared across protocols. +For API-key authentication, provide `api_key` in the model configuration from your secret store. +The SDK retains the existing retry, timeout, cancellation, and credential ownership rules. +There is no dependency on `openai.decisions.create`, which needs a newer major version of the `openai` package; +the shared asynchronous HTTP transport sends the same request to `/openai/v1/decisions`. + +Omitting `decision_protocol` selects the Microsoft Decision protocol. When an Azure OpenAI Decisions request fails +with `model_not_found`, the error says to set `decision_protocol` to `microsoft`, or remove it, if the deployment is a +Microsoft Decision model. An ordinary chat configuration continues to use chat even when +that deployment also supports Decisions; GPT-6 Luna is a reasoning model, so pass `is_reasoning_model=True` to an +evaluator that uses it as a chat judge. + +The SDK's protocol adapter turns each Boolean question into a predicate whose instructions state its true and false +criteria, and each ordinal question into levels labeled `Level 1` to `Level n` from the lowest, the ordinal names that +rubric text such as "imply level 2" uses. Answers are bound back by question name, so +the evaluator scores them exactly as it scores Microsoft Decision answers: ordinal metrics select the +highest-probability level, resolving ties to the lowest level, and the provider's weighted mean is kept only as +`native_mean`. `decision_model.native_answer` holds the answer in this shared shape, and +`decision_model.decision_protocol` is `openai`. Cache and reasoning counters stay under `decision_model.usage_details` +and are not added to token totals. + +A refused question fails the evaluation only when the evaluation reads that answer. For example, when the +applicability question says the metric does not apply, the evaluation is skipped even if its grade was refused. +A refusal that is read raises an `EvaluationException` with known usage and `decision_model.error_kind` set to +`model_refusal`; it never becomes a fabricated score or skip. + +Hosted evaluations select a service-approved deployment by name; the service owns protocol configuration and checks +its live capability and model version. + +References: [Azure Decisions](https://learn.microsoft.com/en-gb/azure/foundry/openai/how-to/decisions), +[OpenAI Decisions contract](https://developers.openai.com/api/reference/resources/decisions/methods/create). diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_batch_persistence.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_batch_persistence.py index 49020631c5eb..813269874579 100644 --- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_batch_persistence.py +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_batch_persistence.py @@ -506,6 +506,7 @@ class ToolUseQualityEvaluator(OutputQualityEvaluator): def test_semantic_failure_retains_measured_usage_and_hashes(self, batch_client, tmp_path, status): class Runner: deployment = "Decision-1" + decision_protocol = "microsoft" async def __call__(self, **kwargs): return payload(status) @@ -540,6 +541,7 @@ async def judge(query): def test_success_and_skip_are_not_stamped_as_errors(self, batch_client, tmp_path, status, expected): class Runner: deployment = "Decision-1" + decision_protocol = "microsoft" async def __call__(self, **kwargs): return payload(status) @@ -659,7 +661,9 @@ def cancel(): assert "total_tokens" not in result["properties"] and "prompt_tokens" not in result["properties"] else: assert "usage" not in result["sample"] and "model" not in result["sample"] - assert result["properties"] == {"decision_model": {"deployment": "Decision-1"}} + assert result["properties"] == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"} + } if "cancel" in case: assert result["sample"]["error"]["blame"] == "UserError" else: diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_conversation_failure.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_conversation_failure.py index 9a444edc50e3..fba9ec1f5454 100644 --- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_conversation_failure.py +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_conversation_failure.py @@ -176,6 +176,7 @@ async def send(self, request, **kwargs): assert {field: properties[field] for field in expected} == expected assert properties["decision_model"]["result_usage_scope"] == "reported_turns_only" assert properties["decision_model"]["usage_available"] is True + assert properties["decision_model"]["decision_protocol"] == "microsoft" assert "execution" not in properties["decision_model"] if mixed: assert "model" not in properties and "model" not in properties["decision_model"] @@ -302,6 +303,115 @@ def test_cancelled_error_without_properties_can_retain_known_turn_metadata(): assert error.properties["decision_model"]["usage_available"] is True +def test_failed_split_conversation_sums_usage_details_with_the_failing_turn(): + """Cache and reasoning counters follow the same reported-turn rule as the token totals.""" + evaluator = SplitDecisionEvaluator() + error = asyncio.CancelledError() + error.properties = { + "prompt_tokens": 20, + "completion_tokens": 3, + "model": "known-model", + "decision_model": { + "usage_details": { + "input_tokens_details": {"cached_tokens": 5}, + "output_tokens_details": {"reasoning_tokens": 7}, + } + }, + } + evaluator._on_evaluation_error( + error, + [ + { + "task_completion_prompt_tokens": 12, + "task_completion_model": "known-model", + "task_completion_properties": { + "decision_model": {"usage_details": {"input_tokens_details": {"cached_tokens": 4}}} + }, + } + ], + ) + assert error.properties["prompt_tokens"] == 32 + assert error.properties["completion_tokens"] == 3 + assert error.properties["decision_model"]["usage_details"] == { + "input_tokens_details": {"cached_tokens": 9}, + "output_tokens_details": {"reasoning_tokens": 7}, + } + + +def test_failed_split_conversation_keeps_each_counter_only_with_its_token_count(): + """Reasoning counters need the turn's completion tokens and cache counters its prompt tokens.""" + evaluator = SplitDecisionEvaluator() + error = asyncio.CancelledError() + error.properties = { + "completion_tokens": 3, + "decision_model": { + "usage_details": { + "input_tokens_details": {"cached_tokens": 50}, + "output_tokens_details": {"reasoning_tokens": 2}, + } + }, + } + evaluator._on_evaluation_error( + error, + [ + { + "task_completion_prompt_tokens": 12, + "task_completion_properties": { + "decision_model": {"usage_details": {"output_tokens_details": {"reasoning_tokens": 40}}} + }, + } + ], + ) + assert error.properties["prompt_tokens"] == 12 and error.properties["completion_tokens"] == 3 + assert error.properties["decision_model"]["usage_details"] == {"output_tokens_details": {"reasoning_tokens": 2}} + + +def test_failed_split_conversation_sums_usage_details_only_with_counted_usage(): + """Counters of a turn whose token usage is not counted could exceed the reported token totals.""" + evaluator = SplitDecisionEvaluator() + error = asyncio.CancelledError() + error.properties = {"decision_model": {"usage_details": {"input_tokens_details": {"cached_tokens": 50}}}} + evaluator._on_evaluation_error( + error, + [ + { + "task_completion_prompt_tokens": 12, + "task_completion_properties": { + "decision_model": {"usage_details": {"input_tokens_details": {"cached_tokens": 4}}} + }, + }, + { + "task_completion_properties": { + "decision_model": {"usage_details": {"output_tokens_details": {"reasoning_tokens": 9}}} + }, + }, + ], + ) + assert error.properties["prompt_tokens"] == 12 + assert error.properties["decision_model"]["usage_details"] == {"input_tokens_details": {"cached_tokens": 4}} + + +def test_failure_after_turns_without_requests_keeps_the_failing_request_metadata(): + """Earlier turns that made no request leave the failing request's usage, counters and summary as reported.""" + evaluator = SplitDecisionEvaluator() + error = asyncio.CancelledError() + error.properties = { + "prompt_tokens": 20, + "completion_tokens": 2, + "total_tokens": 22, + "model": "known-model", + "decision_model": { + "usage_details": {"input_tokens_details": {"cached_tokens": 5}}, + "execution": {"attempt_count": 1, "outcome": "error"}, + }, + } + reported = deepcopy(error.properties) + evaluator._on_evaluation_error( + error, [{"task_completion": None, "task_completion_status": "skipped", "task_completion_properties": None}] + ) + assert error.properties == reported + + def test_prompty_split_failure_preserves_existing_exception(monkeypatch): monkeypatch.setattr(_model_runners.AsyncPrompty, "load", Mock()) error = EvaluationException( diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_default_transport.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_default_transport.py index d8d0a43b9aaf..0ad616f27ee4 100644 --- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_default_transport.py +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_default_transport.py @@ -78,7 +78,7 @@ def invoke(endpoint, timeout=3, **controls): def assert_failure_execution(error, attempts, outcome): assert set(error.properties) == {"decision_model"} native = error.properties["decision_model"] - assert set(native) == {"deployment", "execution"} and native["deployment"] == "Decision-1" + assert set(native) == {"deployment", "execution", "decision_protocol"} and native["deployment"] == "Decision-1" execution = native["execution"] assert validated_execution(execution) == execution assert execution["attempt_count"] == attempts and execution["retry_count"] == max(0, attempts - 1) diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_model_support.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_model_support.py index 5761abc9e135..c1663747ded3 100644 --- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_model_support.py +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_decision_model_support.py @@ -6,10 +6,12 @@ import asyncio import hashlib +import importlib import json import logging import math import os +import pkgutil import subprocess import sys import threading @@ -17,7 +19,7 @@ from logging.handlers import BufferingHandler from concurrent.futures import ThreadPoolExecutor from pathlib import Path -from types import SimpleNamespace +from types import ModuleType, SimpleNamespace from unittest.mock import AsyncMock, Mock, call, patch import pytest @@ -27,11 +29,13 @@ from azure.core.exceptions import ServiceRequestError from azure.ai.evaluation import CoherenceEvaluator, FluencyEvaluator, GroundednessEvaluator +from azure.ai.evaluation import _evaluators as evaluators_package from azure.ai.evaluation._evaluators._common import EvaluatorBase, ModelEvaluatorBase, PromptyEvaluatorBase -from azure.ai.evaluation._evaluators._common import _model_runners +from azure.ai.evaluation._evaluators._common import _decision_strategy, _model_runners from azure.ai.evaluation._evaluators._common._decision_strategy import ( BinaryScoring, DecisionEvaluationStrategy, + DecisionRefusalError, LikertScoring, ) from azure.ai.evaluation._evaluators._common._model_runners import ( @@ -445,16 +449,83 @@ def test_provider_http_error_code_preserves_classification_and_accepted_metadata assert execution["outcome"] == "error" assert raised.value.properties == ( { - "decision_model": {"deployment": "Decision-1"}, + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"}, "prompt_tokens": 120, "completion_tokens": 3, "total_tokens": 123, "model": "microsoft-decision-1", } if response_limit == 1024 - else {"decision_model": {"deployment": "Decision-1"}} + else {"decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"}} ) + def test_an_openai_deployment_that_is_not_found_suggests_the_microsoft_protocol(self, client): + payload = {"error": {"code": "model_not_found", "message": "test-key-value and private-message"}} + client.response = FakeResponse(404, json.dumps(payload).encode()) + with pytest.raises(EvaluationException) as raised: + run(DecisionModelRunner({**DECISION_CONFIG, "decision_protocol": "openai"}, max_retries=0)) + assert str(raised.value) == ( + "(UserError) The decision model returned HTTP 404 (model_not_found). If the deployment is a Microsoft " + "Decision model, set decision_protocol to 'microsoft' or remove it." + ) + assert raised.value.blame == ErrorBlame.USER_ERROR + assert raised.value.properties["decision_model"]["decision_protocol"] == "openai" + + @pytest.mark.parametrize("protocol", [None, "microsoft"]) + def test_a_microsoft_unknown_model_error_has_no_protocol_hint(self, client, protocol): + # unknown_model also answers a mistyped or chat deployment, so the error is the same as in 1.18.11. + payload = {"error": {"code": "unknown_model", "message": "test-key-value and private-message"}} + client.response = FakeResponse(400, json.dumps(payload).encode()) + config = DECISION_CONFIG if protocol is None else {**DECISION_CONFIG, "decision_protocol": protocol} + with pytest.raises(EvaluationException) as raised: + run(DecisionModelRunner(config, max_retries=0)) + assert str(raised.value) == "(UserError) The decision model returned HTTP 400 (unknown_model)." + assert raised.value.blame == ErrorBlame.USER_ERROR + + @pytest.mark.parametrize( + "usage, kept", + [ + ({"input_tokens_details": {"cached_tokens": 50}, "output_tokens_details": {"reasoning_tokens": 5}}, None), + ( + { + "input_tokens": 9, + "input_tokens_details": {"cached_tokens": 4}, + "output_tokens_details": {"reasoning_tokens": 7}, + }, + {"input_tokens_details": {"cached_tokens": 4}}, + ), + ( + { + "output_tokens": 3, + "input_tokens_details": {"cached_tokens": 4}, + "output_tokens_details": {"reasoning_tokens": 2}, + }, + {"output_tokens_details": {"reasoning_tokens": 2}}, + ), + ], + ids=["no_counts", "input_count", "output_count"], + ) + def test_a_failed_request_keeps_usage_details_only_with_their_token_count(self, client, usage, kept): + client.response = FakeResponse(500, json.dumps({"usage": usage}).encode()) + with pytest.raises(EvaluationException) as raised: + run(DecisionModelRunner(DECISION_CONFIG, max_retries=0)) + assert raised.value.properties["decision_model"].get("usage_details") == kept + + @pytest.mark.parametrize( + "protocol, code", + [ + ("microsoft", "model_not_found"), + ("openai", "unknown_model"), + ("microsoft", "DeploymentNotFound"), + ("openai", "DeploymentNotFound"), + ], + ) + def test_other_error_codes_get_no_protocol_hint(self, client, protocol, code): + client.response = FakeResponse(404, json.dumps({"error": {"code": code}}).encode()) + with pytest.raises(EvaluationException) as raised: + run(DecisionModelRunner({**DECISION_CONFIG, "decision_protocol": protocol}, max_retries=0)) + assert str(raised.value) == f"(UserError) The decision model returned HTTP 404 ({code})." + @pytest.mark.parametrize( "detail,kind", [ @@ -946,6 +1017,7 @@ def completion_payload(probability, status=0.95): def strategy_with(payload, scoring, key="tool_call_accuracy", extra_properties=None): runner = AsyncMock(return_value=payload) runner.deployment = "Decision-1" + runner.decision_protocol = "microsoft" return DecisionEvaluationStrategy( runner, result_key=key, @@ -1296,6 +1368,7 @@ async def _do_eval(self, eval_input): } assert properties["model"] == "microsoft-decision-1" assert properties["decision_model"]["result_usage_scope"] == "reported_turns_only" + assert properties["decision_model"]["decision_protocol"] == "microsoft" assert "execution" not in properties["decision_model"] assert persisted["properties"] == properties turns = result["evaluation_per_turn"]["tool_call_accuracy_properties"] @@ -1453,6 +1526,331 @@ def test_chat_aggregation_is_unchanged(self): ] assert evaluator._aggregate_results(turns) == EvaluatorBase._aggregate_results(evaluator, turns) + def test_usage_details_are_summed_over_the_turns_that_reported_them(self): + evaluator = ProbeModelEvaluator(result_key="probe", model_config=DECISION_CONFIG) + first = {"input_tokens_details": {"cached_tokens": 4}, "output_tokens_details": {"reasoning_tokens": 2}} + # The second turn reports no completion tokens, so its reasoning counter is not summed. + second = { + "input_tokens_details": {"cached_tokens": 6, "cache_write_tokens": 1}, + "output_tokens_details": {"reasoning_tokens": 50}, + } + turns = [ + { + "probe": 1, + "probe_prompt_tokens": 12, + "probe_completion_tokens": 3, + "probe_properties": {"decision_model": {"usage_details": first}}, + }, + {"probe": 5, "probe_prompt_tokens": 34, "probe_properties": {"decision_model": {"usage_details": second}}}, + {"probe": 3, "probe_prompt_tokens": 7, "probe_properties": {"decision_model": {}}}, + ] + result = evaluator._aggregate_results(turns) + native = result["probe_properties"]["decision_model"] + assert native["usage_details"] == { + "input_tokens_details": {"cached_tokens": 10, "cache_write_tokens": 1}, + "output_tokens_details": {"reasoning_tokens": 2}, + } + assert native["result_usage_scope"] == "reported_turns_only" + assert result["probe_prompt_tokens"] == 53 + per_turn = result["evaluation_per_turn"]["probe_properties"] + assert [turn["decision_model"].get("usage_details") for turn in per_turn] == [first, second, None] + + def test_last_turn_usage_details_are_not_reported_as_conversation_totals(self): + evaluator = ProbeModelEvaluator(result_key="probe", model_config=DECISION_CONFIG) + invalid = {"input_tokens_details": {"cached_tokens": -1}, "output_tokens_details": {"reasoning_tokens": True}} + turns = [ + {"probe": 1, "probe_prompt_tokens": 12, "probe_properties": {"decision_model": {}}}, + {"probe": 5, "probe_prompt_tokens": 34, "probe_properties": {"decision_model": {"usage_details": invalid}}}, + ] + native = evaluator._aggregate_results(turns)["probe_properties"]["decision_model"] + assert "usage_details" not in native + + def test_usage_details_of_turns_without_counted_usage_are_not_summed(self): + evaluator = ProbeModelEvaluator(result_key="probe", model_config=DECISION_CONFIG) + counted = {"input_tokens_details": {"cached_tokens": 4}} + uncounted = {"input_tokens_details": {"cached_tokens": 90}, "output_tokens_details": {"reasoning_tokens": 9}} + turns = [ + {"probe": 1, "probe_prompt_tokens": 12, "probe_properties": {"decision_model": {"usage_details": counted}}}, + {"probe": 5, "probe_properties": {"decision_model": {"usage_details": uncounted}}}, + {"probe": 3, "probe_prompt_tokens": 7, "probe_properties": {"decision_model": {}}}, + ] + result = evaluator._aggregate_results(turns) + assert result["probe_prompt_tokens"] == 19 + assert result["probe_properties"]["decision_model"]["usage_details"] == counted + + +class TestDecisionEvaluationMetrics: + """Every Decision evaluation records one outcome count and one duration with the same bounded attributes.""" + + EVALUATIONS = "azure.ai.evaluation.decision.evaluations" + DURATION = "azure.ai.evaluation.decision.evaluation.duration" + + @pytest.fixture + def recorded(self): + records = [] + + class Instrument: + def __init__(self, name, **_): + self.name = name + + def add(self, value, attributes): + records.append((self.name, value, dict(attributes))) + + record = add + + meter = SimpleNamespace(create_counter=Instrument, create_histogram=Instrument) + with patch.dict(sys.modules, {"opentelemetry.metrics": SimpleNamespace(get_meter=lambda _: meter)}): + yield records + + def evaluation_attributes(self, records): + evaluations = [record for record in records if record[0] in (self.EVALUATIONS, self.DURATION)] + assert [name for name, _, _ in evaluations] == [self.EVALUATIONS, self.DURATION] + (_, count, attributes), (_, duration, duration_attributes) = evaluations + assert count == 1 and duration >= 0 and attributes == duration_attributes + assert set(attributes) == {"backend", "protocol", "metric", "outcome", "blame", "error_kind"} + return attributes + + def test_completed_evaluation(self, recorded): + result = asyncio.run(strategy_with(level_payload(4), LikertScoring(5)).evaluate({})) + assert result["tool_call_accuracy_status"] != "skipped" + assert self.evaluation_attributes(recorded) == { + "backend": "decision", + "protocol": "microsoft", + "metric": "tool_call_accuracy", + "outcome": "completed", + "blame": "none", + "error_kind": "none", + } + + def test_not_applicable_evaluation_is_skipped(self, recorded): + result = asyncio.run(strategy_with(level_payload(4, status=0.1), LikertScoring(5)).evaluate({})) + assert result["tool_call_accuracy_status"] == "skipped" + assert self.evaluation_attributes(recorded)["outcome"] == "skipped" + + def test_invalid_response_is_a_system_error(self, recorded): + response = level_payload(4) + response["answers"]["status"]["diagnostic"] = math.inf + with pytest.raises(EvaluationException): + asyncio.run(strategy_with(response, LikertScoring(5)).evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ("error", "system", "none") + + @pytest.mark.parametrize("protocol", ["microsoft", "openai"]) + def test_request_error_keeps_its_blame_and_protocol(self, recorded, protocol): + configured = strategy_with(level_payload(4), LikertScoring(5)) + configured.runner.side_effect = EvaluationException( + message="rejected", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.INVALID_VALUE, + target=ErrorTarget.EVALUATE, + ) + configured.runner.decision_protocol = protocol + with pytest.raises(EvaluationException): + asyncio.run(configured.evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["protocol"]) == ("error", "user", protocol) + + def test_refusal_is_a_user_error_with_its_kind(self, recorded): + response = level_payload(4) + response["answers"]["tool_call_accuracy"] = {"type": "refusal"} + with pytest.raises(DecisionRefusalError): + asyncio.run(strategy_with(response, LikertScoring(5)).evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ( + "error", + "user", + "model_refusal", + ) + + def test_content_filter_rejection_keeps_its_kind(self, recorded): + error = EvaluationException( + message="The decision model returned HTTP 400 (content_filter).", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.FAILED_EXECUTION, + target=ErrorTarget.EVALUATE, + ) + error.properties = {"decision_model": {"error_kind": "content_filter"}} + configured = strategy_with(level_payload(4), LikertScoring(5)) + configured.runner.side_effect = error + with pytest.raises(EvaluationException): + asyncio.run(configured.evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ( + "error", + "user", + "content_filter", + ) + + def test_cancellation_is_not_counted_as_an_error(self, recorded): + configured = strategy_with(level_payload(4), LikertScoring(5)) + configured.runner.side_effect = asyncio.CancelledError() + with pytest.raises(asyncio.CancelledError): + asyncio.run(configured.evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"]) == ("cancelled", "none") + + @pytest.mark.parametrize("kind", [*_model_runners.DECISION_ERROR_KINDS, "private text", None]) + def test_request_error_reports_only_a_bounded_error_kind(self, recorded, kind): + error = EvaluationException( + message="rejected", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.FAILED_EXECUTION, + target=ErrorTarget.EVALUATE, + ) + error.properties = {"decision_model": {"error_kind": kind}} + configured = strategy_with(level_payload(4), LikertScoring(5)) + configured.runner.side_effect = error + with pytest.raises(EvaluationException): + asyncio.run(configured.evaluate({})) + expected = kind if kind in _model_runners.DECISION_ERROR_KINDS else "none" + assert self.evaluation_attributes(recorded)["error_kind"] == expected + + def test_a_named_provider_rejection_reports_its_kind(self, client, recorded): + detail = "invalid model request: Question prompt exceeds the 32768-token context limit" + client.response = FakeResponse(422, json.dumps({"detail": detail}).encode()) + runner = DecisionModelRunner(DECISION_CONFIG, max_retries=0) + with pytest.raises(EvaluationException, match="context_window_exceeded"): + asyncio.run(self.runner_strategy(runner).evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ( + "error", + "user", + "context_window_exceeded", + ) + + @pytest.mark.parametrize( + "key, label", + [("coherence", "coherence"), ("deflection_rate", "deflection_rate"), ("customer_rubric_v2", "custom")], + ) + def test_metric_is_a_known_result_key_or_custom(self, recorded, key, label): + configured = strategy_with(level_payload(4), LikertScoring(5), key=key) + configured.runner.side_effect = asyncio.CancelledError() + with pytest.raises(asyncio.CancelledError): + asyncio.run(configured.evaluate({})) + assert self.evaluation_attributes(recorded)["metric"] == label + + @pytest.mark.parametrize( + "protocol, label", + [("absent", "microsoft"), ("future", "unknown"), (Mock(), "unknown")], + ids=["absent", "unrecognized", "not_a_name"], + ) + def test_runner_protocol_is_bounded_in_metrics_and_provenance(self, recorded, protocol, label): + configured = strategy_with(level_payload(4), LikertScoring(5)) + if protocol == "absent": + del configured.runner.decision_protocol + else: + configured.runner.decision_protocol = protocol + result = asyncio.run(configured.evaluate({})) + provenance = result["tool_call_accuracy_properties"]["decision_model"] + # decision_model.protocol keeps the route values that hosted evaluations write. + assert provenance["decision_protocol"] == label and "protocol" not in provenance + assert self.evaluation_attributes(recorded)["protocol"] == label + + def test_every_decision_evaluator_in_this_package_reports_its_result_key(self): + for module in pkgutil.walk_packages(evaluators_package.__path__, evaluators_package.__name__ + "."): + importlib.import_module(module.name) + + def subclasses(cls): + for subclass in cls.__subclasses__(): + yield subclass + yield from subclasses(subclass) + + keys = { + cls._RESULT_KEY + for cls in subclasses(ModelEvaluatorBase) + if getattr(cls, "_supports_decision_models", False) and isinstance(getattr(cls, "_RESULT_KEY", None), str) + } + assert len(keys) >= 15 + assert keys <= _decision_strategy._METRIC_NAMES + + @staticmethod + def runner_strategy(runner): + return DecisionEvaluationStrategy( + runner, + result_key="tool_call_accuracy", + error_target=ErrorTarget.TOOL_CALL_ACCURACY_EVALUATOR, + questions=QUESTIONS, + scoring=LikertScoring(5), + threshold=3, + timeout=5, + ) + + @pytest.mark.parametrize("summary_kept", [True, False], ids=["summary", "summary_over_bounds"]) + def test_callback_cancellation_is_not_counted_as_an_error(self, client, recorded, monkeypatch, summary_kept): + if not summary_kept: + # A summary beyond the hosted bounds, such as one after more than 20 retries, is omitted. + monkeypatch.setattr(_model_runners, "validated_execution", lambda _: None) + runner = DecisionModelRunner(DECISION_CONFIG, cancellation_requested=lambda: True) + with pytest.raises(EvaluationException, match="cancelled") as raised: + asyncio.run(self.runner_strategy(runner).evaluate({})) + # The exception is unchanged; only the evaluation metric reads the runner's cancellation marker. + assert raised.value.blame == ErrorBlame.USER_ERROR and not client.calls + execution = (raised.value.properties.get("decision_model") or {}).get("execution") or {} + assert execution.get("outcome") == ("cancelled" if summary_kept else None) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ("cancelled", "none", "none") + + def test_cancellation_callback_failure_stays_a_user_error(self, client, recorded): + def broken(): + raise RuntimeError("callback down") + + runner = DecisionModelRunner(DECISION_CONFIG, cancellation_requested=broken) + with pytest.raises(EvaluationException, match="cancellation callback failed"): + asyncio.run(self.runner_strategy(runner).evaluate({})) + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["blame"], attributes["error_kind"]) == ("error", "user", "none") + + def test_metric_sink_failure_does_not_change_the_result(self): + broken = SimpleNamespace(get_meter=Mock(side_effect=RuntimeError("sink down"))) + with patch.dict(sys.modules, {"opentelemetry.metrics": broken}): + result = asyncio.run(strategy_with(level_payload(4), LikertScoring(5)).evaluate({})) + assert result["tool_call_accuracy_status"] != "skipped" + broken.get_meter.assert_called() + + def test_broken_opentelemetry_import_does_not_change_the_result_or_error(self): + broken = ModuleType("opentelemetry.metrics") + + def unavailable(name): + raise RuntimeError(f"broken opentelemetry installation: {name}") + + broken.__getattr__ = unavailable + refused = level_payload(4) + refused["answers"]["tool_call_accuracy"] = {"type": "refusal"} + with patch.dict(sys.modules, {"opentelemetry.metrics": broken}): + result = asyncio.run(strategy_with(level_payload(4), LikertScoring(5)).evaluate({})) + with pytest.raises(DecisionRefusalError): + asyncio.run(strategy_with(refused, LikertScoring(5)).evaluate({})) + assert result["tool_call_accuracy_status"] != "skipped" + + def test_missing_opentelemetry_is_ignored(self): + with patch.dict(sys.modules, {"opentelemetry.metrics": None}): + result = asyncio.run(strategy_with(level_payload(4), LikertScoring(5)).evaluate({})) + assert result["tool_call_accuracy_status"] != "skipped" + + def test_public_evaluator_records_request_and_evaluation_metrics(self, client, recorded): + class Judge(ModelEvaluatorBase): + _supports_decision_models = True + + def _derive_singleton_inputs(self): + return [["query", "response"]] + + async def _do_eval(self, eval_input): + return await DecisionEvaluationStrategy( + self._decision_model_runner, + result_key="tool_call_accuracy", + error_target=ErrorTarget.TOOL_CALL_ACCURACY_EVALUATOR, + questions=QUESTIONS, + scoring=LikertScoring(5), + threshold=3, + timeout=self._LLM_CALL_TIMEOUT, + ).evaluate(eval_input) + + client.post = AsyncMock(return_value=FakeResponse(body=json.dumps(level_payload(4)).encode())) + Judge(result_key="tool_call_accuracy", model_config=DECISION_CONFIG)(query="q", response="r") + assert "azure.ai.evaluation.decision.operations" in [name for name, _, _ in recorded] + attributes = self.evaluation_attributes(recorded) + assert (attributes["outcome"], attributes["protocol"]) == ("completed", "microsoft") + class TestDecisionExecutionControls: @pytest.mark.parametrize( @@ -1543,7 +1941,9 @@ def test_final_http_error_keeps_classification_and_only_accepted_known_usage(sel assert validated_execution(execution) == execution assert execution["outcome"] == "error" if response_limit == 1: - assert raised.value.properties == {"decision_model": {"deployment": "Decision-1"}} + assert raised.value.properties == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"} + } else: assert raised.value.properties["total_tokens"] == 123 assert raised.value.properties["model"] == "microsoft-decision-1" @@ -1654,7 +2054,10 @@ def test_unknown_or_invalid_usage_is_not_fabricated(self, client): run(DecisionModelRunner(DECISION_CONFIG)) execution = raised.value.properties["decision_model"].pop("execution") assert execution["usage_state"] == "unknown" - assert raised.value.properties == {"decision_model": {"deployment": "Decision-1"}, "completion_tokens": 3} + assert raised.value.properties == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"}, + "completion_tokens": 3, + } def test_default_pipeline_owns_an_execution_observer(self, client): client.response = FakeResponse(body=json.dumps(level_payload(4)).encode()) @@ -1741,7 +2144,9 @@ def test_authentication_failure_is_sanitized_and_has_no_fabricated_usage(self, c execution = raised.value.properties["decision_model"].pop("execution") assert execution["attempt_count"] == 0 assert execution["usage_state"] == "unknown" - assert raised.value.properties == {"decision_model": {"deployment": "Decision-1"}} + assert raised.value.properties == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"} + } assert "secret" not in str(raised.value) assert not client.calls @@ -1878,7 +2283,9 @@ def cancel(): execution = raised.value.properties["decision_model"].pop("execution") assert execution["attempt_count"] == 1 assert execution["outcome"] == ("deadline" if termination == "timeout" else "cancelled") - assert raised.value.properties == {"decision_model": {"deployment": "Decision-1"}} + assert raised.value.properties == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"} + } finally: handle.cancel() @@ -1917,7 +2324,9 @@ async def send(self, request, **kwargs): assert execution["attempt_count"] == expected assert execution["retry_count"] == max_retries assert execution["outcome"] == "error" - assert raised.value.properties == {"decision_model": {"deployment": "Decision-1"}} + assert raised.value.properties == { + "decision_model": {"deployment": "Decision-1", "decision_protocol": "microsoft"} + } assert "secret" not in str(raised.value) def test_real_pipeline_success_has_no_hook_failure_or_billed_replay(self, monkeypatch): @@ -2078,6 +2487,7 @@ async def request(): assert set(attributes) == { "backend", "route", + "protocol", "phase", "outcome", "usage_state", diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_openai_decisions.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_openai_decisions.py new file mode 100644 index 000000000000..399b7f279601 --- /dev/null +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_openai_decisions.py @@ -0,0 +1,435 @@ +# --------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# --------------------------------------------------------- + +"""The OpenAI Decisions protocol on the shared Decision runner and scoring strategy.""" + +import asyncio +import copy +import json +import logging +import pickle +from logging.handlers import BufferingHandler +from unittest.mock import AsyncMock + +import pytest +from azure.core.pipeline.transport import AsyncHttpTransport, HttpResponse + +from azure.ai.evaluation._evaluate._batch_run._run_submitter_client import _decision_error_metadata +from azure.ai.evaluation._evaluators._common import _model_runners +from azure.ai.evaluation._evaluators._common._decision_protocols import OpenAIDecisionProtocol +from azure.ai.evaluation._evaluators._common._decision_strategy import ( + DecisionEvaluationStrategy, + DecisionRefusalError, + LikertScoring, +) +from azure.ai.evaluation._evaluators._common._model_runners import DecisionExecution, DecisionModelRunner +from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException +from azure.ai.evaluation._http_utils import AsyncHttpPipeline + +pytestmark = pytest.mark.unittest + +CONFIG = { + "type": "decision", + "decision_protocol": "openai", + "azure_endpoint": "https://test.openai.azure.com", + "azure_deployment": "judge", + "api_key": "test-only", +} +QUESTIONS = { + "status": {"type": "noul", "instructions": "Applicable?", "criteria": {"true": "Has evidence", "false": "Empty"}}, + "quality": {"type": "score", "instructions": "Quality?", "criteria": ["Bad", "Okay", "Good"]}, +} + + +def response(status=0.9, quality=None): + """An Azure OpenAI Decisions response; Azure returns the deployment alias as the model.""" + return { + "model": "judge", + "usage": { + "input_tokens": 12, + "output_tokens": 3, + "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 4, "cache_write_tokens": 2}, + "output_tokens_details": {"reasoning_tokens": 1}, + }, + "answers": [ + {"type": "predicate", "name": "status", "probability": status}, + quality + or { + "type": "score", + "name": "quality", + "confidence": 0.4, + "score": 0.8, + "probabilities": [ + {"value": 0, "label": "Level 1", "probability": 0.4}, + {"value": 1, "label": "Level 2", "probability": 0.4}, + {"value": 2, "label": "Level 3", "probability": 0.2}, + ], + }, + ], + } + + +def strategy(payload, config=CONFIG): + runner = DecisionModelRunner(config) + runner._post = AsyncMock(return_value=payload) + return DecisionEvaluationStrategy( + runner, + result_key="quality", + error_target=ErrorTarget.EVALUATE, + questions=QUESTIONS, + scoring=LikertScoring(3), + threshold=2, + timeout=30, + ) + + +def evaluate(payload): + return asyncio.run(strategy(payload).evaluate({"response": "Evidence"})) + + +class FakeResponse(HttpResponse): + def __init__(self, request, status_code, body): + super().__init__(request, None) + self.status_code = status_code + self.headers = {} + self._body = json.dumps(body).encode() + + def body(self): + return self._body + + +class FakeTransport(AsyncHttpTransport): + """Answers each request with the next ``(status, body)`` reply and records the requests.""" + + def __init__(self, *replies): + self.replies = list(replies) + self.requests = [] + + async def open(self): + pass + + async def close(self): + pass + + async def __aexit__(self, *args): + await self.close() + + async def send(self, request, **kwargs): + self.requests.append(request) + status, body = self.replies.pop(0) + return FakeResponse(request, status, body) + + +class TestRequest: + def test_encode_keeps_evidence_and_both_boolean_criteria(self): + request = OpenAIDecisionProtocol().encode("judge", {"response": "\u00e9vidence"}, QUESTIONS) + assert json.loads(request["input"]) == {"response": "\u00e9vidence"} + assert request["model"] == "judge" and "state" not in request + assert request["questions"][0] == { + "name": "status", + "type": "predicate", + "instructions": "Applicable?\n\nThe predicate is true when: Has evidence\nThe predicate is false when: Empty", + } + assert request["questions"][1]["levels"] == [ + {"label": f"Level {position}", "description": value} + for position, value in enumerate(["Bad", "Okay", "Good"], start=1) + ] + + @pytest.mark.parametrize("protocol", ["microsoft", "openai"]) + @pytest.mark.parametrize("state", [{"response": float("nan")}, {"response": object()}]) + def test_invalid_evidence_is_the_same_user_error_on_every_protocol(self, protocol, state): + endpoint = "https://test.services.ai.azure.com" if protocol == "microsoft" else CONFIG["azure_endpoint"] + runner = DecisionModelRunner({**CONFIG, "decision_protocol": protocol, "azure_endpoint": endpoint}) + runner._post = AsyncMock() + with pytest.raises(EvaluationException) as raised: + asyncio.run(runner(state=state, questions=QUESTIONS, timeout=30)) + assert str(raised.value).endswith(") Decision evidence must be finite JSON-serializable data.") + assert (raised.value.category, raised.value.blame) == (ErrorCategory.FAILED_EXECUTION, ErrorBlame.USER_ERROR) + runner._post.assert_not_awaited() + + @pytest.mark.parametrize( + "questions", + [ + {str(i): QUESTIONS["status"] for i in range(401)}, + {"q": {"type": "score", "instructions": "Score", "criteria": ["one"]}}, + {"q": {"type": "noul", "instructions": "Boolean", "criteria": {"true": "yes"}}}, + {"q": {"type": "noul", "instructions": " "}}, + ], + ) + def test_requests_the_protocol_cannot_express_fail_before_transport(self, questions): + runner = DecisionModelRunner(CONFIG) + runner._post = AsyncMock() + with pytest.raises(EvaluationException) as raised: + asyncio.run(runner(state="evidence", questions=questions, timeout=30)) + assert raised.value.category == ErrorCategory.INVALID_VALUE + runner._post.assert_not_awaited() + + @pytest.mark.parametrize( + "endpoint", + [ + "https://test.openai.azure.com", + "https://test.openai.azure.com/openai/", + "https://test.openai.azure.com/openai/v1/", + ], + ) + def test_endpoint_forms(self, endpoint): + runner = DecisionModelRunner({**CONFIG, "azure_endpoint": endpoint}) + assert runner._url == "https://test.openai.azure.com/openai/v1/decisions" + assert runner.decision_protocol == "openai" + + def test_selection_is_explicit_and_preserves_the_microsoft_default(self): + config = {key: value for key, value in CONFIG.items() if key != "decision_protocol"} + runner = DecisionModelRunner({**config, "azure_endpoint": "https://test.services.ai.azure.com/base"}) + assert runner._url == "https://test.services.ai.azure.com/base/providers/microsoft/v1/systemone" + assert runner.decision_protocol == "microsoft" + assert DecisionModelRunner.SUPPORTED_PROTOCOLS == frozenset({"microsoft", "openai"}) + for changes in ( + {"decision_protocol": "invalid"}, + {"decision_protocol": []}, + {"azure_endpoint": "https://test.services.ai.azure.com/api/projects/test"}, + ): + with pytest.raises(EvaluationException): + DecisionModelRunner({**CONFIG, **changes}) + + def test_byte_budgets_measure_the_canonical_request_on_every_protocol(self): + """Budgets bound what the model reads, so escaping evidence into OpenAI's text input admits no less evidence.""" + state = {"response": '"quoted"' * 20} + + def size(value): + return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) + + canonical = size({"model": "judge", "state": state, "questions": QUESTIONS}) + unit = max(size({"state": state, "question": question}) for question in QUESTIONS.values()) + assert size(OpenAIDecisionProtocol().encode("judge", state, QUESTIONS)) > canonical + for limits in ({"max_request_bytes": canonical - 1}, {"max_state_question_bytes": unit - 1}): + runner = DecisionModelRunner(CONFIG, **limits) + runner._post = AsyncMock() + with pytest.raises(EvaluationException, match="byte budget") as raised: + asyncio.run(runner(state=state, questions=QUESTIONS, timeout=30)) + assert raised.value.blame == ErrorBlame.USER_ERROR + runner._post.assert_not_awaited() + runner = DecisionModelRunner(CONFIG, max_request_bytes=canonical, max_state_question_bytes=unit) + runner._post = AsyncMock(return_value=response()) + asyncio.run(runner(state=state, questions=QUESTIONS, timeout=30)) + runner._post.assert_awaited_once() + + +class TestResponse: + def test_answers_bind_by_name_into_the_canonical_shape(self): + payload = response() + payload["answers"].reverse() + payload["answers"][0]["additive_field"] = "ignored" + runner = DecisionModelRunner(CONFIG) + runner._post = AsyncMock(return_value=payload) + result = asyncio.run(runner(state="evidence", questions=QUESTIONS, timeout=30)) + assert result["answers"] == { + "status": {"type": "noul", "noul": 0.9}, + "quality": { + "type": "score", + "score": 0.8, + "confidence": 0.4, + "probabilities": {"0": 0.4, "1": 0.4, "2": 0.2}, + }, + } + assert result["usage"] == {"input_tokens": 12, "output_tokens": 3} + assert result["usage_details"] == { + "input_tokens_details": {"cached_tokens": 4, "cache_write_tokens": 2}, + "output_tokens_details": {"reasoning_tokens": 1}, + } + + @pytest.mark.parametrize( + "mutation", + [ + lambda p: p.update(answers={"status": {"type": "noul", "noul": 0.9}}), + lambda p: p["answers"].pop(), + lambda p: p["answers"].append(dict(p["answers"][0])), + lambda p: p["answers"][0].update(name="unknown"), + lambda p: p["answers"][0].update(name=None), + lambda p: p["answers"][1].update(type="predicate"), + lambda p: p["answers"][0].update(type="choice"), + lambda p: p["answers"][0].update(type="noul"), + lambda p: p["answers"][1].update(type="score_answer"), + ], + ) + def test_answers_that_cannot_bind_to_the_questions_are_invalid(self, mutation): + payload = response() + mutation(payload) + runner = DecisionModelRunner(CONFIG) + runner._post = AsyncMock(return_value=payload) + with pytest.raises(EvaluationException, match="invalid response") as raised: + asyncio.run(runner(state="evidence", questions=QUESTIONS, timeout=30)) + assert raised.value.blame == ErrorBlame.SYSTEM_ERROR + assert raised.value.properties["total_tokens"] == 15 + + @pytest.mark.parametrize("kind", _model_runners.DECISION_ERROR_KINDS) + def test_failed_rows_keep_every_bounded_error_kind_and_both_protocol_fields(self, kind): + error = EvaluationException( + message="failed", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.FAILED_EXECUTION, + target=ErrorTarget.EVALUATE, + ) + native = {"error_kind": kind, "decision_protocol": "openai", "protocol": "shared_sdk"} + error.properties = {"decision_model": dict(native)} + assert _decision_error_metadata(error)["properties"]["decision_model"] == native + + def test_failed_rows_drop_unknown_error_kinds_and_protocols(self): + error = EvaluationException( + message="failed", + blame=ErrorBlame.USER_ERROR, + category=ErrorCategory.FAILED_EXECUTION, + target=ErrorTarget.EVALUATE, + ) + # decision_model.protocol keeps the hosted route vocabulary; the wire protocol is decision_protocol. + error.properties = { + "decision_model": {"error_kind": "private text", "decision_protocol": "custom", "protocol": "microsoft"} + } + assert _decision_error_metadata(error)["properties"]["decision_model"] == {} + + def test_the_microsoft_protocol_still_rejects_a_refusal_answer(self): + config = {key: value for key, value in CONFIG.items() if key != "decision_protocol"} + runner = DecisionModelRunner(config) + runner._post = AsyncMock( + return_value={ + "model": "Microsoft-Decision-1", + "usage": {"input_tokens": 1, "output_tokens": 1}, + "answers": {"status": {"type": "refusal"}, "quality": {"type": "refusal"}}, + } + ) + with pytest.raises(EvaluationException, match="invalid response"): + asyncio.run(runner(state="evidence", questions=QUESTIONS, timeout=30)) + + def test_transport_uses_the_decisions_route_and_key_and_retries(self, monkeypatch): + transport = FakeTransport((429, {"error": {"code": "RateLimitExceeded"}}), (200, response())) + monkeypatch.setattr(_model_runners, "get_async_http_client", lambda **_: AsyncHttpPipeline(transport=transport)) + result = asyncio.run(DecisionModelRunner(CONFIG, max_retries=1)(state="text", questions=QUESTIONS, timeout=30)) + assert len(transport.requests) == 2 + request = transport.requests[0] + assert request.url == "https://test.openai.azure.com/openai/v1/decisions" + assert request.headers["api-key"] == "test-only" and "Authorization" not in request.headers + assert json.loads(request.content) == OpenAIDecisionProtocol().encode("judge", "text", QUESTIONS) + assert result["answers"]["status"] == {"type": "noul", "noul": 0.9} + + def test_a_content_filter_rejection_is_the_callers_content_and_is_marked(self, monkeypatch): + transport = FakeTransport((400, {"error": {"code": "content_filter", "message": "private filter detail"}})) + monkeypatch.setattr(_model_runners, "get_async_http_client", lambda **_: AsyncHttpPipeline(transport=transport)) + with pytest.raises(EvaluationException, match=r"HTTP 400 \(content_filter\)") as raised: + asyncio.run(DecisionModelRunner(CONFIG)(state="text", questions=QUESTIONS, timeout=30)) + assert raised.value.blame == ErrorBlame.USER_ERROR + assert "private filter detail" not in str(raised.value) + persisted = _decision_error_metadata(raised.value)["properties"]["decision_model"] + assert persisted["error_kind"] == "content_filter" and persisted["decision_protocol"] == "openai" + assert persisted["execution"]["last_http_status"] == 400 + + def test_execution_observations_name_the_protocol(self): + handler = BufferingHandler(10) + logger = logging.Logger("decision-execution") + logger.addHandler(handler) + execution = DecisionExecution(logger, None, "openai") + execution.emit("execution", "completed", execution.summary("completed"), 1.0, "known") + dimensions = json.loads(handler.buffer[0].decision_dimensions) + assert dimensions["protocol"] == "openai" and dimensions["route"] == "shared_sdk" + + +class TestScoring: + def test_shared_strategy_keeps_argmax_ties_threshold_and_usage(self): + result = evaluate(response()) + assert result["quality"] == 1 and result["quality_passed"] is False + assert result["quality_total_tokens"] == 15 and result["quality_model"] == "judge" + provenance = result["quality_properties"]["decision_model"] + assert provenance["decision_protocol"] == "openai" + assert provenance["native_status_answer"] == {"type": "noul", "noul": 0.9} + assert provenance["native_answer"] == { + "type": "score", + "score": 0.8, + "confidence": 0.4, + "probabilities": {"0": 0.4, "1": 0.4, "2": 0.2}, + } + assert provenance["usage_details"]["input_tokens_details"]["cached_tokens"] == 4 + + def test_rounded_probabilities_use_the_shared_tolerance(self): + quality = { + "type": "score", + "name": "quality", + "confidence": 0.33, + "score": 1.0, + "probabilities": [{"value": i, "label": str(i), "probability": 0.33} for i in range(3)], + } + assert evaluate(response(quality=quality))["quality"] == 1 + + @pytest.mark.parametrize( + "quality", + [ + {"type": "refusal", "name": "quality"}, + {"type": "score", "name": "quality", "score": 9, "confidence": 2, "probabilities": "invalid"}, + { + "type": "score", + "name": "quality", + "score": 0, + "confidence": 1, + "probabilities": [{"value": 0, "probability": 1}, {"value": 0, "probability": 0}], + }, + { + "type": "score", + "name": "quality", + "score": 0.8, + "confidence": 0.4, + "probabilities": {"0": 0.2, "1": 0.8, "2": 0.0}, + }, + ], + ) + def test_a_skipped_evaluation_never_reads_its_grade(self, quality): + result = evaluate(response(status=0.1, quality=quality)) + assert result["quality_status"] == "skipped" and result["quality"] is None + assert "native_answer" not in result["quality_properties"]["decision_model"] + + @pytest.mark.parametrize("refused", ["status", "quality"]) + def test_a_refused_answer_that_is_read_fails_without_a_score(self, refused): + payload = response() + index = 0 if refused == "status" else 1 + payload["answers"][index] = {"type": "refusal", "name": refused} + with pytest.raises(DecisionRefusalError, match=f"refused to answer question '{refused}'") as raised: + evaluate(payload) + error = raised.value + assert error.blame == ErrorBlame.USER_ERROR and error.category == ErrorCategory.FAILED_EXECUTION + assert error.properties["total_tokens"] == 15 + provenance = error.properties["decision_model"] + assert provenance["decision_protocol"] == "openai" and provenance["error_kind"] == "model_refusal" + persisted = _decision_error_metadata(error) + assert persisted["blame"] == "UserError" + assert persisted["properties"]["decision_model"]["error_kind"] == "model_refusal" + assert persisted["properties"]["decision_model"]["decision_protocol"] == "openai" + assert persisted["properties"]["decision_model"]["usage_details"]["input_tokens_details"]["cached_tokens"] == 4 + + def test_a_refusal_survives_copies_and_pickles_with_its_metadata(self): + payload = response() + payload["answers"][1] = {"type": "refusal", "name": "quality"} + with pytest.raises(DecisionRefusalError) as raised: + evaluate(payload) + for copied in (copy.deepcopy(raised.value), pickle.loads(pickle.dumps(raised.value))): + assert isinstance(copied, DecisionRefusalError) and str(copied) == str(raised.value) + assert copied.properties == raised.value.properties and copied.target == raised.value.target + + @pytest.mark.parametrize( + "probabilities", + [ + [{"value": 0, "probability": 1}, {"value": 0, "probability": 0}], + {"0": 0.2, "1": 0.8, "2": 0.0}, + [{"value": True, "probability": 1}, {"value": 1, "probability": 0}, {"value": 2, "probability": 0}], + ], + ids=["duplicate_level", "canonical_map", "boolean_level"], + ) + def test_a_malformed_grade_that_is_read_is_an_invalid_response(self, probabilities): + quality = { + "type": "score", + "name": "quality", + "score": 0, + "confidence": 1, + "probabilities": probabilities, + } + with pytest.raises(EvaluationException, match="invalid response") as raised: + evaluate(response(quality=quality)) + assert not isinstance(raised.value, DecisionRefusalError) + assert raised.value.blame == ErrorBlame.SYSTEM_ERROR