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

fix(sae): stabilize navigation engine, fix container filtering, and negative reinforcement logic

Browse Source
This commit is contained in:
Marc Mintel
2026-04-25 13:09:12 +02:00
parent ad1af4edfe
commit 77e8251aa7
43 changed files with 4107 additions and 4075 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
GramAddict/__main__.py
View File

@@ -1,8 +1,8 @@
from GramAddict.core.agentic_views import *
import argparse
from os import getcwd, path
from GramAddict import __version__
from GramAddict.core.agentic_views import *
from GramAddict.core.bot_flow import start_bot
from GramAddict.core.download_from_github import download_from_github
@@ -13,9 +13,7 @@ def cmd_init(args):
for username in args.account_name:
if not path.exists("./run.py"):
print("Creating run.py ...")
download_from_github(
"https://github.com/GramAddict/bot/blob/master/run.py"
)
download_from_github("https://github.com/GramAddict/bot/blob/master/run.py")
if not path.exists(f"./accounts/{username}"):
print(
f"Creating 'accounts/{username}' folder with a config starting point inside. You have to edit these files according with https://docs.gramaddict.org/#/configuration"
@@ -53,8 +51,10 @@ def cmd_dump(args):
os.popen("adb shell pkill atx-agent").close()
try:
d = u2.connect(args.device)
except RuntimeError as err:
raise SystemExit(err)
except Exception as err:
raise SystemExit(
f"⚠️ [ADB ConnectError] Could not connect to device: {err}\nPlease check if ADB is running and your device is authorized."
)
def dump_hierarchy(device, path):
xml_dump = device.dump_hierarchy()
@@ -71,11 +71,7 @@ def cmd_dump(args):
dump_hierarchy(d, "dump/cur/hierarchy.xml")
archive_name = int(time.time())
make_archive(archive_name)
print(
Fore.GREEN
+ Style.BRIGHT
+ "\nCurrent screen dump generated successfully! Please, send me this file:"
)
print(Fore.GREEN + Style.BRIGHT + "\nCurrent screen dump generated successfully! Please, send me this file:")
print(Fore.BLUE + Style.BRIGHT + f"{os.getcwd()}\\screen_{archive_name}.zip")
@@ -126,9 +122,7 @@ def main() -> None:
prog="GramAddict",
description="free human-like Instagram bot",
)
parser.add_argument(
"-v", "--version", action="version", version=f"{parser.prog} {__version__}"
)
parser.add_argument("-v", "--version", action="version", version=f"{parser.prog} {__version__}")
subparser = parser.add_subparsers(dest="subparser")
actions = {}
for c in _commands:

2
GramAddict/core/bot_flow.py
View File

