Spaces:
Paused
Paused
File size: 1,827 Bytes
1bc3f18 | 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 | import json
import logging
from typing import Any, Dict, Optional
logger = logging.getLogger("ExamGraph")
def safe_parse(parser_obj, text: str, question_no: int) -> Optional[Dict[str, Any]]:
if not text or text.strip() in ("null", "None", ""):
logger.warning(f"[Parse] q{question_no}: empty/null response")
return None
last_error = None
# Try direct parse
try:
result = parser_obj.parse(text)
return result.model_dump() if hasattr(result, "model_dump") else result
except Exception as e:
last_error = e
logger.debug(f"[Parse] q{question_no}: direct parse failed, trying extraction")
# Try to extract JSON from text (LLM may have wrapped it in prose)
try:
# look for {...} pattern
start = text.rfind("{")
end = text.rfind("}") + 1
if start >= 0 and end > start:
json_str = text[start:end]
json_obj = json.loads(json_str)
result = parser_obj.parse(json.dumps(json_obj))
return result.model_dump() if hasattr(result, "model_dump") else result
except Exception as e:
last_error = e
logger.debug(f"[Parse] q{question_no}: json extraction failed")
# Last resort: if it looks like partial JSON, mark for regen
error_msg = str(last_error) if last_error else "unknown"
logger.error(f"[Parse] q{question_no}: failed all attempts: {error_msg}")
return None
def categorize_error(error_str: str) -> str:
err = error_str.lower()
if "timeout" in err:
return "timeout"
elif "json" in err or "invalid" in err:
return "invalid_json"
elif "field required" in err or "missing" in err:
return "missing_field"
elif "none" in err or "null" in err:
return "null"
return "unknown"
|