fix(perception): enforce visual Set of Mark by passing device context globally
This commit is contained in:
@@ -47,7 +47,7 @@ def verify_and_switch_account(device, nav_graph, target_username):
|
||||
telepath = TelepathicEngine.get_instance()
|
||||
|
||||
# We ask the semantic engine to find the profile tab, ensuring 100% ID-agnostic behavior
|
||||
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3)
|
||||
profile_tab_node = telepath.find_best_node(xml_dump, "tap profile tab", min_threshold=0.3, device=device)
|
||||
if profile_tab_node:
|
||||
profile_tab = (profile_tab_node["x"], profile_tab_node["y"])
|
||||
except Exception as e:
|
||||
@@ -113,7 +113,7 @@ def verify_and_switch_account(device, nav_graph, target_username):
|
||||
dump_ui_state(
|
||||
device, "identity_guard", {"reason": "account_not_found_in_bottom_sheet", "target": target_username}
|
||||
)
|
||||
except:
|
||||
except Exception:
|
||||
pass
|
||||
# Escape the bottom sheet
|
||||
device.press("back")
|
||||
|
||||
@@ -56,7 +56,7 @@ class ObstacleGuardPlugin(BehaviorPlugin):
|
||||
# Check recovery
|
||||
new_xml = ctx.device.dump_hierarchy()
|
||||
tele = TelepathicEngine.get_instance()
|
||||
best_node = tele.find_best_node(new_xml, intent_description="Dismiss obstacle")
|
||||
best_node = tele.find_best_node(new_xml, intent_description="Dismiss obstacle", device=ctx.device)
|
||||
if best_node:
|
||||
ctx.device.click(best_node.get("x", 0), best_node.get("y", 0))
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ class PostDataExtractionPlugin(BehaviorPlugin):
|
||||
|
||||
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
|
||||
logger.debug("🧩 [PostDataExtraction] Extracting post metadata...")
|
||||
post_data = extract_post_content(ctx.context_xml)
|
||||
post_data = extract_post_content(ctx.context_xml, device=ctx.device)
|
||||
|
||||
if post_data:
|
||||
ctx.post_data = post_data
|
||||
|
||||
@@ -1,10 +1,46 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from GramAddict.core.perception.spatial_parser import SpatialNode
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _parse_yes_no(response: str) -> Optional[bool]:
|
||||
"""Parses a VLM response to find a definitive YES or NO without substring-matching 'not' or 'now'."""
|
||||
text = response.strip()
|
||||
|
||||
# Try parsing as JSON first
|
||||
if text.startswith("{"):
|
||||
try:
|
||||
data = json.loads(text)
|
||||
for k, v in data.items():
|
||||
if str(k).strip().upper() == "YES" or str(v).strip().upper() == "YES":
|
||||
return True
|
||||
if str(k).strip().upper() == "NO" or str(v).strip().upper() == "NO":
|
||||
return False
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
text_lower = text.lower()
|
||||
if text_lower.startswith("yes"):
|
||||
return True
|
||||
if text_lower.startswith("no") and not text_lower.startswith("now") and not text_lower.startswith("not"):
|
||||
return False
|
||||
|
||||
has_yes = re.search(r"\byes\b", text_lower) is not None
|
||||
has_no = re.search(r"\bno\b", text_lower) is not None
|
||||
|
||||
if has_yes and not has_no:
|
||||
return True
|
||||
if has_no and not has_yes:
|
||||
return False
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════
|
||||
# Semantic Match Keywords — SSOT for intent → element validation
|
||||
# ═══════════════════════════════════════════════════════
|
||||
@@ -189,10 +225,12 @@ class ActionMemory:
|
||||
raise ValueError("No screenshot available from device")
|
||||
response = evaluator._query_vlm(prompt, screenshot)
|
||||
|
||||
if response and "yes" in response.lower() and "no" not in response.lower():
|
||||
decision = _parse_yes_no(response) if response else None
|
||||
|
||||
if decision is True:
|
||||
logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
|
||||
return True
|
||||
elif response and "no" in response.lower() and "yes" not in response.lower():
|
||||
elif decision is False:
|
||||
logger.warning(
|
||||
f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'"
|
||||
)
|
||||
@@ -265,10 +303,13 @@ class ActionMemory:
|
||||
prompt = f"The user just attempted to perform the action: '{intent}'. Does the current screen match the expected outcome? Answer ONLY with the word YES or NO."
|
||||
try:
|
||||
response = evaluator._query_vlm(prompt, device.get_screenshot_b64())
|
||||
if response and "yes" in response.lower() and "no" not in response.lower():
|
||||
decision = _parse_yes_no(response) if response else None
|
||||
if decision is True:
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"⚠️ [ActionMemory] VLM rejected success for abstract intent '{intent}'.")
|
||||
logger.warning(
|
||||
f"⚠️ [ActionMemory] VLM rejected success for abstract intent '{intent}'. Response: '{response}'"
|
||||
)
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"VLM visual verification failed: {e}")
|
||||
|
||||
@@ -46,7 +46,7 @@ def has_carousel_in_view(xml_dump: str) -> bool:
|
||||
return any(ind in xml_dump for ind in CAROUSEL_INDICATORS)
|
||||
|
||||
|
||||
def extract_post_content(context_xml: str) -> dict:
|
||||
def extract_post_content(context_xml: str, device=None) -> dict:
|
||||
"""
|
||||
Extracts meaningful content data from the current feed post's XML.
|
||||
This is the BOT'S EYES — what it actually "sees" about each post.
|
||||
@@ -62,14 +62,16 @@ def extract_post_content(context_xml: str) -> dict:
|
||||
telepath = TelepathicEngine.get_instance()
|
||||
|
||||
# 1. Learn/extract post author dynamically
|
||||
author_node = telepath.find_best_node(context_xml, "post author username header", min_confidence=0.75)
|
||||
author_node = telepath.find_best_node(
|
||||
context_xml, "post author username header", min_confidence=0.75, device=device
|
||||
)
|
||||
|
||||
# 🛡️ Anti-Hallucination Guard: Ensure we actually found text.
|
||||
if author_node and author_node.get("original_attribs", {}).get("text"):
|
||||
result["username"] = author_node["original_attribs"]["text"].strip()
|
||||
|
||||
# 2. Learn/extract post media description dynamically
|
||||
media_node = telepath.find_best_node(context_xml, "post media content", min_confidence=0.35)
|
||||
media_node = telepath.find_best_node(context_xml, "post media content", min_confidence=0.35, device=device)
|
||||
if media_node and media_node.get("original_attribs", {}).get("desc"):
|
||||
result["description"] = media_node["original_attribs"]["desc"].strip()
|
||||
|
||||
|
||||
@@ -365,6 +365,12 @@ class IntentResolver:
|
||||
)
|
||||
data = json.loads(res)
|
||||
box_idx = data.get("box")
|
||||
if box_idx is None:
|
||||
box_idx = data.get("selected_index")
|
||||
if box_idx is None:
|
||||
box_idx = data.get("box_index")
|
||||
if box_idx is None:
|
||||
box_idx = data.get("index")
|
||||
|
||||
if box_idx is not None and box_idx in box_map:
|
||||
selected = box_map[box_idx]
|
||||
|
||||
Reference in New Issue
Block a user