@@ -849,6 +849,7 @@ def _run_zero_latency_feed_loop(
consecutive_marker_misses += 1
if consecutive_marker_misses >= 3:
logger.error("❌ Lost context completely. Aborting feed loop to force reset.")
sae.unlearn_current_state(context_xml)
dump_ui_state(device, "context_lost", {"feed": job_target, "misses": consecutive_marker_misses})
return "CONTEXT_LOST"
@@ -888,6 +889,7 @@ def _run_zero_latency_feed_loop(
consecutive_marker_misses += 1
if consecutive_marker_misses >= 3:
logger.error("❌ Lost context completely. Aborting feed loop to force reset.")
sae.unlearn_current_state(context_xml)
dump_ui_state(device, "context_lost", {"feed": job_target, "misses": consecutive_marker_misses})
return "CONTEXT_LOST"

98
GramAddict/core/device_facade.py
View File

@@ -1,19 +1,17 @@
import logging
import json
import os
import re
import uiautomator2 as u2
from time import sleep, time
from random import uniform
from GramAddict.core.utils import random_sleep
from functools import wraps
from random import uniform
from time import sleep
from GramAddict.core.physics.biomechanics import PhysicsBody, BezierGesture
import uiautomator2 as u2
from GramAddict.core.physics.biomechanics import BezierGesture, PhysicsBody
from GramAddict.core.physics.sendevent_injector import SendEventInjector
logger = logging.getLogger(__name__)
def adb_retry(retries=3, delay=2.0):
def decorator(func):
@wraps(func)
@@ -28,18 +26,38 @@ def adb_retry(retries=3, delay=2.0):
sleep(delay * (attempt + 1)) # Exponential backoff
logger.error(f"❌ ADB action {func.__name__} failed after {retries} retries. Crashing gracefully.")
raise last_err
return wrapper
return decorator
def create_device(device_id, app_id, args=None):
try:
return DeviceFacade(device_id, app_id, args)
except Exception as e:
err_str = str(e)
err_type = str(type(e))
if (
"ConnectError" in err_type
or "ConnectionRefusedError" in err_type
or "ConnectionError" in err_type
or "Timeout" in err_type
):
logger.error(f"⚠️ [ADB ConnectError] Could not connect to device '{device_id}'.")
logger.error("👉 Please verify:")
logger.error(" 1. Your phone is connected via USB or Wi-Fi.")
logger.error(" 2. 'USB Debugging' is enabled in Developer Options.")
logger.error(" 3. You have authorized this computer on your phone's screen.")
logger.error(" 4. The adb server is running ('adb devices').")
raise SystemExit(1)
logger.error(f"Failed to create device: {e}")
# We don't want to just return None and crash later.
# We don't want to just return None and crash later.
# We should raise so the orchestrator knows it's a fatal boot error.
raise e
def get_device_info(device):
if not device or not device.deviceV2:
logger.error("Cannot get device info: Device not initialized.")
@@ -47,32 +65,29 @@ def get_device_info(device):
info = device.info
logger.debug(f"Device Info: {info.get('productName')} | SDK: {info.get('sdkInt')}")
class DeviceFacade:
deviceV2 = None
app_id = None
device_id = None
def __init__(self, device_id, app_id, args):
self.device_id = device_id
self.app_id = app_id
self.args = args
self.deviceV2 = u2.connect(device_id)
# Configure uiautomator2
self.deviceV2.settings["wait_timeout"] = 3.0
self.deviceV2.settings["post_delay"] = 0.5
# System dialog handler (language-agnostic via resource-id, not text)
try:
# u2 v3.x: named watchers with xpath selectors
# android:id/aerr_close = App crash "Close" button (all languages)
self.deviceV2.watcher("crash_dialog").when(
xpath='//*[@resource-id="android:id/aerr_close"]'
).click()
self.deviceV2.watcher("crash_dialog").when(xpath='//*[@resource-id="android:id/aerr_close"]').click()
# android:id/button1 = positive system dialog button (all languages)
self.deviceV2.watcher("system_dialog").when(
xpath='//*[@resource-id="android:id/button1"]'
).click()
self.deviceV2.watcher("system_dialog").when(xpath='//*[@resource-id="android:id/button1"]').click()
self.deviceV2.watcher.start()
except Exception as e:
logger.debug(f"Could not start system watcher: {e}")
@@ -88,7 +103,7 @@ class DeviceFacade:
@adb_retry()
def cm_to_pixels(self, cm: float) -> int:
info = self.deviceV2.info
dpx = info.get("displaySizeDpX", 400)
dpx = info.get("displaySizeDpX", 400)
width = info.get("displayWidth", 1080)
# Android baseline: 1 dp = 1/160 inch. 1 inch = 2.54 cm
# PPCM (Pixels Per CM) = (width / dpx) * (160 / 2.54)
@@ -106,7 +121,6 @@ class DeviceFacade:
def unlock(self):
self.deviceV2.unlock()
@property
def info(self):
return self.deviceV2.info
@@ -132,7 +146,8 @@ class DeviceFacade:
def swipe(self, sx, sy, ex, ey, duration=None):
"""Pass-through strictly for non-biological bezier swiping (e.g., darwin_engine noise correction)"""
kwargs = {}
if duration is not None: kwargs["duration"] = duration
if duration is not None:
kwargs["duration"] = duration
self.deviceV2.swipe(sx, sy, ex, ey, **kwargs)
@adb_retry()
@@ -146,14 +161,14 @@ class DeviceFacade:
@adb_retry()
def click(self, x=None, y=None, obj=None):
if obj:
if isinstance(obj, dict) and 'x' in obj and 'y' in obj:
self.human_click(obj['x'], obj['y'])
if isinstance(obj, dict) and "x" in obj and "y" in obj:
self.human_click(obj["x"], obj["y"])
return
try:
left, top, right, bottom = obj.bounds()
w = right - left
h = bottom - top
# Biological fingerprint via PhysicsBody
body = PhysicsBody.get_session_instance(self)
# Thumb bias: right-handers land slightly left-below center
@@ -163,20 +178,21 @@ class DeviceFacade:
else:
cx_base = left + (w * 0.55)
cy_base = top + (h * 0.55)
from random import gauss
# Fatigue increases spread
fatigue_mult = 1.0 + body.fatigue * 0.3
sigma_x = max(1, w * 0.15 * fatigue_mult)
sigma_y = max(1, h * 0.15 * fatigue_mult)
cx = int(gauss(cx_base, sigma_x))
cy = int(gauss(cy_base, sigma_y))
# Math constraint to ensure it physically lands on the button
cx = max(left + 1, min(cx, right - 1))
cy = max(top + 1, min(cy, bottom - 1))
self.human_click(cx, cy)
except Exception as e:
logger.debug(f"Bounds extraction failed, fallback to native click: {e}")
@@ -193,7 +209,6 @@ class DeviceFacade:
self.deviceV2.shell(f"input tap {int(x)} {int(y)}")
return
from random import uniform
try:
body = PhysicsBody.get_session_instance(self)
injector = SendEventInjector.get_instance(self)
@@ -201,7 +216,6 @@ class DeviceFacade:
tap_duration = uniform(40, 90)
timing = BezierGesture.compute_sigmoid_timing(len(points), tap_duration)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
except Exception as e:
logger.debug(f"human_click biomechanics failed, fallback: {e}")
@@ -234,18 +248,17 @@ class DeviceFacade:
try:
body = PhysicsBody.get_session_instance(self)
injector = SendEventInjector.get_instance(self)
# Use scroll_curve for vertical swipes, horizontal_swipe_curve for horizontal
is_horizontal = abs(end_x - start_x) > abs(end_y - start_y)
if is_horizontal:
points = BezierGesture.horizontal_swipe_curve((start_x, start_y), (end_x, end_y), body)
else:
points = BezierGesture.scroll_curve((start_x, start_y), (end_x, end_y), body)
# Use fling timing (J-curve) to ensure high terminal velocity so Android scroll physics works natively
timing = BezierGesture.compute_fling_timing(len(points), dur_ms)
injector.inject_gesture(points, timing, touch_major=body.get_touch_major())
except Exception as e:
logger.debug(f"human_swipe biomechanics failed, fallback to native swipe: {e}")
@@ -263,14 +276,14 @@ class DeviceFacade:
pkg = self.deviceV2.app_current().get("package")
if pkg == self.app_id:
return pkg
# Brief retry: many false positives come from <500ms notification banners
# A single short wait handles ALL transient overlays regardless of source app
sleep(0.5)
pkg = self.deviceV2.app_current().get("package")
# If still not our app, check if it's just SystemUI (always present, never a real takeover)
if pkg in ('com.android.systemui', 'android'):
if pkg in ("com.android.systemui", "android"):
return self.app_id
return pkg
@@ -284,38 +297,41 @@ class DeviceFacade:
def dump_hierarchy(self):
# Compressed=True dramatically speeds up UIAutomator2 dumps by skipping invisible elements!
xml = self.deviceV2.dump_hierarchy(compressed=True)
# Continuous Session Tracing
from datetime import datetime
try:
if not hasattr(self, "_trace_counter"):
self._trace_counter = 0
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
self._trace_dir = os.path.join("debug", "session_traces", ts)
os.makedirs(self._trace_dir, exist_ok=True)
self._trace_counter += 1
trace_path = os.path.join(self._trace_dir, f"{self._trace_counter:05d}.xml")
with open(trace_path, "w", encoding="utf-8") as f:
f.write(xml)
except Exception as e:
logger.debug(f"Failed to write session trace: {e}")
return xml
@adb_retry()
def get_screenshot_b64(self):
import base64
from io import BytesIO
img = self.deviceV2.screenshot()
buffered = BytesIO()
img.save(buffered, format="JPEG", quality=70) # Compressed for target latency
return base64.b64encode(buffered.getvalue()).decode('utf-8')
img.save(buffered, format="JPEG", quality=70) # Compressed for target latency
return base64.b64encode(buffered.getvalue()).decode("utf-8")
# Telepathic Semantic UI Integration
@adb_retry()
def find_semantic(self, intent_description: str):
from GramAddict.core.telepathic_engine import TelepathicEngine
engine = TelepathicEngine.get_instance()
xml = self.dump_hierarchy()
# Passing self (DeviceFacade) enables the Vision Cortex VLM fallback

849
GramAddict/core/goap.py
View File

@@ -13,858 +13,25 @@ Like a GPS navigation system:
- It remembers shortcuts (learn)
"""
import hashlib
import logging
import re
import time
import xml.etree.ElementTree as ET
from enum import Enum
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List
from GramAddict.core.qdrant_memory import QdrantBase
from GramAddict.core.utils import random_sleep
logger = logging.getLogger(__name__)
# ══════════════════════════════════════════════════════
# 1. SCREEN IDENTITY — "Where am I?"
# ══════════════════════════════════════════════════════
class ScreenType(Enum):
HOME_FEED = "home_feed"
EXPLORE_GRID = "explore_grid"
REELS_FEED = "reels_feed"
OWN_PROFILE = "own_profile"
OTHER_PROFILE = "other_profile"
POST_DETAIL = "post_detail"
STORY_VIEW = "story_view"
DM_INBOX = "dm_inbox"
DM_THREAD = "dm_thread"
SEARCH_RESULTS = "search_results"
FOLLOW_LIST = "follow_list"
COMMENTS = "comments"
MODAL = "modal"
FOREIGN_APP = "foreign_app"
UNKNOWN = "unknown"
class ScreenIdentity:
"""
Understands what screen the bot is on by analyzing the XML dump.
NO hardcoded states — purely structural analysis.
This is the bot's EYES. It answers: "What do I see right now?"
"""
def __init__(self, bot_username: str):
self.bot_username = bot_username.lower()
try:
from GramAddict.core.qdrant_memory import ScreenMemoryDB
self.screen_memory = ScreenMemoryDB()
except ImportError:
self.screen_memory = None
def identify(self, xml_dump: str) -> Dict[str, Any]:
"""
Analyzes an XML dump and returns a complete screen description.
Returns:
{
'screen_type': ScreenType,
'available_actions': ['tap like button', 'tap explore tab', ...],
'selected_tab': 'feed_tab' | 'search_tab' | ...,
'context': {'username': '...', 'post_count': '...', ...}
}
"""
if not xml_dump or not isinstance(xml_dump, str):
return self._empty_screen()
try:
clean = re.sub(r"<\?xml.*?\?>", "", xml_dump).strip()
root = ET.fromstring(clean)
except Exception:
return self._empty_screen()
# Extract structural signals
packages = set()
resource_ids = set()
content_descs = []
texts = []
selected_tab = None
clickable_elements = []
app_id = "com.instagram.android"
for elem in root.iter("node"):
pkg = elem.get("package", "")
if pkg:
packages.add(pkg)
rid = elem.get("resource-id", "").strip()
text = elem.get("text", "").strip()
desc = elem.get("content-desc", "").strip()
clickable = elem.get("clickable", "false") == "true"
selected = elem.get("selected", "false") == "true"
bounds = elem.get("bounds", "")
if rid:
# Normalize: "com.instagram.android:id/feed_tab" → "feed_tab"
short_id = rid.split("/")[-1] if "/" in rid else rid
resource_ids.add(short_id)
# Track which tab is selected
if selected and short_id in ("feed_tab", "search_tab", "clips_tab", "profile_tab", "direct_tab"):
selected_tab = short_id
if text:
texts.append(text)
if desc:
content_descs.append(desc)
if clickable and bounds:
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds)
if match:
left, t, r, b = map(int, match.groups())
cx, cy = (left + r) // 2, (t + b) // 2
clickable_elements.append(
{
"text": text,
"desc": desc,
"id": rid.split("/")[-1] if "/" in rid else rid,
"x": cx,
"y": cy,
"bounds": bounds,
}
)
# ── Foreign app check ──
if app_id not in packages:
return {
"screen_type": ScreenType.FOREIGN_APP,
"available_actions": ["press back", "force start instagram"],
"selected_tab": None,
"context": {"packages": list(packages)},
"signature": self._compute_signature(resource_ids, content_descs, texts),
}
desc_lower = " ".join(content_descs).lower()
text_lower = " ".join(texts).lower()
ids_str = " ".join(resource_ids).lower()
signature = self._compute_signature(resource_ids, content_descs, texts)
# ── Identify screen type from structural signals ──
screen_type = self._classify_screen(
resource_ids, content_descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature
)
# ── Extract available actions from clickable elements ──
available_actions = self._extract_available_actions(
clickable_elements, resource_ids, content_descs, texts, screen_type
)
# ── Extract context ──
context = self._extract_context(content_descs, texts, resource_ids, screen_type)
return {
"screen_type": screen_type,
"available_actions": available_actions,
"selected_tab": selected_tab,
"context": context,
"signature": signature,
}
def _classify_screen(self, ids, descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature=None):
"""Classify screen type using Semantic Memory with LLM fallback — NO hardcoded states."""
# Priority 0: Content-creation overlays that block ALL navigation.
# These full-screen Instagram UIs have no navigation tabs and trap the bot.
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
# Priority 1: Check Qdrant Semantic Cache
if signature and self.screen_memory and self.screen_memory.is_connected:
cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92)
if cached_type_str:
try:
return ScreenType[cached_type_str]
except KeyError:
pass
# Priority 2: Structural Heuristics (Instant, for core tabs)
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
return ScreenType.FOLLOW_LIST
if "profile_header_container" in ids:
return ScreenType.OTHER_PROFILE
# Reels structural markers — present even when Instagram hides the tab bar
# in full-screen Reels viewing. Without this, selected_tab=None → UNKNOWN.
REELS_MARKERS = ("clips_viewer_container", "root_clips_layout", "clips_linear_layout_container")
if any(marker in ids for marker in REELS_MARKERS):
return ScreenType.REELS_FEED
# DM thread detection — structural markers present inside DM conversations
if "direct_thread_header" in ids or "row_thread_composer_edittext" in ids:
return ScreenType.DM_THREAD
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab:
return ScreenType.POST_DETAIL
if selected_tab == "feed_tab":
return ScreenType.HOME_FEED
if selected_tab == "clips_tab":
return ScreenType.REELS_FEED
if selected_tab == "search_tab":
return ScreenType.EXPLORE_GRID
if selected_tab == "profile_tab":
return ScreenType.OWN_PROFILE
if selected_tab == "direct_tab":
return ScreenType.DM_INBOX
if "message_input" in ids:
return ScreenType.DM_INBOX # Fallback for DM thread as inbox
# Priority 3: Semantic VLM Classification Fallback
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_llm
cfg = Config()
url = (
getattr(cfg.args, "ai_embedding_url", "http://localhost:11434/api/chat")
if hasattr(cfg, "args")
else "http://localhost:11434/api/chat"
)
model = getattr(cfg.args, "ai_embedding_model", "llama3") if hasattr(cfg, "args") else "llama3"
layout_context = (
f"Selected Tab: {selected_tab}\nResource IDs: {list(ids)}\nVisible Texts context: {texts[:10]}\n"
)
prompt = (
f"Identify the Instagram screen layout type based on these DOM structural signals.\n"
f"Valid types: {[t.name for t in ScreenType]}\n"
f"Context:\n{layout_context}\n"
f"Reply ONLY with the exact matching enum Type Name string, or 'UNKNOWN' if no type matches."
)
try:
response = query_llm(
url=url, model=model, prompt="Classify this screen layout.", system=prompt, format_json=False
)
if response and isinstance(response, str):
result = response.strip().upper()
elif response and isinstance(response, dict) and "response" in response:
result = response["response"].strip().upper()
else:
return ScreenType.UNKNOWN
for t in ScreenType:
if t.name in result:
if signature and self.screen_memory:
self.screen_memory.store_screen(signature, t.name)
return t
except Exception as e:
import logging
logging.getLogger(__name__).debug(f"LLM Classification failed: {e}")
return ScreenType.UNKNOWN
def _extract_available_actions(self, clickable_elements, resource_ids, content_descs, texts, screen_type):
"""Discover what actions are possible on this screen."""
actions = []
# Navigation tabs (always available when visible)
tab_map = {
"feed_tab": "tap home tab",
"search_tab": "tap explore tab",
"clips_tab": "tap reels tab",
"profile_tab": "tap profile tab",
"direct_tab": "tap messages tab",
}
for tab_id, action in tab_map.items():
if tab_id in resource_ids:
actions.append(action)
# Screen-specific actions
desc_lower = " ".join(content_descs).lower()
text_lower = " ".join(texts).lower()
if "like" in desc_lower:
actions.append("tap like button")
if "comment" in desc_lower:
actions.append("tap comment button")
if "share" in desc_lower:
actions.append("tap share button")
if "save" in desc_lower or "bookmark" in desc_lower:
actions.append("tap save button")
if "back" in desc_lower:
actions.append("tap back button")
if any("follow" in e.get("text", "").lower() for e in clickable_elements):
actions.append("tap follow button")
if screen_type == ScreenType.OWN_PROFILE or screen_type == ScreenType.OTHER_PROFILE:
if "message" in desc_lower or "nachricht" in desc_lower:
actions.append("tap message button")
if (
"following" in desc_lower
or "abonniert" in desc_lower
or "following" in text_lower
or "profile_header_following" in " ".join(resource_ids).lower()
):
actions.append("tap following list")
# Grid items
if screen_type == ScreenType.EXPLORE_GRID:
actions.append("tap first grid item")
# Scroll
actions.append("scroll down")
actions.append("press back")
return list(set(actions)) # Deduplicate
def _extract_context(self, content_descs, texts, resource_ids, screen_type):
"""Extract meaningful context from the screen."""
context = {}
desc_text = " ".join(content_descs)
# Username on profile
username_match = re.search(r"(\w+)'s (?:profile|story|unseen story)", desc_text)
if username_match:
context["username"] = username_match.group(1)
# Post/follower counts
for d in content_descs:
m = re.match(r"([\d,.]+K?M?)(\s*)(posts?|followers?|following)", d, re.IGNORECASE)
if m:
context[m.group(3).lower()] = m.group(1)
# Like state
for d in content_descs:
if d.lower() == "liked":
context["is_liked"] = True
elif d.lower() == "like":
context["is_liked"] = False
return context
def _compute_signature(self, resource_ids, content_descs, texts):
"""Compute a stable hash for this screen state (for Qdrant lookup)."""
# Use sorted IDs + key content for stability
sig_parts = sorted(resource_ids)[:20]
sig_parts.extend(sorted(set(d.lower()[:30] for d in content_descs if len(d) > 2))[:10])
sig = "|".join(sig_parts)
return hashlib.sha256(sig.encode()).hexdigest()[:24]
def _empty_screen(self):
return {
"screen_type": ScreenType.FOREIGN_APP,
"available_actions": ["press back", "force start instagram"],
"selected_tab": None,
"context": {},
"signature": "empty",
}
from GramAddict.core.navigation.knowledge import NavigationKnowledge
from GramAddict.core.navigation.path_memory import PathMemory
from GramAddict.core.navigation.planner import GoalPlanner
from GramAddict.core.perception.screen_identity import ScreenIdentity, ScreenType
# Re-export for backward compatibility (optional but helps minimize import breakage)
__all__ = ["GoalExecutor", "ScreenIdentity", "ScreenType", "PathMemory", "NavigationKnowledge", "GoalPlanner"]
# ══════════════════════════════════════════════════════
# 2. PATH MEMORY — "How did I get there last time?"
# ══════════════════════════════════════════════════════
class PathMemory:
"""
Qdrant-backed memory for successful navigation paths.
Stores: goal → [step1, step2, ...] → success
Enables instant recall for known goals.
"""
def __init__(self, username: str = ""):
self.username = username
try:
suffix = f"_{username}" if username else ""
self._db = QdrantBase(f"goap_paths_v1{suffix}", vector_size=768)
except Exception:
self._db = None
def wipe(self):
"""Wipe all learned navigation paths from Qdrant."""
if self._db and self._db.is_connected:
try:
self._db.wipe_collection()
except Exception as e:
logger.warning(f"⚠️ [PathMemory] Could not wipe collection: {e}")
def recall_path(self, goal: str, current_screen_type: str) -> Optional[List[Dict]]:
"""
Recall a previously successful path for this goal from this screen type.
Returns list of steps or None.
"""
if not self._db or not self._db.is_connected:
return None
query = f"goal: {goal} | from: {current_screen_type}"
vec = self._db._get_embedding(query)
if not vec:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.query_points(
collection_name=self._db.collection_name,
query=vec,
query_filter=Filter(
must=[FieldCondition(key="start_screen", match=MatchValue(value=current_screen_type))]
),
limit=3,
score_threshold=0.85,
).points
for r in results:
p = r.payload
if p.get("success") and p.get("steps"):
logger.info(
f"🧠 [GOAP Recall] Found path for '{goal}': "
f"{len(p['steps'])} steps (confidence: {p.get('confidence', 0):.2f})"
)
return p["steps"]
return None
except Exception as e:
logger.debug(f"GOAP recall error: {e}")
return None
def learn_path(self, goal: str, start_screen: str, steps: List[Dict], success: bool):
"""Store a navigation path in Qdrant."""
if not self._db or not self._db.is_connected:
return
query = f"goal: {goal} | from: {start_screen}"
vec = self._db._get_embedding(query)
if not vec:
return
seed = f"{goal}|{start_screen}"
payload = {
"goal": goal,
"start_screen": start_screen,
"steps": steps,
"step_count": len(steps),
"success": success,
"confidence": 0.85 if success else 0.0,
"timestamp": time.time(),
}
outcome = "✅" if success else "❌"
self._db.upsert_point(
seed,
payload,
vector=vec,
log_success=f"🧠 [GOAP Learn] {outcome} Path for '{goal}': {len(steps)} steps from {start_screen}",
)
def forget_path(self, goal: str, start_screen: str):
"""Remove a cached path to force re-discovery."""
if not self._db or not self._db.is_connected:
return
seed = f"{goal}|{start_screen}"
try:
from qdrant_client import models
point_id = self._db._get_id(seed)
self._db.client.delete(
collection_name=self._db.collection_name, points_selector=models.PointIdsList(points=[point_id])
)
except Exception as e:
logger.debug(f"Failed to forget path: {e}")
# ══════════════════════════════════════════════════════
# 3. GOAL PLANNER — "What should I do next?"
# ══════════════════════════════════════════════════════
class NavigationKnowledge:
"""
Manages the bot's learned understanding of the Instagram UI.
Discovered dynamically through exploration and success.
"""
def __init__(self, username: str):
self.username = username
try:
self._db = QdrantBase("navigation_knowledge", vector_size=768)
except Exception:
self._db = None
# In-memory cache for rapidly avoiding traps during exploration
# In-memory cache for rapidly avoiding traps during exploration
self._learned_screen_mappings = {}
self._learned_traps = set()
def wipe(self):
"""Wipe all learned knowledge from Qdrant."""
if self._db and self._db.is_connected:
try:
self._db.wipe_collection()
except Exception as e:
logger.warning(f"⚠️ [NavigationKnowledge] Could not wipe knowledge: {e}")
def update_username(self, username: str):
"""Update username and reconnect DB if needed."""
if self.username != username:
self.username = username
try:
self._db = QdrantBase("navigation_knowledge", vector_size=768)
except Exception:
self._db = None
def get_requirements(self, goal: str) -> List[ScreenType]:
"""Get required screens for a goal. Returns known requirements or empty list."""
if not self._db or not self._db.is_connected:
return []
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="goal", match=MatchValue(value=goal))]),
limit=1,
)[0]
if results:
screen_name = results[0].payload.get("required_screen")
logger.debug(f"🧠 [Nav Knowledge] Found requirement for '{goal}': {screen_name}")
if screen_name:
return [ScreenType[screen_name]]
except Exception as e:
logger.warning(f"⚠️ [Nav Knowledge] Search error: {e}")
return []
def learn_goal_requirement(self, goal: str, screen_type: ScreenType):
"""Learn that achieving 'goal' lands us on 'screen_type'."""
if not self._db or not self._db.is_connected:
logger.warning("⚠️ [Nav Knowledge] Cannot learn: DB not connected")
return
seed = f"req_{goal}"
vec = self._db._get_embedding(f"goal_requirement: {goal}")
payload = {"goal": goal, "required_screen": screen_type.name, "timestamp": time.time()}
self._db.upsert_point(seed, payload, vector=vec)
logger.info(f"🧠 [Nav Knowledge] Learned: '{goal}' → {screen_type.name}")
def get_action_for_screen(self, target_screen: ScreenType) -> Optional[str]:
"""Find which action leads to this screen."""
for action, screen in self._learned_screen_mappings.items():
if screen == target_screen:
return action
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(
must=[FieldCondition(key="result_screen", match=MatchValue(value=target_screen.name))]
),
limit=1,
)[0]
if results:
return results[0].payload.get("action")
except Exception:
pass
return None
def get_screen_for_action(self, action: str) -> Optional[ScreenType]:
"""Find where this action leads to to avoid looping traps."""
if action in self._learned_screen_mappings:
return self._learned_screen_mappings[action]
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="action", match=MatchValue(value=action))]),
limit=1,
)[0]
if results:
screen_name = results[0].payload.get("result_screen")
if screen_name:
return ScreenType[screen_name]
except Exception:
pass
return None
def learn_screen_mapping(self, action: str, result_screen: ScreenType):
"""Learn that taking 'action' leads to 'result_screen'."""
if not self._db or not self._db.is_connected:
return
seed = f"map_{action}"
vec = self._db._get_embedding(f"screen_mapping: {result_screen.name}")
payload = {"action": action, "result_screen": result_screen.name, "timestamp": time.time()}
self._learned_screen_mappings[action] = result_screen
self._db.upsert_point(seed, payload, vector=vec)
logger.info(f"🧠 [Nav Knowledge] Learned Mapping: '{action}' → {result_screen.name}")
def get_screen_for_tab(self, tab_id: str) -> Optional[ScreenType]:
"""Find where this tab leads to to avoid looping traps."""
if tab_id in self._learned_screen_mappings:
return self._learned_screen_mappings[tab_id]
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="tab_id", match=MatchValue(value=tab_id))]),
limit=1,
)[0]
if results:
s_name = results[0].payload.get("result_screen")
if s_name:
return ScreenType[s_name]
except Exception:
pass
return None
def learn_trap(self, screen_type: ScreenType, action: str, trap_reason: str = "softlock"):
"""Aversively learn that an action on a screen is dangerous/useless."""
trap_key = f"{screen_type.name}_{action}"
self._learned_traps.add(trap_key)
if not self._db or not self._db.is_connected:
return
seed = f"trap_{trap_key}"
# Aversive vector is completely orthogonal to normal goals to prevent retrieval overlap
vec = self._db._get_embedding(f"trap_avoidance: {trap_key} {trap_reason}")
payload = {
"trap_screen": screen_type.name,
"trap_action": action,
"trap_reason": trap_reason,
"timestamp": time.time(),
}
self._db.upsert_point(seed, payload, vector=vec)
logger.error(f"💀 [Aversive Learning] BURNED action '{action}' on {screen_type.name} due to: {trap_reason}")
def is_trap(self, screen_type: ScreenType, action: str) -> bool:
"""Check if an action on this screen is a known trap."""
trap_key = f"{screen_type.name}_{action}"
if trap_key in self._learned_traps:
return True
if not self._db or not self._db.is_connected:
return False
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(
must=[
FieldCondition(key="trap_screen", match=MatchValue(value=screen_type.name)),
FieldCondition(key="trap_action", match=MatchValue(value=action)),
]
),
limit=1,
)[0]
if results:
self._learned_traps.add(trap_key)
return True
except Exception:
pass
return False
class GoalPlanner:
"""
Given a goal and current screen state, plans the next action.
Uses Dynamic Discovery to navigate without hardcoded maps.
"""
def __init__(self, username: str):
self.knowledge = NavigationKnowledge(username)
def plan_next_step(self, goal: str, screen: Dict[str, Any], explored_nav_actions: set = None) -> Optional[str]:
"""Plans the NEXT single action to take toward the goal."""
screen_type = screen["screen_type"]
available = screen.get("available_actions", [])
context = screen.get("context", {})
goal_lower = goal.lower()
# ── 1. Check if goal is ALREADY achieved ──
if self._is_goal_achieved(goal_lower, screen_type, context):
logger.info(f"🎯 [GOAP] Goal '{goal}' already achieved on {screen_type.value}!")
return None
# (Phase 5: legacy _plan_goal_action static heuristics purged,
# all intents fall through to VLM-driven Discovery in _plan_navigation)
# ── 3. Am I on the right screen? If not, navigate there ──
selected_tab = screen.get("selected_tab")
nav_action = self._plan_navigation(goal_lower, screen_type, available, selected_tab, explored_nav_actions)
if nav_action:
return nav_action
# Final fallback: back-track, UNLESS back-tracking is a known trap on this screen!
if not self.knowledge.is_trap(screen_type, "press back"):
return "press back"
# We are trapped! Can't go forward, can't go back!
logger.error(f"💀 [GOAP] Completely trapped on {screen_type.name}. Forcing Instagram restart.")
return "force start instagram"
def _is_goal_achieved(self, goal: str, screen_type: ScreenType, context: dict) -> bool:
"""Check if the goal is already satisfied. Delegates to ScreenTopology SSOT."""
from GramAddict.core.screen_topology import ScreenTopology
# Interaction goals (context-specific, not navigation)
if "like" in goal and context.get("is_liked") is True:
return True
if "view profile" in goal and screen_type in (ScreenType.OWN_PROFILE, ScreenType.OTHER_PROFILE):
return True
# Navigation goals — delegate to SSOT
target = ScreenTopology.goal_to_target_screen(goal)
if target and screen_type == target:
return True
return False
def _plan_navigation(
self,
goal: str,
screen_type: ScreenType,
available: List[str],
selected_tab: Optional[str] = None,
explored_nav_actions: set = None,
) -> Optional[str]:
"""If we're on the wrong screen, figure out how to navigate.
Strategy (priority order):
1. HD Map (ScreenTopology BFS) — deterministic, pre-computed routes
2. Learned Knowledge (Qdrant) — dynamic discovery from past sessions
3. Autonomous Discovery — linguistic matching + VLM intent
"""
from GramAddict.core.screen_topology import ScreenTopology
# 0. Aversive Filter: Remove known traps from available actions
safe_available = []
for action in available:
if not self.knowledge.is_trap(screen_type, action):
safe_available.append(action)
else:
logger.debug(f"🛡️ [Aversive Filter] Masking trapped action: '{action}'")
available = safe_available
# ── 1. HD Map Routing (Primary Strategy) ──
target_screen = ScreenTopology.goal_to_target_screen(goal)
if target_screen and target_screen != screen_type:
route = ScreenTopology.find_route(screen_type, target_screen)
if route:
next_action, next_screen = route[0]
# Verify action isn't explored/trapped
if next_action not in (explored_nav_actions or set()):
if not self.knowledge.is_trap(screen_type, next_action):
route_desc = " → ".join(s.name for _, s in route)
logger.info(
f"🗺️ [HD Map] Route: {screen_type.name} → {route_desc}. " f"Next action: '{next_action}'"
)
return next_action
else:
logger.warning(f"🛡️ [HD Map] Route action '{next_action}' is trapped. Falling back.")
else:
logger.debug(f"🛡️ [HD Map] Route action '{next_action}' already explored. Falling back.")
# ── 2. Learned Knowledge (Qdrant) ──
required_screens = self.knowledge.get_requirements(goal)
# ── 3. Autonomous Discovery (Blank Start fallback) ──
if not required_screens:
logger.info(f"🧠 [Nav Discovery] No known requirements for '{goal}'. Will attempt autonomous discovery.")
# Return raw intent for TelepathicEngine discovery (VLM)
if explored_nav_actions and goal in explored_nav_actions:
logger.info(
f"🛑 [Nav Discovery] Autonomous intent '{goal}' already tried and failed/trapped. Yielding to back-tracking."
)
return None # Don't return goal again — force fallback to press back
else:
return goal
# 4. If we're already on an acceptable screen, no navigation needed
if screen_type in required_screens:
return None
# 5. Find the action we need to take (from learned knowledge or HD map)
for target_screen in required_screens:
# Try HD Map first!
route = ScreenTopology.find_route(screen_type, target_screen)
if route:
next_action, next_screen = route[0]
if next_action not in (explored_nav_actions or set()):
if not self.knowledge.is_trap(screen_type, next_action):
logger.info(f"🧭 [Nav HD Map] Routing to required {target_screen.name} via '{next_action}'")
return next_action
known_action = self.knowledge.get_action_for_screen(target_screen)
if not known_action:
logger.info(f"🧭 [Nav Discovery] Don't know action to reach {target_screen.name}. Asking VLM...")
screen_friendly_name = target_screen.name.replace("_", " ").lower()
goal_words = [w.rstrip("s") for w in screen_friendly_name.split() if len(w) > 3]
for action in available:
if any(w in action.lower() for w in goal_words):
known_target = self.knowledge.get_screen_for_action(action)
if known_target and known_target != target_screen:
continue
logger.info(
f"🎯 [Nav Discovery] Linguistic match on available action! '{action}' aligns with '{screen_friendly_name}'"
)
return action
return f"navigate to {screen_friendly_name}"
else:
if known_action in available:
logger.info(f"🧭 [Nav Knowledge] Navigating to {target_screen.name} via '{known_action}'")
return known_action
# If no targeted navigation works, try going back first
if "press back" in available:
return "press back"
return None
# ══════════════════════════════════════════════════════
# 4. GOAL EXECUTOR — The Main Brain Loop
# GOAL EXECUTOR — The Main Brain Loop
# ══════════════════════════════════════════════════════

216
GramAddict/core/llm_provider.py
View File

@@ -1,71 +1,77 @@
import re
import os
import json
import requests
import logging
from typing import Optional, List, Dict
import os
import re
from typing import List, Optional
import requests
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
logger = logging.getLogger(__name__)
def extract_json(text: str) -> Optional[str]:
"""
Robustly extracts the first JSON object or array from a string that may contain
Robustly extracts the first JSON object or array from a string that may contain
natural language prefix/suffix. Also purges <think> blocks and markdown ticks.
"""
if not text:
return None
# 100% Autonomous: Scrub model's internal thinking process
if "<think>" in text:
text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL).strip()
text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
logger.debug("🧠 [LLM] Scoped thinking block detected and purged.")
# Remove markdown code block formats
text = re.sub(r'^```json\s*', '', text, flags=re.MULTILINE)
text = re.sub(r'^```\s*', '', text, flags=re.MULTILINE)
text = re.sub(r"^```json\s*", "", text, flags=re.MULTILINE)
text = re.sub(r"^```\s*", "", text, flags=re.MULTILINE)
# Try perfect json block extraction first
match = re.search(r'(\{.*\}|\[.*\])', text, re.DOTALL)
match = re.search(r"(\{.*\}|\[.*\])", text, re.DOTALL)
if match:
candidate = match.group(0)
try:
import json
json.loads(candidate)
return candidate
except Exception:
pass
# Smart Fallback: Truncated JSON Healing
# If standard validation fails (e.g., due to EOF truncation by local models),
# run a regex extraction pass over the raw generated text to safely salvage
# If standard validation fails (e.g., due to EOF truncation by local models),
# run a regex extraction pass over the raw generated text to safely salvage
# all key-value pairs that *were* successfully completed before the truncation.
import json
matches = re.findall(r'"([a-zA-Z0-9_]+)"\s*:\s*(?:([0-9.-]+)|"([^"\\]*(?:\\.[^"\\]*)*)")', text)
if matches:
res = {}
for k, num, obj in matches:
if num:
try:
res[k] = float(num) if '.' in num else int(num)
res[k] = float(num) if "." in num else int(num)
except ValueError:
res[k] = num
else:
res[k] = obj.replace('\\"', '"')
recovered_json = json.dumps(res)
logger.warning(f"🔧 [Fuzzy Parse] Successfully salvaged {len(res)} keys from heavily truncated LLM output.")
return recovered_json
return None
_MODEL_PRICING_CACHE = None
def get_model_pricing(model_id: str) -> dict:
global _MODEL_PRICING_CACHE
if _MODEL_PRICING_CACHE is None:
@@ -78,118 +84,128 @@ def get_model_pricing(model_id: str) -> dict:
_MODEL_PRICING_CACHE = {}
except Exception:
_MODEL_PRICING_CACHE = {}
# Check if exact match exists, if not, try partial matches (e.g., if version suffixes differ)
if _MODEL_PRICING_CACHE and model_id not in _MODEL_PRICING_CACHE:
for k, v in _MODEL_PRICING_CACHE.items():
if model_id in k or k in model_id:
return v
return _MODEL_PRICING_CACHE.get(model_id, {})
def prewarm_ollama_models(configs):
"""
Sends a dummy request to the configured local Ollama API endpoints via a background thread
Sends a dummy request to the configured local Ollama API endpoints via a background thread
to force the models to load into VRAM during bot startup, minimizing initial connection latency
and avoiding timeouts downstream.
"""
args = configs.args
def _warmup():
import threading
models_to_warm = set()
# Collect unique local models
for attr, url_attr in [
("ai_telepathic_model", "ai_telepathic_url"),
("ai_fallback_model", "ai_fallback_url"),
("ai_condenser_model", "ai_condenser_url"),
("ai_model", "ai_model_url")
("ai_model", "ai_model_url"),
]:
url = getattr(args, url_attr, "")
model = getattr(args, attr, "")
if model and url and ("localhost" in url or "127.0.0.1" in url):
models_to_warm.add((url, model))
for url, model in models_to_warm:
logger.info(f"🔥 [VRAM Pre-Warm] Instructing local Ollama engine to load {model} into memory in the background...")
logger.info(
f"🔥 [VRAM Pre-Warm] Instructing local Ollama engine to load {model} into memory in the background..."
)
try:
# Fire an ultra-short generation to force it into VRAM
requests.post(
url,
json={"model": model, "prompt": "Hi", "stream": False, "options": {"num_predict": 1}},
timeout=120
url,
json={"model": model, "prompt": "Hi", "stream": False, "options": {"num_predict": 1}},
timeout=120,
)
except Exception:
pass
if hasattr(args, "ai_telepathic_model"):
import threading
threading.Thread(target=_warmup, daemon=True).start()
def unload_ollama_models(configs):
"""
Sends keep_alive: 0 to all configured local Ollama API endpoints via a background thread
Sends keep_alive: 0 to all configured local Ollama API endpoints via a background thread
to force the models to unload from VRAM during bot shutdown.
"""
args = configs.args
def _unload():
import threading
models_to_unload = set()
# Collect unique local models
for attr, url_attr in [
("ai_telepathic_model", "ai_telepathic_url"),
("ai_fallback_model", "ai_fallback_url"),
("ai_condenser_model", "ai_condenser_url"),
("ai_model", "ai_model_url")
("ai_model", "ai_model_url"),
]:
url = getattr(args, url_attr, "")
model = getattr(args, attr, "")
if model and url and ("localhost" in url or "127.0.0.1" in url):
models_to_unload.add((url, model))
for url, model in models_to_unload:
logger.info(f"❄️ [VRAM Cleanup] Instructing local Ollama engine to unload {model} from memory...")
try:
# Fire keep_alive: 0 to unload it from VRAM
requests.post(
url,
json={"model": model, "keep_alive": 0},
timeout=5
)
requests.post(url, json={"model": model, "keep_alive": 0}, timeout=5)
except Exception as e:
logger.debug(f"Failed to unload {model}: {e}")
if hasattr(args, "ai_telepathic_model"):
import threading
threading.Thread(target=_unload, daemon=True).start()
def log_openrouter_burn():
"""Fetches and logs the current OpenRouter API key usage (money burned) ONLY if OpenRouter is actively used."""
key = os.environ.get("OPENROUTER_API_KEY")
if not key:
return
try:
from GramAddict.core.config import Config
args = Config().args
uses_openrouter = False
# Check all possible model/url endpoints for 'openrouter'
for attr in ["ai_model", "ai_model_url", "ai_telepathic_model", "ai_telepathic_url",
"ai_fallback_model", "ai_fallback_url", "ai_condenser_model", "ai_condenser_url"]:
for attr in [
"ai_model",
"ai_model_url",
"ai_telepathic_model",
"ai_telepathic_url",
"ai_fallback_model",
"ai_fallback_url",
"ai_condenser_model",
"ai_condenser_url",
]:
val = getattr(args, attr, "")
if val and "openrouter" in str(val).lower():
uses_openrouter = True
break
if not uses_openrouter:
return
except Exception:
pass
try:
r = requests.get("https://openrouter.ai/api/v1/auth/key", headers={"Authorization": f"Bearer {key}"}, timeout=5)
if r.status_code == 200:
@@ -197,11 +213,16 @@ def log_openrouter_burn():
total_spent = data.get("usage", 0.0)
daily_spent = data.get("usage_daily", 0.0)
limit = data.get("limit")
logger.info(f"🔥 [OpenRouter Burn Rate] Daily: ${daily_spent:.4f} | Total: ${total_spent:.4f}" + (f" | Limit: ${limit}" if limit else ""), extra={"color": "\x1b[38;5;208m\x1b[1m"})
logger.info(
f"🔥 [OpenRouter Burn Rate] Daily: ${daily_spent:.4f} | Total: ${total_spent:.4f}"
+ (f" | Limit: ${limit}" if limit else ""),
extra={"color": "\x1b[38;5;208m\x1b[1m"},
)
except Exception as e:
logger.debug(f"Could not fetch OpenRouter burn rate: {e}")
def query_llm(
url: str,
model: str,
@@ -213,16 +234,16 @@ def query_llm(
fallback_model: Optional[str] = None,
fallback_url: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None
max_tokens: Optional[int] = None,
) -> Optional[dict]:
"""
Unified LLM API Caller with configurable fallback.
"""
openrouter_key = os.environ.get("OPENROUTER_API_KEY")
# URL-based provider detection (not model-name based — works for any model)
is_openai_compat = "/v1/chat/completions" in url or "openrouter.ai" in url.lower() or "openai.com" in url.lower()
# If using a cloud model but a local URL was passed, fix it
if not is_openai_compat and ("openrouter" in model.lower() or "/" in model):
# Model looks like "org/model-name" which is OpenRouter format
@@ -230,54 +251,43 @@ def query_llm(
url = "https://openrouter.ai/api/v1/chat/completions"
headers = {"Content-Type": "application/json"}
if is_openai_compat:
if openrouter_key:
headers["Authorization"] = f"Bearer {openrouter_key}"
messages = []
if system:
messages.append({"role": "system", "content": system})
user_content = []
if prompt:
user_content.append({"type": "text", "text": prompt})
if images_b64:
for img in images_b64:
user_content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img}"}
})
user_content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img}"}})
messages.append({"role": "user", "content": user_content if len(user_content) > 1 else prompt})
req_data = {
"model": model,
"messages": messages,
"stream": False
}
req_data = {"model": model, "messages": messages, "stream": False}
if format_json:
req_data["response_format"] = {"type": "json_object"}
if temperature is not None:
req_data["temperature"] = temperature
if max_tokens is not None:
req_data["max_tokens"] = max_tokens
else:
# Ollama /generate API
req_data = {
"model": model,
"prompt": prompt,
"stream": False
}
req_data = {"model": model, "prompt": prompt, "stream": False}
if system:
req_data["system"] = system
if images_b64:
req_data["images"] = images_b64
if format_json:
req_data["format"] = "json"
# Ollama passes configs inside 'options'
if temperature is not None or max_tokens is not None:
req_data["options"] = {}
@@ -290,12 +300,12 @@ def query_llm(
response = requests.post(url, json=req_data, headers=headers, timeout=timeout)
response.raise_for_status()
resp_json = response.json()
# Normalize response payload so callers don't have to distinguish
if is_openai_compat:
# OpenRouter returns choices[0].message.content
content = resp_json.get("choices", [{}])[0].get("message", {}).get("content", "")
usage = resp_json.get("usage", {})
if usage:
cost_str = ""
@@ -306,26 +316,29 @@ def query_llm(
pricing = get_model_pricing(model)
if pricing:
try:
p_cost = float(pricing.get("prompt", 0)) * usage.get('prompt_tokens', 0)
c_cost = float(pricing.get("completion", 0)) * usage.get('completion_tokens', 0)
p_cost = float(pricing.get("prompt", 0)) * usage.get("prompt_tokens", 0)
c_cost = float(pricing.get("completion", 0)) * usage.get("completion_tokens", 0)
calc_cost = p_cost + c_cost
if calc_cost > 0:
cost_str = f" | 💸 Cost: ${calc_cost:.6f}"
except Exception:
pass
p_tokens = usage.get('prompt_tokens', 0)
c_tokens = usage.get('completion_tokens', 0)
t_tokens = usage.get('total_tokens', 0)
p_tokens = usage.get("prompt_tokens", 0)
c_tokens = usage.get("completion_tokens", 0)
t_tokens = usage.get("total_tokens", 0)
# Make it stand out!
logger.info(f"🪙 [LLM Burn] {model} -> In: {p_tokens} | Out: {c_tokens} | Total: {t_tokens}{cost_str}", extra={"color": "\x1b[38;5;208m\x1b[1m"})
logger.info(
f"🪙 [LLM Burn] {model} -> In: {p_tokens} | Out: {c_tokens} | Total: {t_tokens}{cost_str}",
extra={"color": "\x1b[38;5;208m\x1b[1m"},
)
# Validation: if JSON was expected, try to extract it
if format_json:
extracted = extract_json(content)
if not extracted:
raise ValueError(f"OpenRouter returned non-JSON content when JSON was expected: {content[:100]}...")
raise ValueError(f"OpenRouter returned non-JSON content when JSON was expected: {content[:100]}...")
content = extracted
return {"response": content}
@@ -337,31 +350,34 @@ def query_llm(
if not extracted:
# Log more context if JSON extraction fails
logger.debug(f"Ollama raw content (for JSON extraction): {content[:200]}...")
raise ValueError(f"Ollama returned non-JSON content when JSON was expected.")
raise ValueError("Ollama returned non-JSON content when JSON was expected.")
resp_json["response"] = extracted
return resp_json
except requests.exceptions.ConnectionError:
logger.error(f"⚠️ [LLM Provider] Connection refused for {model} at {url}. Is the service running?")
except Exception as e:
logger.error(f"LLM Provider Error with {model}: {e}")
# Prevent infinite fallback loops
if getattr(query_llm, "_is_fallback", False):
return None
# Decide on fallback model/url
f_model = fallback_model
f_url = fallback_url
# Read fallback config from args if available
if not f_model or not f_url:
from GramAddict.core.config import Config
try:
args = Config().args
f_model = f_model or getattr(args, "ai_fallback_model", None)
f_url = f_url or getattr(args, "ai_fallback_url", None)
except Exception:
pass
# Last resort defaults
if not f_model or not f_url:
if is_openai_compat:
@@ -388,12 +404,13 @@ def query_llm(
format_json=format_json,
timeout=timeout,
temperature=temperature,
max_tokens=max_tokens
max_tokens=max_tokens,
)
finally:
query_llm._is_fallback = False
return None
def query_telepathic_llm(
model: str,
url: str,
@@ -401,7 +418,7 @@ def query_telepathic_llm(
user_prompt: str,
temperature: float = 0.0,
use_local_edge: bool = False,
images_b64: Optional[List[str]] = None
images_b64: Optional[List[str]] = None,
) -> str:
"""
Routes UI Telepathic requests purely based on textual interpretation of the screen's XML nodes.
@@ -415,10 +432,13 @@ def query_telepathic_llm(
if use_local_edge:
is_already_local = "localhost" in url or "127.0.0.1" in url
if is_already_local:
logger.debug(f"⚡ [Edge Inference] Primary model {model} is already local. Using it directly to prevent VRAM thrashing.")
logger.debug(
f"⚡ [Edge Inference] Primary model {model} is already local. Using it directly to prevent VRAM thrashing."
)
else:
logger.info("⚡ [Edge Inference] Routing telepathic request to local Ollama host (0ms latency target).")
from GramAddict.core.config import Config
try:
args = Config().args
target_url = getattr(args, "ai_fallback_url", "http://localhost:11434/api/generate")
@@ -426,10 +446,10 @@ def query_telepathic_llm(
except Exception:
target_url = "http://localhost:11434/api/generate"
target_model = "llama3.2:1b"
is_local = "localhost" in target_url or "127.0.0.1" in target_url
calc_timeout = 180 if is_local else 45
ans = query_llm(
url=target_url,
model=target_model,
@@ -439,7 +459,7 @@ def query_telepathic_llm(
format_json=True,
timeout=calc_timeout, # Navigation VLM must fail fast for Cloud, but wait for Local VRAM loads
temperature=temperature,
max_tokens=150 # Hard stop to prevent VLM from endlessly hallucinating UI elements
max_tokens=150, # Hard stop to prevent VLM from endlessly hallucinating UI elements
)
if ans and "response" in ans:
return ans["response"]

1
GramAddict/core/navigation/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Navigation domain package

214
GramAddict/core/navigation/knowledge.py Normal file
View File

@@ -0,0 +1,214 @@
import logging
import time
from typing import List, Optional
from GramAddict.core.perception.screen_identity import ScreenType
from GramAddict.core.qdrant_memory import QdrantBase
logger = logging.getLogger(__name__)
class NavigationKnowledge:
"""
Manages the bot's learned understanding of the Instagram UI.
Discovered dynamically through exploration and success.
"""
def __init__(self, username: str):
self.username = username
try:
self._db = QdrantBase("navigation_knowledge", vector_size=768)
except Exception:
self._db = None
# In-memory cache for rapidly avoiding traps during exploration
# In-memory cache for rapidly avoiding traps during exploration
self._learned_screen_mappings = {}
self._learned_traps = set()
def wipe(self):
"""Wipe all learned knowledge from Qdrant."""
if self._db and self._db.is_connected:
try:
self._db.wipe_collection()
except Exception as e:
logger.warning(f"⚠️ [NavigationKnowledge] Could not wipe knowledge: {e}")
def update_username(self, username: str):
"""Update username and reconnect DB if needed."""
if self.username != username:
self.username = username
try:
self._db = QdrantBase("navigation_knowledge", vector_size=768)
except Exception:
self._db = None
def get_requirements(self, goal: str) -> List[ScreenType]:
"""Get required screens for a goal. Returns known requirements or empty list."""
if not self._db or not self._db.is_connected:
return []
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="goal", match=MatchValue(value=goal))]),
limit=1,
)[0]
if results:
screen_name = results[0].payload.get("required_screen")
logger.debug(f"🧠 [Nav Knowledge] Found requirement for '{goal}': {screen_name}")
if screen_name:
return [ScreenType[screen_name]]
except Exception as e:
logger.warning(f"⚠️ [Nav Knowledge] Search error: {e}")
return []
def learn_goal_requirement(self, goal: str, screen_type: ScreenType):
"""Learn that achieving 'goal' lands us on 'screen_type'."""
if not self._db or not self._db.is_connected:
logger.warning("⚠️ [Nav Knowledge] Cannot learn: DB not connected")
return
seed = f"req_{goal}"
vec = self._db._get_embedding(f"goal_requirement: {goal}")
payload = {"goal": goal, "required_screen": screen_type.name, "timestamp": time.time()}
self._db.upsert_point(seed, payload, vector=vec)
logger.info(f"🧠 [Nav Knowledge] Learned: '{goal}' → {screen_type.name}")
def get_action_for_screen(self, target_screen: ScreenType) -> Optional[str]:
"""Find which action leads to this screen."""
for action, screen in self._learned_screen_mappings.items():
if screen == target_screen:
return action
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(
must=[FieldCondition(key="result_screen", match=MatchValue(value=target_screen.name))]
),
limit=1,
)[0]
if results:
return results[0].payload.get("action")
except Exception:
pass
return None
def get_screen_for_action(self, action: str) -> Optional[ScreenType]:
"""Find where this action leads to to avoid looping traps."""
if action in self._learned_screen_mappings:
return self._learned_screen_mappings[action]
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="action", match=MatchValue(value=action))]),
limit=1,
)[0]
if results:
screen_name = results[0].payload.get("result_screen")
if screen_name:
return ScreenType[screen_name]
except Exception:
pass
return None
def learn_screen_mapping(self, action: str, result_screen: ScreenType):
"""Learn that taking 'action' leads to 'result_screen'."""
if not self._db or not self._db.is_connected:
return
seed = f"map_{action}"
vec = self._db._get_embedding(f"screen_mapping: {result_screen.name}")
payload = {"action": action, "result_screen": result_screen.name, "timestamp": time.time()}
self._learned_screen_mappings[action] = result_screen
self._db.upsert_point(seed, payload, vector=vec)
logger.info(f"🧠 [Nav Knowledge] Learned Mapping: '{action}' → {result_screen.name}")
def get_screen_for_tab(self, tab_id: str) -> Optional[ScreenType]:
"""Find where this tab leads to to avoid looping traps."""
if tab_id in self._learned_screen_mappings:
return self._learned_screen_mappings[tab_id]
if not self._db or not self._db.is_connected:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(must=[FieldCondition(key="tab_id", match=MatchValue(value=tab_id))]),
limit=1,
)[0]
if results:
s_name = results[0].payload.get("result_screen")
if s_name:
return ScreenType[s_name]
except Exception:
pass
return None
def learn_trap(self, screen_type: ScreenType, action: str, trap_reason: str = "softlock"):
"""Aversively learn that an action on a screen is dangerous/useless."""
trap_key = f"{screen_type.name}_{action}"
self._learned_traps.add(trap_key)
if not self._db or not self._db.is_connected:
return
seed = f"trap_{trap_key}"
# Aversive vector is completely orthogonal to normal goals to prevent retrieval overlap
vec = self._db._get_embedding(f"trap_avoidance: {trap_key} {trap_reason}")
payload = {
"trap_screen": screen_type.name,
"trap_action": action,
"trap_reason": trap_reason,
"timestamp": time.time(),
}
self._db.upsert_point(seed, payload, vector=vec)
logger.error(f"💀 [Aversive Learning] BURNED action '{action}' on {screen_type.name} due to: {trap_reason}")
def is_trap(self, screen_type: ScreenType, action: str) -> bool:
"""Check if an action on this screen is a known trap."""
trap_key = f"{screen_type.name}_{action}"
if trap_key in self._learned_traps:
return True
if not self._db or not self._db.is_connected:
return False
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.scroll(
collection_name=self._db.collection_name,
scroll_filter=Filter(
must=[
FieldCondition(key="trap_screen", match=MatchValue(value=screen_type.name)),
FieldCondition(key="trap_action", match=MatchValue(value=action)),
]
),
limit=1,
)[0]
if results:
self._learned_traps.add(trap_key)
return True
except Exception:
pass
return False

