All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 8m52s
首席架構師 2026-04-19 Review (92/100 Grade A) 指出 P1 優化:
1. LLM JSON 3-path parse 邏輯在 4 scanner 重複 (~80 行 × 4 = 320 行)
2. coverage red>=20 觸發閾值偏低,生產 bootstrap 必觸發浪費 token
P1.1+1.2 新增 services/llm_json_parser.py (~90 行):
parse_llm_json_response(text, required_key, logger_context)
3-path fallback:
Path 1: 剝 markdown fence + 直接 JSON 含 required_key
Path 2: NemoTron wrapper (description/action_title/reasoning 內嵌 JSON)
Path 3: 所有失敗 return None + logger.warning
失敗永不 raise,呼叫者決定 fallback.
4 個 LLM scanner 改用 helper:
- hermes_rule_quality_job: required_key='recommended_actions'
- capacity_forecaster_job: required_key='priority_actions'
- compliance_scanner_job: required_key='posture_grade'
- coverage_evaluator_job: required_key='worst_dimension'
每個減少約 20 行重複.
P1.3 coverage 觸發條件改雙條件:
原: total_red >= 20 (bootstrap 必觸發)
新: red_ratio > 30% AND total_scanned >= 50
_fetch_red_summary 加 total_scanned 回傳供計算.
5/5 單元測試 parse_llm_json_response:
✅ direct / markdown fence / NemoTron wrapper / invalid / missing key
P1.4 capacity_scanner + rule_catalog_sync: 檢查後已有完整作者註解 (Review 誤判).
其他 P1 (Prom HTTP helper / first_delay 錯開 / LLM budget guard) 留下 session.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
"""
|
||
LLM JSON Response Parser — 共用 helper (Gap 3 Review P1 優化)
|
||
==============================================================
|
||
4 個 LLM scanner 原本各自重複 3-path JSON parse 邏輯 (~80 行 × 4 = 320 行).
|
||
抽成統一 helper,未來擴加 LLM 的 service 直接呼叫.
|
||
|
||
Origin (2026-04-19 pr-review-toolkit code review 指出):
|
||
- hermes_rule_quality_job:167-186
|
||
- capacity_forecaster_job:204-221
|
||
- compliance_scanner_job:446-463
|
||
- coverage_evaluator_job:241-257
|
||
全部重複 strip ``` → direct parse → NemoTron wrapper → description 內嵌 JSON 的 pattern.
|
||
|
||
2026-04-19 ogt + Claude Opus 4.7 (1M context) Asia/Taipei
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json as _json
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
|
||
def parse_llm_json_response(
|
||
text: str,
|
||
required_key: str,
|
||
logger_context: str = "",
|
||
) -> dict[str, Any] | None:
|
||
"""
|
||
解析 LLM JSON 回應,含 3-path fallback.
|
||
|
||
Args:
|
||
text: LLM 回傳的原始文字
|
||
required_key: JSON 必須含的 key (e.g. 'posture_grade', 'priority_actions')
|
||
用來區分「真 JSON」vs 「NemoTron wrapper」
|
||
logger_context: log warning 時附加的 context (e.g. 'compliance_scan')
|
||
|
||
Returns:
|
||
dict | None — 解析成功且含 required_key 回 dict,否則 None
|
||
|
||
3-path fallback:
|
||
Path 1: 直接 JSON → 有 required_key 就用
|
||
Path 2: NemoTron wrapper (description/action_title/reasoning 內嵌 JSON 字串)
|
||
Path 3: Path 2 的 description 含 required_key 直接採用
|
||
|
||
失敗時 log warning 但不 raise,呼叫者決定如何 fallback.
|
||
"""
|
||
if not text:
|
||
return None
|
||
|
||
_raw = text.strip()
|
||
# 剝去 markdown code fence ```json...``` 或 ```...```
|
||
if _raw.startswith("```"):
|
||
_raw = _raw.strip("`").lstrip("json").strip()
|
||
|
||
try:
|
||
parsed = _json.loads(_raw)
|
||
except (_json.JSONDecodeError, ValueError) as e:
|
||
logger.warning(
|
||
"llm_json_parse_raw_failed",
|
||
context=logger_context,
|
||
error=str(e),
|
||
raw_prefix=_raw[:200],
|
||
)
|
||
return None
|
||
|
||
if not isinstance(parsed, dict):
|
||
return None
|
||
|
||
# Path 1: 直接 JSON 含 required_key
|
||
if required_key in parsed:
|
||
return parsed
|
||
|
||
# Path 2+3: NemoTron wrapper — description / action_title / reasoning 可能是內嵌 JSON
|
||
for wrapper_key in ("description", "action_title", "reasoning"):
|
||
desc = parsed.get(wrapper_key)
|
||
if not desc:
|
||
continue
|
||
desc_str = str(desc).strip()
|
||
# Path 2: description 是內嵌 JSON 字串
|
||
if desc_str.startswith("{"):
|
||
try:
|
||
inner = _json.loads(desc_str)
|
||
if isinstance(inner, dict) and required_key in inner:
|
||
return inner
|
||
except (_json.JSONDecodeError, ValueError):
|
||
pass
|
||
|
||
# Path 3: 所有嘗試都失敗
|
||
logger.warning(
|
||
"llm_json_missing_required_key",
|
||
context=logger_context,
|
||
required_key=required_key,
|
||
keys_found=list(parsed.keys())[:10],
|
||
)
|
||
return None
|