diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/llm_judge.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/llm_judge.py index f47f69b556..8c8eaa1235 100644 --- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/llm_judge.py +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/metrics/llm_judge.py @@ -25,7 +25,13 @@ sample_template_payload, ) from nemo_evaluator_sdk.resolver_protocols import ModelResolver, SecretResolver -from nemo_evaluator_sdk.structured_output import InferenceStructuredOutput, detect_structured_output_mode +from nemo_evaluator_sdk.structured_output import ( + InferenceStructuredOutput, + StructuredOutputMode, + detect_structured_output_mode, + looks_like_unsupported_structured_output_error, + next_structured_output_mode, +) from nemo_evaluator_sdk.templates import render_request from nemo_evaluator_sdk.values.common import SecretRef, SupportedJobTypes from nemo_evaluator_sdk.values.llm_judge_defaults import ( @@ -82,6 +88,7 @@ class LLMJudgeMetric(HooksBase, LLMJudge): _parsers: dict[str, ScoreParser] = PrivateAttr(default_factory=dict) _score_dumps: dict[str, dict[str, Any]] = PrivateAttr(default_factory=dict) _prompt_template_is_default: bool = PrivateAttr(default=False) + _rejected_structured_output_modes: set[StructuredOutputMode] = PrivateAttr(default_factory=set) job_type: Literal[SupportedJobTypes.ONLINE, SupportedJobTypes.OFFLINE] = SupportedJobTypes.ONLINE @property @@ -222,12 +229,7 @@ async def preflight(self) -> None: if model.format != ModelFormat.NVIDIA_NIM or not self.structured_output: return - structured_hook: InferenceStructuredOutput | None = None - for hook in self._preprocess_hooks: - if isinstance(hook, InferenceStructuredOutput): - structured_hook = hook - break - + structured_hook = self._structured_output_hook() if structured_hook is None: return @@ -298,6 +300,9 @@ def _initialize_score_parsers(self) -> None: self._score_dumps[score.name] = score.model_dump(mode="json", exclude={"parser"}) def _render_request(self, item: dict, sample: TemplateSample) -> dict: + return self._apply_preprocess_hooks(self._render_base_request(item, sample)) + + def _render_base_request(self, item: dict, sample: TemplateSample) -> dict: sample_payload = sample_template_payload(sample) overlapping_keys = set(item.keys()) & set(sample_payload.keys()) if overlapping_keys: @@ -326,7 +331,38 @@ def _render_request(self, item: dict, sample: TemplateSample) -> dict: request["max_completion_tokens"] = request["max_tokens"] del request["max_tokens"] - return self._apply_preprocess_hooks(request) + return request + + def _structured_output_hook(self) -> InferenceStructuredOutput | None: + for hook in self._preprocess_hooks: + if isinstance(hook, InferenceStructuredOutput): + return hook + return None + + def _downgrade_structured_output(self, error: Exception, rendered_mode: StructuredOutputMode | None) -> bool: + """Latch the next structured-output mode when the backend rejects *rendered_mode*.""" + message = str(error) + if not looks_like_unsupported_structured_output_error(message): + return False + + hook = self._structured_output_hook() + if hook is None: + return False + + if ( + rendered_mode is not None + and rendered_mode != StructuredOutputMode.UNSUPPORTED + and rendered_mode not in self._rejected_structured_output_modes + ): + next_mode = next_structured_output_mode(rendered_mode, message, self._rejected_structured_output_modes) + self._rejected_structured_output_modes.add(rendered_mode) + _logger.warning( + "Judge model rejected structured output mode %s; using %s for all future requests.", + rendered_mode.value, + next_mode.value, + ) + hook.set_mode(next_mode) + return True def _retry_with_max_completion_tokens(self, request: dict) -> dict: if not self._use_max_completion_tokens: @@ -338,24 +374,49 @@ def _retry_with_max_completion_tokens(self, request: dict) -> dict: del request["max_tokens"] return request + def _retry_request( + self, + error: Exception, + request: dict, + base_request: dict, + rendered_mode: StructuredOutputMode | None, + ) -> dict | None: + """Request to retry the rejected call with, or None when no retry can help.""" + if "max_tokens" in request and "'max_tokens' is not supported with this model" in error.args[0]: + return self._retry_with_max_completion_tokens(request) + if self._downgrade_structured_output(error, rendered_mode): + retried = self._apply_preprocess_hooks(deepcopy(base_request)) + if retried != request: + return retried + return None + async def compute_scores(self, input: MetricInput) -> MetricResult: """Compute structured score output for one item/sample pair.""" item = input.row.data sample = input.candidate - request = self._render_request(item, sample) + base_request = self._render_base_request(item, sample) + request = self._apply_preprocess_hooks(deepcopy(base_request)) + hook = self._structured_output_hook() + rendered_mode = hook.mode if hook else None try: response = await self.inference_fn(self._require_model(), request, 3, client=self.client) except inference.ClientInferenceError as error: - if "max_tokens" in request and "'max_tokens' is not supported with this model" in error.args[0]: - request = self._retry_with_max_completion_tokens(request) - response = await self.inference_fn(self._require_model(), request, 3, client=self.client) - else: + retry_request = self._retry_request(error, request, base_request, rendered_mode) + if retry_request is None: return self._handle_invalid_output( error, self._nan_result(), "Inference failed with LLM judge, marking as NaN", ) + try: + response = await self.inference_fn(self._require_model(), retry_request, 3, client=self.client) + except inference.ClientInferenceError as retry_error: + return self._handle_invalid_output( + retry_error, + self._nan_result(), + "Inference failed with LLM judge, marking as NaN", + ) try: output_text = self._validate_output_text( diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/structured_output.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/structured_output.py index 087b646d88..db17fd4d31 100644 --- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/structured_output.py +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/structured_output.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import json +from collections.abc import Collection from enum import Enum from jsonschema.exceptions import SchemaError @@ -106,18 +107,46 @@ def default_structured_output_mode(format: str) -> StructuredOutputMode: raise ValueError(f"Unsupported structured output format: {format}") -def _looks_like_unsupported_guided_json_error(message: str) -> bool: +_STRUCTURED_OUTPUT_PARAMS = ("guided_json", "nvext", "extra_body", "response_format") + + +def _rejected_field_text(lowered_message: str) -> str: + """Drop the trailing list of accepted fields so only the rejected one is matched.""" + return lowered_message.split("expected one of", 1)[0] + + +def looks_like_unsupported_structured_output_error(message: str) -> bool: + """Whether *message* reads as a backend rejecting a structured-output parameter.""" lowered = message.lower() signatures = ( - "guided_json is unsupported", - "unexpected keyword argument 'guided_json'", - "unexpected keyword argument 'nvext'", + "is unsupported", + "is not supported", + "unexpected keyword argument", "extra_forbidden", "extra inputs are not permitted", + "unknown field", ) - if any(sig in lowered for sig in signatures): - return "guided_json" in lowered or "nvext" in lowered or "extra_body" in lowered - return False + if not any(sig in lowered for sig in signatures): + return False + rejected = _rejected_field_text(lowered) + return any(param in rejected for param in _STRUCTURED_OUTPUT_PARAMS) + + +def next_structured_output_mode( + current: StructuredOutputMode, + message: str, + rejected_modes: Collection[StructuredOutputMode], +) -> StructuredOutputMode: + """Mode to try after *current* was rejected, ending at prompt-level JSON instruction.""" + rejected = _rejected_field_text(message.lower()) + if ( + current == StructuredOutputMode.NVEXT_GUIDED_JSON + and StructuredOutputMode.ROOT_GUIDED_JSON not in rejected_modes + and "nvext" in rejected + and "guided_json" not in rejected + ): + return StructuredOutputMode.ROOT_GUIDED_JSON + return StructuredOutputMode.UNSUPPORTED def _extract_chat_content(response: dict) -> str | None: @@ -176,7 +205,7 @@ async def detect_structured_output_mode( if content and _is_probe_valid_json(content, probe_schema): return mode except Exception as e: - if _looks_like_unsupported_guided_json_error(str(e)): + if looks_like_unsupported_structured_output_error(str(e)): continue # Probe failures should not abort evaluation startup. If no mode works, # caller will fall back to prompt-level strict JSON instruction. diff --git a/packages/nemo_evaluator_sdk/tests/metrics/test_llm_judge.py b/packages/nemo_evaluator_sdk/tests/metrics/test_llm_judge.py index f7ef7de8e2..f78386f372 100644 --- a/packages/nemo_evaluator_sdk/tests/metrics/test_llm_judge.py +++ b/packages/nemo_evaluator_sdk/tests/metrics/test_llm_judge.py @@ -1,6 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio +import json import math from copy import deepcopy from types import SimpleNamespace @@ -845,6 +847,257 @@ async def inference_fn(*args, **kwargs): assert "max_tokens" in captured_requests[0] assert "max_completion_tokens" in captured_requests[1] + @pytest.mark.asyncio + async def test_compute_scores_falls_back_when_backend_rejects_structured_output(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.ROOT_GUIDED_JSON) + error = ClientInferenceError( + mocker.Mock( + status_code=400, + response=mocker.Mock(text="unknown field `guided_json`, expected one of `greed_sampling`"), + ) + ) + captured_requests: list[dict] = [] + + async def inference_fn(*args, **kwargs): + request = kwargs.get("request", args[1]) + captured_requests.append(deepcopy(request)) + if len(captured_requests) == 1: + raise error + return {"choices": [{"message": {"content": '{"helpfulness": 4}'}}]} + + metric.set_inference_fn(inference_fn) + + result = await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert result.outputs[0].value == 4 + assert "guided_json" in json.dumps(captured_requests[0]) + assert "guided_json" not in json.dumps(captured_requests[1]) + assert hook.mode == StructuredOutputMode.UNSUPPORTED + + @pytest.mark.asyncio + async def test_compute_scores_falls_back_for_concurrent_in_flight_requests(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.ROOT_GUIDED_JSON) + error = ClientInferenceError( + mocker.Mock( + status_code=400, + response=mocker.Mock(text="unknown field `guided_json`, expected one of `greed_sampling`"), + ) + ) + captured_requests: list[dict] = [] + both_requests_in_flight = asyncio.Event() + structured_request_count = 0 + + async def inference_fn(*args, **kwargs): + nonlocal structured_request_count + request = kwargs.get("request", args[1]) + captured_requests.append(deepcopy(request)) + if "guided_json" in json.dumps(request): + structured_request_count += 1 + if structured_request_count == 2: + both_requests_in_flight.set() + await both_requests_in_flight.wait() + raise error + return {"choices": [{"message": {"content": '{"helpfulness": 4}'}}]} + + metric.set_inference_fn(inference_fn) + + results = await asyncio.wait_for( + asyncio.gather( + compute_scores(metric, {"prompt": "hello"}, {"output_text": "first"}), + compute_scores(metric, {"prompt": "hello"}, {"output_text": "second"}), + ), + timeout=10, + ) + + assert [result.outputs[0].value for result in results] == [4, 4] + assert structured_request_count == 2 + assert len(captured_requests) == 4 + assert all("guided_json" not in json.dumps(request) for request in captured_requests[2:]) + assert hook.mode == StructuredOutputMode.UNSUPPORTED + + @pytest.mark.asyncio + async def test_compute_scores_does_not_retry_unrelated_400(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.ROOT_GUIDED_JSON) + error = ClientInferenceError(mocker.Mock(status_code=400, response=mocker.Mock(text="unknown field `top_k`"))) + calls = 0 + + async def inference_fn(*args, **kwargs): + nonlocal calls + calls += 1 + raise error + + metric.set_inference_fn(inference_fn) + + with pytest.raises(ClientInferenceError): + await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert calls == 1 + assert hook.mode == StructuredOutputMode.ROOT_GUIDED_JSON + + @pytest.mark.asyncio + async def test_compute_scores_does_not_retry_when_structured_param_is_only_accepted(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.ROOT_GUIDED_JSON) + error = ClientInferenceError( + mocker.Mock( + status_code=400, + response=mocker.Mock(text="unknown field `top_k`, expected one of `guided_json`, `nvext`"), + ) + ) + calls = 0 + + async def inference_fn(*args, **kwargs): + nonlocal calls + calls += 1 + raise error + + metric.set_inference_fn(inference_fn) + + with pytest.raises(ClientInferenceError): + await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert calls == 1 + assert hook.mode == StructuredOutputMode.ROOT_GUIDED_JSON + + @pytest.mark.asyncio + async def test_compute_scores_does_not_retry_an_unchanged_request(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.UNSUPPORTED) + error = ClientInferenceError( + mocker.Mock(status_code=400, response=mocker.Mock(text="unknown field `guided_json`")) + ) + calls = 0 + + async def inference_fn(*args, **kwargs): + nonlocal calls + calls += 1 + raise error + + metric.set_inference_fn(inference_fn) + + with pytest.raises(ClientInferenceError): + await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert calls == 1 + + @pytest.mark.asyncio + async def test_compute_scores_marks_nan_when_the_fallback_retry_fails(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ignore_request_failure=True, + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.ROOT_GUIDED_JSON) + rejection = ClientInferenceError( + mocker.Mock(status_code=400, response=mocker.Mock(text="unknown field `guided_json`")) + ) + transient = ClientInferenceError(mocker.Mock(status_code=502, response=mocker.Mock(text="bad gateway"))) + calls = 0 + + async def inference_fn(*args, **kwargs): + nonlocal calls + calls += 1 + raise rejection if calls == 1 else transient + + metric.set_inference_fn(inference_fn) + + result = await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert calls == 2 + assert math.isnan(result.outputs[0].value) + + @pytest.mark.asyncio + async def test_compute_scores_tries_root_guided_json_before_prompt_fallback(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.NVEXT_GUIDED_JSON) + error = ClientInferenceError(mocker.Mock(status_code=400, response=mocker.Mock(text="unknown field `nvext`"))) + captured_requests: list[dict] = [] + + async def inference_fn(*args, **kwargs): + request = kwargs.get("request", args[1]) + captured_requests.append(deepcopy(request)) + if len(captured_requests) == 1: + raise error + return {"choices": [{"message": {"content": '{"helpfulness": 4}'}}]} + + metric.set_inference_fn(inference_fn) + + result = await compute_scores(metric, {"prompt": "hello"}, {"output_text": "world"}) + + assert result.outputs[0].value == 4 + assert "nvext" in captured_requests[0]["extra_body"] + assert "guided_json" in captured_requests[1]["extra_body"] + assert hook.mode == StructuredOutputMode.ROOT_GUIDED_JSON + + @pytest.mark.asyncio + async def test_concurrent_nvext_rejections_still_try_root_guided_json(self, mocker: MockerFixture): + metric = LLMJudgeMetric( + model=_make_model().model_copy(update={"format": ModelFormat.NVIDIA_NIM}), + scores=[_make_metric_score()], + ) + hook = next(h for h in metric._preprocess_hooks if isinstance(h, InferenceStructuredOutput)) + hook.set_mode(StructuredOutputMode.NVEXT_GUIDED_JSON) + error = ClientInferenceError(mocker.Mock(status_code=400, response=mocker.Mock(text="unknown field `nvext`"))) + captured_requests: list[dict] = [] + both_requests_in_flight = asyncio.Event() + nvext_request_count = 0 + + async def inference_fn(*args, **kwargs): + nonlocal nvext_request_count + request = kwargs.get("request", args[1]) + captured_requests.append(deepcopy(request)) + if "nvext" in json.dumps(request): + nvext_request_count += 1 + if nvext_request_count == 2: + both_requests_in_flight.set() + await both_requests_in_flight.wait() + raise error + return {"choices": [{"message": {"content": '{"helpfulness": 4}'}}]} + + metric.set_inference_fn(inference_fn) + + results = await asyncio.wait_for( + asyncio.gather( + compute_scores(metric, {"prompt": "hello"}, {"output_text": "first"}), + compute_scores(metric, {"prompt": "hello"}, {"output_text": "second"}), + ), + timeout=10, + ) + + assert [result.outputs[0].value for result in results] == [4, 4] + assert nvext_request_count == 2 + assert len(captured_requests) == 4 + assert all("guided_json" in request["extra_body"] for request in captured_requests[2:]) + assert hook.mode == StructuredOutputMode.ROOT_GUIDED_JSON + @pytest.mark.asyncio async def test_compute_scores_raises_invalid_output_when_ignore_failure_disabled(self, mocker: MockerFixture): metric = LLMJudgeMetric(model=_make_model(), scores=[_make_metric_score()]) diff --git a/packages/nemo_evaluator_sdk/tests/test_structured_output.py b/packages/nemo_evaluator_sdk/tests/test_structured_output.py index 4f1da5c6b1..d22a753263 100644 --- a/packages/nemo_evaluator_sdk/tests/test_structured_output.py +++ b/packages/nemo_evaluator_sdk/tests/test_structured_output.py @@ -9,8 +9,9 @@ Model, ModelFormat, StructuredOutputMode, - _looks_like_unsupported_guided_json_error, detect_structured_output_mode, + looks_like_unsupported_structured_output_error, + next_structured_output_mode, ) from pydantic import ValidationError @@ -218,7 +219,62 @@ async def inference_fn(model, request, max_retries, **kwargs): ("unexpected keyword argument 'guided_json'", True), ("unexpected keyword argument 'nvext'", True), ("extra inputs are not permitted", False), + ("unknown field `guided_json`, expected one of `greed_sampling`, `use_raw_prompt`", True), + ("unknown field `temperature`", False), + ("unknown field `top_k`, expected one of `guided_json`, `nvext`, `use_raw_prompt`", False), + ("unknown field `response_format`", True), + ("response_format is not supported", True), + ("Error code: 400 - response_format is not supported by this model", True), + ("create() got an unexpected keyword argument 'response_format'", True), + ("create() got an unexpected keyword argument 'extra_body'", True), + ("guided_json is not supported", True), + ("create() got an unexpected keyword argument 'temperature'", False), + ("model nvidia/does-not-exist is not supported", False), ], ) -def test_looks_like_unsupported_guided_json_error(message: str, expected: bool): - assert _looks_like_unsupported_guided_json_error(message) is expected +def test_looks_like_unsupported_structured_output_error(message: str, expected: bool): + assert looks_like_unsupported_structured_output_error(message) is expected + + +@pytest.mark.parametrize( + ("current", "message", "rejected_modes", "expected"), + [ + ( + StructuredOutputMode.NVEXT_GUIDED_JSON, + "unknown field `nvext`", + (), + StructuredOutputMode.ROOT_GUIDED_JSON, + ), + ( + StructuredOutputMode.NVEXT_GUIDED_JSON, + "unknown field `nvext`", + (StructuredOutputMode.ROOT_GUIDED_JSON,), + StructuredOutputMode.UNSUPPORTED, + ), + ( + StructuredOutputMode.NVEXT_GUIDED_JSON, + "unknown field `guided_json`", + (), + StructuredOutputMode.UNSUPPORTED, + ), + ( + StructuredOutputMode.ROOT_GUIDED_JSON, + "unknown field `guided_json`", + (), + StructuredOutputMode.UNSUPPORTED, + ), + ( + StructuredOutputMode.OPENAI_RESPONSE_FORMAT, + "unknown field `response_format`", + (), + StructuredOutputMode.UNSUPPORTED, + ), + ], +) +def test_next_structured_output_mode( + current: StructuredOutputMode, + message: str, + rejected_modes: tuple[StructuredOutputMode, ...], + expected: StructuredOutputMode, +): + assert next_structured_output_mode(current, message, rejected_modes) is expected diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/metrics/llm_judge.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/metrics/llm_judge.py index f22e5222c7..6ea4a660cd 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/metrics/llm_judge.py +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/metrics/llm_judge.py @@ -25,7 +25,13 @@ sample_template_payload, ) from nemo_platform.beta.evaluator.resolver_protocols import ModelResolver, SecretResolver -from nemo_platform.beta.evaluator.structured_output import InferenceStructuredOutput, detect_structured_output_mode +from nemo_platform.beta.evaluator.structured_output import ( + InferenceStructuredOutput, + StructuredOutputMode, + detect_structured_output_mode, + looks_like_unsupported_structured_output_error, + next_structured_output_mode, +) from nemo_platform.beta.evaluator.templates import render_request from nemo_platform.beta.evaluator.values.common import SecretRef, SupportedJobTypes from nemo_platform.beta.evaluator.values.llm_judge_defaults import ( @@ -82,6 +88,7 @@ class LLMJudgeMetric(HooksBase, LLMJudge): _parsers: dict[str, ScoreParser] = PrivateAttr(default_factory=dict) _score_dumps: dict[str, dict[str, Any]] = PrivateAttr(default_factory=dict) _prompt_template_is_default: bool = PrivateAttr(default=False) + _rejected_structured_output_modes: set[StructuredOutputMode] = PrivateAttr(default_factory=set) job_type: Literal[SupportedJobTypes.ONLINE, SupportedJobTypes.OFFLINE] = SupportedJobTypes.ONLINE @property @@ -222,12 +229,7 @@ async def preflight(self) -> None: if model.format != ModelFormat.NVIDIA_NIM or not self.structured_output: return - structured_hook: InferenceStructuredOutput | None = None - for hook in self._preprocess_hooks: - if isinstance(hook, InferenceStructuredOutput): - structured_hook = hook - break - + structured_hook = self._structured_output_hook() if structured_hook is None: return @@ -298,6 +300,9 @@ def _initialize_score_parsers(self) -> None: self._score_dumps[score.name] = score.model_dump(mode="json", exclude={"parser"}) def _render_request(self, item: dict, sample: TemplateSample) -> dict: + return self._apply_preprocess_hooks(self._render_base_request(item, sample)) + + def _render_base_request(self, item: dict, sample: TemplateSample) -> dict: sample_payload = sample_template_payload(sample) overlapping_keys = set(item.keys()) & set(sample_payload.keys()) if overlapping_keys: @@ -326,7 +331,38 @@ def _render_request(self, item: dict, sample: TemplateSample) -> dict: request["max_completion_tokens"] = request["max_tokens"] del request["max_tokens"] - return self._apply_preprocess_hooks(request) + return request + + def _structured_output_hook(self) -> InferenceStructuredOutput | None: + for hook in self._preprocess_hooks: + if isinstance(hook, InferenceStructuredOutput): + return hook + return None + + def _downgrade_structured_output(self, error: Exception, rendered_mode: StructuredOutputMode | None) -> bool: + """Latch the next structured-output mode when the backend rejects *rendered_mode*.""" + message = str(error) + if not looks_like_unsupported_structured_output_error(message): + return False + + hook = self._structured_output_hook() + if hook is None: + return False + + if ( + rendered_mode is not None + and rendered_mode != StructuredOutputMode.UNSUPPORTED + and rendered_mode not in self._rejected_structured_output_modes + ): + next_mode = next_structured_output_mode(rendered_mode, message, self._rejected_structured_output_modes) + self._rejected_structured_output_modes.add(rendered_mode) + _logger.warning( + "Judge model rejected structured output mode %s; using %s for all future requests.", + rendered_mode.value, + next_mode.value, + ) + hook.set_mode(next_mode) + return True def _retry_with_max_completion_tokens(self, request: dict) -> dict: if not self._use_max_completion_tokens: @@ -338,24 +374,49 @@ def _retry_with_max_completion_tokens(self, request: dict) -> dict: del request["max_tokens"] return request + def _retry_request( + self, + error: Exception, + request: dict, + base_request: dict, + rendered_mode: StructuredOutputMode | None, + ) -> dict | None: + """Request to retry the rejected call with, or None when no retry can help.""" + if "max_tokens" in request and "'max_tokens' is not supported with this model" in error.args[0]: + return self._retry_with_max_completion_tokens(request) + if self._downgrade_structured_output(error, rendered_mode): + retried = self._apply_preprocess_hooks(deepcopy(base_request)) + if retried != request: + return retried + return None + async def compute_scores(self, input: MetricInput) -> MetricResult: """Compute structured score output for one item/sample pair.""" item = input.row.data sample = input.candidate - request = self._render_request(item, sample) + base_request = self._render_base_request(item, sample) + request = self._apply_preprocess_hooks(deepcopy(base_request)) + hook = self._structured_output_hook() + rendered_mode = hook.mode if hook else None try: response = await self.inference_fn(self._require_model(), request, 3, client=self.client) except inference.ClientInferenceError as error: - if "max_tokens" in request and "'max_tokens' is not supported with this model" in error.args[0]: - request = self._retry_with_max_completion_tokens(request) - response = await self.inference_fn(self._require_model(), request, 3, client=self.client) - else: + retry_request = self._retry_request(error, request, base_request, rendered_mode) + if retry_request is None: return self._handle_invalid_output( error, self._nan_result(), "Inference failed with LLM judge, marking as NaN", ) + try: + response = await self.inference_fn(self._require_model(), retry_request, 3, client=self.client) + except inference.ClientInferenceError as retry_error: + return self._handle_invalid_output( + retry_error, + self._nan_result(), + "Inference failed with LLM judge, marking as NaN", + ) try: output_text = self._validate_output_text( diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/structured_output.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/structured_output.py index b9565c39e4..ff01f496c5 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/structured_output.py +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/structured_output.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import json +from collections.abc import Collection from enum import Enum from jsonschema.exceptions import SchemaError @@ -106,18 +107,46 @@ def default_structured_output_mode(format: str) -> StructuredOutputMode: raise ValueError(f"Unsupported structured output format: {format}") -def _looks_like_unsupported_guided_json_error(message: str) -> bool: +_STRUCTURED_OUTPUT_PARAMS = ("guided_json", "nvext", "extra_body", "response_format") + + +def _rejected_field_text(lowered_message: str) -> str: + """Drop the trailing list of accepted fields so only the rejected one is matched.""" + return lowered_message.split("expected one of", 1)[0] + + +def looks_like_unsupported_structured_output_error(message: str) -> bool: + """Whether *message* reads as a backend rejecting a structured-output parameter.""" lowered = message.lower() signatures = ( - "guided_json is unsupported", - "unexpected keyword argument 'guided_json'", - "unexpected keyword argument 'nvext'", + "is unsupported", + "is not supported", + "unexpected keyword argument", "extra_forbidden", "extra inputs are not permitted", + "unknown field", ) - if any(sig in lowered for sig in signatures): - return "guided_json" in lowered or "nvext" in lowered or "extra_body" in lowered - return False + if not any(sig in lowered for sig in signatures): + return False + rejected = _rejected_field_text(lowered) + return any(param in rejected for param in _STRUCTURED_OUTPUT_PARAMS) + + +def next_structured_output_mode( + current: StructuredOutputMode, + message: str, + rejected_modes: Collection[StructuredOutputMode], +) -> StructuredOutputMode: + """Mode to try after *current* was rejected, ending at prompt-level JSON instruction.""" + rejected = _rejected_field_text(message.lower()) + if ( + current == StructuredOutputMode.NVEXT_GUIDED_JSON + and StructuredOutputMode.ROOT_GUIDED_JSON not in rejected_modes + and "nvext" in rejected + and "guided_json" not in rejected + ): + return StructuredOutputMode.ROOT_GUIDED_JSON + return StructuredOutputMode.UNSUPPORTED def _extract_chat_content(response: dict) -> str | None: @@ -176,7 +205,7 @@ async def detect_structured_output_mode( if content and _is_probe_valid_json(content, probe_schema): return mode except Exception as e: - if _looks_like_unsupported_guided_json_error(str(e)): + if looks_like_unsupported_structured_output_error(str(e)): continue # Probe failures should not abort evaluation startup. If no mode works, # caller will fall back to prompt-level strict JSON instruction.