117
GramAddict/core/navigation/path_memory.py Normal file
View File

@@ -0,0 +1,117 @@
import logging
import time
from typing import Dict, List, Optional
from GramAddict.core.qdrant_memory import QdrantBase
logger = logging.getLogger(__name__)
class PathMemory:
"""
Qdrant-backed memory for successful navigation paths.
Stores: goal → [step1, step2, ...] → success
Enables instant recall for known goals.
"""
def __init__(self, username: str = ""):
self.username = username
try:
suffix = f"_{username}" if username else ""
self._db = QdrantBase(f"goap_paths_v1{suffix}", vector_size=768)
except Exception:
self._db = None
def wipe(self):
"""Wipe all learned navigation paths from Qdrant."""
if self._db and self._db.is_connected:
try:
self._db.wipe_collection()
except Exception as e:
logger.warning(f"⚠️ [PathMemory] Could not wipe collection: {e}")
def recall_path(self, goal: str, current_screen_type: str) -> Optional[List[Dict]]:
"""
Recall a previously successful path for this goal from this screen type.
Returns list of steps or None.
"""
if not self._db or not self._db.is_connected:
return None
query = f"goal: {goal} | from: {current_screen_type}"
vec = self._db._get_embedding(query)
if not vec:
return None
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
results = self._db.client.query_points(
collection_name=self._db.collection_name,
query=vec,
query_filter=Filter(
must=[FieldCondition(key="start_screen", match=MatchValue(value=current_screen_type))]
),
limit=3,
score_threshold=0.85,
).points
for r in results:
p = r.payload
if p.get("success") and p.get("steps"):
logger.info(
f"🧠 [GOAP Recall] Found path for '{goal}': "
f"{len(p['steps'])} steps (confidence: {p.get('confidence', 0):.2f})"
)
return p["steps"]
return None
except Exception as e:
logger.debug(f"GOAP recall error: {e}")
return None
def learn_path(self, goal: str, start_screen: str, steps: List[Dict], success: bool):
"""Store a navigation path in Qdrant."""
if not self._db or not self._db.is_connected:
return
query = f"goal: {goal} | from: {start_screen}"
vec = self._db._get_embedding(query)
if not vec:
return
seed = f"{goal}|{start_screen}"
payload = {
"goal": goal,
"start_screen": start_screen,
"steps": steps,
"step_count": len(steps),
"success": success,
"confidence": 0.85 if success else 0.0,
"timestamp": time.time(),
}
outcome = "✅" if success else "❌"
self._db.upsert_point(
seed,
payload,
vector=vec,
log_success=f"🧠 [GOAP Learn] {outcome} Path for '{goal}': {len(steps)} steps from {start_screen}",
)
def forget_path(self, goal: str, start_screen: str):
"""Remove a cached path to force re-discovery."""
if not self._db or not self._db.is_connected:
return
seed = f"{goal}|{start_screen}"
try:
from qdrant_client import models
point_id = self._db._get_id(seed)
self._db.client.delete(
collection_name=self._db.collection_name, points_selector=models.PointIdsList(points=[point_id])
)
except Exception as e:
logger.debug(f"Failed to forget path: {e}")

