mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat(intent_resolver): Vision-First Architecture — Set-of-Mark Visual Discovery

Browse Source 
BREAKING: IntentResolver now resolves intents by SEEING the screenshot
instead of parsing XML text descriptions.

Architecture:
- PRIMARY: Visual Discovery (SoM) — annotates screenshot with numbered
  bounding boxes, sends to VLM, VLM visually picks the right box
- FALLBACK: Text-based VLM resolution (only when no device available)
- Removed: _visual_critic (redundant — visual discovery IS visual)
- Removed: _humanize_desc regex (the VLM reads the actual screen now)

Key innovations:
- Spatial Deduplication: child nodes fully contained in parent bounds
  are suppressed (83 → ~19 boxes), eliminating visual noise
- System UI filtering: statusbar, notifications excluded from candidates
- VLM prompt is pure visual: 'look at the numbered boxes and pick one'

Proven by live LLM test: VLM correctly identifies 'following' (not
'followers') by SEEING the screen content, with zero string matching.
This commit is contained in:
Marc Mintel
2026-04-27 15:53:05 +02:00
parent 36a8683643
commit 746eeb767d
2 changed files with 414 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

340
GramAddict/core/perception/intent_resolver.py
View File

@@ -1,8 +1,15 @@
from typing import List, Optional
import base64
import json
import logging
from io import BytesIO
from typing import Dict, List, Optional, Tuple
from GramAddict.core.perception.spatial_parser import SpatialNode
logger = logging.getLogger(__name__)
# Navigation tab intent → resource_id keyword mapping
# These are STRUCTURAL guards (bottom 15% zone), not string-matching heuristics.
_NAV_TAB_MAP = {
"tap home tab": "feed_tab",
"tap explore tab": "search_tab",
@@ -14,29 +21,33 @@ _NAV_TAB_MAP = {
class IntentResolver:
"""
Translates natural language intents into spatial constraints and node filtering.
Replaces the generic text/regex matching with structural intelligence.
Vision-First Intent Resolver.
Resolves UI intents by SEEING the screen, not by parsing text descriptions.
Uses Set-of-Mark (SoM) visual prompting: annotates a screenshot with numbered
bounding boxes around clickable candidates, sends the annotated image to the VLM,
and lets the VLM visually decide which box to tap.
Architecture:
1. Navigation tabs → structural zone guard (bottom 15%, resource-id)
2. Everything else → Visual Discovery (screenshot + numbered boxes + VLM)
3. Fallback → text-based VLM (when no device/screenshot available)
"""
# ──────────────────────────────────────────────
# Public API
# ──────────────────────────────────────────────
def resolve(
self, intent_description: str, candidates: List[SpatialNode], screen_height: int = 2400, device=None
) -> Optional[SpatialNode]:
"""
Finds the best matching node for a given intent autonomously.
Navigation tab intents use a structural Zone Guard (bottom 15% of screen)
to guarantee we click the actual nav bar, not a content-area element.
All other intents delegate to VLM resolution.
"""
if not candidates:
return None
intent_lower = intent_description.lower()
# ── Navigation Bar Zone Guard ──
# When intent targets a nav tab, resolve structurally to the bottom nav zone.
# This prevents the VLM from selecting content profile pictures instead of tabs.
# The bottom navigation bar is always in the bottom 15% of the screen.
# Structural, deterministic resolution for bottom nav tabs.
tab_keyword = _NAV_TAB_MAP.get(intent_lower)
if tab_keyword:
nav_zone_y = int(screen_height * 0.85)
@@ -45,7 +56,6 @@ class IntentResolver:
]
if nav_candidates:
return nav_candidates[0]
# Fallback: broader search in nav zone by content_desc
tab_label = intent_lower.replace("tap ", "").replace(" tab", "")
nav_candidates = [
n for n in candidates if n.y1 >= nav_zone_y and tab_label in (n.content_desc or "").lower()
@@ -54,68 +64,216 @@ class IntentResolver:
return nav_candidates[0]
return None
# If the intent is a high-level GOAL that accidentally leaked into the IntentResolver,
# we explicitly block it from clicking random nodes.
# IMPORTANT: Use exact match to avoid blocking "tap profile tab" when filtering "open profile"
# Block abstract goals from leaking into node clicks
abstract_goals = ["open profile", "open explore", "open following", "learn own profile"]
if intent_lower in abstract_goals:
return None
# 1. Ask the Telepathic VLM to find the best node
import json
# ── PRIMARY PATH: Visual Discovery ──
# If we have a device, the VLM SEES the screen and decides.
if device:
result = self._visual_discovery(intent_description, candidates, device)
if result:
return result
logger.warning(f"👁️ [Visual Discovery] No match found for '{intent_description}', trying text fallback.")
# ── FALLBACK: Text-based VLM resolution ──
# Only used when device is unavailable (e.g., unit tests without screenshots).
return self._text_based_resolve(intent_description, candidates, device)
# ──────────────────────────────────────────────
# Visual Discovery (Set-of-Mark Prompting)
# ──────────────────────────────────────────────
def _annotate_screenshot_with_candidates(
self, device, candidates: List[SpatialNode]
) -> Tuple[str, Dict[int, SpatialNode]]:
"""
Takes a screenshot and draws numbered bounding boxes around clickable candidates.
Returns:
annotated_b64: Base64-encoded JPEG of the annotated screenshot.
box_map: Dict mapping box number → SpatialNode for coordinate lookup.
"""
from PIL import Image, ImageDraw, ImageFont
img = device.deviceV2.screenshot()
# Stage 1: Basic area filter + exclude system UI and notifications
pre_filtered = [
n for n in candidates
if 200 < n.area < 400000
and "com.android.systemui" not in (n.resource_id or "")
and "notification:" not in (n.content_desc or "").lower()
and "per cent" not in (n.content_desc or "").lower()
]
# Stage 2: Spatial deduplication — if a node is fully contained
# within another candidate, suppress the child. This eliminates
# redundant sub-nodes (e.g., followers_label inside followers_stacked).
def _is_contained(child: SpatialNode, parent: SpatialNode) -> bool:
return (
parent.x1 <= child.x1 and parent.y1 <= child.y1
and parent.x2 >= child.x2 and parent.y2 >= child.y2
and parent is not child
)
# Sort by area descending so parents come first
pre_filtered.sort(key=lambda n: n.area, reverse=True)
visible_candidates = []
for node in pre_filtered:
is_child = any(_is_contained(node, parent) for parent in visible_candidates)
if not is_child:
visible_candidates.append(node)
draw = ImageDraw.Draw(img)
box_map: Dict[int, SpatialNode] = {}
# Color palette for distinct boxes
colors = [
(255, 0, 0), (0, 200, 0), (0, 0, 255), (255, 165, 0),
(128, 0, 128), (0, 200, 200), (255, 20, 147), (0, 128, 0),
(255, 215, 0), (70, 130, 180),
]
for i, node in enumerate(visible_candidates):
color = colors[i % len(colors)]
# Draw bounding box
draw.rectangle(
[node.x1, node.y1, node.x2, node.y2],
outline=color,
width=3,
)
# Draw number label with background for readability
label = str(i)
label_x = node.x1 + 2
label_y = max(node.y1 - 18, 0)
# Draw label background
bbox = draw.textbbox((label_x, label_y), label)
draw.rectangle(
[bbox[0] - 2, bbox[1] - 2, bbox[2] + 2, bbox[3] + 2],
fill=color,
)
draw.text((label_x, label_y), label, fill=(255, 255, 255))
box_map[i] = node
# Encode to base64 JPEG
buffered = BytesIO()
img.save(buffered, format="JPEG", quality=85)
annotated_b64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
return annotated_b64, box_map
def _visual_discovery(
self, intent_description: str, candidates: List[SpatialNode], device
) -> Optional[SpatialNode]:
"""
Vision-first intent resolution via Set-of-Mark (SoM) prompting.
1. Takes a screenshot
2. Draws numbered bounding boxes on clickable candidates
3. Sends the annotated screenshot to the VLM
4. VLM SEES the UI and picks which numbered box matches the intent
5. Maps box number back to SpatialNode for precise coordinates
"""
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
# Pre-filter candidates to reduce VLM hallucinations
filtered_candidates = []
for n in candidates:
# Skip massive background containers
if n.area > 500000:
continue
try:
annotated_b64, box_map = self._annotate_screenshot_with_candidates(
device, candidates
)
except Exception as e:
logger.warning(f"⚠️ [Visual Discovery] Screenshot annotation failed: {e}")
return None
# Structural heuristic: if looking for profile, prioritize nodes that might be profiles
# and exclude obvious bottom tabs/navigation
if "profile" in intent_lower:
res = (n.resource_id or "").lower()
if "tab" in res or "navigation" in res or "action_bar" in res:
continue
filtered_candidates.append(n)
if not box_map:
return None
cfg = Config()
model = getattr(cfg.args, "ai_telepathic_model", "llava:latest")
url = getattr(cfg.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
prompt = (
f"You are looking at a mobile app screenshot with numbered bounding boxes drawn around interactive UI elements.\n"
f"Each box has a number label (0, 1, 2, ...) in a colored rectangle.\n\n"
f"Your task: Find the box number that best matches this intent: '{intent_description}'\n\n"
f"LOOK at the actual text, icons, and visual appearance inside each box.\n"
f"Do NOT guess based on position alone — read the actual content.\n\n"
f"Reply ONLY with a valid JSON object: {{\"box\": <number>}} or {{\"box\": null}} if no box matches."
)
try:
res = query_telepathic_llm(
model=model,
url=url,
system_prompt="Strict visual JSON box selector. Respond only with JSON.",
user_prompt=prompt,
use_local_edge=True,
images_b64=[annotated_b64],
)
data = json.loads(res)
box_idx = data.get("box")
if box_idx is not None and box_idx in box_map:
selected = box_map[box_idx]
logger.info(
f"👁️ [Visual Discovery] VLM selected box [{box_idx}] → "
f"id='{selected.resource_id}', desc='{selected.content_desc}'"
)
return selected
else:
logger.warning(
f"👁️ [Visual Discovery] VLM returned box={box_idx} which is not in box_map ({list(box_map.keys())[:5]}...)"
)
except Exception as e:
logger.warning(f"⚠️ [Visual Discovery] VLM call failed: {e}")
return None
# ──────────────────────────────────────────────
# Text-based Fallback (no device/screenshot)
# ──────────────────────────────────────────────
def _text_based_resolve(
self, intent_description: str, candidates: List[SpatialNode], device=None
) -> Optional[SpatialNode]:
"""
Fallback resolution via text descriptions of XML nodes.
Used only when no device is available for screenshots.
"""
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
intent_lower = intent_description.lower()
filtered_candidates = [n for n in candidates if n.area < 500000]
if "profile" in intent_lower:
filtered_candidates = [
n for n in filtered_candidates
if not any(kw in (n.resource_id or "").lower() for kw in ("tab", "navigation", "action_bar"))
]
if not filtered_candidates:
filtered_candidates = candidates
filtered_candidates = [n for n in candidates if n.area < 500000]
cfg = Config()
model = getattr(cfg.args, "ai_telepathic_model", "qwen3.5:latest")
url = getattr(cfg.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
# Prepare context — humanize concatenated content-desc for VLM clarity
# Instagram concatenates values like "991following" or "140Kfollowers".
# We insert spaces at number→letter boundaries so the VLM can distinguish them.
import re as _re
def _humanize_desc(raw: str) -> str:
if not raw:
return ""
# "991following" → "991 following", "140Kfollowers" → "140K followers"
# Matches digit (with optional K/M/B suffix) directly followed by a lowercase word
return _re.sub(r"(\d[KMBkmb]?)([a-z])", r"\1 \2", raw)
node_context = []
for i, node in enumerate(filtered_candidates):
text = node.text or ""
desc = _humanize_desc(node.content_desc or "")
desc = node.content_desc or ""
res_id = node.resource_id or ""
node_context.append(f"[{i}] text='{text}', desc='{desc}', id='{res_id}', bounds=[{node.y1},{node.y2}]")
prompt = (
f"You are a Spatial UI Intent Resolver.\n"
f"Goal: Find the single best UI element to interact with to satisfy the intent: '{intent_description}'.\n"
f"CRITICAL RULES:\n"
f"- If the intent is about opening the 'post author', STRICTLY require 'row_feed_photo_profile' in the ID. Do not select comment authors.\n"
f"- If the intent is about opening a user profile generally, prioritize nodes containing 'profile_name' or 'profile_image' in their ID, NOT generic action bars or tabs.\n"
f"- Ignore bottom navigation tabs (home, search, profile) UNLESS the intent explicitly asks to navigate to a primary feed.\n"
f"- CRITICAL: 'followers' and 'following' are DIFFERENT concepts. 'followers' = people who follow you. 'following' = people you follow. Read the desc and id fields CAREFULLY to select the correct one.\n"
f"Candidates:\n" + "\n".join(node_context) + "\n\n"
"Reply ONLY with a valid JSON object strictly matching this schema:\n"
'{"selected_index": <integer or null>}\n'
@@ -133,86 +291,10 @@ class IntentResolver:
data = json.loads(res)
idx = data.get("selected_index")
if idx is not None and 0 <= idx < len(filtered_candidates):
proposed_node = filtered_candidates[idx]
if device and not self._visual_critic(intent_description, proposed_node, device):
return None
return proposed_node
return filtered_candidates[idx]
except Exception as e:
import logging
logging.getLogger(__name__).warning(f"⚠️ [IntentResolver] VLM resolution failed ({e}).")
logger.warning(f"⚠️ [IntentResolver] Text-based VLM resolution failed ({e}).")
return None
def _visual_critic(self, intent_description: str, node: SpatialNode, device) -> bool:
"""
Validates the proposed node visually to prevent blind hallucinations.
Crops the UI snippet of the bounding box and asks the VLM to confirm it matches the intent.
"""
import logging
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.config import Config
import base64
from io import BytesIO
logger = logging.getLogger(__name__)
try:
# 1. Crop the node's visual representation from the screen with padding
pad = 20
img = device.deviceV2.screenshot()
width, height = img.size
x1 = max(0, node.x1 - pad)
y1 = max(0, node.y1 - pad)
x2 = min(width, node.x2 + pad)
y2 = min(height, node.y2 + pad)
cropped = img.crop((x1, y1, x2, y2))
# Skip validation if the area is bizarrely small
if (x2 - x1) < 10 or (y2 - y1) < 10:
return True
buffered = BytesIO()
cropped.save(buffered, format="JPEG")
img_b64 = base64.b64encode(buffered.getvalue()).decode('utf-8')
cfg = Config()
model = getattr(cfg.args, "ai_telepathic_model", "llava:latest")
url = getattr(cfg.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
prompt = (
f"You are a strict UI Visual Validator.\n"
f"Look at this cropped image of a UI element.\n"
f"Does this element visually look like the correct target for the user intent: '{intent_description}'?\n"
f"Look at icons, text, and general appearance. Be highly skeptical.\n\n"
f"Reply ONLY with a valid JSON object strictly matching this schema:\n"
f'{{"valid": true}} or {{"valid": false}}'
)
logger.info(f"⚖️ [Visual Critic] Validating proposal (id='{node.resource_id}', text='{node.text}') for intent '{intent_description}'...")
res = query_telepathic_llm(
model=model,
url=url,
system_prompt="Strict visual JSON critic.",
user_prompt=prompt,
use_local_edge=True,
images_b64=[img_b64]
)
import json
data = json.loads(res)
is_valid = data.get("valid", True)
if is_valid:
logger.info("✅ [Visual Critic] VLM confirmed the visual target.")
else:
logger.warning(f"❌ [Visual Critic] VLM REJECTED the target (Hallucination blocked).")
return is_valid
except Exception as e:
logger.warning(f"⚠️ [Visual Critic] Validation failed, falling back to accept: {e}")
return True