"""Core evaluation pipeline — evaluator dispatch, retry, parallel execution.""" import json import os import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, Iterator, List, Optional, Tuple from azure.ai.evaluation import ( AzureOpenAIModelConfiguration, RelevanceEvaluator, CoherenceEvaluator, GroundednessEvaluator, SimilarityEvaluator, ToolCallAccuracyEvaluator, ) from api_clients.base_agent_client import BaseAgentClient from cli_logging.cli_logger import emit_structured_log from cli_logging.logging_utils import Operation from common import ( RELEVANCE, COHERENCE, GROUNDEDNESS, SIMILARITY, TOOL_CALL_ACCURACY, CITATIONS, EXACT_MATCH, PARTIAL_MATCH, RETRIEVAL_QUERY, RETRIEVAL_RESULT, METRIC_IDS, MAX_CONCURRENCY, DEFAULT_PASS_THRESHOLD, DEFAULT_NON_LLM_PASS_THRESHOLD, STATUS_PASS, STATUS_FAIL, STATUS_ERROR, STATUS_PARTIAL, MAX_TURNS_PER_THREAD, LONG_THREAD_WARNING_THRESHOLD, RunConfig, resolve_max_attempts, ) from error_messages import ( agent_request_failed, evaluation_crashed, evaluator_error, evaluator_failed, turn_skipped, AGENT_REQUEST_ERROR_CODES, EVAL_ERROR_CODES, ERROR_CODE_AGENT_REQUEST_FAILED, ERROR_CODE_AGENT_AUTH, ERROR_CODE_AGENT_RATE_LIMITED, ERROR_CODE_AGENT_TIMEOUT, ERROR_CODE_AGENT_RESPONSE_UNPARSEABLE, EVAL_ERROR_CODE_EMPTY_AGENT_RESPONSE, EVAL_ERROR_CODE_JUDGE_UNPARSEABLE, EVAL_ERROR_CODE_JUDGE_AUTH, EVAL_ERROR_CODE_JUDGE_RATE_LIMITED, EVAL_ERROR_CODE_JUDGE_TIMEOUT, EVAL_ERROR_CODE_EVALUATOR_CONFIG, EVAL_ERROR_CODE_EVALUATOR_LOAD, EVAL_ERROR_CODE_INVALID_RESULT, EVAL_ERROR_CODE_EVALUATOR_ERROR, ) from judge_backend import JudgeBackend from status_derivation import rollup_thread_status, status_for_response from custom_evaluators.citations_evaluator import CitationsEvaluator, CitationFormat from custom_evaluators.exact_match_evaluator import ExactMatchEvaluator from custom_evaluators.partial_match_evaluator import PartialMatchEvaluator from custom_evaluators.retrieval import ( RetrievalQueryEvaluator, RetrievalResultEvaluator, sanitize_retrieval_diagnostics, ) from evaluator_resolver import ( EVALUATOR_REGISTRY, get_custom_evaluator_spec, validate_evaluator_names, validate_evaluator_options, resolve_evaluators_for_prompt, get_evaluator_threshold, ) from custom_evaluators.discovery import CustomEvaluatorDiscoveryError from api_clients.A2A.constants import AcceptedOutputMode from parallel_executor import execute_in_parallel from response_extractor import ( build_enhanced_response_from_eval_item, get_response_text_for_evaluation, get_retrieval_telemetry_for_evaluation, ) from retry_policy import ( is_retryable_status, is_timeout_error, get_backoff_seconds, get_retry_after_seconds, ) from throttle_gate import ThrottleGate from auth.azure_ai_auth_handler import wrap_credential_error from foundry_cloud_evaluator import ( FoundryCloudEvaluator, FoundryEvalCollector, FOUNDRY_LLM_EVALUATORS, ) _RETRIEVAL_EVALUATOR_NAMES = frozenset({RETRIEVAL_QUERY, RETRIEVAL_RESULT}) # Truthy values for the diagnostics kill-switch env var (case-insensitive). _DIAGNOSTICS_SUPPRESS_VALUES = frozenset({"1", "true"}) def _diagnostics_suppressed() -> bool: """Return True when ``EVALS_SUPPRESS_DIAGNOSTICS`` is set to a truthy value.""" value = os.environ.get("EVALS_SUPPRESS_DIAGNOSTICS") return bool(value) and value.strip().lower() in _DIAGNOSTICS_SUPPRESS_VALUES def _build_diagnostics( resolved_evaluators: Dict[str, Any], retrieval_telemetry: Optional[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: """Build the extensible ``diagnostics`` container for a prompt/turn. Returns the container (``{"retrieval_executions": [...]}``) when the resolved evaluator list contains a retrieval evaluator and diagnostics are not suppressed, otherwise ``None``. Future diagnostic categories merge their own sibling keys into the returned dict. """ if _diagnostics_suppressed(): return None if not _RETRIEVAL_EVALUATOR_NAMES.intersection(resolved_evaluators): return None diagnostics: Dict[str, Any] = {} diagnostics.update(sanitize_retrieval_diagnostics(retrieval_telemetry)) return diagnostics def _truncate_prompt(prompt: str, limit: int = 80) -> str: """Collapse whitespace and cap a prompt for single-line log readability.""" text = " ".join((prompt or "").split()) if len(text) > limit: return text[: limit - 3].rstrip() + "..." return text def _attempts_phrase(n: int) -> str: """Return '1 attempt' / 'N attempts' with correct pluralization.""" return f"{n} attempt" if n == 1 else f"{n} attempts" def _resolve_accepted_output_modes(resolved_evaluators: Dict[str, Any]) -> List[AcceptedOutputMode]: """Return the A2A output modes this evaluator set needs. Used at send time to populate ``configuration.acceptedOutputModes``. Returns an empty list when no opt-in artifacts are required (default prompts stay on the lighter response payload path). Extending for future diagnostic artifacts: add another guarded ``append`` here that activates the matching :class:`AcceptedOutputMode` member. List ordering is significant; duplicates are removed downstream by the A2A client. """ modes: List[AcceptedOutputMode] = [] if any(name in _RETRIEVAL_EVALUATOR_NAMES for name in resolved_evaluators): modes.append(AcceptedOutputMode.RETRIEVAL) return modes @dataclass class PipelineConfig: """Runtime configuration for the evaluation pipeline.""" agent_client: Optional[BaseAgentClient] model_config: AzureOpenAIModelConfiguration has_azure_openai: bool default_evaluators: Dict[str, Any] selected_auth_mode: str auth_selection_source: str judge_backend: Optional[JudgeBackend] = None foundry_evaluator: Optional[FoundryCloudEvaluator] = None chat_gate: ThrottleGate = field(default_factory=lambda: ThrottleGate("chat_api")) is_retryable_status: Any = field(default=is_retryable_status) get_backoff_seconds: Any = field(default=get_backoff_seconds) class ItemType(Enum): SINGLE_TURN = "single_turn" MULTI_TURN = "multi_turn" class PipelineMode(Enum): FULL = "full" EVALUATE_ONLY = "evaluate_only" @dataclass class SingleTurnAcquisition: """Response acquisition outcome for one single-turn item.""" response: Optional[Dict[str, Any]] = None terminal_result: Optional[Dict[str, Any]] = None @dataclass class MultiTurnAcquisition: """Responses acquired for a multi-turn item.""" turns: List[Dict[str, Any]] conversation_id: str = "" def detect_item_type(item: dict) -> ItemType: """Determine if an evaluation item is single-turn or multi-turn. Returns ItemType.SINGLE_TURN if item has 'prompt' without 'turns', ItemType.MULTI_TURN if item has 'turns' array. Raises ValueError for invalid items (both, neither, or invalid turns). """ has_turns = "turns" in item has_prompt = "prompt" in item if has_turns and has_prompt: raise ValueError( "Invalid evaluation item: cannot have both 'turns' and 'prompt'. " "Use 'turns' for multi-turn threads or 'prompt' for single-turn." ) if has_turns and not isinstance(item["turns"], list): raise ValueError("Invalid evaluation item: 'turns' must be a list") if has_turns: if len(item["turns"]) == 0: raise ValueError("Invalid multi-turn thread: 'turns' array cannot be empty") return ItemType.MULTI_TURN if has_prompt: return ItemType.SINGLE_TURN raise ValueError( "Invalid evaluation item: must have either 'turns' array (multi-turn) " "or 'prompt' field (single-turn)" ) def _invoke_custom_evaluator( *, eval_name: str, eval_options: Dict[str, Any], threshold: Optional[float], prompt: str, actual_response: str, expected_response: str, context: str, model_config: AzureOpenAIModelConfiguration, is_llm: bool = True, ) -> Dict[str, Any]: """Instantiate and invoke a discovered custom evaluator. The user's ``.py`` class is instantiated with ``threshold`` and ``options`` (plus ``model_config`` for LLM-judge evaluators only), then called with the eval-document fields. The framework owns the ``result`` (derived from ``score >= threshold``) and ``threshold`` of the returned metric; the author returns just ``{"score", "reason"}`` on success, or signals failure by raising or returning ``{"error": ""}``. ``is_llm`` is a hint from the registry; the authoritative kind is ``spec.is_llm`` (resolved during load) and overrides it. The kind selects: * LLM-judge: constructed with ``model_config``; score validated on the strict 1-5 integer scale. * Code-only / non-LLM: constructed WITHOUT ``model_config``; score may be any number (no 1-5 clamp). The evaluator is imported lazily on first use (FR-004). If the import or class resolution fails (syntax error, missing dependency, no top-level class), the failure is returned as an inline ``error`` result per FR-026 rather than crashing the run. """ try: spec = get_custom_evaluator_spec(eval_name) except CustomEvaluatorDiscoveryError as exc: entry = dict(evaluator_error(EVAL_ERROR_CODE_EVALUATOR_LOAD, evaluator_failed(str(exc)))) entry["threshold"] = threshold return entry # spec.is_llm is the authoritative kind, resolved during load. Prefer it # over the registry-derived `is_llm` argument, which can be stale if a # `.prompty` was added/removed between enumeration and this first load. is_llm = spec.is_llm options_payload = eval_options.get("options") or {} if is_llm: evaluator = spec.user_class( model_config=model_config, threshold=threshold, options=options_payload, ) else: # Code-only evaluators never receive model_config. evaluator = spec.user_class( threshold=threshold, options=options_payload, ) # Pass all eval-document fields; the evaluator's __call__ signature # determines which it actually consumes. raw_result = evaluator( prompt=prompt, response=actual_response, expected_response=expected_response, context=context, ) return _validate_custom_evaluator_result( eval_name, raw_result, threshold, relaxed=not is_llm ) def _validate_custom_evaluator_result( eval_name: str, result: Any, threshold: Optional[float], *, relaxed: bool = False, ) -> Dict[str, Any]: """Validate the dict returned by a custom evaluator's ``__call__``. Customer-facing contract (documented in docs/custom-evaluators/README.md). On success the author returns just:: { "score": , "reason": "" } The framework derives ``result`` (``"pass"`` when ``score >= threshold``, else ``"fail"``) and attaches ``threshold`` — the author never returns either. To signal a failure the author raises, or returns:: { "error": "" } which becomes ``{"result": "error", "error": ..., "threshold": ...}``. For backward compatibility a returned ``{"result": "error", ...}`` is still accepted, and a returned ``result``/``threshold`` on success is ignored (the framework's derived values win). Score rules depend on ``relaxed``: * ``relaxed=False`` (LLM-judge): ``score`` MUST be an integer in ``[1, 5]`` (floats that are mathematically integers are normalized). This keeps the custom LLM scale consistent with the built-in evaluators. * ``relaxed=True`` (code-only / non-LLM): ``score`` may be any real number on a higher-is-better scale; no integer-only or 1-5 range constraint. Booleans are still rejected — use ``0``/``1`` integers, not ``True``/ ``False``. Anything that doesn't conform — missing/invalid score, non-dict return — produces an error result so the per-item loop can mark the evaluator as errored without crashing the run (per FR-026). """ # Kind-aware fallback: a None threshold defaults to 1 for code-only # (relaxed) evaluators and to the 1-5 LLM default otherwise, so error # metadata stays consistent with the evaluator's scale. default_threshold = ( DEFAULT_NON_LLM_PASS_THRESHOLD if relaxed else DEFAULT_PASS_THRESHOLD ) threshold_for_error = threshold if threshold is not None else default_threshold if not isinstance(result, dict): return { **evaluator_error( EVAL_ERROR_CODE_INVALID_RESULT, ( f"Custom evaluator '{eval_name}' returned {type(result).__name__} " "instead of a dict. Return a dict shaped like " '{"score": , "reason": "..."} on success, or ' '{"error": "..."} to signal a failure.' ), ), "threshold": threshold_for_error, } # Dedicated error channel: a returned {"error": ...} (or the legacy # {"result": "error", ...}) marks the evaluator as errored. The author owns # the message; the framework attaches result/threshold. if "error" in result or result.get("result") == STATUS_ERROR: message = result.get("error") or f"Custom evaluator '{eval_name}' signalled an error." return { **evaluator_error(EVAL_ERROR_CODE_EVALUATOR_ERROR, message), "threshold": threshold_for_error, } if "score" not in result: return { **evaluator_error( EVAL_ERROR_CODE_INVALID_RESULT, ( f"Custom evaluator '{eval_name}' result is missing the " "required 'score' key. Return a dict shaped like " '{"score": , "reason": "..."} on success, or ' '{"error": "..."} to signal a failure.' ), ), "threshold": threshold_for_error, } score = result["score"] if isinstance(score, bool) or not isinstance(score, (int, float)): expected = "a number" if relaxed else "integer in [1, 5]" return { **evaluator_error( EVAL_ERROR_CODE_INVALID_RESULT, ( f"Custom evaluator '{eval_name}' returned non-numeric score " f"{score!r}; expected {expected}." ), ), "threshold": threshold_for_error, } if relaxed: # Code-only evaluators may use any numeric score. Normalize a # mathematically-integer float to int for cleaner output. score_out: Any = int(score) if isinstance(score, float) and score.is_integer() else score else: if isinstance(score, float) and not score.is_integer(): return { **evaluator_error( EVAL_ERROR_CODE_INVALID_RESULT, ( f"Custom evaluator '{eval_name}' returned non-integer score " f"{score}; expected integer in [1, 5]." ), ), "threshold": threshold_for_error, } if not (1 <= score <= 5): return { **evaluator_error( EVAL_ERROR_CODE_INVALID_RESULT, ( f"Custom evaluator '{eval_name}' returned score {score} " "outside the allowed range [1, 5]. Adjust the evaluator's " "scoring rubric to use the 1-5 LLM scale." ), ), "threshold": threshold_for_error, } score_out = int(score) # Framework owns result + threshold. Derive pass/fail from score vs the # resolved threshold (consistent with built-in evaluators), and drop any # author-supplied "score"/"result"/"threshold" so they can't diverge. pass_threshold = threshold if threshold is not None else default_threshold rewritten = { k: v for k, v in result.items() if k not in ("score", "result", "threshold") } rewritten[eval_name] = score_out rewritten["threshold"] = pass_threshold rewritten["result"] = STATUS_PASS if score_out >= pass_threshold else STATUS_FAIL return rewritten def _decorate_metric(metric_id: str, data, threshold: Optional[float] = None) -> Dict[str, Any]: """Augment raw evaluator output with standardized threshold + pass/fail result. Raises ValueError if the SDK returned a malformed result (no numeric score under ``metric_id``). The outer try/except in :func:`_run_evaluators_for_item` catches it and emits a standard ``evaluator_failed`` error entry. """ pass_threshold = threshold if threshold is not None else DEFAULT_PASS_THRESHOLD payload = {} if isinstance(data, dict): payload.update(data) else: payload['raw'] = data score_val = None if isinstance(data, dict): if metric_id in data: score_val = data[metric_id] if not isinstance(score_val, (int, float)): raise ValueError( f"non-numeric score from evaluator (metric_id={metric_id!r}, score={score_val!r})" ) payload['threshold'] = pass_threshold payload['result'] = STATUS_PASS if score_val >= pass_threshold else STATUS_FAIL return payload # Evaluators that require a non-empty agent response to produce a meaningful # score. Retrieval evaluators are excluded — they score retrieval telemetry and # can be meaningful even when the rendered answer text is empty. _RESPONSE_REQUIRED_EVALUATORS = frozenset( { RELEVANCE, COHERENCE, GROUNDEDNESS, SIMILARITY, TOOL_CALL_ACCURACY, CITATIONS, EXACT_MATCH, PARTIAL_MATCH, } ) def _is_empty_response(text: Any) -> bool: """True when the agent response is None, empty, or whitespace-only.""" return text is None or (isinstance(text, str) and text.strip() == "") def _exception_chain(exc: Optional[BaseException]) -> "Iterator[BaseException]": """Yield ``exc`` and every exception in its ``__cause__``/``__context__`` chain, guarding against cycles. Classification walks the whole chain because subsystems wrap the original failure (e.g. an SDK ``RateLimitError`` chained via ``from`` into our own exception), so the discriminating signal is often not on the outermost object. """ seen: set[int] = set() current: Optional[BaseException] = exc while current is not None and id(current) not in seen: seen.add(id(current)) yield current current = current.__cause__ or current.__context__ def _http_status_from_exc(exc: BaseException) -> Optional[int]: """Best-effort extraction of an HTTP status code from an exception chain. Reads the integer status that HTTP SDK exceptions expose under a variety of attribute names — OpenAI ``status_code``, Azure ``status_code``, urllib ``HTTPError.code``, and a nested ``response.status_code`` (requests-style). This is **type/attribute based**: it never parses the human-readable message, so it cannot be thrown off by incidental wording. """ for e in _exception_chain(exc): for attr in ("status_code", "http_status", "status", "code"): val = getattr(e, attr, None) if isinstance(val, bool): continue if isinstance(val, int) and 100 <= val <= 599: return val resp = getattr(e, "response", None) if resp is not None: for attr in ("status_code", "status"): val = getattr(resp, attr, None) if isinstance(val, int) and 100 <= val <= 599: return val return None def _exc_type_names(exc: BaseException) -> "frozenset[str]": """Return the lowercased class names across the exception chain's MRO. Lets us recognise well-known third-party exception **types** (OpenAI ``RateLimitError`` / ``APITimeoutError`` / ``AuthenticationError``, Azure ``ClientAuthenticationError``, etc.) without importing those optional SDKs, and — crucially — without matching on the human-readable message, which is what caused prior misclassification. Base classes are included so a subclass like ``APITimeoutError`` (whose name contains "timeout") is caught even when the concrete leaf type is unknown to us. """ names: set[str] = set() for e in _exception_chain(exc): for klass in type(e).__mro__: names.add(klass.__name__.lower()) return frozenset(names) class FailureSignal(Enum): """A subsystem-neutral failure category detected from an exception's *structure* — its type (across the ``__cause__``/``__context__`` chain) and any HTTP status attribute — never from its human-readable message. Both the evaluator and the agent classifier share the same structural detection (:func:`_detect_failure_signal`) and then map the signal into their own code taxonomy (``judge*`` / ``agent*``). This keeps the deterministic type/status logic in exactly one place while preserving the two distinct code namespaces. :data:`NONE` means no structural signal was found, so the caller falls back to its subsystem catch-all code (``evaluatorError`` / ``agentRequestFailed``). """ TIMEOUT = "timeout" RATE_LIMITED = "rate_limited" AUTHENTICATION = "authentication" UNPARSEABLE = "unparseable" NONE = "none" def _detect_failure_signal(exc: BaseException) -> FailureSignal: """Detect a subsystem-neutral :class:`FailureSignal` from an exception's type and HTTP status, deterministically and independent of its message. Checks are mutually exclusive in practice (an exception is not both a JSON-decode error and a timeout), so the order only formalises precedence for pathological chains: unparseable → rate-limited → timeout → authentication. Returns :data:`FailureSignal.NONE` when nothing structural matches. """ type_names = _exc_type_names(exc) if isinstance(exc, json.JSONDecodeError) or "jsondecodeerror" in type_names: return FailureSignal.UNPARSEABLE if any("ratelimit" in n for n in type_names): return FailureSignal.RATE_LIMITED if is_timeout_error(exc) or any("timeout" in n for n in type_names): return FailureSignal.TIMEOUT if any("authentication" in n or "permissiondenied" in n for n in type_names): return FailureSignal.AUTHENTICATION status = _http_status_from_exc(exc) if status == 429: return FailureSignal.RATE_LIMITED if status in (401, 403): return FailureSignal.AUTHENTICATION return FailureSignal.NONE # Per-subsystem mapping from a structural signal to that subsystem's taxonomy # code. UNPARSEABLE maps to the judge's *output* vs the agent's *response* # variant, honoring the originating-subsystem naming. _EVALUATOR_SIGNAL_CODES: "dict[FailureSignal, str]" = { FailureSignal.TIMEOUT: EVAL_ERROR_CODE_JUDGE_TIMEOUT, FailureSignal.RATE_LIMITED: EVAL_ERROR_CODE_JUDGE_RATE_LIMITED, FailureSignal.AUTHENTICATION: EVAL_ERROR_CODE_JUDGE_AUTH, FailureSignal.UNPARSEABLE: EVAL_ERROR_CODE_JUDGE_UNPARSEABLE, } _AGENT_SIGNAL_CODES: "dict[FailureSignal, str]" = { FailureSignal.TIMEOUT: ERROR_CODE_AGENT_TIMEOUT, FailureSignal.RATE_LIMITED: ERROR_CODE_AGENT_RATE_LIMITED, FailureSignal.AUTHENTICATION: ERROR_CODE_AGENT_AUTH, FailureSignal.UNPARSEABLE: ERROR_CODE_AGENT_RESPONSE_UNPARSEABLE, } def _classify_evaluator_exception(exc: BaseException) -> str: """Map an evaluator exception to a per-evaluator error code. Naming honors the originating subsystem so a reader knows where to look: ``agent*`` = the WorkIQ agent, ``judge*`` = the LLM judge / model backend, ``evaluator*`` = the evaluator framework or custom evaluator code. Classification is **deterministic and layered**, from most to least trustworthy signal — a lower layer is consulted only when the layer above it is silent: 1. **Structured code** — our own typed exceptions (:class:`JudgeError`) carry a validated ``code`` set at the raise site; it is authoritative. 2. **Structural signal** — :func:`_detect_failure_signal` inspects the exception's type (across the cause chain) and any HTTP status, yielding a subsystem-neutral :class:`FailureSignal` that is mapped to the ``judge*`` code via :data:`_EVALUATOR_SIGNAL_CODES` (``UNPARSEABLE`` → ``judgeOutputUnparseable``, ``RATE_LIMITED`` → ``judgeRateLimited``, ``TIMEOUT`` → ``judgeTimeout``, ``AUTHENTICATION`` → ``judgeAuthenticationError``). Anything unmatched → ``evaluatorError``; the full exception message is preserved in the errored entry's ``error`` field to explain the cause. We deliberately do **not** guess from message text (e.g. substring-matching ``"429"`` or ``"unauthorized"``), which is prone to false positives. (Load and configuration failures are categorized at their call sites, not here.) """ # 1. Authoritative structured code from a typed exception we raised. code = getattr(exc, "code", None) if isinstance(code, str) and code in EVAL_ERROR_CODES: return code # 2. Deterministic structural signal (exception type + HTTP status). signal = _detect_failure_signal(exc) if signal is not FailureSignal.NONE: return _EVALUATOR_SIGNAL_CODES[signal] return EVAL_ERROR_CODE_EVALUATOR_ERROR def _classify_agent_exception(exc: BaseException) -> str: """Map an agent (WorkIQ) request failure to a turn-level ``ErrorObject`` code. Mirrors :func:`_classify_evaluator_exception`'s layered, deterministic design and shares the same structural detector, so the outcome does not depend on incidental message wording: 1. **Structured code** — an :class:`AgentRequestError` we raised carries a validated ``code`` set where the HTTP status / exception type / JSON-RPC error was known; it is authoritative. Every response-parsing failure the A2A client can produce (non-JSON body, missing ``result``, JSON-RPC error, unexpected result kind) already raises with ``agentResponseUnparseable`` here. 2. **Structural signal** — :func:`_detect_failure_signal` (exception type + HTTP status across the cause chain) mapped to the ``agent*`` code via :data:`_AGENT_SIGNAL_CODES` (``TIMEOUT`` → ``agentTimeout``, ``RATE_LIMITED`` → ``agentRateLimited``, ``AUTHENTICATION`` → ``agentAuthenticationError``, ``UNPARSEABLE`` → ``agentResponseUnparseable``). Anything unmatched → ``agentRequestFailed`` (catch-all); the full exception message is preserved in the ``ErrorObject.message`` to explain the cause. We deliberately do **not** guess from message text, which is prone to false positives (e.g. ``"429"`` inside a request id, ``"unauthorized"`` inside an unrelated filesystem error). """ # 1. Authoritative structured code from a typed exception we raised. code = getattr(exc, "code", None) if isinstance(code, str) and code in AGENT_REQUEST_ERROR_CODES: return code # 2. Deterministic structural signal (exception type + HTTP status). signal = _detect_failure_signal(exc) if signal is not FailureSignal.NONE: return _AGENT_SIGNAL_CODES[signal] return ERROR_CODE_AGENT_REQUEST_FAILED def _errored_entry(code: str, detail: str, threshold: Optional[float]) -> Dict[str, Any]: """Structured per-evaluator errored entry with the transient ``threshold``. ``detail`` is a raw exception fragment, so it is wrapped with :func:`evaluator_failed` to get the ``"Evaluator failed: "`` prefix. """ entry = dict(evaluator_error(code, evaluator_failed(detail))) entry["threshold"] = threshold return entry def _run_evaluators_for_item( prompt: str, actual_response: str, expected_response: str, enhanced_response: Dict[str, Any], resolved_evaluators: Dict[str, Any], model_config: AzureOpenAIModelConfiguration, selected_auth_mode: str, auth_selection_source: str, has_azure_openai: bool = True, context_label: str = "", item_context: str = "", judge_backend: Optional[JudgeBackend] = None, foundry_collector: Optional[Any] = None, ) -> Tuple[Dict[str, Dict[str, Any]], List[str], Optional[Dict[str, Any]]]: """Run resolved evaluators against a single item/turn. When a judge_backend is provided, LLM evaluators (Relevance, Coherence, Groundedness, Similarity) are routed through it instead of using the Azure AI Evaluation SDK evaluators directly. When a foundry_collector is provided (Foundry cloud evaluation mode), those same LLM evaluators are deferred: a pending placeholder is recorded and the row is registered with the collector for a later batch cloud run. Non-LLM evaluators always run locally. Each value in results_dict is a decorated metric dict on success or an errored entry ``{result: "error", error: "Evaluator failed: ", threshold}`` on crash. The ``threshold`` is included on errored entries so the aggregate report can still display it; the persisted ErroredScore shape strips it out at write time (see ``_as_errored_score`` in result_writer). Returns ``(results_dict, evaluators_ran, retrieval_telemetry)``. The normalized ``retrieval_telemetry`` (or ``None``) is returned so callers can build the output ``diagnostics`` container without re-extracting it. """ results_dict: Dict[str, Dict[str, Any]] = {} evaluators_ran: List[str] = [] retrieval_telemetry = get_retrieval_telemetry_for_evaluation(enhanced_response) # LLM evaluator names that can be routed through the judge backend _JUDGE_ROUTABLE = {RELEVANCE, COHERENCE, GROUNDEDNESS, SIMILARITY} for eval_name, eval_options in resolved_evaluators.items(): threshold = get_evaluator_threshold(eval_name, eval_options) try: # Pre-flight guard: an empty/None/whitespace agent response cannot be # scored by evaluators that require it. Record a categorized errored # entry (result="error", code="emptyAgentResponse") instead of invoking # the evaluator and letting it raise a cryptic SDK error. This makes # the outcome deterministic across runs when the agent intermittently # returns an empty response. if _is_empty_response(actual_response) and ( eval_name in _RESPONSE_REQUIRED_EVALUATORS or ( (entry := EVALUATOR_REGISTRY.get(eval_name)) is not None and entry.type in ("custom-llm", "custom-non-llm") ) ): results_dict[eval_name] = _errored_entry( EVAL_ERROR_CODE_EMPTY_AGENT_RESPONSE, "empty agent response", threshold, ) evaluators_ran.append(eval_name) continue # Defer LLM evaluators to Foundry cloud evaluation when active. The # actual score is filled in by a batch run after all responses are # collected (see run_pipeline); record a pending placeholder + row. if foundry_collector is not None and eval_name in FOUNDRY_LLM_EVALUATORS: metric_id = METRIC_IDS[eval_name] row_id = foundry_collector.register( query=prompt, response=actual_response, context=expected_response, ground_truth=expected_response, metric=eval_name, ) results_dict[eval_name] = { "_foundry_pending": True, "_foundry_row_id": row_id, "_foundry_metric_id": metric_id, "threshold": threshold, } evaluators_ran.append(eval_name) continue # Route LLM evaluators through judge backend if available if judge_backend and eval_name in _JUDGE_ROUTABLE: raw_score = judge_backend.evaluate( eval_name=eval_name, prompt=prompt, response=actual_response, context=expected_response if eval_name in (GROUNDEDNESS, SIMILARITY) else None, ) if raw_score is not None: results_dict[eval_name] = _decorate_metric(METRIC_IDS[eval_name], raw_score, threshold) else: results_dict[eval_name] = _errored_entry( EVAL_ERROR_CODE_JUDGE_UNPARSEABLE, f"judge backend returned no parseable score for '{eval_name}'", threshold, ) evaluators_ran.append(eval_name) continue if eval_name == RELEVANCE: raw_score = RelevanceEvaluator(model_config=model_config)(query=prompt, response=actual_response) results_dict[RELEVANCE] = _decorate_metric(METRIC_IDS[RELEVANCE], raw_score, threshold) elif eval_name == COHERENCE: raw_score = CoherenceEvaluator(model_config=model_config)(query=prompt, response=actual_response) results_dict[COHERENCE] = _decorate_metric(METRIC_IDS[COHERENCE], raw_score, threshold) elif eval_name == GROUNDEDNESS: raw_score = GroundednessEvaluator(model_config=model_config)(response=actual_response, context=expected_response) results_dict[GROUNDEDNESS] = _decorate_metric(METRIC_IDS[GROUNDEDNESS], raw_score, threshold) elif eval_name == SIMILARITY: raw_score = SimilarityEvaluator(model_config=model_config)(query=prompt, response=actual_response, ground_truth=expected_response) results_dict[SIMILARITY] = _decorate_metric(METRIC_IDS[SIMILARITY], raw_score, threshold) elif eval_name == TOOL_CALL_ACCURACY: raw_score = ToolCallAccuracyEvaluator(model_config)( query=prompt, response=enhanced_response.get("response", actual_response), tool_definitions=enhanced_response.get("tool_definitions", []), ) results_dict[TOOL_CALL_ACCURACY] = _decorate_metric(METRIC_IDS[TOOL_CALL_ACCURACY], raw_score, threshold) elif eval_name == CITATIONS: fmt_str = eval_options.get("citation_format", "oai_unicode") fmt_map = { "oai_unicode": CitationFormat.OAI_UNICODE, "bracket": CitationFormat.LEGACY_BRACKET, "markdown": CitationFormat.MARKDOWN_LINK, "mixed": CitationFormat.AUTO, } raw_score = CitationsEvaluator(citation_format=fmt_map.get(fmt_str, CitationFormat.OAI_UNICODE))(response=actual_response) results_dict[CITATIONS] = _decorate_metric(METRIC_IDS[CITATIONS], raw_score, threshold) elif eval_name == EXACT_MATCH: case_sensitive = eval_options.get("case_sensitive", False) raw_score = ExactMatchEvaluator(case_sensitive=case_sensitive)(response=actual_response, expected_answer=expected_response) # ExactMatch is binary — the evaluator already sets 'result' # so _decorate_metric (which computes result from score vs threshold) is not needed. results_dict[EXACT_MATCH] = raw_score elif eval_name == PARTIAL_MATCH: case_sensitive = eval_options.get("case_sensitive", False) raw_score = PartialMatchEvaluator(case_sensitive=case_sensitive)(response=actual_response, expected_answer=expected_response) results_dict[PARTIAL_MATCH] = _decorate_metric(METRIC_IDS[PARTIAL_MATCH], raw_score, threshold) elif eval_name == RETRIEVAL_QUERY: evaluator = RetrievalQueryEvaluator( capability=eval_options["capability"], selector=eval_options["selector"], includes=eval_options.get("includes"), excludes=eval_options.get("excludes"), ) # Retrieval evaluators emit their own decorated score with # threshold/result fields; no _decorate_metric needed. results_dict[RETRIEVAL_QUERY] = evaluator( response=actual_response, retrieval_telemetry=retrieval_telemetry, ) elif eval_name == RETRIEVAL_RESULT: evaluator = RetrievalResultEvaluator( capability=eval_options["capability"], expected_items=eval_options.get("expected_items"), min_expected_count=eval_options.get("min_expected_count"), max_rank=eval_options.get("max_rank", 10), model_config=model_config if has_azure_openai else None, ) results_dict[RETRIEVAL_RESULT] = evaluator( response=actual_response, retrieval_telemetry=retrieval_telemetry, ) elif (entry := EVALUATOR_REGISTRY.get(eval_name)) is not None and entry.type in ("custom-llm", "custom-non-llm"): # Custom LLM-judge evaluators require Azure OpenAI (model_config) # and cannot yet be routed through the Copilot SDK judge backend. if entry.type == "custom-llm" and judge_backend and not has_azure_openai: emit_structured_log( "warning", f"⚠️ Skipping custom evaluator '{eval_name}': custom evaluators " f"require Azure OpenAI credentials and are not yet supported with " f"the Copilot SDK judge backend. Configure Azure OpenAI env vars " f"or remove '{eval_name}' from this run.", operation=Operation.EVALUATE, ) results_dict[eval_name] = _errored_entry( EVAL_ERROR_CODE_EVALUATOR_CONFIG, ( f"Custom evaluator '{eval_name}' is not supported with the Copilot SDK " f"judge backend. Custom evaluators require Azure OpenAI credentials " f"(AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_VERSION, AZURE_OPENAI_MODEL_NAME). " f"Use --judge-backend azure or provide Azure OpenAI env vars alongside " f"the Copilot backend." ), threshold, ) evaluators_ran.append(eval_name) continue # Custom evaluator (LLM-judge or code-only). The framework # derives result from score>=threshold and attaches threshold; # the author returns {score, reason} on success or signals an # error via raise / {"error": ...}. custom_result = _invoke_custom_evaluator( eval_name=eval_name, eval_options=eval_options, threshold=threshold, prompt=prompt, actual_response=actual_response, expected_response=expected_response, context=item_context, model_config=model_config, is_llm=entry.type == "custom-llm", ) results_dict[eval_name] = custom_result evaluators_ran.append(eval_name) except Exception as e: auth_failure = wrap_credential_error(e, selected_mode=selected_auth_mode, selection_source=auth_selection_source) if auth_failure: emit_structured_log( "error", str(auth_failure), operation=Operation.EVALUATE, ) exc_msg = str(auth_failure) error_code = EVAL_ERROR_CODE_JUDGE_AUTH else: # Log full detail (at error level); persisted output uses the # scrubbed message + a machine-readable code only. "failed to # produce a score" — not "crashed": the run is not aborted and # partial results are preserved. error_code = _classify_evaluator_exception(e) where = f" on response for {context_label}" if context_label else "" emit_structured_log( "error", f"Evaluator '{eval_name}' failed to produce a score{where} [{error_code}]: {e}", operation=Operation.EVALUATE, ) exc_msg = getattr(e, "message", None) or str(e) results_dict[eval_name] = _errored_entry(error_code, exc_msg, threshold) return results_dict, evaluators_ran, retrieval_telemetry def _collect_evaluator_results(results_dict: Dict[str, Dict[str, Any]]) -> List[str]: """Extract per-evaluator ``result`` values (one of pass/fail/error) for status derivation.""" return [ d["result"] for d in results_dict.values() if d.get("result") in (STATUS_PASS, STATUS_FAIL, STATUS_ERROR) ] def _evaluate_multi_turn_responses( turns: List[Dict], effective_log_level: str, default_evaluators: Dict[str, Any], model_config: AzureOpenAIModelConfiguration, selected_auth_mode: str, auth_selection_source: str, has_azure_openai: bool = True, thread_name: str = "", judge_backend: Optional[JudgeBackend] = None, foundry_collector: Optional[Any] = None, ) -> Tuple[List[Dict], Dict]: """Run per-turn evaluations and build evaluated turn results with summary. Returns: Tuple of (evaluated_turns, summary). Each evaluated turn contains prompt, response, expected_response, status, evaluators_ran, results, and optionally error. Does not mutate the input turns. """ evaluated_turns: List[Dict] = [] for i, turn in enumerate(turns): evaluated_turn: Dict[str, Any] = { "prompt": turn.get("prompt", ""), } if "expected_response" in turn: evaluated_turn["expected_response"] = turn["expected_response"] if "response" in turn: evaluated_turn["response"] = turn["response"] if "evaluators" in turn: evaluated_turn["evaluators"] = turn["evaluators"] if "evaluators_mode" in turn: evaluated_turn["evaluators_mode"] = turn["evaluators_mode"] for key in ("tags", "extensions"): if key in turn: evaluated_turn[key] = turn[key] if turn.get("status") == STATUS_ERROR: # Request-failure or downstream-skip turn — error already set upstream. evaluated_turn["status"] = STATUS_ERROR if "error" in turn: evaluated_turn["error"] = turn["error"] evaluated_turns.append(evaluated_turn) continue enhanced_response = turn.get("_enhanced_response", {}) actual_response = get_response_text_for_evaluation(enhanced_response) resolved = resolve_evaluators_for_prompt( turn.get("evaluators"), turn.get("evaluators_mode", "extend"), turn.get("prompt", ""), default_evaluators, ) thread_part = f" of '{thread_name}'" if thread_name else "" turn_label = f"turn {i + 1}/{len(turns)}{thread_part}" results_dict, evaluators_ran, retrieval_telemetry = _run_evaluators_for_item( turn.get("prompt", ""), actual_response, turn.get("expected_response", ""), enhanced_response, resolved, model_config, context_label=turn_label, has_azure_openai=has_azure_openai, item_context=turn.get("context", ""), selected_auth_mode=selected_auth_mode, auth_selection_source=auth_selection_source, judge_backend=judge_backend, foundry_collector=foundry_collector, ) evaluator_result_values = _collect_evaluator_results(results_dict) status, error_obj = status_for_response(evaluator_result_values) evaluated_turn["results"] = results_dict evaluated_turn["evaluators_ran"] = evaluators_ran evaluated_turn["status"] = status if error_obj is not None: evaluated_turn["error"] = error_obj # Per-turn diagnostics from this turn's own retrieval telemetry. diagnostics = _build_diagnostics(resolved, retrieval_telemetry) if diagnostics is not None: evaluated_turn["_diagnostics"] = diagnostics if effective_log_level == "debug": emit_structured_log( "debug", f"Evaluation completed for turn {i + 1} prompt='{turn.get('prompt', '')}'. " f"Evaluators: {', '.join(evaluators_ran)}. " f"Scores: {results_dict}", operation=Operation.EVALUATE, ) evaluated_turns.append(evaluated_turn) turn_statuses = [t.get("status", STATUS_ERROR) for t in evaluated_turns] turns_total = len(evaluated_turns) summary = { "turns_total": turns_total, "turns_passed": sum(1 for s in turn_statuses if s == STATUS_PASS), "turns_failed": sum(1 for s in turn_statuses if s == STATUS_FAIL), "turns_partial": sum(1 for s in turn_statuses if s == STATUS_PARTIAL), "turns_errored": sum(1 for s in turn_statuses if s == STATUS_ERROR), "overall_status": rollup_thread_status(turn_statuses), } return evaluated_turns, summary def _evaluate_single_response( enhanced_response: Dict[str, Any], eval_item: Dict, effective_log_level: str, model_config: AzureOpenAIModelConfiguration, default_evaluators: Dict[str, Any], selected_auth_mode: str, auth_selection_source: str, has_azure_openai: bool = True, judge_backend: Optional[JudgeBackend] = None, foundry_collector: Optional[Any] = None, ) -> Dict[str, Any]: """Run all evaluators for a single prompt/response pair and return the result dict.""" actual_response_text = get_response_text_for_evaluation(enhanced_response) prompt = eval_item.get("prompt", "") expected_response = eval_item.get("expected_response", "") resolved = resolve_evaluators_for_prompt( eval_item.get("evaluators"), eval_item.get("evaluators_mode", "extend"), prompt, default_evaluators, ) results_dict, evaluators_ran, retrieval_telemetry = _run_evaluators_for_item( prompt, actual_response_text, expected_response, enhanced_response, resolved, model_config, context_label=f"prompt '{prompt[:60]}'" if prompt else "", has_azure_openai=has_azure_openai, item_context=eval_item.get("context", ""), selected_auth_mode=selected_auth_mode, auth_selection_source=auth_selection_source, judge_backend=judge_backend, foundry_collector=foundry_collector, ) evaluator_result_values = _collect_evaluator_results(results_dict) status, error_obj = status_for_response(evaluator_result_values) evaluation_result: Dict[str, Any] = { "prompt": prompt, "response": enhanced_response.get( "display_response_text", actual_response_text ), "expected_response": expected_response, "evaluators_ran": evaluators_ran, "results": results_dict, "status": status, } if error_obj is not None: evaluation_result["error"] = error_obj # Auto-attach retrieval diagnostics when a retrieval evaluator was resolved # (suppressed by EVALS_SUPPRESS_DIAGNOSTICS). diagnostics = _build_diagnostics(resolved, retrieval_telemetry) if diagnostics is not None: evaluation_result["_diagnostics"] = diagnostics if "evaluators" in eval_item: evaluation_result["evaluators"] = eval_item["evaluators"] if "evaluators_mode" in eval_item: evaluation_result["evaluators_mode"] = eval_item["evaluators_mode"] for key in ("tags", "extensions"): if key in eval_item: evaluation_result[key] = eval_item[key] if effective_log_level == "debug": emit_structured_log( "debug", f"Evaluation completed for prompt='{evaluation_result['prompt']}'. " f"Evaluators: {', '.join(evaluators_ran)}. " f"Scores: {evaluation_result['results']}", operation=Operation.EVALUATE, ) return evaluation_result def get_effective_worker_count(prompt_count: int, concurrency: int) -> int: """Compute safe worker count for prompt processing.""" if prompt_count <= 0: return 1 try: requested_int = int(concurrency) except (TypeError, ValueError): requested_int = MAX_CONCURRENCY bounded = max(1, min(requested_int, MAX_CONCURRENCY)) return min(bounded, prompt_count) def _patch_foundry_results_dict( results_dict: Dict[str, Any], scores: Dict[int, Dict[str, Dict[str, Any]]], ) -> None: """Replace pending Foundry placeholders in a results dict with real scores. A placeholder (``{"_foundry_pending": True, ...}``) is replaced with either a decorated metric dict (score + threshold + pass/fail result) or a standard errored-score entry when the batch produced no usable score for that cell. """ for eval_name, entry in list(results_dict.items()): if not (isinstance(entry, dict) and entry.get("_foundry_pending")): continue row_id = entry.get("_foundry_row_id") metric_id = entry.get("_foundry_metric_id") threshold = entry.get("threshold") score_entry = scores.get(row_id, {}).get(metric_id) if isinstance(score_entry, dict) and "score" in score_entry: raw: Dict[str, Any] = {metric_id: score_entry["score"]} reason = score_entry.get("reason") if reason: raw[f"{metric_id}_reason"] = reason try: results_dict[eval_name] = _decorate_metric(metric_id, raw, threshold) except ValueError as e: results_dict[eval_name] = _errored_entry( EVAL_ERROR_CODE_EVALUATOR_ERROR, str(e), threshold ) else: detail = score_entry.get("error") if isinstance(score_entry, dict) else None results_dict[eval_name] = _errored_entry( EVAL_ERROR_CODE_JUDGE_UNPARSEABLE, detail or f"no Foundry score for '{eval_name}'", threshold, ) def _apply_foundry_scores( ordered_results: List[Dict[str, Any]], scores: Dict[int, Dict[str, Dict[str, Any]]], ) -> None: """Patch deferred Foundry scores into results and re-derive status/summary. Items/turns that failed at request time (empty results) keep their existing error status; only entries that actually ran evaluators are re-derived. """ for result in ordered_results: if result.get("type") == "multi_turn": for turn in result.get("turns", []): rd = turn.get("results") if not (isinstance(rd, dict) and rd): continue _patch_foundry_results_dict(rd, scores) status, error_obj = status_for_response(_collect_evaluator_results(rd)) turn["status"] = status if error_obj is not None: turn["error"] = error_obj elif "error" in turn: del turn["error"] turn_statuses = [t.get("status", STATUS_ERROR) for t in result.get("turns", [])] result["summary"] = { "turns_total": len(turn_statuses), "turns_passed": sum(1 for s in turn_statuses if s == STATUS_PASS), "turns_failed": sum(1 for s in turn_statuses if s == STATUS_FAIL), "turns_partial": sum(1 for s in turn_statuses if s == STATUS_PARTIAL), "turns_errored": sum(1 for s in turn_statuses if s == STATUS_ERROR), "overall_status": rollup_thread_status(turn_statuses), } else: rd = result.get("results") if not (isinstance(rd, dict) and rd): continue _patch_foundry_results_dict(rd, scores) status, error_obj = status_for_response(_collect_evaluator_results(rd)) result["status"] = status if error_obj is not None: result["error"] = error_obj elif "error" in result: del result["error"] def _validate_and_classify_items( default_evaluators: Dict[str, Any], eval_items: List[Dict], ) -> List[ItemType]: """Validate evaluator configuration and return each item's type.""" all_evaluator_maps = [default_evaluators] for eval_item in eval_items: if "evaluators" in eval_item: all_evaluator_maps.append(eval_item["evaluators"]) for turn in eval_item.get("turns", []): if "evaluators" in turn: all_evaluator_maps.append(turn["evaluators"]) for evaluator_map in all_evaluator_maps: validate_evaluator_names(evaluator_map) validate_evaluator_options(evaluator_map) item_types: List[ItemType] = [] for index, eval_item in enumerate(eval_items): try: item_type = detect_item_type(eval_item) except ValueError as exc: raise ValueError( f"Invalid evaluation item at index {index}: {exc}" ) from exc if item_type == ItemType.MULTI_TURN: turn_count = len(eval_item["turns"]) if turn_count > MAX_TURNS_PER_THREAD: raise ValueError( f"Invalid evaluation item at index {index}: 'turns' array has " f"{turn_count} items (max {MAX_TURNS_PER_THREAD})" ) item_types.append(item_type) return item_types def _build_multi_turn_result( eval_item: Dict, evaluated_turns: List[Dict], summary: Dict, conversation_id: str, ) -> Dict[str, Any]: """Build the common result shape for an evaluated multi-turn item.""" return { "type": "multi_turn", "name": eval_item.get("name", ""), "description": eval_item.get("description", ""), "conversation_id": conversation_id, "turns": evaluated_turns, "summary": summary, **{ key: eval_item[key] for key in ("tags", "extensions") if key in eval_item }, } def _finalize_foundry_scores( pipeline: PipelineConfig, foundry_collector: Optional[FoundryEvalCollector], ordered_results: List[Dict[str, Any]], ) -> None: """Run deferred Foundry evaluators and patch their scores in place.""" if ( pipeline.foundry_evaluator is None or foundry_collector is None or foundry_collector.is_empty() ): return try: foundry_scores = pipeline.foundry_evaluator.evaluate_batch( foundry_collector.rows ) except Exception as exc: # noqa: BLE001 — degrade to per-row errored scores emit_structured_log( "error", f"Foundry cloud evaluation batch failed: {exc}", operation=Operation.EVALUATE, ) foundry_scores = {} _apply_foundry_scores(ordered_results, foundry_scores) def _resolve_pipeline_mode(config: RunConfig) -> PipelineMode: """Resolve the current execution mode from CLI configuration.""" if config.evaluate_only: return PipelineMode.EVALUATE_ONLY return PipelineMode.FULL def _load_single_turn_response( pipeline: PipelineConfig, eval_item: Dict, config: RunConfig, max_attempts: int, ) -> SingleTurnAcquisition: """Load a captured single-turn response from a v1 eval item. Accepts the same parameter list as :func:`_invoke_single_turn_response` (unused here — no agent is invoked) so both can be bound to a single ``acquire_single`` reference in :func:`run_pipeline` instead of a ternary at each call site. """ del pipeline, config, max_attempts # unused; signature parity with _invoke_* return SingleTurnAcquisition( response=build_enhanced_response_from_eval_item(eval_item) ) def _invoke_single_turn_response( pipeline: PipelineConfig, eval_item: Dict, config: RunConfig, max_attempts: int, ) -> SingleTurnAcquisition: """Invoke the agent for one single-turn item with retry and throttling.""" prompt = eval_item.get("prompt", "") resolved_for_send = resolve_evaluators_for_prompt( eval_item.get("evaluators"), eval_item.get("evaluators_mode", "extend"), prompt, pipeline.default_evaluators, ) accepted_output_modes = _resolve_accepted_output_modes(resolved_for_send) for attempt in range(1, max_attempts + 1): pipeline.chat_gate.wait_if_blocked() try: response, _ = pipeline.agent_client.send_prompt( prompt, agent_id=config.m365_agent_id, accepted_output_modes=accepted_output_modes, ) return SingleTurnAcquisition(response=response) except Exception as exc: cause = exc.__cause__ status = ( int(getattr(cause, "code", 0) or 0) or None if cause else None ) retry_after = get_retry_after_seconds( cause.headers.get("Retry-After") if cause and getattr(cause, "headers", None) else None ) timed_out = is_timeout_error(exc) retryable = pipeline.is_retryable_status(status) or timed_out if retry_after is not None and pipeline.is_retryable_status(status): pipeline.chat_gate.apply_retry_after(retry_after) if not retryable or attempt >= max_attempts: emit_structured_log( "error", f'Prompt "{_truncate_prompt(prompt)}" failed after ' f"{_attempts_phrase(attempt)}: {exc}", operation=Operation.SEND_PROMPT, ) return SingleTurnAcquisition( terminal_result={ "prompt": prompt, "response": "", "expected_response": eval_item.get( "expected_response", "" ), "evaluators_ran": [], "results": {}, "status": STATUS_ERROR, "error": agent_request_failed( getattr(exc, "message", None) or str(exc), code=_classify_agent_exception(exc), ), **{ key: eval_item[key] for key in ("tags", "extensions") if key in eval_item }, } ) delay = ( retry_after if retry_after is not None else pipeline.get_backoff_seconds(attempt) ) reason = ( "timeout" if timed_out else (f"HTTP {status}" if status else "transient error") ) emit_structured_log( "warning", f'Prompt "{_truncate_prompt(prompt)}": ' f"attempt {attempt}/{max_attempts} failed ({reason}); " f"retrying in {delay}s.", operation=Operation.SEND_PROMPT, ) time.sleep(delay) raise RuntimeError("Agent response acquisition ended without a result") def _load_multi_turn_responses( pipeline: PipelineConfig, eval_item: Dict, config: RunConfig, max_attempts: int, ) -> MultiTurnAcquisition: """Load captured responses for every turn in a v1 multi-turn item. Accepts the same parameter list as :func:`_invoke_multi_turn_responses` (unused here — no agent is invoked) so both can be bound to a single ``acquire_multi`` reference in :func:`run_pipeline` instead of a ternary at each call site. """ del pipeline, config, max_attempts # unused; signature parity with _invoke_* captured_turns: List[Dict[str, Any]] = [] for turn in eval_item["turns"]: captured_turn = { key: value for key, value in turn.items() if key not in ("status", "error", "scores") } captured_turn["_enhanced_response"] = ( build_enhanced_response_from_eval_item(turn) ) captured_turns.append(captured_turn) return MultiTurnAcquisition( turns=captured_turns, conversation_id=eval_item.get("conversation_id", ""), ) def _invoke_multi_turn_responses( pipeline: PipelineConfig, eval_item: Dict, config: RunConfig, max_attempts: int, ) -> MultiTurnAcquisition: """Invoke each turn sequentially while preserving conversation state.""" turns = eval_item["turns"] thread_name = eval_item.get("name", "Unnamed thread") conversation_context = None conversation_id = None enriched_turns: List[Dict[str, Any]] = [] failure_exception: Optional[Exception] = None for i, turn in enumerate(turns): prompt = turn["prompt"] emit_structured_log( "debug", f"Sending turn {i + 1}/{len(turns)} of '{thread_name}'.", operation=Operation.SEND_PROMPT, ) resolved_for_send = resolve_evaluators_for_prompt( turn.get("evaluators"), turn.get("evaluators_mode", "extend"), prompt, pipeline.default_evaluators, ) accepted_output_modes = _resolve_accepted_output_modes( resolved_for_send ) response = None for attempt in range(1, max_attempts + 1): pipeline.chat_gate.wait_if_blocked() try: response, conversation_context = ( pipeline.agent_client.send_prompt( prompt, agent_id=config.m365_agent_id, conversation_context=conversation_context, accepted_output_modes=accepted_output_modes, ) ) break except Exception as exc: cause = exc.__cause__ status = ( int(getattr(cause, "code", 0) or 0) or None if cause else None ) retry_after = get_retry_after_seconds( cause.headers.get("Retry-After") if cause and getattr(cause, "headers", None) else None ) timed_out = is_timeout_error(exc) retryable = pipeline.is_retryable_status(status) or timed_out if ( retry_after is not None and pipeline.is_retryable_status(status) ): pipeline.chat_gate.apply_retry_after(retry_after) if not retryable or attempt >= max_attempts: emit_structured_log( "error", f"Turn {i + 1} of '{thread_name}' " f'(prompt: "{_truncate_prompt(prompt)}") failed after ' f"{_attempts_phrase(attempt)}: {exc}", operation=Operation.SEND_PROMPT, ) failure_exception = exc break delay = ( retry_after if retry_after is not None else pipeline.get_backoff_seconds(attempt) ) reason = ( "timeout" if timed_out else (f"HTTP {status}" if status else "transient error") ) emit_structured_log( "warning", f"Turn {i + 1} of '{thread_name}' " f'(prompt: "{_truncate_prompt(prompt)}"): ' f"attempt {attempt}/{max_attempts} failed ({reason}); " f"retrying in {delay}s.", operation=Operation.SEND_PROMPT, ) time.sleep(delay) if failure_exception is not None: exc_msg = ( getattr(failure_exception, "message", None) or str(failure_exception) ) enriched_turns.append( { **turn, "response": "", "status": STATUS_ERROR, "error": agent_request_failed( exc_msg, code=_classify_agent_exception(failure_exception), ), } ) for j in range(i + 1, len(turns)): enriched_turns.append({ **turns[j], "response": "", "status": STATUS_ERROR, "error": turn_skipped(), }) break response_text = get_response_text_for_evaluation(response) enriched_turns.append( { **turn, "response": response.get( "display_response_text", response_text ), "_enhanced_response": response, } ) if conversation_id is None: conversation_id = response.get("metadata", {}).get( "conversation_id" ) return MultiTurnAcquisition( turns=enriched_turns, conversation_id=conversation_id or "", ) def run_pipeline( pipeline: PipelineConfig, eval_items: List[Dict], config: RunConfig, ) -> List[Dict[str, Any]]: """Run response acquisition and evaluation in parallel. Full mode composes invoke → evaluate. Evaluate-only mode composes load captured response → evaluate. Each worker processes one top-level item end-to-end, while turns inside a multi-turn item remain sequential. Results are returned in original prompt order (FR-006). """ mode = _resolve_pipeline_mode(config) item_types = _validate_and_classify_items( pipeline.default_evaluators, eval_items, ) total = len(eval_items) worker_count = get_effective_worker_count(total, config.concurrency) multi_turn_count = sum(1 for t in item_types if t == ItemType.MULTI_TURN) single_turn_count = total - multi_turn_count max_attempts = ( resolve_max_attempts() if mode == PipelineMode.FULL else 0 ) emit_structured_log( "info", f"Running {mode.value} pipeline with {worker_count} worker(s) " f"for {total} item(s) " f"({single_turn_count} single-turn, {multi_turn_count} multi-turn).", operation=Operation.EVALUATE, ) if mode == PipelineMode.FULL: emit_structured_log( "info", f"Max attempts per agent request set to {max_attempts}.", operation=Operation.EVALUATE, ) # Bind the acquisition strategy once per mode instead of branching at each # call site. `_load_*` and `_invoke_*` share a signature so either can be # assigned here; PipelineMode is otherwise used only for logging/messaging # below, not for further control-flow branching. acquire_single = ( _invoke_single_turn_response if mode == PipelineMode.FULL else _load_single_turn_response ) acquire_multi = ( _invoke_multi_turn_responses if mode == PipelineMode.FULL else _load_multi_turn_responses ) # In Foundry cloud evaluation mode, LLM evaluators are deferred into this # collector during Phase B and scored in a single batch run afterwards. foundry_collector = FoundryEvalCollector() if pipeline.foundry_evaluator else None def _process_item(eval_item: Dict, index: int) -> Dict[str, Any]: if item_types[index] == ItemType.MULTI_TURN: return _process_multi_turn(eval_item, index) return _process_single_turn(eval_item, index) def _process_single_turn(eval_item: Dict, index: int) -> Dict[str, Any]: emit_structured_log( "info", f"Processing item {index + 1}/{total} (single-turn).", operation=( Operation.SEND_PROMPT if mode == PipelineMode.FULL else Operation.EVALUATE ), ) acquisition = acquire_single( pipeline, eval_item, config, max_attempts, ) if acquisition.terminal_result is not None: return acquisition.terminal_result return _evaluate_single_response( acquisition.response, eval_item, config.effective_log_level, pipeline.model_config, pipeline.default_evaluators, has_azure_openai=pipeline.has_azure_openai, selected_auth_mode=pipeline.selected_auth_mode, auth_selection_source=pipeline.auth_selection_source, judge_backend=pipeline.judge_backend, foundry_collector=foundry_collector, ) def _process_multi_turn(eval_item: Dict, index: int) -> Dict[str, Any]: turns = eval_item["turns"] thread_name = eval_item.get("name", "Unnamed thread") emit_structured_log( "info", f"Processing item {index + 1}/{total} (multi-turn: '{thread_name}').", operation=( Operation.SEND_PROMPT if mode == PipelineMode.FULL else Operation.EVALUATE ), ) if ( mode == PipelineMode.FULL and len(turns) > LONG_THREAD_WARNING_THRESHOLD ): emit_structured_log( "warning", f"Thread '{thread_name}' has {len(turns)} turns (>{LONG_THREAD_WARNING_THRESHOLD}). This may take a while.", operation=Operation.SEND_PROMPT, ) acquisition = acquire_multi( pipeline, eval_item, config, max_attempts, ) evaluated_turns, summary = _evaluate_multi_turn_responses( acquisition.turns, config.effective_log_level, pipeline.default_evaluators, thread_name=thread_name, model_config=pipeline.model_config, has_azure_openai=pipeline.has_azure_openai, selected_auth_mode=pipeline.selected_auth_mode, auth_selection_source=pipeline.auth_selection_source, judge_backend=pipeline.judge_backend, foundry_collector=foundry_collector, ) return _build_multi_turn_result( eval_item, evaluated_turns, summary, acquisition.conversation_id, ) execution_results = execute_in_parallel( eval_items, _process_item, max_workers=worker_count, ) # Unwrap WorkerResult objects into plain dicts, with error fallback ordered_results: List[Dict[str, Any]] = [] for wr in execution_results: if wr.error: idx = wr.index item = eval_items[idx] exc_msg = getattr(wr.error, "message", None) or str(wr.error) # PipelineMode only selects *which* error-object builder to use # (messaging) — the structure below (turn building, turn_skipped # for downstream turns, tags/extensions passthrough) is identical # for both modes. if mode == PipelineMode.FULL: emit_structured_log( "error", f"Worker failed for item {idx}: {wr.error}", operation=Operation.SEND_PROMPT, ) cause_error = agent_request_failed( exc_msg, code=_classify_agent_exception(wr.error) ) else: emit_structured_log( "error", f"Evaluate-only worker failed for item {idx}: {wr.error}", operation=Operation.EVALUATE, ) cause_error = evaluation_crashed(exc_msg) if item_types[idx] == ItemType.MULTI_TURN: # Worker raised before any turn ran. Turn 1 carries the cause; # remaining turns are downstream-skipped. All turns errored → # thread overall_status="error". turns = item.get("turns", []) turn_dicts = [] for j, t in enumerate(turns): turn_dicts.append({ **t, "response": "", "results": {}, "status": STATUS_ERROR, "error": cause_error if j == 0 else turn_skipped(), }) summary = { "turns_total": len(turns), "turns_passed": 0, "turns_failed": 0, "turns_partial": 0, "turns_errored": len(turns), "overall_status": STATUS_ERROR, } ordered_results.append( _build_multi_turn_result( item, turn_dicts, summary, item.get("conversation_id", ""), ) ) else: ordered_results.append({ "prompt": item.get("prompt", ""), "response": "", "expected_response": item.get("expected_response", ""), "evaluators_ran": [], "results": {}, "status": STATUS_ERROR, "error": cause_error, **{k: item[k] for k in ("tags", "extensions") if k in item}, }) else: ordered_results.append(wr.value) _finalize_foundry_scores(pipeline, foundry_collector, ordered_results) return ordered_results