171
GramAddict/core/navigation/planner.py Normal file
View File

@@ -0,0 +1,171 @@
import logging
from typing import Any, Dict, List, Optional
from GramAddict.core.navigation.knowledge import NavigationKnowledge
from GramAddict.core.perception.screen_identity import ScreenType
logger = logging.getLogger(__name__)
class GoalPlanner:
"""
Given a goal and current screen state, plans the next action.
Uses Dynamic Discovery to navigate without hardcoded maps.
"""
def __init__(self, username: str):
self.knowledge = NavigationKnowledge(username)
def plan_next_step(self, goal: str, screen: Dict[str, Any], explored_nav_actions: set = None) -> Optional[str]:
"""Plans the NEXT single action to take toward the goal."""
screen_type = screen["screen_type"]
available = screen.get("available_actions", [])
context = screen.get("context", {})
goal_lower = goal.lower()
# ── 1. Check if goal is ALREADY achieved ──
if self._is_goal_achieved(goal_lower, screen_type, context):
logger.info(f"🎯 [GOAP] Goal '{goal}' already achieved on {screen_type.value}!")
return None
# (Phase 5: legacy _plan_goal_action static heuristics purged,
# all intents fall through to VLM-driven Discovery in _plan_navigation)
# ── 3. Am I on the right screen? If not, navigate there ──
selected_tab = screen.get("selected_tab")
nav_action = self._plan_navigation(goal_lower, screen_type, available, selected_tab, explored_nav_actions)
if nav_action:
return nav_action
# Final fallback: back-track, UNLESS back-tracking is a known trap on this screen!
if not self.knowledge.is_trap(screen_type, "press back"):
return "press back"
# We are trapped! Can't go forward, can't go back!
logger.error(f"💀 [GOAP] Completely trapped on {screen_type.name}. Forcing Instagram restart.")
return "force start instagram"
def _is_goal_achieved(self, goal: str, screen_type: ScreenType, context: dict) -> bool:
"""Check if the goal is already satisfied. Delegates to ScreenTopology SSOT."""
from GramAddict.core.screen_topology import ScreenTopology
# Interaction goals (context-specific, not navigation)
if "like" in goal and context.get("is_liked") is True:
return True
if "view profile" in goal and screen_type in (ScreenType.OWN_PROFILE, ScreenType.OTHER_PROFILE):
return True
# Navigation goals — delegate to SSOT
target = ScreenTopology.goal_to_target_screen(goal)
if target and screen_type == target:
return True
return False
def _plan_navigation(
self,
goal: str,
screen_type: ScreenType,
available: List[str],
selected_tab: Optional[str] = None,
explored_nav_actions: set = None,
) -> Optional[str]:
"""If we're on the wrong screen, figure out how to navigate.
Strategy (priority order):
1. HD Map (ScreenTopology BFS) — deterministic, pre-computed routes
2. Learned Knowledge (Qdrant) — dynamic discovery from past sessions
3. Autonomous Discovery — linguistic matching + VLM intent
"""
from GramAddict.core.screen_topology import ScreenTopology
# 0. Aversive Filter: Remove known traps from available actions
safe_available = []
for action in available:
if not self.knowledge.is_trap(screen_type, action):
safe_available.append(action)
else:
logger.debug(f"🛡️ [Aversive Filter] Masking trapped action: '{action}'")
available = safe_available
# ── 1. HD Map Routing (Primary Strategy) ──
target_screen = ScreenTopology.goal_to_target_screen(goal)
if target_screen and target_screen != screen_type:
route = ScreenTopology.find_route(screen_type, target_screen)
if route:
next_action, next_screen = route[0]
# Verify action isn't explored/trapped
if next_action not in (explored_nav_actions or set()):
if not self.knowledge.is_trap(screen_type, next_action):
route_desc = " → ".join(s.name for _, s in route)
logger.info(
f"🗺️ [HD Map] Route: {screen_type.name} → {route_desc}. " f"Next action: '{next_action}'"
)
return next_action
else:
logger.warning(f"🛡️ [HD Map] Route action '{next_action}' is trapped. Falling back.")
else:
logger.debug(f"🛡️ [HD Map] Route action '{next_action}' already explored. Falling back.")
# ── 2. Learned Knowledge (Qdrant) ──
required_screens = self.knowledge.get_requirements(goal)
# ── 3. Autonomous Discovery (Blank Start fallback) ──
if not required_screens:
logger.info(f"🧠 [Nav Discovery] No known requirements for '{goal}'. Will attempt autonomous discovery.")
# Return raw intent for TelepathicEngine discovery (VLM)
if explored_nav_actions and goal in explored_nav_actions:
logger.info(
f"🛑 [Nav Discovery] Autonomous intent '{goal}' already tried and failed/trapped. Yielding to back-tracking."
)
return None # Don't return goal again — force fallback to press back
else:
return goal
# 4. If we're already on an acceptable screen, no navigation needed
if screen_type in required_screens:
return None
# 5. Find the action we need to take (from learned knowledge or HD map)
for target_screen in required_screens:
# Try HD Map first!
route = ScreenTopology.find_route(screen_type, target_screen)
if route:
next_action, next_screen = route[0]
if next_action not in (explored_nav_actions or set()):
if not self.knowledge.is_trap(screen_type, next_action):
logger.info(f"🧭 [Nav HD Map] Routing to required {target_screen.name} via '{next_action}'")
return next_action
known_action = self.knowledge.get_action_for_screen(target_screen)
if not known_action:
logger.info(f"🧭 [Nav Discovery] Don't know action to reach {target_screen.name}. Asking VLM...")
screen_friendly_name = target_screen.name.replace("_", " ").lower()
goal_words = [w.rstrip("s") for w in screen_friendly_name.split() if len(w) > 3]
for action in available:
if any(w in action.lower() for w in goal_words):
known_target = self.knowledge.get_screen_for_action(action)
if known_target and known_target != target_screen:
continue
logger.info(
f"🎯 [Nav Discovery] Linguistic match on available action! '{action}' aligns with '{screen_friendly_name}'"
)
return action
return f"navigate to {screen_friendly_name}"
else:
if known_action in available:
logger.info(f"🧭 [Nav Knowledge] Navigating to {target_screen.name} via '{known_action}'")
return known_action
# If no targeted navigation works, try going back first
if "press back" in available:
return "press back"
return None

