|
| 1 | +import json |
| 2 | +import re |
| 3 | +from typing import Dict, Any, List, Optional |
| 4 | + |
| 5 | +# ============================================================================== |
| 6 | +# HELPER FUNCTIONS (Based on your provided robust fixers) |
| 7 | +# ============================================================================== |
| 8 | +# For clarity, these are named as internal functions (prefixed with an underscore). |
| 9 | + |
| 10 | +def _fix_json_quotes(s: str) -> str: |
| 11 | + """First-stage repair: handle incorrect quotes and basic structure.""" |
| 12 | + # Replace Python-style booleans/None with JSON standard |
| 13 | + s = re.sub(r'\bTrue\b', 'true', s) |
| 14 | + s = re.sub(r'\bFalse\b', 'false', s) |
| 15 | + s = re.sub(r'\bNone\b', 'null', s) |
| 16 | + |
| 17 | + # Attempt to replace single quotes with double quotes (a common VLM error) |
| 18 | + # This is a high-risk operation that might break the reasoning content, |
| 19 | + # but it's worth trying early on. |
| 20 | + try: |
| 21 | + temp_s = s.replace("'", '"') |
| 22 | + json.loads(temp_s) |
| 23 | + return temp_s |
| 24 | + except json.JSONDecodeError: |
| 25 | + # If it's still invalid after replacement, return the original string for the next repair step. |
| 26 | + pass |
| 27 | + |
| 28 | + # Add double quotes to keys (e.g., {reasoning: ...} -> {"reasoning": ...}) |
| 29 | + s = re.sub(r'([\{\s,])(\w+)\s*:', r'\1"\2":', s) |
| 30 | + return s |
| 31 | + |
| 32 | +def _repair_reasoning_field_robust(json_str: str) -> str: |
| 33 | + """Second-stage repair: specifically fix unescaped double quotes inside the 'reasoning' field.""" |
| 34 | + pattern = re.compile( |
| 35 | + r'("reasoning"\s*:\s*")' # --- Group 1: "reasoning" : " |
| 36 | + r'(.*?)' # --- Group 2: The content (non-greedy) |
| 37 | + r'(?="\s*[,}])', # --- Lookahead: find " followed by , or } |
| 38 | + re.DOTALL |
| 39 | + ) |
| 40 | + |
| 41 | + def replacer(match): |
| 42 | + prefix = match.group(1) |
| 43 | + content = match.group(2) |
| 44 | + # In the content, replace all unescaped " with \" |
| 45 | + fixed_content = content.replace('"', '\\"') |
| 46 | + return prefix + fixed_content |
| 47 | + |
| 48 | + return pattern.sub(replacer, json_str) |
| 49 | + |
| 50 | +def _fallback_extract_and_rebuild(input_str: str) -> str: |
| 51 | + """Final fallback strategy: abandon repair, directly extract information, and rebuild a valid JSON.""" |
| 52 | + # 1. Extract reasoning |
| 53 | + # Find all content between "reasoning": and ,"score": |
| 54 | + reasoning_text = "" |
| 55 | + reason_match = re.search(r'["\']reasoning["\']\s*:\s*["\']?(.*?)["\']?\s*,\s*["\']score["\']', input_str, re.DOTALL | re.IGNORECASE) |
| 56 | + if reason_match: |
| 57 | + reasoning_text = reason_match.group(1).strip() |
| 58 | + # Clean up any potentially remaining escape characters |
| 59 | + reasoning_text = reasoning_text.replace('\\"', '"') |
| 60 | + else: |
| 61 | + # If not found, assume all text besides the score part is the reasoning. |
| 62 | + # First, remove the score part. |
| 63 | + score_part_match = re.search(r'["\']score["\']\s*:.*', input_str, re.IGNORECASE) |
| 64 | + if score_part_match: |
| 65 | + reasoning_text = input_str[:score_part_match.start()].strip() |
| 66 | + else: |
| 67 | + # If even 'score' cannot be found, assume the entire string is the reasoning. |
| 68 | + reasoning_text = input_str |
| 69 | + |
| 70 | + # 2. Extract scores |
| 71 | + scores = [] |
| 72 | + # Prioritize searching after the 'score' keyword. |
| 73 | + score_match = re.search(r'["\']score["\']\s*:\s*(.*)', input_str, re.DOTALL | re.IGNORECASE) |
| 74 | + search_area = score_match.group(1) if score_match else input_str |
| 75 | + |
| 76 | + # Find all integers or floats. |
| 77 | + numbers = re.findall(r'[-+]?\d*\.?\d+', search_area) |
| 78 | + if numbers: |
| 79 | + scores = [float(num) for num in numbers] |
| 80 | + |
| 81 | + # 3. Rebuild into a standard dictionary and return a JSON string. |
| 82 | + rebuilt_data = { |
| 83 | + "reasoning": reasoning_text, |
| 84 | + "score": scores |
| 85 | + } |
| 86 | + return json.dumps(rebuilt_data, ensure_ascii=False) |
| 87 | + |
| 88 | + |
| 89 | +def _format_and_validate_dict(data: Dict[str, Any]) -> Optional[Dict[str, Any]]: |
| 90 | + """Validate and format the parsed dictionary to ensure it meets the final output standard.""" |
| 91 | + if not isinstance(data, dict): |
| 92 | + return None |
| 93 | + |
| 94 | + # Extract reasoning, tolerating case and spelling variations. |
| 95 | + reasoning = "" |
| 96 | + for key in ["reasoning", "reason", "rationale"]: |
| 97 | + if key in data and isinstance(data[key], str): |
| 98 | + reasoning = data[key] |
| 99 | + break |
| 100 | + |
| 101 | + # Extract score, and ensure it is a list of floats. |
| 102 | + scores = [] |
| 103 | + if 'score' in data: |
| 104 | + score_val = data['score'] |
| 105 | + if isinstance(score_val, list): |
| 106 | + scores = [float(s) for s in score_val if isinstance(s, (int, float, str))] |
| 107 | + elif isinstance(score_val, (int, float)): |
| 108 | + scores = [float(score_val)] |
| 109 | + |
| 110 | + # If any field was found, consider it a success. |
| 111 | + if reasoning or scores: |
| 112 | + return {"score": scores, "reasoning": reasoning} |
| 113 | + |
| 114 | + return None |
| 115 | + |
| 116 | +# ============================================================================== |
| 117 | +# MAIN PARSING FUNCTION |
| 118 | +# ============================================================================== |
| 119 | + |
| 120 | +def parse_vlm_output_to_dict(input_string: str) -> Dict[str, Any]: |
| 121 | + """ |
| 122 | + A highly robust function to parse a VLM's output string into a dictionary |
| 123 | + containing 'score' and 'reasoning'. |
| 124 | +
|
| 125 | + It uses a multi-stage repair pipeline, progressively degrading from standard |
| 126 | + JSON parsing to a final information extraction fallback. |
| 127 | + """ |
| 128 | + # --- 0. Preprocessing --- |
| 129 | + if not input_string or not input_string.strip(): |
| 130 | + return {"score": [], "reasoning": "Input was empty."} |
| 131 | + |
| 132 | + # Find the substring enclosed by `{}`, which is often the core of the VLM output. |
| 133 | + json_match = re.search(r'\{.*\}', input_string, re.DOTALL) |
| 134 | + target_str = json_match.group(0) if json_match else input_string.strip() |
| 135 | + |
| 136 | + # --- Repair Pipeline --- |
| 137 | + # Apply fixers in order, attempting to parse after each one. |
| 138 | + |
| 139 | + fixer_pipeline = [ |
| 140 | + lambda s: s, # 1. Try the original string. |
| 141 | + _fix_json_quotes, # 2. Fix basic quotes and keywords. |
| 142 | + _repair_reasoning_field_robust, # 3. Fix internal quotes in the reasoning field. |
| 143 | + ] |
| 144 | + |
| 145 | + for fixer in fixer_pipeline: |
| 146 | + try: |
| 147 | + fixed_str = fixer(target_str) |
| 148 | + data = json.loads(fixed_str) |
| 149 | + validated_data = _format_and_validate_dict(data) |
| 150 | + if validated_data is not None: |
| 151 | + return validated_data |
| 152 | + except (json.JSONDecodeError, TypeError): |
| 153 | + # If it fails, continue to the next fixer. |
| 154 | + continue |
| 155 | + |
| 156 | + # --- Final Fallback Strategy --- |
| 157 | + # If all repair and parsing attempts fail, activate the information extraction mode. |
| 158 | + try: |
| 159 | + fallback_str = _fallback_extract_and_rebuild(target_str) |
| 160 | + # This function guarantees a valid JSON string, so we can load it directly. |
| 161 | + data = json.loads(fallback_str) |
| 162 | + # Still run it through the validator to standardize the format. |
| 163 | + validated_data = _format_and_validate_dict(data) |
| 164 | + if validated_data: |
| 165 | + return validated_data |
| 166 | + except Exception: |
| 167 | + # If even the final fallback strategy fails, return an error message. |
| 168 | + pass |
| 169 | + |
| 170 | + return { |
| 171 | + "score": [], |
| 172 | + "reasoning": f"Failed to parse after all strategies. Original output: '{input_string}'" |
| 173 | + } |
0 commit comments