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

feat: implement brain-driven dynamic decision making to prevent goap traps

Browse Source
This commit is contained in:
Marc Mintel
2026-04-27 22:28:53 +02:00
parent 93175b7caf
commit da804b174a
4 changed files with 132 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
GramAddict/core/goap.py
View File

@@ -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

59
GramAddict/core/navigation/brain.py Normal file
View File

@@ -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

25
GramAddict/core/navigation/planner.py
View File

@@ -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)

33
tests/test_planner_dynamic_brain.py Normal file
View File

@@ -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"