feat(llm): robust_json_parse 五级兜底——QA 范式+tickflow 尾部修复,纯函数 [vps] [no-doc]

sanguo_api/llm/json_utils.py

@@ -0,0 +1,74 @@
+# sanguo_api/llm/json_utils.py
+"""LLM 输出 JSON 五级兜底解析(纯函数,零 IO).
+
+搬运: QuantaAlpha llm/client.py:36-121 robust_json_parse 五级降级
++ tickflow custom_signals_ai.py:198-214 尾随垃圾/尾逗号修复(调研报告 §1/§2).
+"""
+from __future__ import annotations
+
+import json
+import re
+
+_FENCE = re.compile(r"```(?:json)?\s*(.+?)\s*```", re.S)
+_TRAILING_COMMA = re.compile(r",\s*([}\]])")
+
+
+class JSONParseError(Exception):
+    """文本无法解析为 JSON object."""
+
+
+def _loads_dict(text: str) -> dict:
+    obj = json.loads(text)
+    if not isinstance(obj, dict):
+        raise JSONParseError(f"顶层不是 JSON object: {type(obj).__name__}")
+    return obj
+
+
+def _balanced_braces(text: str) -> str | None:
+    """首个字符串感知的配平 {} 子串(tickflow 三级候选之三)."""
+    start = text.find("{")
+    if start < 0:
+        return None
+    depth, in_str, esc = 0, False, False
+    for i in range(start, len(text)):
+        ch = text[i]
+        if in_str:
+            if esc:
+                esc = False
+            elif ch == "\\":
+                esc = True
+            elif ch == '"':
+                in_str = False
+            continue
+        if ch == '"':
+            in_str = True
+        elif ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                return text[start:i + 1]
+    return None
+
+
+def robust_json_parse(text: str) -> dict:
+    """五级降级:直读→围栏→配平截取→反斜杠修复→宽松(尾逗号/尾垃圾).全败 raise."""
+    candidates: list[str] = [text]
+    m = _FENCE.search(text)
+    if m:
+        candidates.append(m.group(1))
+    balanced = _balanced_braces(text)
+    if balanced:
+        candidates.append(balanced)
+        # L4: 常见 LaTeX 转义残留破坏 json.loads
+        candidates.append(re.sub(r"\\([()\[\]])", r"\1", balanced))
+        # L5: 尾逗号清洗 + 尾随垃圾截断(重截一次)
+        cleaned = _TRAILING_COMMA.sub(r"\1", balanced)
+        candidates.append(cleaned)
+    last_err: Exception = JSONParseError("空文本")
+    for cand in candidates:
+        try:
+            return _loads_dict(cand)
+        except (json.JSONDecodeError, JSONParseError) as e:
+            last_err = e
+    raise JSONParseError(f"无法解析为 JSON object: {last_err}")