102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Reflexion gate for deep-research.
|
|
|
|
Runs after `critique` has reviewed the current research findings. If the
|
|
critique's verdict is REVISE and the reflexion budget (the graph's
|
|
`reflexion_budget` variable, default 2) is not spent,
|
|
loops back to `research` with the critique attached as
|
|
`research_feedback`, so the retry is informed rather than a blind
|
|
re-run. Otherwise it proceeds to `synthesize`.
|
|
|
|
Routing (`_next`):
|
|
- verdict PASS -> synthesize
|
|
- verdict REVISE, budget remaining -> research_each_question (+ research_feedback)
|
|
- verdict REVISE, budget spent -> synthesize
|
|
|
|
Reflexion is a best-effort quality booster, not a hard gate: once the
|
|
budget is spent the workflow proceeds anyway, and the human approval
|
|
step is the final backstop.
|
|
|
|
Fail-safe: a crash here fails FORWARD to `synthesize` — the same
|
|
direction as the existing malformed-critique PASS default — with a
|
|
PIPELINE-FAULT entry recorded (best-effort preserving existing faults),
|
|
so a broken gate costs the run its automated critique routing, never the
|
|
run itself.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
# Default automated revision passes when the `reflexion_budget` variable
|
|
# is absent or malformed. `research` runs at most budget + 1 times per
|
|
# user pass. Callers raise it via --variables {"reflexion_budget": "6"}
|
|
# to let the run churn until the critique passes.
|
|
DEFAULT_REFLEXION_REVISIONS = 2
|
|
|
|
|
|
def load_state():
|
|
path = os.environ.get("GRAPH_STATE_FILE")
|
|
if path:
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
return json.loads(os.environ.get("GRAPH_STATE", "{}"))
|
|
|
|
|
|
def safe_faults():
|
|
try:
|
|
state = load_state()
|
|
return [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
|
|
except Exception: # noqa: BLE001 — best-effort preservation only
|
|
return []
|
|
|
|
|
|
def as_int(value, default=0):
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def parse_verdict(critique):
|
|
"""Pull PASS/REVISE from the critique's `VERDICT:` line. Defaults to
|
|
PASS when no verdict line is found, so a malformed critique lets the
|
|
workflow proceed instead of burning the whole revision budget."""
|
|
match = re.search(r"VERDICT:\s*([A-Za-z]+)", critique, re.IGNORECASE)
|
|
if not match:
|
|
return "PASS"
|
|
return match.group(1).upper()
|
|
|
|
|
|
def main():
|
|
state = load_state()
|
|
critique = state.get("critique") or ""
|
|
verdict = parse_verdict(critique)
|
|
attempts = as_int(state.get("research_attempts"))
|
|
budget = as_int(state.get("reflexion_budget"), DEFAULT_REFLEXION_REVISIONS)
|
|
|
|
if verdict == "REVISE" and attempts < budget:
|
|
feedback = (
|
|
"A reviewer judged the previous research pass incomplete. "
|
|
"Address every point in the critique below:\n\n" + critique
|
|
)
|
|
output = {
|
|
"_next": "research_each_question",
|
|
"research_attempts": attempts + 1,
|
|
"research_feedback": feedback,
|
|
}
|
|
else:
|
|
output = {"_next": "synthesize"}
|
|
|
|
print(json.dumps(output))
|
|
|
|
|
|
try:
|
|
main()
|
|
except Exception as e: # noqa: BLE001 — a crashed gate fails forward, never dies
|
|
faults = safe_faults()
|
|
faults.append(
|
|
f"PIPELINE-FAULT: reflexion gate crashed — {e}; proceeded to synthesis "
|
|
"without automated critique routing"
|
|
)
|
|
print(json.dumps({"_next": "synthesize", "pipeline_faults": faults}))
|