File size: 7,784 Bytes
d13a83d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | from __future__ import annotations
import json
import math
import os
import re
from typing import Any, Mapping
LEGACY_ACTOR_SCHEMA = "reason_rating_suggestion"
REASONS_RATING_ACTOR_SCHEMA = "reasons_rating"
REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA = (
"reasoning_evidence_solution_rating"
)
SUPPORTED_ACTOR_SCHEMAS = {
LEGACY_ACTOR_SCHEMA,
REASONS_RATING_ACTOR_SCHEMA,
REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA,
}
TOP_LEVEL_FIELDS = ("reason", "rating", "suggestion")
REASONS_RATING_FIELDS = ("reasons", "rating")
REASONING_RATING_FIELDS = ("reasoning", "rating")
REASONING_FIELDS = ("evidence", "solution")
NUMBER_RE = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)$")
QWEN35_NON_THINKING_PREFIX = "<think>\n\n</think>\n\n"
def actor_schema() -> str:
schema = os.environ.get("VF_ACTOR_SCHEMA", LEGACY_ACTOR_SCHEMA).strip()
if schema not in SUPPORTED_ACTOR_SCHEMAS:
raise ValueError(f"unsupported VF_ACTOR_SCHEMA: {schema}")
return schema
def active_top_level_fields() -> tuple[str, ...]:
if actor_schema() == REASONS_RATING_ACTOR_SCHEMA:
return REASONS_RATING_FIELDS
if actor_schema() == REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA:
return REASONING_RATING_FIELDS
return TOP_LEVEL_FIELDS
def strip_qwen35_non_thinking_prefix(text: str) -> str:
raw = str(text or "")
if raw.startswith(QWEN35_NON_THINKING_PREFIX):
return raw[len(QWEN35_NON_THINKING_PREFIX):]
return raw
def actor_rating_number(value: object) -> float | None:
number = unbounded_rating_number(value)
if number is None or not 1.0 <= number <= 5.0:
return None
return number
def unbounded_rating_number(value: object) -> float | None:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
number = float(value)
elif isinstance(value, str) and NUMBER_RE.fullmatch(value.strip()):
number = float(value.strip())
else:
return None
if not math.isfinite(number):
return None
return number
def score_number(value: object) -> float | None:
"""Parse a bounded external score without repairing invalid values."""
return actor_rating_number(value)
def parse_actor_json(text: str) -> dict[str, Any] | None:
raw = strip_qwen35_non_thinking_prefix(text).strip()
if not raw:
return None
decoder = json.JSONDecoder()
try:
payload, end = decoder.raw_decode(raw)
except (TypeError, ValueError, json.JSONDecodeError):
return None
if raw[end:].strip() or not isinstance(payload, dict):
return None
return payload
def parse_tokenizable_actor_json(text: str) -> dict[str, Any] | None:
"""Return ordered field values when semantic rating credit can be located."""
payload = parse_actor_json(text)
fields = active_top_level_fields()
if payload is None or tuple(payload) != fields:
return None
schema = actor_schema()
if schema == REASONS_RATING_ACTOR_SCHEMA:
if not isinstance(payload.get("reasons"), str) or not payload["reasons"].strip():
return None
elif schema == REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA:
# Rating credit remains locatable when the nested reasoning payload is
# malformed. Editor/Judge eligibility is validated separately.
pass
elif not isinstance(payload.get("reason"), str) or not isinstance(payload.get("suggestion"), str):
return None
if unbounded_rating_number(payload.get("rating")) is None:
return None
return payload
def actor_payload_errors(payload: object) -> list[str]:
if not isinstance(payload, dict):
return ["payload:not_object"]
fields = active_top_level_fields()
keys = set(payload)
expected = set(fields)
errors = [f"top_level:missing:{key}" for key in fields if key not in keys]
errors.extend(f"top_level:unexpected:{key}" for key in sorted(keys - expected))
if not errors and tuple(payload) != fields:
errors.append("top_level:order")
if errors:
return errors
schema = actor_schema()
if schema == REASONS_RATING_ACTOR_SCHEMA:
if not isinstance(payload.get("reasons"), str):
errors.append("reasons:not_string")
elif not payload["reasons"].strip():
errors.append("reasons:empty")
elif schema == REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA:
reasoning = payload.get("reasoning")
if not isinstance(reasoning, dict):
errors.append("reasoning:not_object")
else:
reasoning_keys = set(reasoning)
expected_reasoning = set(REASONING_FIELDS)
errors.extend(
f"reasoning:missing:{key}"
for key in REASONING_FIELDS
if key not in reasoning_keys
)
errors.extend(
f"reasoning:unexpected:{key}"
for key in sorted(reasoning_keys - expected_reasoning)
)
if not any(error.startswith("reasoning:") for error in errors):
if tuple(reasoning) != REASONING_FIELDS:
errors.append("reasoning:order")
evidence = reasoning.get("evidence")
solution = reasoning.get("solution")
if not isinstance(evidence, str):
errors.append("evidence:not_string")
elif not evidence.strip():
errors.append("evidence:empty")
if not isinstance(solution, str):
errors.append("solution:not_string")
elif not solution.strip():
errors.append("solution:empty")
else:
if not isinstance(payload.get("reason"), str):
errors.append("reason:not_string")
if actor_rating_number(payload.get("rating")) is None:
errors.append("rating:invalid")
if actor_schema() == LEGACY_ACTOR_SCHEMA and not isinstance(payload.get("suggestion"), str):
errors.append("suggestion:not_string")
return errors
def to_internal_actor_payload(payload: Mapping[str, Any]) -> dict[str, Any]:
"""Map the strict actor-visible schema onto stable reward/editor names."""
schema = actor_schema()
if schema == REASONS_RATING_ACTOR_SCHEMA:
return {
"think": payload["reasons"],
"rating": payload["rating"],
"editing": "",
}
if schema == REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA:
reasoning = payload["reasoning"]
evidence = reasoning["evidence"]
solution = reasoning["solution"]
return {
"think": f"{evidence}\n{solution}",
"rating": payload["rating"],
"editing": solution,
"evidence": evidence,
"solution": solution,
}
return {
"think": payload["reason"],
"rating": payload["rating"],
"editing": payload["suggestion"],
}
def parse_valid_actor_json(text: str) -> tuple[dict[str, Any] | None, list[str]]:
payload = parse_actor_json(text)
errors = actor_payload_errors(payload)
return (payload if not errors else None), errors
def parse_valid_reasoning_component_json(
text: str,
) -> tuple[dict[str, Any] | None, list[str]]:
"""Validate nested reasoning independently from rating value semantics."""
payload = parse_actor_json(text)
errors = actor_payload_errors(payload)
if actor_schema() != REASONING_EVIDENCE_SOLUTION_RATING_ACTOR_SCHEMA:
return (payload if not errors else None), errors
reasoning_errors = [
error for error in errors if error != "rating:invalid"
]
return (
payload if payload is not None and not reasoning_errors else None,
reasoning_errors,
)
|