96
GramAddict/core/perception/action_memory.py Normal file
View File

@@ -0,0 +1,96 @@
import logging
from typing import Any, Dict, Optional
from GramAddict.core.perception.spatial_parser import SpatialNode
logger = logging.getLogger(__name__)
class ActionMemory:
"""
Handles the caching, tracking, and negative reinforcement (unlearning) of UI interactions.
Decouples the memory layer from the core parsing engine.
"""
def __init__(self, ui_memory=None):
# We optionally inject UIMemoryDB to decouple tests
if ui_memory is None:
from GramAddict.core.qdrant_memory import UIMemoryDB
self.ui_memory = UIMemoryDB()
else:
self.ui_memory = ui_memory
self._last_click_context: Optional[Dict[str, Any]] = None
def track_click(self, intent: str, node: SpatialNode, xml_context: str = ""):
"""Stores the context of a click before it's actually performed."""
semantic_string = f"text: '{node.text}', desc: '{node.content_desc}', id: '{node.resource_id}'"
self._last_click_context = {
"intent": intent,
"node_dict": node.to_dict(),
"semantic_string": semantic_string,
"xml_context": xml_context,
}
logger.debug(f"🧠 [ActionMemory] Tracking tentative click for intent: '{intent}' -> {semantic_string}")
def confirm_click(self, intent: str = None):
"""Positive Reinforcement: Confirms the last click was successful."""
ctx = self._last_click_context
if not ctx:
return
if intent and ctx["intent"] != intent:
return
logger.info(f"✅ [ActionMemory] Confirming success for '{ctx['intent']}'. Boosting confidence.")
# Store or boost in Qdrant
try:
# Check if it exists first
existing = self.ui_memory.retrieve_memory(ctx["intent"], ctx["xml_context"])
if existing:
self.ui_memory.boost_confidence(ctx["intent"], ctx["xml_context"])
else:
self.ui_memory.store_memory(ctx["intent"], ctx["xml_context"], ctx["node_dict"])
except Exception as e:
logger.warning(f"Failed to confirm click in Qdrant: {e}")
self._last_click_context = None
def reject_click(self, intent: str = None):
"""Negative Reinforcement: Penalizes a failed click (Unlearning)."""
ctx = self._last_click_context
if not ctx:
return
if intent and ctx["intent"] != intent:
return
logger.warning(f"❌ [ActionMemory] Click failed for '{ctx['intent']}'. Applying penalty.")
try:
self.ui_memory.decay_confidence(ctx["intent"], ctx["xml_context"])
except Exception as e:
logger.warning(f"Failed to decay confidence in Qdrant: {e}")
self._last_click_context = None
def verify_success(self, intent: str, pre_click_xml: str, post_click_xml: str) -> Optional[bool]:
"""
Structural verification: Did the UI actually change after the click?
"""
# Specific check for explore grid
if "first image in explore grid" in intent or "grid item" in intent:
if "row_feed_photo_imageview" in post_click_xml or "row_feed_button_like" in post_click_xml:
return True
if "explore_action_bar" in post_click_xml and "row_feed_button_like" not in post_click_xml:
return None # Still on grid, inconclusive
if abs(len(pre_click_xml) - len(post_click_xml)) > 50:
logger.debug(f"🧠 [ActionMemory] Structural change detected for '{intent}'. Verification PASS.")
return True
logger.warning(f"⚠️ [ActionMemory] No structural change detected for '{intent}'. Verification FAIL.")
return False

63
GramAddict/core/perception/feed_analysis.py
View File

@@ -1,7 +1,7 @@
"""
Perception — Feed Content Analysis.
Structural analysis of the feed: detecting markers, carousels,
Structural analysis of the feed: detecting markers, carousels,
extracting post content. Zero-AI, pure structural parsing.
Extracted from bot_flow.py to enable isolated testing.
@@ -27,14 +27,14 @@ FEED_MARKERS = [
"clips_linear_layout_container",
"zoomable_view_container",
"feed_action_row",
"carousel_viewpager"
"carousel_viewpager",
]
# ── Carousel Detection ──
CAROUSEL_INDICATORS = [
"com.instagram.android:id/carousel_page_indicator",
"com.instagram.android:id/carousel_media_group",
"com.instagram.android:id/carousel_viewpager"
"com.instagram.android:id/carousel_viewpager",
]
@@ -50,26 +50,29 @@ def extract_post_content(context_xml: str) -> 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.
Returns:
{'username': str, 'description': str, 'caption': str}
"""
result = {"username": "", "description": "", "caption": ""}
try:
from GramAddict.core.telepathic_engine import TelepathicEngine
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)
if author_node and author_node.get("original_attribs", {}).get("text"):
# 🛡️ Anti-Hallucination Guard: The author header is always near the top. Ignore names in the comment section.
if author_node and author_node.get("y", 0) < 1000 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)
if media_node and media_node.get("original_attribs", {}).get("desc"):
result["description"] = media_node["original_attribs"]["desc"].strip()
# 3. Visible caption text (heuristic fallback if node isn't explicitly found)
# Search all nodes for text that contains the username to find the caption body
root = ET.fromstring(context_xml)
@@ -78,13 +81,51 @@ def extract_post_content(context_xml: str) -> dict:
if result["username"] and len(text) > 20 and result["username"] in text:
result["caption"] = text
break
except Exception as e:
logger.warning(f"Error extracting post content autonomously: {e}")
return result
def _parse_number_from_text(text: str) -> int:
"""Extracts numeric value from strings like '1,234 likes', '1.5M views', 'Gefällt 12.345 Mal'."""
text = text.lower()
# Clean up purely thousands separators but keep decimals
# If there is a 'm' or 'k', a period is usually a decimal (e.g. 1.5m).
# If no 'm' or 'k', a period might be a German thousands separator (12.345).
# We will let the regex handle decimals.
# Remove commas (usually thousands separator in English)
text = text.replace(",", "")
# Find all numbers, potentially with k or m
matches = re.findall(r"(\d+(?:\.\d+)?)\s*([km])?", text)
if not matches:
return 0
best_val = 0
for num_str, multiplier in matches:
val = float(num_str)
if multiplier == "k":
val *= 1000
elif multiplier == "m":
val *= 1000000
else:
# If no multiplier, a period in num_str might be a German thousands separator
if "." in num_str and val < 1000:
# E.g. '12.345' became 12.345. Since no multiplier, it's actually 12345.
# Heuristic: If it has 3 decimal places, it's a thousands separator.
parts = num_str.split(".")
if len(parts[1]) == 3:
val = float(num_str.replace(".", ""))
best_val = max(best_val, int(val))
return best_val
def has_feed_markers(xml_dump: str) -> bool:
"""Quick check: does this XML contain any feed presence markers?"""
return any(marker in xml_dump for marker in FEED_MARKERS)

349
GramAddict/core/perception/screen_identity.py Normal file
View File

