Skills › Scientific Agent Skills › clinical-decision-support › evidence_profile_check.py
evidence_profile_check.py
263 lines · skills/clinical-decision-support/scripts/evidence_profile_check.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check a human-authored GRADE evidence profile without assigning grades."""
4from __future__ import annotations
6import argparse
7import sys
8from typing import Any
10from _common import (
11 InputError,
12 IssueLog,
13 load_json_object,
14 print_report,
15 require_list,
16 require_nonempty_text,
17 source_ids,
18 validate_references,
19 write_json,
20)
22DOMAINS = (
23 "risk_of_bias",
24 "inconsistency",
25 "indirectness",
26 "imprecision",
27 "publication_bias",
28)
29DOMAIN_JUDGMENTS = {"not_serious", "serious", "very_serious", "unassessed"}
30UPGRADING = ("large_effect", "dose_response", "residual_confounding")
31UPGRADING_JUDGMENTS = {"none", "present", "unassessed", "not_applicable"}
32CERTAINTY_LEVELS = {"high", "moderate", "low", "very_low", "unassessed"}
33PLACEHOLDER_MARKERS = ("REPLACE_", "REQUIRES_", "YYYY-MM-DD")
36def _profile_has_recommendation_key(value: Any) -> bool:
37 if isinstance(value, dict):
38 for key, nested in value.items():
39 if str(key).strip().lower() in {
40 "recommendation",
41 "recommendation_strength",
42 "treatment_recommendation",
43 }:
44 return True
45 if _profile_has_recommendation_key(nested):
46 return True
47 elif isinstance(value, list):
48 return any(_profile_has_recommendation_key(item) for item in value)
49 return False
52def _find_placeholders(value: Any, location: str = "$") -> list[str]:
53 findings: list[str] = []
54 if isinstance(value, dict):
55 for key, nested in value.items():
56 findings.extend(_find_placeholders(nested, f"{location}.{key}"))
57 elif isinstance(value, list):
58 for index, nested in enumerate(value):
59 findings.extend(_find_placeholders(nested, f"{location}[{index}]"))
60 elif isinstance(value, str) and any(marker in value for marker in PLACEHOLDER_MARKERS):
61 findings.append(location)
62 return findings
65def check_profile(document: dict[str, Any]) -> tuple[IssueLog, list[dict[str, str]]]:
66 log = IssueLog()
67 summaries: list[dict[str, str]] = []
68 try:
69 require_nonempty_text(document.get("schema_version"), "schema_version")
70 require_nonempty_text(document.get("profile_id"), "profile_id")
71 require_nonempty_text(document.get("title"), "title")
72 if not isinstance(document.get("data_level"), str) or document.get("data_level") not in {"published_aggregate_evidence", "synthetic"}:
73 log.errors.append(
74 "data_level must be published_aggregate_evidence or synthetic"
75 )
76 if document.get("human_judgments_required") is not True:
77 log.errors.append("human_judgments_required must be true")
78 if document.get("auto_grade") is not False:
79 log.errors.append("auto_grade must be false")
80 if _profile_has_recommendation_key(document):
81 log.errors.append(
82 "Recommendation fields are outside the evidence-profile checker"
83 )
84 governance = document.get("governance")
85 if not isinstance(governance, dict):
86 raise InputError("governance must be an object")
87 for field in (
88 "version",
89 "owner",
90 "change_summary",
91 "auditability",
92 "update_plan",
93 ):
94 require_nonempty_text(governance.get(field), f"governance.{field}")
96 question = document.get("question")
97 if not isinstance(question, dict):
98 raise InputError("question must be an object")
99 for field in ("population", "intervention_or_exposure", "comparator", "setting"):
100 require_nonempty_text(question.get(field), f"question.{field}")
102 known_sources = source_ids(document)
103 outcomes = require_list(document.get("outcomes"), "outcomes", maximum=50)
104 if not outcomes:
105 log.errors.append("At least one outcome is required")
107 names: set[str] = set()
108 for index, outcome in enumerate(outcomes):
109 prefix = f"outcomes[{index}]"
110 if not isinstance(outcome, dict):
111 raise InputError(f"{prefix} must be an object")
112 name = require_nonempty_text(outcome.get("name"), f"{prefix}.name")
113 if name in names:
114 log.errors.append(f"Duplicate outcome name: {name}")
115 names.add(name)
116 importance = require_nonempty_text(
117 outcome.get("importance"), f"{prefix}.importance"
118 )
119 if importance not in {"critical", "important", "not_important"}:
120 log.errors.append(f"{prefix}.importance has an unsupported value")
121 for field in (
122 "effect_measure",
123 "effect_estimate",
124 "uncertainty_interval",
125 "participants",
126 "studies",
127 "time_horizon",
128 ):
129 require_nonempty_text(outcome.get(field), f"{prefix}.{field}")
131 domains = outcome.get("domains")
132 if not isinstance(domains, dict):
133 raise InputError(f"{prefix}.domains must be an object")
134 for domain_name in DOMAINS:
135 domain = domains.get(domain_name)
136 field = f"{prefix}.domains.{domain_name}"
137 if not isinstance(domain, dict):
138 raise InputError(f"{field} must be an object")
139 judgment = require_nonempty_text(
140 domain.get("judgment"), f"{field}.judgment"
141 )
142 allowed_judgments = DOMAIN_JUDGMENTS
143 if domain_name == "imprecision":
144 # Current GRADE Book permits a human three-level downgrade.
145 allowed_judgments = DOMAIN_JUDGMENTS | {"extremely_serious"}
146 if judgment not in allowed_judgments:
147 log.errors.append(f"{field}.judgment is unsupported")
148 if judgment == "unassessed":
149 log.errors.append(f"{field} requires a human judgment")
150 require_nonempty_text(domain.get("rationale"), f"{field}.rationale")
151 require_nonempty_text(
152 domain.get("reviewer_role"), f"{field}.reviewer_role"
153 )
154 validate_references(
155 domain.get("source_ids"), known_sources, f"{field}.source_ids"
156 )
158 upgrading = outcome.get("upgrading")
159 if not isinstance(upgrading, dict):
160 raise InputError(f"{prefix}.upgrading must be an object")
161 for item_name in UPGRADING:
162 item = upgrading.get(item_name)
163 field = f"{prefix}.upgrading.{item_name}"
164 if not isinstance(item, dict):
165 raise InputError(f"{field} must be an object")
166 judgment = require_nonempty_text(
167 item.get("judgment"), f"{field}.judgment"
168 )
169 if judgment not in UPGRADING_JUDGMENTS:
170 log.errors.append(f"{field}.judgment is unsupported")
171 if judgment == "unassessed":
172 log.errors.append(f"{field} requires a human judgment")
173 require_nonempty_text(item.get("rationale"), f"{field}.rationale")
174 if judgment == "present":
175 validate_references(
176 item.get("source_ids"), known_sources, f"{field}.source_ids"
177 )
179 certainty = outcome.get("certainty")
180 if not isinstance(certainty, dict):
181 raise InputError(f"{prefix}.certainty must be an object")
182 if certainty.get("human_judgment") is not True:
183 log.errors.append(f"{prefix}.certainty.human_judgment must be true")
184 level = require_nonempty_text(
185 certainty.get("level"), f"{prefix}.certainty.level"
186 )
187 if level not in CERTAINTY_LEVELS:
188 log.errors.append(f"{prefix}.certainty.level is unsupported")
189 if level == "unassessed":
190 log.errors.append(f"{prefix}.certainty requires a human judgment")
191 require_nonempty_text(
192 certainty.get("rationale"), f"{prefix}.certainty.rationale"
193 )
194 require_nonempty_text(
195 certainty.get("reviewer_role"), f"{prefix}.certainty.reviewer_role"
196 )
197 require_nonempty_text(
198 certainty.get("judgment_date"), f"{prefix}.certainty.judgment_date"
199 )
200 validate_references(
201 certainty.get("source_ids"),
202 known_sources,
203 f"{prefix}.certainty.source_ids",
204 )
205 summaries.append(
206 {
207 "outcome": name,
208 "importance": importance,
209 "human_entered_certainty": level,
210 }
211 )
213 review = document.get("profile_review")
214 if not isinstance(review, dict):
215 raise InputError("profile_review must be an object")
216 for field in ("prepared_by_role", "reviewed_by_role", "review_date"):
217 require_nonempty_text(review.get(field), f"profile_review.{field}")
218 if review.get("completed") is not True:
219 log.errors.append("profile_review.completed must be true")
220 placeholders = _find_placeholders(document)
221 if placeholders:
222 log.errors.append(
223 "Unresolved template placeholders remain: "
224 + ", ".join(placeholders[:10])
225 )
226 except InputError as exc:
227 log.errors.append(str(exc))
229 if log.ok:
230 log.info.append(
231 "Profile is structurally complete; all certainty labels remain human judgments"
232 )
233 return log, summaries
236def main() -> int:
237 parser = argparse.ArgumentParser(
238 description=(
239 "Check completeness and citation traceability of a human-authored "
240 "GRADE evidence profile. This tool never assigns certainty."
241 )
242 )
243 parser.add_argument("input", help="Local evidence-profile JSON")
244 parser.add_argument("-o", "--output", help="Optional local JSON report")
245 args = parser.parse_args()
247 try:
248 document = load_json_object(args.input)
249 log, summaries = check_profile(document)
250 report = log.as_dict()
251 report["outcomes"] = summaries
252 report["auto_grade_performed"] = False
253 if args.output:
254 write_json(args.output, report)
255 print_report(report)
256 except InputError as exc:
257 print_report(IssueLog(errors=[str(exc)]).as_dict())
258 return 2
259 return 0 if report["status"] == "pass" else 1
262if __name__ == "__main__":
263 sys.exit(main())