File size: 6,689 Bytes
ed79a7f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """
Flow Reasoning Engine with Fixed-Point Forcing (FPF).
Synthesized from:
"Flow Reasoning Models: Turning Flows Into Efficient Recurrent Reasoners"
(Georgia Tech / MIT / MIT-IBM / IBM Research, Sep 2026).
Treats reasoning as iterative solution refinement toward a stable attractor.
Rather than committing irrevocably to one-shot greedy decisions, candidate
browser actions and spatial grounding coordinates undergo recurrent flow
refinement over held states:
s^{(k+1)} = Refine(s^{(k)}, Observation)
until converging to a globally consistent fixed point.
"""
from __future__ import annotations
import math
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
from miniswardbower.core.schemas import BrowserAction, BrowserActionType, InteractiveElement, PrunedAXTree
@dataclass
class FlowRefinementState:
"""Represents the recurrent held-state evolving toward a stable attractor."""
step_index: int
candidate_action: BrowserAction
spatial_coords: Tuple[float, float]
confidence: float
residual_norm: float = 1.0
converged: bool = False
iterations_run: int = 0
refinement_trace: List[Dict[str, Any]] = field(default_factory=list)
class FlowReasoningRefiner:
"""
Recurrent Flow Reasoner with Fixed-Point Iteration for Browser Grounding & Actions.
Iteratively refines candidate coordinates and parameter schemas to prevent cascade failures.
"""
def __init__(
self,
max_refinement_depth: int = 4,
convergence_threshold: float = 0.05,
damping_factor: float = 0.65,
):
self.max_depth = max_refinement_depth
self.threshold = convergence_threshold
self.damping = damping_factor
@staticmethod
def _find_element(tree: Optional[PrunedAXTree], target_id: Optional[str]) -> Optional[InteractiveElement]:
if not tree or not target_id:
return None
for el in tree.elements:
if el.id == target_id or el.selector == target_id:
return el
return None
def refine_action(
self,
initial_action: BrowserAction,
tree: Optional[PrunedAXTree] = None,
viewport_dims: Tuple[int, int] = (1280, 900),
) -> FlowRefinementState:
"""
Runs recurrent fixed-point refinement on candidate action and coordinates.
Converges candidate toward the optimal, collision-free attractor.
"""
t0 = time.perf_counter()
vw, vh = viewport_dims
# Extract initial spatial coordinates
cur_x, cur_y = 0.0, 0.0
if getattr(initial_action, "coords", None):
cur_x, cur_y = float(initial_action.coords[0]), float(initial_action.coords[1])
elif tree and initial_action.target:
target_el = self._find_element(tree, initial_action.target)
if target_el and target_el.bbox:
bx, by, bw, bh = target_el.bbox
cur_x, cur_y = bx + bw / 2.0, by + bh / 2.0
else:
cur_x, cur_y = float(vw / 2.0), float(vh / 2.0)
else:
cur_x, cur_y = float(vw / 2.0), float(vh / 2.0)
cur_action = initial_action.model_copy(deep=True)
cur_confidence = 0.50
state = FlowRefinementState(
step_index=0,
candidate_action=cur_action,
spatial_coords=(cur_x, cur_y),
confidence=cur_confidence,
)
trace = []
# Recurrent refinement loop
for depth in range(1, self.max_depth + 1):
prev_x, prev_y = cur_x, cur_y
prev_conf = cur_confidence
# 1. Attractor gradient from DOM layout
target_el: Optional[InteractiveElement] = None
if tree and cur_action.target:
target_el = self._find_element(tree, cur_action.target)
attractor_x, attractor_y = prev_x, prev_y
target_valid = False
if target_el:
target_valid = True
if target_el.bbox:
bx, by, bw, bh = target_el.bbox
attractor_x = bx + bw / 2.0
attractor_y = by + bh / 2.0
# Type mismatch refinement (e.g. TYPE requested on non-editable tag)
if cur_action.op == BrowserActionType.TYPE:
if target_el.tag not in ("input", "textarea") and target_el.element_type not in ("textbox", "search"):
# Search for adjacent editable child/sibling in tree
for alt_el in tree.elements:
if alt_el.tag in ("input", "textarea") and alt_el.bbox:
cur_action.target = alt_el.id
cur_action.selector = alt_el.selector
abx, aby, abw, abh = alt_el.bbox
attractor_x = abx + abw / 2.0
attractor_y = aby + abh / 2.0
break
# 2. Viewport boundary attraction (keep coordinates within visible viewport)
attractor_x = max(10.0, min(float(vw - 10.0), attractor_x))
attractor_y = max(10.0, min(float(vh - 10.0), attractor_y))
# 3. Fixed-point update with damping
dx = (attractor_x - cur_x) * self.damping
dy = (attractor_y - cur_y) * self.damping
cur_x += dx
cur_y += dy
# 4. Confidence refinement
if target_valid:
cur_confidence = min(0.99, cur_confidence + 0.15)
else:
cur_confidence = max(0.20, cur_confidence - 0.10)
# Residual norm
delta_dist = math.sqrt(dx * dx + dy * dy)
residual = delta_dist / max(1.0, math.sqrt(float(vw * vw + vh * vh)))
trace.append({
"depth": depth,
"coords": (round(cur_x, 2), round(cur_y, 2)),
"residual": round(residual, 5),
"confidence": round(cur_confidence, 3),
})
if residual < self.threshold:
state.converged = True
state.iterations_run = depth
state.residual_norm = residual
break
if not state.converged:
state.iterations_run = self.max_depth
state.residual_norm = residual
cur_action.coords = (cur_x, cur_y)
state.candidate_action = cur_action
state.spatial_coords = (cur_x, cur_y)
state.confidence = cur_confidence
state.refinement_trace = trace
return state
|