diff --git a/GramAddict/core/goap.py b/GramAddict/core/goap.py index de4abed..4e09881 100644 --- a/GramAddict/core/goap.py +++ b/GramAddict/core/goap.py @@ -18,15 +18,13 @@ import time from typing import Any, Dict, List from GramAddict.core.utils import random_sleep - -logger = logging.getLogger(__name__) - - 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 +logger = logging.getLogger(__name__) + # Re-export for backward compatibility (optional but helps minimize import breakage) __all__ = ["GoalExecutor", "ScreenIdentity", "ScreenType", "PathMemory", "NavigationKnowledge", "GoalPlanner"] @@ -173,7 +171,9 @@ class GoalExecutor: continue # PLAN - action = self.planner.plan_next_step(goal, screen, explored_nav_actions=explored_nav_actions, action_failures=self.action_failures) + action = self.planner.plan_next_step( + goal, screen, explored_nav_actions=explored_nav_actions, action_failures=self.action_failures + ) if action is None: # Goal achieved! @@ -199,6 +199,21 @@ class GoalExecutor: # Reset failures for this action since it eventually succeeded self.action_failures[action] = 0 + if "scroll" in action.lower(): + logger.debug( + "📍 [GOAP State] Scrolled successfully. Clearing explored actions to allow retrying off-screen elements." + ) + explored_nav_actions.clear() + # Keep action_failures for synthetic intents, but clear them for structural actions + # so that the HD Map can retry route actions that might now be visible! + from GramAddict.core.screen_topology import ScreenTopology + + keys_to_clear = [ + k for k in self.action_failures.keys() if ScreenTopology.is_structural_action(screen_type, k) + ] + for k in keys_to_clear: + del self.action_failures[k] + # ── Back-Press Circuit Breaker ── if action == "press back": consecutive_back_presses += 1 diff --git a/GramAddict/core/navigation/brain.py b/GramAddict/core/navigation/brain.py new file mode 100644 index 0000000..27f16d7 --- /dev/null +++ b/GramAddict/core/navigation/brain.py @@ -0,0 +1,59 @@ +import logging +from typing import List, Optional + +from GramAddict.core.config import Config +from GramAddict.core.llm_provider import query_llm + +logger = logging.getLogger(__name__) + + +def ask_brain_for_action( + goal: str, screen_type: str, available_actions: List[str], explored_actions: set, context: dict = None +) -> Optional[str]: + """Asks the VLM to decide the best available action to reach the goal, considering failures.""" + if not available_actions: + return None + + 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" + + prompt = ( + f"You are an autonomous Instagram agent. Your ultimate goal is: '{goal}'.\n" + f"You are currently on the screen: {screen_type}.\n" + f"These actions are available to you right now: {available_actions}\n" + ) + if explored_actions: + prompt += f"You recently tried these actions but they failed or didn't help: {list(explored_actions)}\n" + if context: + prompt += f"Context: {context}\n" + + prompt += ( + "Based on this, which of the available actions should you take next to make progress towards your goal? " + "If you need to find an element that is likely off-screen, choosing to scroll is a smart move. " + "If you are stuck, choosing to go back or to a main tab is acceptable to reset. " + "Reply ONLY with the exact string from the available actions list." + ) + + try: + response = query_llm( + url=url, model=model, prompt="Choose the next best action.", system=prompt, format_json=False + ) + if response: + result = response if isinstance(response, str) else response.get("response", "") + result = result.strip().strip("'\"") + + # Fuzzy match to available actions just in case + for act in available_actions: + if act.lower() in result.lower(): + return act + + logger.warning(f"🧠 [Brain] LLM returned an invalid action: '{result}'. Falling back.") + except Exception as e: + logger.debug(f"🧠 [Brain] Error querying LLM: {e}") + + return None diff --git a/GramAddict/core/navigation/planner.py b/GramAddict/core/navigation/planner.py index cc8bdd5..25e4f93 100644 --- a/GramAddict/core/navigation/planner.py +++ b/GramAddict/core/navigation/planner.py @@ -17,7 +17,9 @@ class GoalPlanner: 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, action_failures: dict = None) -> Optional[str]: + def plan_next_step( + self, goal: str, screen: Dict[str, Any], explored_nav_actions: set = None, action_failures: dict = None + ) -> Optional[str]: """Plans the NEXT single action to take toward the goal.""" screen_type = screen["screen_type"] available = screen.get("available_actions", []) @@ -34,7 +36,9 @@ class GoalPlanner: # ── 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, action_failures) + nav_action = self._plan_navigation( + goal_lower, screen_type, available, selected_tab, explored_nav_actions, action_failures + ) if nav_action: return nav_action @@ -89,7 +93,7 @@ class GoalPlanner: else: logger.debug(f"🛡️ [Aversive Filter] Masking trapped action: '{action}'") available = safe_available - + # Build avoid_actions for HD Map route planning avoid_actions = (explored_nav_actions or set()).copy() if action_failures: @@ -112,9 +116,20 @@ class GoalPlanner: ) return next_action else: - logger.warning(f"🛡️ [HD Map] Route action '{next_action}' is trapped. Falling back.") + logger.warning(f"🛡️ [HD Map] Route action '{next_action}' is trapped. Falling back to brain.") else: - logger.debug(f"🛡️ [HD Map] Route action '{next_action}' already explored. Falling back.") + logger.debug( + f"🛡️ [HD Map] Route action '{next_action}' already explored and failed. Falling back to brain." + ) + + # ── 1.5 Brain-Driven Decision Making (Preventing Trapped States) ── + # If HD Map didn't yield a valid action, or we're stuck, ask the AI brain. + from GramAddict.core.navigation.brain import ask_brain_for_action + + brain_action = ask_brain_for_action(goal, screen_type.name, available, explored_nav_actions) + if brain_action: + logger.info(f"🧠 [Brain] Decided dynamically to execute: '{brain_action}'") + return brain_action # ── 2. Learned Knowledge (Qdrant) ── required_screens = self.knowledge.get_requirements(goal) diff --git a/tests/test_planner_dynamic_brain.py b/tests/test_planner_dynamic_brain.py new file mode 100644 index 0000000..e6b61b1 --- /dev/null +++ b/tests/test_planner_dynamic_brain.py @@ -0,0 +1,33 @@ +from unittest.mock import patch + +from GramAddict.core.navigation.planner import GoalPlanner +from GramAddict.core.perception.screen_identity import ScreenType + + +def test_planner_falls_back_to_brain_when_hd_map_fails(): + """ + Test that if HD Map routing fails because the structural target is not visible + (and thus in explored_nav_actions), the planner falls back to the Brain + to dynamically pick an action like 'scroll down'. + """ + planner = GoalPlanner("testuser") + + # Simulate a screen where the target isn't visible + screen = { + "screen_type": ScreenType.OWN_PROFILE, + "available_actions": ["press back", "scroll down", "tap profile tab"], + "context": {}, + } + + # Simulate that 'tap following list' failed previously and is in explored actions + explored = {"tap following list"} + + # The brain should realize that 'scroll down' is the best way to uncover the target + with patch("GramAddict.core.navigation.brain.ask_brain_for_action", return_value="scroll down") as mock_brain: + action = planner.plan_next_step("go to followers/following list", screen, explored_nav_actions=explored) + + # Verify the brain was queried + mock_brain.assert_called_once() + + # Verify the brain's decision is respected + assert action == "scroll down"