woogiandClaude Opus 4.8
1d248a8808
Cross-discipline design-contradiction checker for construction drawing sets. Standalone tool broken out from Iron_Bid; a pipeline stage may later fold back into Iron_Bid. Pipeline: PDF->images -> per-sheet assertion extraction -> deterministic clustering by location -> per-cluster reasoning -> report. Includes CLI (cli/run_check.py) and web UI (backend/main.py). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
"""
|
|
risk.py - Stage 10: risk scoring and prioritization (LLM, text-only).
|
|
|
|
Scores each validated issue 1-100, annotates it with risk_score,
|
|
recommended_priority and risk_drivers, and returns the list sorted
|
|
highest-risk first. Uses TEXT_MODEL (defaults to MODEL). On failure it falls
|
|
back to a deterministic severity-based ordering so the pipeline still produces
|
|
a prioritized list.
|
|
"""
|
|
|
|
from typing import Dict, List
|
|
|
|
from backend import config
|
|
from backend.pipeline._serialize import dumps
|
|
from backend.pipeline._stage import call_stage
|
|
from backend.prompts import RISK_SYSTEM_PROMPT, RISK_USER_INSTRUCTION
|
|
|
|
_SEV_RANK = {"critical": 90, "high": 70, "medium": 40, "low": 15}
|
|
|
|
|
|
def _ensure_ids(issues: List[Dict]) -> None:
|
|
for i, issue in enumerate(issues, 1):
|
|
if not issue.get("issue_id"):
|
|
issue["issue_id"] = f"ISSUE-{i:03d}"
|
|
|
|
|
|
def score_and_prioritize(validated: List[Dict]) -> List[Dict]:
|
|
if not validated:
|
|
return []
|
|
_ensure_ids(validated)
|
|
|
|
parsed = call_stage(
|
|
RISK_SYSTEM_PROMPT,
|
|
RISK_USER_INSTRUCTION,
|
|
subs={"validated_issues": dumps(validated)},
|
|
max_tokens=config.RISK_MAX_TOKENS,
|
|
)
|
|
|
|
if isinstance(parsed, list):
|
|
rows = parsed
|
|
elif isinstance(parsed, dict):
|
|
rows = parsed.get("prioritized_issues") or []
|
|
else:
|
|
rows = []
|
|
scores: Dict[str, Dict] = {}
|
|
for p in rows:
|
|
if isinstance(p, dict) and p.get("issue_id"):
|
|
scores[p["issue_id"]] = p
|
|
|
|
for issue in validated:
|
|
p = scores.get(issue["issue_id"])
|
|
if p and isinstance(p.get("overall_risk_score"), (int, float)):
|
|
issue["risk_score"] = int(p["overall_risk_score"])
|
|
issue["recommended_priority"] = p.get("recommended_priority")
|
|
issue["risk_drivers"] = p.get("risk_drivers") or []
|
|
else:
|
|
# Deterministic fallback from severity.
|
|
issue["risk_score"] = _SEV_RANK.get(issue.get("severity"), 40)
|
|
|
|
validated.sort(key=lambda i: i.get("risk_score", 0), reverse=True)
|
|
print(f"[Risk] scored {len(validated)} issue(s)"
|
|
f" ({len(scores)} from model, rest by severity)")
|
|
return validated
|