@@ -0,0 +1,349 @@
import hashlib
import logging
import re
import xml.etree.ElementTree as ET
from enum import Enum
from typing import Any, Dict
logger = logging.getLogger(__name__)
class ScreenType(Enum):
HOME_FEED = "home_feed"
EXPLORE_GRID = "explore_grid"
REELS_FEED = "reels_feed"
OWN_PROFILE = "own_profile"
OTHER_PROFILE = "other_profile"
POST_DETAIL = "post_detail"
STORY_VIEW = "story_view"
DM_INBOX = "dm_inbox"
DM_THREAD = "dm_thread"
SEARCH_RESULTS = "search_results"
FOLLOW_LIST = "follow_list"
COMMENTS = "comments"
MODAL = "modal"
FOREIGN_APP = "foreign_app"
UNKNOWN = "unknown"
class ScreenIdentity:
"""
Understands what screen the bot is on by analyzing the XML dump.
NO hardcoded states — purely structural analysis.
This is the bot's EYES. It answers: "What do I see right now?"
"""
def __init__(self, bot_username: str):
self.bot_username = bot_username.lower()
try:
from GramAddict.core.qdrant_memory import ScreenMemoryDB
self.screen_memory = ScreenMemoryDB()
except ImportError:
self.screen_memory = None
def identify(self, xml_dump: str) -> Dict[str, Any]:
"""
Analyzes an XML dump and returns a complete screen description.
Returns:
{
'screen_type': ScreenType,
'available_actions': ['tap like button', 'tap explore tab', ...],
'selected_tab': 'feed_tab' | 'search_tab' | ...,
'context': {'username': '...', 'post_count': '...', ...}
}
"""
if not xml_dump or not isinstance(xml_dump, str):
return self._empty_screen()
try:
clean = re.sub(r"<\?xml.*?\?>", "", xml_dump).strip()
root = ET.fromstring(clean)
except Exception:
return self._empty_screen()
# Extract structural signals
packages = set()
resource_ids = set()
content_descs = []
texts = []
selected_tab = None
clickable_elements = []
app_id = "com.instagram.android"
for elem in root.iter("node"):
pkg = elem.get("package", "")
if pkg:
packages.add(pkg)
rid = elem.get("resource-id", "").strip()
text = elem.get("text", "").strip()
desc = elem.get("content-desc", "").strip()
clickable = elem.get("clickable", "false") == "true"
selected = elem.get("selected", "false") == "true"
bounds = elem.get("bounds", "")
if rid:
# Normalize: "com.instagram.android:id/feed_tab" → "feed_tab"
short_id = rid.split("/")[-1] if "/" in rid else rid
resource_ids.add(short_id)
# Track which tab is selected
if selected and short_id in ("feed_tab", "search_tab", "clips_tab", "profile_tab", "direct_tab"):
selected_tab = short_id
if text:
texts.append(text)
if desc:
content_descs.append(desc)
if clickable and bounds:
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds)
if match:
left, t, r, b = map(int, match.groups())
cx, cy = (left + r) // 2, (t + b) // 2
clickable_elements.append(
{
"text": text,
"desc": desc,
"id": rid.split("/")[-1] if "/" in rid else rid,
"x": cx,
"y": cy,
"bounds": bounds,
}
)
# ── Foreign app check ──
if app_id not in packages:
return {
"screen_type": ScreenType.FOREIGN_APP,
"available_actions": ["press back", "force start instagram"],
"selected_tab": None,
"context": {"packages": list(packages)},
"signature": self._compute_signature(resource_ids, content_descs, texts),
}
desc_lower = " ".join(content_descs).lower()
text_lower = " ".join(texts).lower()
ids_str = " ".join(resource_ids).lower()
signature = self._compute_signature(resource_ids, content_descs, texts)
# ── Identify screen type from structural signals ──
screen_type = self._classify_screen(
resource_ids, content_descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature
)
# ── Extract available actions from clickable elements ──
available_actions = self._extract_available_actions(
clickable_elements, resource_ids, content_descs, texts, screen_type
)
# ── Extract context ──
context = self._extract_context(content_descs, texts, resource_ids, screen_type)
return {
"screen_type": screen_type,
"available_actions": available_actions,
"selected_tab": selected_tab,
"context": context,
"signature": signature,
}
def _classify_screen(self, ids, descs, texts, selected_tab, desc_lower, text_lower, ids_str, signature=None):
"""Classify screen type using Semantic Memory with LLM fallback — NO hardcoded states."""
# Priority 0: Content-creation overlays that block ALL navigation.
# These full-screen Instagram UIs have no navigation tabs and trap the bot.
# Structural detection is O(1), zero LLM calls, and cannot be fooled.
creation_flow_markers = ("quick_capture", "gallery_cancel_button", "creation_flow", "reel_camera")
if any(marker in ids_str for marker in creation_flow_markers):
logger.info("🛡️ [ScreenIdentity] Content-creation overlay detected → MODAL")
return ScreenType.MODAL
# Priority 1: Check Qdrant Semantic Cache
if signature and self.screen_memory and self.screen_memory.is_connected:
cached_type_str = self.screen_memory.get_screen_type(signature, similarity_threshold=0.92)
if cached_type_str:
try:
return ScreenType[cached_type_str]
except KeyError:
pass
# Priority 2: Structural Heuristics (Instant, for core tabs)
if "unified_follow_list_tab_layout" in ids or "follow_list_container" in ids:
return ScreenType.FOLLOW_LIST
if "profile_header_container" in ids:
return ScreenType.OTHER_PROFILE
# Reels structural markers — present even when Instagram hides the tab bar
# in full-screen Reels viewing. Without this, selected_tab=None → UNKNOWN.
REELS_MARKERS = ("clips_viewer_container", "root_clips_layout", "clips_linear_layout_container")
if any(marker in ids for marker in REELS_MARKERS):
return ScreenType.REELS_FEED
# DM thread detection — structural markers present inside DM conversations
if "direct_thread_header" in ids or "row_thread_composer_edittext" in ids:
return ScreenType.DM_THREAD
if "row_feed_button_like" in ids and "row_feed_photo_profile_name" in ids and not selected_tab:
return ScreenType.POST_DETAIL
if selected_tab == "feed_tab":
return ScreenType.HOME_FEED
if selected_tab == "clips_tab":
return ScreenType.REELS_FEED
if selected_tab == "search_tab":
return ScreenType.EXPLORE_GRID
if selected_tab == "profile_tab":
return ScreenType.OWN_PROFILE
if selected_tab == "direct_tab":
return ScreenType.DM_INBOX
if "message_input" in ids:
return ScreenType.DM_INBOX # Fallback for DM thread as inbox
# Priority 3: Semantic VLM Classification Fallback
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_llm
cfg = Config()
url = (
getattr(cfg.args, "ai_embedding_url", "http://localhost:11434/api/chat")
if hasattr(cfg, "args")
else "http://localhost:11434/api/chat"
)
model = getattr(cfg.args, "ai_embedding_model", "llama3") if hasattr(cfg, "args") else "llama3"
layout_context = (
f"Selected Tab: {selected_tab}\nResource IDs: {list(ids)}\nVisible Texts context: {texts[:10]}\n"
)
prompt = (
f"Identify the Instagram screen layout type based on these DOM structural signals.\n"
f"Valid types: {[t.name for t in ScreenType]}\n"
f"Context:\n{layout_context}\n"
f"Reply ONLY with the exact matching enum Type Name string, or 'UNKNOWN' if no type matches."
)
try:
response = query_llm(
url=url, model=model, prompt="Classify this screen layout.", system=prompt, format_json=False
)
if response and isinstance(response, str):
result = response.strip().upper()
elif response and isinstance(response, dict) and "response" in response:
result = response["response"].strip().upper()
else:
return ScreenType.UNKNOWN
for t in ScreenType:
if t.name in result:
if signature and self.screen_memory:
self.screen_memory.store_screen(signature, t.name)
return t
except Exception as e:
import logging
logging.getLogger(__name__).debug(f"LLM Classification failed: {e}")
return ScreenType.UNKNOWN
def _extract_available_actions(self, clickable_elements, resource_ids, content_descs, texts, screen_type):
"""Discover what actions are possible on this screen."""
actions = []
# Navigation tabs (always available when visible)
tab_map = {
"feed_tab": "tap home tab",
"search_tab": "tap explore tab",
"clips_tab": "tap reels tab",
"profile_tab": "tap profile tab",
"direct_tab": "tap messages tab",
}
for tab_id, action in tab_map.items():
if tab_id in resource_ids:
actions.append(action)
# Screen-specific actions
desc_lower = " ".join(content_descs).lower()
text_lower = " ".join(texts).lower()
if "like" in desc_lower:
actions.append("tap like button")
if "comment" in desc_lower:
actions.append("tap comment button")
if "share" in desc_lower:
actions.append("tap share button")
if "save" in desc_lower or "bookmark" in desc_lower:
actions.append("tap save button")
if "back" in desc_lower:
actions.append("tap back button")
if any("follow" in e.get("text", "").lower() for e in clickable_elements):
actions.append("tap follow button")
if screen_type == ScreenType.OWN_PROFILE or screen_type == ScreenType.OTHER_PROFILE:
if "message" in desc_lower or "nachricht" in desc_lower:
actions.append("tap message button")
if (
"following" in desc_lower
or "abonniert" in desc_lower
or "following" in text_lower
or "profile_header_following" in " ".join(resource_ids).lower()
):
actions.append("tap following list")
# Grid items
if screen_type == ScreenType.EXPLORE_GRID:
actions.append("tap first grid item")
# Scroll
actions.append("scroll down")
actions.append("press back")
return list(set(actions)) # Deduplicate
def _extract_context(self, content_descs, texts, resource_ids, screen_type):
"""Extract meaningful context from the screen."""
context = {}
desc_text = " ".join(content_descs)
# Username on profile
username_match = re.search(r"(\w+)'s (?:profile|story|unseen story)", desc_text)
if username_match:
context["username"] = username_match.group(1)
# Post/follower counts
for d in content_descs:
m = re.match(r"([\d,.]+K?M?)(\s*)(posts?|followers?|following)", d, re.IGNORECASE)
if m:
context[m.group(3).lower()] = m.group(1)
# Like state
for d in content_descs:
if d.lower() == "liked":
context["is_liked"] = True
elif d.lower() == "like":
context["is_liked"] = False
return context
def _compute_signature(self, resource_ids, content_descs, texts):
"""Compute a stable hash for this screen state (for Qdrant lookup)."""
# Use sorted IDs + key content for stability
sig_parts = sorted(resource_ids)[:20]
sig_parts.extend(sorted(set(d.lower()[:30] for d in content_descs if len(d) > 2))[:10])
sig = "|".join(sig_parts)
return hashlib.sha256(sig.encode()).hexdigest()[:24]
def _empty_screen(self):
return {
"screen_type": ScreenType.FOREIGN_APP,
"available_actions": ["press back", "force start instagram"],
"selected_tab": None,
"context": {},
"signature": "empty",
}

144
GramAddict/core/perception/semantic_evaluator.py Normal file
View File

@@ -0,0 +1,144 @@
import json
import logging
import re
from typing import List, Optional
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.perception.spatial_parser import SpatialNode
logger = logging.getLogger(__name__)
class SemanticEvaluator:
"""
Handles LLM/VLM interaction for high-level semantic analysis of the UI.
Delegates vision processing and prompt engineering out of the core routing engine.
"""
def __init__(self):
from GramAddict.core.config import Config
try:
self.args = Config().args
except Exception:
self.args = None
def _query_vlm(self, prompt: str, screenshot_b64: str) -> Optional[str]:
if not self.args:
logger.warning("👁️ [Vision Core] No config available. Cannot query VLM.")
return None
model = getattr(self.args, "ai_telepathic_model", "llama3.2-vision")
url = getattr(self.args, "ai_telepathic_url", "http://localhost:11434/api/generate")
try:
res = query_telepathic_llm(
model=model,
url=url,
system_prompt="You are an expert Instagram assistant.",
user_prompt=prompt,
images_b64=[screenshot_b64],
)
return res
except Exception as e:
logger.error(f"👁️ [Vision Core] LLM query failed: {e}")
return None
def evaluate_grid_visuals(
self, device, persona_interests: list[str], grid_nodes: List[SpatialNode]
) -> Optional[SpatialNode]:
"""
Takes the spatial grid nodes and asks the VLM which one best matches the persona.
"""
logger.info(f"👁️ [Vision Core] Analyzing grid aesthetics against niche interests: {persona_interests}...")
if not grid_nodes:
return None
# Take a screenshot
try:
screenshot_b64 = device.get_screenshot_b64()
except Exception as e:
logger.error(f"👁️ [Vision Core] Failed to capture screenshot: {e}")
return None
simplified_nodes = []
for i, node in enumerate(grid_nodes[:9]): # Limit to 9 to save tokens
simplified_nodes.append({"index": i, "bounds": node.bounds})
prompt = f"""
You are a highly perceptive Instagram user with the following interests: {', '.join(persona_interests)}.
Look at the provided screenshot of the Instagram Explore/Profile grid.
Below are the bounding boxes for the top grid posts currently visible.
{simplified_nodes}
Your task:
1. Identify which of these posts visually aligns BEST with your interests.
2. Reply ONLY in JSON format: {{"index": <int>}}
3. If absolutely none of them are relevant, reply with {{"index": -1}}.
"""
try:
response = self._query_vlm(prompt, screenshot_b64)
if not response:
return None
try:
data = json.loads(response)
idx = data.get("index", -1)
if idx == -1:
logger.info("👁️ [Vision Core] VLM rejected all grid items. Will scroll down.")
return None
if 0 <= idx < len(grid_nodes):
logger.info(f"👁️ [Vision Core] VLM selected grid item index [{idx}] as the best match.")
return grid_nodes[idx]
except json.JSONDecodeError:
# Fallback to fuzzy
clean_res = response.strip().upper()
match = re.search(r"\d+", clean_res)
if match:
idx = int(match.group())
if 0 <= idx < len(grid_nodes):
logger.info(f"👁️ [Vision Core] VLM selected grid item index [{idx}] as the best match.")
return grid_nodes[idx]
except Exception as e:
logger.warning(f"👁️ [Vision Core] Exception during grid evaluation: {e}")
return None
def evaluate_post_vibe(self, device, persona_interests: list[str]) -> Optional[dict]:
"""Evaluates whether the currently viewed post aligns with persona interests."""
logger.info(f"👁️ [Vision Core] Evaluating post vibe against: {persona_interests}")
try:
screenshot_b64 = device.get_screenshot_b64()
prompt = f"""
You are a user with the following interests: {', '.join(persona_interests)}.
You are looking at an Instagram post.
Evaluate if this post is highly relevant to your interests and if you should like/comment on it.
Reply ONLY in valid JSON format:
{{
"should_like": true/false,
"should_comment": true/false,
"reasoning": "brief explanation"
}}
"""
response = self._query_vlm(prompt, screenshot_b64)
if response:
if "```json" in response:
json_str = response.split("```json")[1].split("```")[0].strip()
else:
json_str = response.strip()
return json.loads(json_str)
except Exception as e:
logger.warning(f"Failed to evaluate post vibe: {e}")
return None
def evaluate_profile_vibe(self, device, persona_interests: list[str]) -> Optional[dict]:
"""Evaluates if a profile is worth following."""
pass
def classify_screen_content(self, xml_hierarchy: str, target_class: str) -> Optional[str]:
pass

194
GramAddict/core/perception/spatial_parser.py Normal file
View File

@@ -0,0 +1,194 @@
import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
@dataclass
class SpatialNode:
"""A single node in the Spatial Graph, representing a UI element and its geometry."""
bounds: Tuple[int, int, int, int] # (x1, y1, x2, y2)
node_id: str = ""
class_name: str = ""
text: str = ""
content_desc: str = ""
resource_id: str = ""
clickable: bool = False
scrollable: bool = False
# Spatial Properties
children: List["SpatialNode"] = field(default_factory=list)
parent: Optional["SpatialNode"] = None
@property
def x1(self) -> int:
return self.bounds[0]
@property
def y1(self) -> int:
return self.bounds[1]
@property
def x2(self) -> int:
return self.bounds[2]
@property
def y2(self) -> int:
return self.bounds[3]
@property
def width(self) -> int:
return self.x2 - self.x1
@property
def height(self) -> int:
return self.y2 - self.y1
@property
def center_x(self) -> int:
return self.x1 + (self.width // 2)
@property
def center_y(self) -> int:
return self.y1 + (self.height // 2)
@property
def area(self) -> int:
return self.width * self.height
def contains(self, other: "SpatialNode") -> bool:
"""Returns True if this node completely encompasses the other node geometrically."""
return self.x1 <= other.x1 and self.y1 <= other.y1 and self.x2 >= other.x2 and self.y2 >= other.y2
def intersects(self, other: "SpatialNode") -> bool:
"""Returns True if this node's bounding box overlaps with the other's bounding box."""
if self.x1 >= other.x2 or other.x1 >= self.x2:
return False
if self.y1 >= other.y2 or other.y1 >= self.y2:
return False
return True
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.node_id,
"class": self.class_name,
"text": self.text,
"content_desc": self.content_desc,
"resource_id": self.resource_id,
"bounds": self.bounds,
"clickable": self.clickable,
"scrollable": self.scrollable,
"center": (self.center_x, self.center_y),
}
class SpatialParser:
"""
Parses Android UI XML into a structured 2D Spatial Tree.
Calculates parent-child relationships structurally, not just based on XML nesting.
"""
def __init__(self):
self._node_counter = 0
def parse(self, xml_string: str) -> Optional[SpatialNode]:
"""Parses the raw XML dump into a Spatial Graph."""
try:
clean_xml = re.sub(r"<\?xml.*?\?>", "", xml_string).strip()
if not clean_xml:
return None
root_elem = ET.fromstring(clean_xml)
# 1. First Pass: Create flat list of spatial nodes
all_nodes = []
self._flatten_xml(root_elem, all_nodes)
if not all_nodes:
return None
# 2. Second Pass: Reconstruct tree based on strict spatial containment
# Sort nodes by area descending (largest first)
all_nodes.sort(key=lambda n: n.area, reverse=True)
root_node = all_nodes[0]
for i in range(1, len(all_nodes)):
child = all_nodes[i]
# Find the smallest node that contains this child
# Since we sorted by area descending, we search backwards to find the tightest fit
parent_found = False
for j in range(i - 1, -1, -1):
potential_parent = all_nodes[j]
if potential_parent.contains(child):
potential_parent.children.append(child)
child.parent = potential_parent
parent_found = True
break
# Fallback to root if no parent found (floating node)
if not parent_found and child != root_node:
root_node.children.append(child)
child.parent = root_node
return root_node
except ET.ParseError:
return None
def _flatten_xml(self, element: ET.Element, nodes_list: List[SpatialNode]):
"""Recursively traverses the XML and creates a flat list of SpatialNodes."""
attrib = element.attrib
bounds_str = attrib.get("bounds", "")
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds_str)
if match:
left, top, right, bottom = map(int, match.groups())
# Filter zero-area nodes early
if right > left and bottom > top:
self._node_counter += 1
node = SpatialNode(
node_id=f"n_{self._node_counter}",
class_name=attrib.get("class", ""),
text=attrib.get("text", "").strip(),
content_desc=attrib.get("content-desc", "").strip(),
resource_id=attrib.get("resource-id", "").strip(),
bounds=(left, top, right, bottom),
clickable=attrib.get("clickable", "false") == "true",
scrollable=attrib.get("scrollable", "false") == "true",
)
nodes_list.append(node)
for child in element:
self._flatten_xml(child, nodes_list)
def get_all_nodes(self, root: SpatialNode) -> List[SpatialNode]:
"""Flattens the Spatial Tree into a list for easy filtering."""
result = [root]
for child in root.children:
result.extend(self.get_all_nodes(child))
return result
def get_clickable_nodes(self, root: SpatialNode) -> List[SpatialNode]:
"""Returns all nodes that are clickable or have strong semantic meaning."""
all_nodes = self.get_all_nodes(root)
clickables = []
for n in all_nodes:
has_semantic = bool(n.text or n.content_desc)
semantic_res = n.resource_id and any(
x in n.resource_id.lower() for x in ["button", "tab", "icon", "action", "menu"]
)
if n.clickable or n.scrollable or semantic_res or (has_semantic and n.area < 500000 and n.area > 0):
# Filter out pure massive containers (like whole screen) if they aren't explicitly clickable
if not n.clickable and not n.scrollable and n.area > 2000000:
continue
# Also exclude if it's just a ViewGroup with a description but no action
if not n.clickable and n.class_name == "android.view.ViewGroup":
continue
clickables.append(n)
return clickables

494
GramAddict/core/qdrant_memory.py
View File

File diff suppressed because it is too large Load Diff

17
GramAddict/core/screen_topology.py
View File

