Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 34 additions & 1 deletion resources_servers/genrm_compare/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,21 @@

logger = logging.getLogger(__name__)


def _output_budget_exhausted(raw_response: Any) -> bool:
"""True if a Responses API object was cut off by ``max_output_tokens``.

The model spent its whole output budget (typically on reasoning) and never emitted a
verdict: ``status="incomplete"`` with ``incomplete_details.reason="max_output_tokens"``.
Gym's chat->Responses conversion produces this for ``finish_reason="length"``; hosted
Responses API backends emit it natively.
"""
if not isinstance(raw_response, dict) or raw_response.get("status") != "incomplete":
return False
details = raw_response.get("incomplete_details") or {}
return isinstance(details, dict) and details.get("reason") == "max_output_tokens"


GROUP_ID_KEY_NAME = "_ng_group_id"
GROUP_ATTEMPT_KEY_NAME = "_ng_group_attempt"

Expand Down Expand Up @@ -961,6 +976,7 @@ async def call():

max_attempts = max(1, int(cfg.genrm_parse_retries) + 1)
saw_completed_answer = False
budget_exhausted_attempts = 0
for attempt_idx in range(max_attempts):
try:
raw_response = await call()
Expand All @@ -984,6 +1000,16 @@ async def call():
and bool(answer.strip())
)
saw_completed_answer |= usable
if _output_budget_exhausted(raw_response):
budget_exhausted_attempts += 1
logger.warning(
"GenRM output budget exhausted for pair %s (attempt %s/%s, max_output_tokens=%s): "
"response is incomplete (reason=max_output_tokens) and contains no verdict",
pair_idx,
attempt_idx + 1,
max_attempts,
responses_create_params.max_output_tokens,
)
try:
if not usable:
raise GenRMOutputParseError("Judge returned an empty or unsuccessful response")
Expand All @@ -993,7 +1019,14 @@ async def call():
await asyncio.sleep(float(cfg.genrm_parse_retry_sleep_s))
continue
if not saw_completed_answer:
raise JudgeError(f"Judge returned no completed answer after {max_attempts} attempts") from error
message = f"Judge returned no completed answer after {max_attempts} attempts"
if budget_exhausted_attempts:
message += (
f" ({budget_exhausted_attempts} of them exhausted max_output_tokens="
f"{responses_create_params.max_output_tokens} without emitting a verdict; "
"raise the budget or constrain the judge's reasoning)"
)
raise JudgeError(message) from error
# Preserve the existing fallback for completed, nonempty but malformed answers.
logger.warning(
"GenRM parse failed for pair %s after %s attempts; using defaults", pair_idx, max_attempts
Expand Down
90 changes: 90 additions & 0 deletions resources_servers/genrm_compare/tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
ROLLOUT_INDEX_KEY_NAME,
TASK_INDEX_KEY_NAME,
)
from nemo_gym.judge import JudgeError
from nemo_gym.openai_utils import (
NeMoGymEasyInputMessage,
NeMoGymResponse,
Expand All @@ -45,6 +46,7 @@
GenRMCompareResponse,
GenRMCompareVerifyRequest,
_input_to_conversation_history,
_output_budget_exhausted,
)
from resources_servers.genrm_compare.utils import get_prompt_key_from_input

Expand Down Expand Up @@ -1096,3 +1098,91 @@ def test_principle_absent_from_metadata_when_disabled(self):

body = self._get_sent_body(mock_client)
assert "principle" not in body.metadata

@staticmethod
def _http_response(body: dict):
response = AsyncMock(ok=True)
response.read = AsyncMock(return_value=json.dumps(body).encode())
return response

@staticmethod
def _incomplete_response():
"""A Responses API object cut off by max_output_tokens: reasoning present, no verdict text."""
return {
"status": "incomplete",
"incomplete_details": {"reason": "max_output_tokens"},
"output": [{"type": "reasoning", "summary": [], "content": [{"type": "reasoning_text", "text": "..."}]}],
}

def test_output_budget_exhausted_detection(self):
assert _output_budget_exhausted(self._incomplete_response())
assert not _output_budget_exhausted({"status": "completed", "output": []})
assert not _output_budget_exhausted(
{"status": "incomplete", "incomplete_details": {"reason": "content_filter"}}
)
assert not _output_budget_exhausted({"status": "incomplete"})
assert not _output_budget_exhausted(None)

def test_budget_exhaustion_is_named_in_judge_error(self, caplog):
"""Every attempt hits max_output_tokens -> JudgeError says so, with the budget, instead of a generic message."""
server, mock_client = self._make_server()
server.config.genrm_parse_retries = 1
server.config.genrm_parse_retry_sleep_s = 0.0
mock_client.post = AsyncMock(return_value=self._http_response(self._incomplete_response()))

with caplog.at_level("WARNING"), pytest.raises(JudgeError) as error:
asyncio.run(
server._run_single_comparison(
[{"role": "user", "content": "q"}], self._make_response_obj("a"), self._make_response_obj("b")
)
)

assert mock_client.post.await_count == 2
assert "no completed answer after 2 attempts" in str(error.value)
assert "2 of them exhausted max_output_tokens=1024" in str(error.value)
assert sum("output budget exhausted" in r.getMessage() for r in caplog.records) == 2

def test_budget_exhaustion_then_verdict_recovers(self):
"""One exhausted attempt followed by a completed verdict parses normally."""
server, mock_client = self._make_server()
server.config.genrm_parse_retries = 1
server.config.genrm_parse_retry_sleep_s = 0.0
completed = {
"status": "completed",
"output": [
{
"type": "message",
"content": [{"type": "output_text", "text": '{"score_1": 4, "score_2": 2, "ranking": 2}'}],
}
],
}
mock_client.post = AsyncMock(
side_effect=[self._http_response(self._incomplete_response()), self._http_response(completed)]
)

result = asyncio.run(
server._run_single_comparison(
[{"role": "user", "content": "q"}], self._make_response_obj("a"), self._make_response_obj("b")
)
)

assert result == (4.0, 2.0, 2.0)

def test_empty_completed_answer_keeps_generic_judge_error(self):
"""A completed but empty answer is not budget exhaustion; the message stays generic."""
server, mock_client = self._make_server()
server.config.genrm_parse_retries = 0
empty = {
"status": "completed",
"output": [{"type": "message", "content": [{"type": "output_text", "text": ""}]}],
}
mock_client.post = AsyncMock(return_value=self._http_response(empty))

with pytest.raises(JudgeError) as error:
asyncio.run(
server._run_single_comparison(
[{"role": "user", "content": "q"}], self._make_response_obj("a"), self._make_response_obj("b")
)
)

assert str(error.value) == "Judge returned no completed answer after 1 attempts"
Loading