@@ -8,8 +8,8 @@ This is the bot's GPS: it knows HOW to get from screen A to screen B
before the bot starts moving. The GOAP planner consults this map
as its primary routing strategy.
"""
from collections import deque
from enum import Enum
from typing import Dict, List, Optional, Tuple
from GramAddict.core.goap import ScreenType
@@ -38,6 +38,7 @@ class ScreenTopology:
"tap home tab": ScreenType.HOME_FEED,
"tap profile tab": ScreenType.OWN_PROFILE,
"tap reels tab": ScreenType.REELS_FEED,
"view a post": ScreenType.POST_DETAIL,
},
ScreenType.REELS_FEED: {
"tap home tab": ScreenType.HOME_FEED,
@@ -78,12 +79,16 @@ class ScreenTopology:
"open messages": ScreenType.DM_INBOX,
"open following list": ScreenType.FOLLOW_LIST,
"open followers list": ScreenType.FOLLOW_LIST,
"view a post": ScreenType.POST_DETAIL,
"open post": ScreenType.POST_DETAIL,
"open post author profile": ScreenType.OTHER_PROFILE,
"view the user profile": ScreenType.OTHER_PROFILE,
"view user profile": ScreenType.OTHER_PROFILE,
"open user profile": ScreenType.OTHER_PROFILE,
}
@classmethod
def find_route(
cls, from_screen: ScreenType, to_screen: ScreenType
) -> Optional[List[Tuple[str, ScreenType]]]:
def find_route(cls, from_screen: ScreenType, to_screen: ScreenType) -> Optional[List[Tuple[str, ScreenType]]]:
"""
BFS shortest path from from_screen to to_screen.
@@ -171,9 +176,7 @@ class ScreenTopology:
return f"navigate to {screen_name}"
@classmethod
def expected_screen_for_action(
cls, action: str, from_screen: ScreenType
) -> Optional[ScreenType]:
def expected_screen_for_action(cls, action: str, from_screen: ScreenType) -> Optional[ScreenType]:
"""What screen should we land on after this action from this screen?
Used by _execute_action to validate INTERMEDIATE navigation steps.

85
GramAddict/core/session_state.py
View File

@@ -80,64 +80,40 @@ class SessionState:
self,
):
"""set the limits for current session"""
self.args.current_likes_limit = get_value(
getattr(self.args, "total_likes_limit", 300), None, 300
)
self.args.current_follow_limit = get_value(
getattr(self.args, "total_follows_limit", 50), None, 50
)
self.args.current_unfollow_limit = get_value(
getattr(self.args, "total_unfollows_limit", 50), None, 50
)
self.args.current_comments_limit = get_value(
getattr(self.args, "total_comments_limit", 10), None, 10
)
self.args.current_likes_limit = get_value(getattr(self.args, "total_likes_limit", 300), None, 300)
self.args.current_follow_limit = get_value(getattr(self.args, "total_follows_limit", 50), None, 50)
self.args.current_unfollow_limit = get_value(getattr(self.args, "total_unfollows_limit", 50), None, 50)
self.args.current_comments_limit = get_value(getattr(self.args, "total_comments_limit", 10), None, 10)
self.args.current_pm_limit = get_value(getattr(self.args, "total_pm_limit", 10), None, 10)
self.args.current_watch_limit = get_value(
getattr(self.args, "total_watches_limit", 50), None, 50
)
self.args.current_watch_limit = get_value(getattr(self.args, "total_watches_limit", 50), None, 50)
self.args.current_success_limit = get_value(
getattr(self.args, "total_successful_interactions_limit", 100), None, 100
)
self.args.current_total_limit = get_value(
getattr(self.args, "total_interactions_limit", 1000), None, 1000
)
self.args.current_scraped_limit = get_value(
getattr(self.args, "total_scraped_limit", 200), None, 200
)
self.args.current_crashes_limit = get_value(
getattr(self.args, "total_crashes_limit", 5), None, 5
)
self.args.current_total_limit = get_value(getattr(self.args, "total_interactions_limit", 1000), None, 1000)
self.args.current_scraped_limit = get_value(getattr(self.args, "total_scraped_limit", 200), None, 200)
self.args.current_crashes_limit = get_value(getattr(self.args, "total_crashes_limit", 5), None, 5)
def check_limit(self, limit_type=None, output=False):
"""Returns True if limit reached - else False"""
limit_type = SessionState.Limit.ALL if limit_type is None else limit_type
# check limits
total_likes = self.totalLikes >= int(self.args.current_likes_limit)
total_followed = sum(self.totalFollowed.values()) >= int(
self.args.current_follow_limit
)
total_followed = sum(self.totalFollowed.values()) >= int(self.args.current_follow_limit)
total_unfollowed = self.totalUnfollowed >= int(self.args.current_unfollow_limit)
total_comments = self.totalComments >= int(self.args.current_comments_limit)
total_pm = self.totalPm >= int(self.args.current_pm_limit)
total_watched = self.totalWatched >= int(self.args.current_watch_limit)
total_successful = sum(self.successfulInteractions.values()) >= int(
self.args.current_success_limit
)
total_interactions = sum(self.totalInteractions.values()) >= int(
self.args.current_total_limit
)
total_successful = sum(self.successfulInteractions.values()) >= int(self.args.current_success_limit)
total_interactions = sum(self.totalInteractions.values()) >= int(self.args.current_total_limit)
total_scraped = sum(self.totalScraped.values()) >= int(
self.args.current_scraped_limit
)
total_scraped = sum(self.totalScraped.values()) >= int(self.args.current_scraped_limit)
total_crashes = self.totalCrashes >= int(self.args.current_crashes_limit)
session_info = [
"Checking session limits:",
f"- Total Likes:\t\t\t\t{'Limit Reached' if total_likes else 'OK'} ({self.totalLikes}/{self.args.current_likes_limit})",
f"- Total Comments:\t\t\t\t{'Limit Reached' if total_comments else 'OK'} ({self.totalComments}/{self.args.current_comments_limit})",
f"- Session Likes Given:\t\t{'Limit Reached' if total_likes else 'OK'} ({self.totalLikes}/{self.args.current_likes_limit})",
f"- Session Comments Given:\t{'Limit Reached' if total_comments else 'OK'} ({self.totalComments}/{self.args.current_comments_limit})",
f"- Total PM:\t\t\t\t\t{'Limit Reached' if total_pm else 'OK'} ({self.totalPm}/{self.args.current_pm_limit})",
f"- Total Followed:\t\t\t\t{'Limit Reached' if total_followed else 'OK'} ({sum(self.totalFollowed.values())}/{self.args.current_follow_limit})",
f"- Total Unfollowed:\t\t\t\t{'Limit Reached' if total_unfollowed else 'OK'} ({self.totalUnfollowed}/{self.args.current_unfollow_limit})",
@@ -154,11 +130,16 @@ class SessionState:
logger.info(line)
return (
total_likes and getattr(self.args, "end_if_likes_limit_reached", False)
or total_followed and getattr(self.args, "end_if_follows_limit_reached", False)
or total_watched and getattr(self.args, "end_if_watches_limit_reached", False)
or total_comments and getattr(self.args, "end_if_comments_limit_reached", False)
or total_pm and getattr(self.args, "end_if_pm_limit_reached", False),
total_likes
and getattr(self.args, "end_if_likes_limit_reached", False)
or total_followed
and getattr(self.args, "end_if_follows_limit_reached", False)
or total_watched
and getattr(self.args, "end_if_watches_limit_reached", False)
or total_comments
and getattr(self.args, "end_if_comments_limit_reached", False)
or total_pm
and getattr(self.args, "end_if_pm_limit_reached", False),
total_unfollowed,
total_interactions or total_successful or total_scraped,
)
@@ -247,20 +228,20 @@ class SessionState:
delta = timedelta(seconds=delta_sec)
if not working_hours:
return True, 0
for n in working_hours:
today = current_time.strftime("%Y-%m-%d")
# 100% Autonomous: Hybrid Time Format Support (Legacy . vs Modern :)
h_start = n.split('-')[0].replace(":", ".")
h_end = n.split('-')[1].replace(":", ".")
h_start = n.split("-")[0].replace(":", ".")
h_end = n.split("-")[1].replace(":", ".")
inf_value = f"{h_start} {today}"
inf = datetime.strptime(inf_value, "%H.%M %Y-%m-%d") + delta
sup_value = f"{h_end} {today}"
sup = datetime.strptime(sup_value, "%H.%M %Y-%m-%d") + delta
if sup - inf + timedelta(minutes=1) == timedelta(
days=1
) or sup - inf + timedelta(minutes=1) == timedelta(days=0):
if sup - inf + timedelta(minutes=1) == timedelta(days=1) or sup - inf + timedelta(minutes=1) == timedelta(
days=0
):
logger.debug("Whole day mode.")
return True, 0
if time_in_range(inf.time(), sup.time(), current_time.time()):
@@ -300,9 +281,7 @@ class SessionStateEncoder(JSONEncoder):
return {
"id": session_state.id,
"total_interactions": sum(session_state.totalInteractions.values()),
"successful_interactions": sum(
session_state.successfulInteractions.values()
),
"successful_interactions": sum(session_state.successfulInteractions.values()),
"total_followed": sum(session_state.totalFollowed.values()),
"total_likes": session_state.totalLikes,
"total_comments": session_state.totalComments,

276
GramAddict/core/situational_awareness.py
View File

@@ -10,13 +10,13 @@ After initial learning, 95%+ of situations are handled from memory
alone with ZERO LLM calls. This is "Tesla fleet learning" for bots.
"""
import logging
import hashlib
import time
import logging
import re
import time
import xml.etree.ElementTree as ET
from typing import Optional, Dict, Any
from enum import Enum
from typing import Dict, Optional
from GramAddict.core.utils import random_sleep
@@ -34,6 +34,7 @@ class SituationType(Enum):
class EscapeAction:
"""Represents a planned escape action."""
def __init__(self, action_type: str, x: int = 0, y: int = 0, reason: str = "", resource_id: str = ""):
self.action_type = action_type # 'click', 'back', 'app_start', 'home_then_app'
self.x = x
@@ -42,11 +43,19 @@ class EscapeAction:
self.resource_id = resource_id
def to_dict(self) -> dict:
return {"action_type": self.action_type, "x": self.x, "y": self.y, "reason": self.reason, "resource_id": self.resource_id}
return {
"action_type": self.action_type,
"x": self.x,
"y": self.y,
"reason": self.reason,
"resource_id": self.resource_id,
}
@classmethod
def from_dict(cls, d: dict) -> "EscapeAction":
return cls(d.get("action_type", "back"), d.get("x", 0), d.get("y", 0), d.get("reason", ""), d.get("resource_id", ""))
return cls(
d.get("action_type", "back"), d.get("x", 0), d.get("y", 0), d.get("reason", ""), d.get("resource_id", "")
)
class SituationEpisodeDB:
@@ -56,8 +65,10 @@ class SituationEpisodeDB:
Enables instant recall for known situations (0 LLM calls).
Stores BOTH positive and negative episodes for full learning.
"""
def __init__(self):
from GramAddict.core.qdrant_memory import QdrantBase
self._db = QdrantBase("sae_episodes_v1", vector_size=768)
def recall(self, situation_signature: str) -> Optional[Dict]:
@@ -126,9 +137,31 @@ class SituationEpisodeDB:
if not vec:
return
# Unique key: situation + action type + success flag
seed = f"{situation_signature}|{action.action_type}|{action.x},{action.y}|{success}"
confidence = 0.8 if success else 0.0
# Unique key: situation + action type (ignoring success flag for the seed so we update the same entry)
seed = f"{situation_signature}|{action.action_type}|{action.x},{action.y}"
point_id = self._db.generate_uuid(seed)
current_conf = 0.0
has_existing = False
try:
points = self._db.client.retrieve(
collection_name=self._db.collection_name, ids=[point_id], with_payload=True, with_vectors=False
)
if points:
has_existing = True
current_conf = points[0].payload.get("confidence", 0.0)
except Exception:
pass
if success:
confidence = min(1.0, current_conf + 0.5) if has_existing else 0.8
else:
confidence = current_conf - 0.5 if has_existing else -0.5
if confidence < 0.1 and not success:
self._db.client.delete(collection_name=self._db.collection_name, points_selector=[point_id])
logger.info("🗑️ [SAE Learn] Action decayed below threshold. Deleted from memory.")
return
payload = {
"situation": situation_signature[:500],
@@ -141,8 +174,10 @@ class SituationEpisodeDB:
outcome = "✅ SUCCESS" if success else "❌ FAILURE"
self._db.upsert_point(
seed, payload, vector=vec,
log_success=f"🧠 [SAE Learn] {outcome}: '{action.reason}' → Stored for future recall"
seed,
payload,
vector=vec,
log_success=f"🧠 [SAE Learn] {outcome}: '{action.reason}' → Stored for future recall",
)
def boost(self, situation_signature: str, action: EscapeAction):
@@ -193,7 +228,7 @@ class SituationalAwarenessEngine:
try:
# Remove XML declaration
clean = re.sub(r'<\?xml.*?\?>', '', xml_dump).strip()
clean = re.sub(r"<\?xml.*?\?>", "", xml_dump).strip()
root = ET.fromstring(clean)
except Exception:
# If XML is broken, extract what we can with regex
@@ -205,17 +240,17 @@ class SituationalAwarenessEngine:
packages = set()
elements = []
for elem in root.iter('node'):
for elem in root.iter("node"):
a = elem.attrib
pkg = a.get('package', '')
pkg = a.get("package", "")
if pkg:
packages.add(pkg)
rid = a.get('resource-id', '').strip()
text = a.get('text', '').strip()
desc = a.get('content-desc', '').strip()
bounds = a.get('bounds', '')
clickable = a.get('clickable', 'false')
rid = a.get("resource-id", "").strip()
text = a.get("text", "").strip()
desc = a.get("content-desc", "").strip()
bounds = a.get("bounds", "")
clickable = a.get("clickable", "false")
# Only keep nodes with meaningful content
if not rid and not text and not desc:
@@ -225,10 +260,14 @@ class SituationalAwarenessEngine:
if rid:
parts.append(f"id={rid.split('/')[-1]}")
if text:
parts.append(f"text='{text[:60]}'")
if len(text) > 20:
text = text[:10] + "..." + text[-10:]
parts.append(f"text='{text}'")
if desc:
parts.append(f"desc='{desc[:60]}'")
if clickable == 'true':
if len(desc) > 20:
desc = desc[:10] + "..." + desc[-10:]
parts.append(f"desc='{desc}'")
if clickable == "true":
parts.append("CLICKABLE")
if bounds:
parts.append(f"bounds={bounds}")
@@ -236,14 +275,14 @@ class SituationalAwarenessEngine:
elements.append(" | ".join(parts))
sig = f"PACKAGES: {', '.join(sorted(packages))}\n"
sig += "\n".join(elements[:50]) # Cap at 50 elements
sig += "\n".join(elements[-50:]) # Keep the last 50 elements (highest Z-index/foreground)
return sig[:3000]
def _compute_situation_hash(self, compressed: str) -> str:
"""Deterministic hash for situation dedup."""
# Remove volatile parts (timestamps, counters) but keep structural identity
stable = re.sub(r'\d{2}:\d{2}', 'HH:MM', compressed)
stable = re.sub(r'Battery \d+ per cent', 'Battery NN per cent', stable)
stable = re.sub(r"\d{2}:\d{2}", "HH:MM", compressed)
stable = re.sub(r"Battery \d+ per cent", "Battery NN per cent", stable)
return hashlib.sha256(stable.encode()).hexdigest()[:32]
def perceive(self, xml_dump: str) -> SituationType:
@@ -257,12 +296,17 @@ class SituationalAwarenessEngine:
xml_lower = xml_dump.lower()
blocked_markers = [
"try again later", "action blocked", "restrict certain activity",
"help us confirm you own", "confirm it's you",
"später erneut versuchen", "bestätige, dass du es bist",
"handlung blockiert", "eingeschränkt",
"try again later",
"action blocked",
"restrict certain activity",
"help us confirm you own",
"confirm it's you",
"später erneut versuchen",
"bestätige, dass du es bist",
"handlung blockiert",
"eingeschränkt",
]
# Guard: Check if the text matches are relatively isolated (e.g. short strings).
# If the string is buried inside a 200-character caption, it's a false positive.
# We can regex match text="..." attributes that are less than 60 characters total,
@@ -274,18 +318,18 @@ class SituationalAwarenessEngine:
return SituationType.DANGER_ACTION_BLOCKED
# ── Hardware Guard: Screen Off / Locked ──
if not getattr(self.device.deviceV2, 'info', {}).get("screenOn", True):
if not getattr(self.device.deviceV2, "info", {}).get("screenOn", True):
logger.info("📱 [SAE Perceive] Screen is physically OFF.")
return SituationType.OBSTACLE_LOCKED_SCREEN
# ── System Dialog / Permission Detect (Fast Path) ──
packages = set(re.findall(r'package=["\']([^"\']+)["\']', xml_dump))
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
app_id = getattr(self.device, "app_id", "com.instagram.android")
system_dialog_pkgs = {
'com.google.android.permissioncontroller',
'com.android.permissioncontroller',
'com.samsung.android.permissioncontroller'
"com.google.android.permissioncontroller",
"com.android.permissioncontroller",
"com.samsung.android.permissioncontroller",
}
if any(pkg in system_dialog_pkgs for pkg in packages):
logger.info("📱 [SAE Perceive] System permission dialog explicitly detected.")
@@ -294,7 +338,7 @@ class SituationalAwarenessEngine:
# ── Foreign Environment Detection (package-based) ──
# If the main app package is completely absent from the UI hierarchy,
# OR if there's a dominant foreign package and no app package, we might have lost the app.
# If our app is on screen, we trust we are in the app (even if a custom keyboard is open).
# We only trigger foreign app classification if our app is completely missing from the screen.
is_foreign = False
@@ -305,11 +349,11 @@ class SituationalAwarenessEngine:
# We explicitly ask the TelepathicEngine to classify this to avoid writing brittle substring hacks
# for Android System UI variations across different device manufacturers.
try:
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.config import Config
screen_off = not getattr(self.device.deviceV2, 'info', {}).get("screenOn", True)
from GramAddict.core.llm_provider import query_telepathic_llm
screen_off = not getattr(self.device.deviceV2, "info", {}).get("screenOn", True)
prompt = (
"You are a Situation Classifier for a mobile automation agent.\n"
"Analyze the given Android UI XML dump. Is this a physical DEVICE_LOCK_SCREEN, "
@@ -319,18 +363,27 @@ class SituationalAwarenessEngine:
"{\"situation\": \"OBSTACLE_LOCKED_SCREEN\" | \"OBSTACLE_SYSTEM\" | \"OBSTACLE_FOREIGN_APP\"}\n\n"
f"XML:\n{self._compress_xml(xml_dump)[:2500]}"
)
args = {}
try: args = Config().args
except Exception: pass
try:
args = Config().args
except Exception:
pass
model = getattr(args, "ai_telepathic_model", "qwen3.5:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
res = query_telepathic_llm(model=model, url=url, system_prompt="Strict JSON classifier.", user_prompt=prompt, use_local_edge=True)
res = query_telepathic_llm(
model=model,
url=url,
system_prompt="Strict JSON classifier.",
user_prompt=prompt,
use_local_edge=True,
)
import json
data = json.loads(res)
situ_str = data.get("situation", "")
if situ_str == "OBSTACLE_LOCKED_SCREEN":
logger.info("🧠 [Smart Perceive] SystemUI definitively classified as: LOCKED_SCREEN.")
return SituationType.OBSTACLE_LOCKED_SCREEN
@@ -348,8 +401,9 @@ class SituationalAwarenessEngine:
# We explicitly query ScreenMemoryDB. If unknown, we ask the LLM.
# This replaces ALL brittle string/ID matching for modals.
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
compressed = self._compress_xml(xml_dump)
# ── Structural Fast-Check: Content-Creation Overlays ──
@@ -358,21 +412,21 @@ class SituationalAwarenessEngine:
# and frequently fool the LLM into thinking they are "normal" browsing.
# Detecting them structurally is O(1) and requires ZERO LLM calls.
creation_flow_markers = (
'quick_capture', # Camera / story capture overlay
'gallery_cancel_button', # Story gallery "Back to Home" button
'creation_flow', # Post creation wizard
'reel_camera', # Reel recording interface
"quick_capture", # Camera / story capture overlay
"gallery_cancel_button", # Story gallery "Back to Home" button
"creation_flow", # Post creation wizard
"reel_camera", # Reel recording interface
)
# Guard: Check against compressed string to ensure these markers ONLY appear
# as resource IDs (e.g. "id=quick_capture_...") and not as plain text in
# Guard: Check against compressed string to ensure these markers ONLY appear
# as resource IDs (e.g. "id=quick_capture_...") and not as plain text in
# user comments/bios (which would look like "text='... creation_flow ...'")
if any(re.search(rf'id=[^\s|]*{marker}', compressed, re.IGNORECASE) for marker in creation_flow_markers):
if any(re.search(rf"id=[^\s|]*{marker}", compressed, re.IGNORECASE) for marker in creation_flow_markers):
logger.info("🧠 [SAE Perceive] Content-creation overlay detected structurally → OBSTACLE_MODAL")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
return SituationType.OBSTACLE_MODAL
cached_type = screen_memory.get_screen_type(compressed)
if cached_type:
if cached_type == "OBSTACLE_MODAL":
return SituationType.OBSTACLE_MODAL
@@ -381,9 +435,9 @@ class SituationalAwarenessEngine:
# If not cached, query LLM for autonomous structural classification
try:
from GramAddict.core.llm_provider import query_telepathic_llm
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_telepathic_llm
prompt = (
"You are a Situation Classifier for a mobile automation agent.\n"
"Analyze the given Android UI XML dump. Is there a blocking MODAL, DIALOG, or POPUP "
@@ -394,21 +448,26 @@ class SituationalAwarenessEngine:
"or ANY content-creation flow (reel recording, post editor, live mode) is an OBSTACLE_MODAL — "
"it blocks normal navigation and must be dismissed.\n"
"Respond ONLY with a valid JSON object strictly matching this schema: "
"{\"situation\": \"OBSTACLE_MODAL\" | \"NORMAL\"}\n\n"
'{"situation": "OBSTACLE_MODAL" | "NORMAL"}\n\n'
f"XML:\n{compressed[:2500]}"
)
args = {}
try: args = Config().args
except Exception: pass
try:
args = Config().args
except Exception:
pass
model = getattr(args, "ai_telepathic_model", "qwen3.5:latest")
url = getattr(args, "ai_telepathic_url", "http://localhost:11434/api/generate")
res = query_telepathic_llm(model=model, url=url, system_prompt="Strict JSON classifier.", user_prompt=prompt, use_local_edge=True)
res = query_telepathic_llm(
model=model, url=url, system_prompt="Strict JSON classifier.", user_prompt=prompt, use_local_edge=True
)
import json
data = json.loads(res)
situ_str = data.get("situation", "NORMAL")
if situ_str == "OBSTACLE_MODAL":
logger.info("🧠 [Smart Perceive] Screen classified as: OBSTACLE_MODAL.")
screen_memory.store_screen(compressed, "OBSTACLE_MODAL")
@@ -422,19 +481,28 @@ class SituationalAwarenessEngine:
return SituationType.NORMAL
def unlearn_current_state(self, xml_dump: str):
"""Purges the current screen's signature from Qdrant to self-heal from hallucinations."""
compressed = self._compress_xml(xml_dump)
from GramAddict.core.qdrant_memory import ScreenMemoryDB
screen_memory = ScreenMemoryDB()
screen_memory.purge_screen(compressed)
logger.info("🗑️ [Smart Perceive] Purged cached screen signature to force autonomous re-evaluation.")
# ──────────────────────────────────────────────
# 2. PLAN: AI-driven escape strategy
# ──────────────────────────────────────────────
def _plan_escape_via_llm(self, xml_dump: str, compressed: str, situation_type: SituationType, failed_actions: set = None) -> Optional[EscapeAction]:
def _plan_escape_via_llm(
self, xml_dump: str, compressed: str, situation_type: SituationType, failed_actions: set = None
) -> Optional[EscapeAction]:
"""
LLM-powered escape planning for situations where structural scan fails.
Called ONLY when recall AND structural planning both miss.
"""
from GramAddict.core.llm_provider import query_llm
from GramAddict.core.config import Config
from GramAddict.core.llm_provider import query_llm
try:
args = Config().args
@@ -455,29 +523,35 @@ class SituationalAwarenessEngine:
"- If there is NO obstacle and the screen is a normal Instagram view (false positive), action must be 'false_positive'\n"
"- If nothing else works, suggest 'app_start' to force-reopen Instagram\n"
"- NEVER click 'OK'/'Confirm'/'Accept' on surveys or prompts\n"
"- Return ONLY valid JSON: {\"action\": \"click\"|\"back\"|\"app_start\"|\"unlock\"|\"kill_foreign_apps\"|\"false_positive\", \"x\": N, \"y\": N, \"reason\": \"...\"}"
'- Return ONLY valid JSON: {"action": "click"|"back"|"app_start"|"unlock"|"kill_foreign_apps"|"false_positive", "x": N, "y": N, "reason": "..."}'
)
user_prompt = (
f"Situation type: {situation_type.value}\n\n"
f"Screen content:\n{compressed}\n\n"
)
user_prompt = f"Situation type: {situation_type.value}\n\n" f"Screen content:\n{compressed}\n\n"
if failed_actions:
user_prompt += f"Failed actions this session (DO NOT REPEAT): {list(failed_actions)}\n\n"
user_prompt += "What action should I take to clear this obstacle and return to Instagram? Return JSON only."
try:
resp = query_llm(url=url, model=model, prompt=user_prompt, system=system_prompt,
format_json=True, timeout=30, max_tokens=300, temperature=0.0)
resp = query_llm(
url=url,
model=model,
prompt=user_prompt,
system=system_prompt,
format_json=True,
timeout=30,
max_tokens=300,
temperature=0.0,
)
if resp and "response" in resp:
import json
data = json.loads(resp["response"])
return EscapeAction(
action_type=data.get("action", "back"),
x=int(data.get("x", 0)),
y=int(data.get("y", 0)),
reason=data.get("reason", "LLM-planned escape")
reason=data.get("reason", "LLM-planned escape"),
)
except Exception as e:
logger.warning(f"🧠 [SAE] LLM escape planning failed: {e}")
@@ -504,24 +578,24 @@ class SituationalAwarenessEngine:
logger.info(f"🔓 [SAE Act] Unlocking device: {action.reason}")
self.device.unlock()
random_sleep(1.0, 2.0)
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
app_id = getattr(self.device, "app_id", "com.instagram.android")
self.device.app_start(app_id, use_monkey=True)
random_sleep(1.5, 2.5)
elif action.action_type == "app_start":
logger.info(f"🚀 [SAE Act] Force-starting app: {action.reason}")
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
app_id = getattr(self.device, "app_id", "com.instagram.android")
self.device.app_start(app_id, use_monkey=True)
random_sleep(2.0, 3.5)
elif action.action_type == "kill_foreign_apps":
logger.info(f"🔪 [SAE Act] Killing foreign apps: {action.reason}")
# The reason string will contain the package name or 'all'
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
app_id = getattr(self.device, "app_id", "com.instagram.android")
try:
# We can dump current package again, or just get it from device
current_pkg = self.device.deviceV2.app_current().get("package")
if current_pkg and current_pkg != app_id and current_pkg not in ('com.android.systemui', 'android'):
if current_pkg and current_pkg != app_id and current_pkg not in ("com.android.systemui", "android"):
logger.info(f"🔪 Stopping {current_pkg}")
self.device.app_stop(current_pkg)
random_sleep(1.0, 2.0)
@@ -535,7 +609,7 @@ class SituationalAwarenessEngine:
logger.info(f"🏠 [SAE Act] HOME → App Start: {action.reason}")
self.device.press("home")
random_sleep(0.5, 1.0)
app_id = getattr(self.device, 'app_id', 'com.instagram.android')
app_id = getattr(self.device, "app_id", "com.instagram.android")
self.device.app_start(app_id, use_monkey=True)
random_sleep(2.0, 3.5)
@@ -550,9 +624,10 @@ class SituationalAwarenessEngine:
Returns True if an obstacle was successfully cleared, False if already clear or failed.
"""
from GramAddict.core.exceptions import ActionBlockedError
failed_this_session = set()
cleared_something = False
last_situation = None
situation_attempts = 0
@@ -562,9 +637,9 @@ class SituationalAwarenessEngine:
xml_dump = initial_xml
else:
xml_dump = self.device.dump_hierarchy()
situation = self.perceive(xml_dump)
if last_situation != situation:
situation_attempts = 0
last_situation = situation
@@ -582,13 +657,11 @@ class SituationalAwarenessEngine:
logger.error("🚫 [SAE CRITICAL] Instagram Action Block detected! Halting to protect account.")
raise ActionBlockedError("Instagram action block detected by SAE.")
logger.warning(
f"🔍 [SAE] Obstacle detected: {situation.value} (attempt {attempt + 1}/{max_attempts})"
)
logger.warning(f"🔍 [SAE] Obstacle detected: {situation.value} (attempt {attempt + 1}/{max_attempts})")
# ── COMPRESS for memory lookup ──
compressed = self._compress_xml(xml_dump)
# ── RECALL from memory ──
recalled = self.episodes.recall(compressed)
if recalled:
@@ -597,7 +670,7 @@ class SituationalAwarenessEngine:
action = EscapeAction.from_dict(recalled)
logger.info(f"🧠 [SAE] Using recalled strategy: {action.reason}")
else:
logger.info(f"🧠 [SAE] Recalled strategy already failed this session. Using LLM planning.")
logger.info("🧠 [SAE] Recalled strategy already failed this session. Using LLM planning.")
recalled = None
if not recalled:
@@ -605,14 +678,23 @@ class SituationalAwarenessEngine:
logger.info("🧠 [SAE] Autonomous Blank Start: Escalating to LLM-assisted escape planning...")
action = self._plan_escape_via_llm(xml_dump, compressed, situation, failed_this_session)
elif situation_attempts == 3:
action = EscapeAction("app_start", reason=f"Escalation level 4: force app restart after {situation_attempts} failed attempts on this situation")
action = EscapeAction(
"app_start",
reason=f"Escalation level 4: force app restart after {situation_attempts} failed attempts on this situation",
)
else:
action = EscapeAction("home_then_app", reason=f"Nuclear escalation: HOME + app_start after {situation_attempts} failed attempts on this situation")
action = EscapeAction(
"home_then_app",
reason=f"Nuclear escalation: HOME + app_start after {situation_attempts} failed attempts on this situation",
)
# ── EXECUTE ──
if action.action_type == "false_positive":
logger.warning(f"🧠 [SAE Unlearn] LLM identified false positive obstacle. Overwriting Qdrant memory to NORMAL.")
logger.warning(
"🧠 [SAE Unlearn] LLM identified false positive obstacle. Overwriting Qdrant memory to NORMAL."
)
from GramAddict.core.qdrant_memory import ScreenMemoryDB
ScreenMemoryDB().store_screen(compressed, "NORMAL")
self._consecutive_failures = 0
return True
@@ -623,8 +705,8 @@ class SituationalAwarenessEngine:
# ── VERIFY ──
post_xml = self.device.dump_hierarchy()
post_situation = self.perceive(post_xml)
reached_normal = (post_situation == SituationType.NORMAL)
situation_changed = (post_situation != situation)
reached_normal = post_situation == SituationType.NORMAL
situation_changed = post_situation != situation
if reached_normal:
# ── LEARN FULL SUCCESS ──
@@ -635,7 +717,9 @@ class SituationalAwarenessEngine:
elif situation_changed:
# ── LEARN PARTIAL SUCCESS ──
self.episodes.learn(compressed, action, True)
logger.info(f"🔄 [SAE] Situation changed from {situation.value} to {post_situation.value}. Continuing recovery...")
logger.info(
f"🔄 [SAE] Situation changed from {situation.value} to {post_situation.value}. Continuing recovery..."
)
# We do not increment consecutive_failures or situation_attempts because we made progress
# The next loop iteration will clear failed_this_session since last_situation != situation
else:

2307
GramAddict/core/telepathic_engine.py
View File

File diff suppressed because it is too large Load Diff