Upload app/agents/gesture_action_agent.py with huggingface_hub

app/agents/gesture_action_agent.py

@@ -0,0 +1,768 @@
+"""
+Gesture Action Agent
+
+Maps hand gestures to LLM actions with OpenClaw-RL inspired learning.
+
+Gesture-to-Action Mapping:
+- 2 fingers + swipe right: Query ChatGPT + Gemini
+- 2 fingers + swipe left: Query specific topic
+- 1 finger + tap: Trigger RL learning loop
+- Pinch gesture: Capture and query
+- Open palm: Pause/stop
+
+Inspired by:
+- GestLLM: LLM-powered gesture interpretation
+- GestOS: Multi-robot gesture orchestration
+- GestureGPT: Free-form gesture understanding
+- OpenClaw-RL: Personal agent optimization
+"""
+
+import numpy as np
+from typing import Dict, List, Any, Optional, Callable
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class GestureAction(Enum):
+    """Actions triggered by gestures"""
+    QUERY_MULTI_LLM = "query_multi_llm"
+    QUERY_CHATGPT = "query_chatgpt"
+    QUERY_GEMINI = "query_gemini"
+    TRIGGER_RL_LOOP = "trigger_rl_loop"
+    CAPTURE_CONTENT = "capture_content"
+    GENERATE_PROMPT = "generate_prompt"
+    PAUSE_SESSION = "pause_session"
+    RESUME_SESSION = "resume_session"
+    SAVE_PROGRESS = "save_progress"
+    SWIPE_LEFT = "swipe_left"
+    SWIPE_RIGHT = "swipe_right"
+    SWIPE_UP = "swipe_up"
+    SWIPE_DOWN = "swipe_down"
+    PINCH = "pinch"
+    CUSTOM = "custom"
+
+
+@dataclass
+class SwipeGesture:
+    """Represents a swipe gesture with direction and speed"""
+    direction: str  # "left", "right", "up", "down"
+    speed: float
+    start_x: float
+    start_y: float
+    end_x: float
+    end_y: float
+    finger_count: int
+    timestamp: datetime = field(default_factory=datetime.now)
+
+
+@dataclass
+class PinchGesture:
+    """Represents a pinch gesture"""
+    thumb_tip: List[float]
+    index_tip: List[float]
+    distance: float
+    gesture_type: str  # "zoom_in", "zoom_out", "grab"
+    timestamp: datetime = field(default_factory=datetime.now)
+
+
+@dataclass
+class GestureActionEvent:
+    """An event triggered by a gesture"""
+    action: GestureAction
+    gesture_name: str
+    parameters: Dict[str, Any]
+    confidence: float
+    timestamp: datetime = field(default_factory=datetime.now)
+    llm_responses: List[Dict] = field(default_factory=list)
+    rl_feedback: Optional[Dict] = None
+
+
+class SwipeDetector:
+    """
+    Detects swipe gestures from hand landmarks.
+    
+    Uses finger position and movement to detect:
+    - Swipe direction (left, right, up, down)
+    - Number of fingers involved
+    - Swipe speed
+    """
+    
+    def __init__(self):
+        self.position_history: List[Dict] = []
+        self.max_history = 30
+        
+        self.swipe_threshold = 0.15
+        self.min_swipe_speed = 0.005
+        self.finger_count_window = 5
+        
+        self._reset()
+    
+    def _reset(self):
+        """Reset detector state"""
+        self.position_history = []
+        self.swipe_start = None
+        self.swipe_in_progress = False
+    
+    def count_extended_fingers(self, landmarks: List[List[float]]) -> int:
+        """Count how many fingers are extended"""
+        if not landmarks or len(landmarks) < 21:
+            return 0
+        
+        finger_tips = [4, 8, 12, 16, 20]
+        finger_bases = [3, 6, 10, 14, 18]
+        
+        extended = 0
+        
+        for tip, base in zip(finger_tips, finger_bases):
+            tip_y = landmarks[tip][1]
+            base_y = landmarks[base][1]
+            
+            if tip < len(landmarks) and base < len(landmarks):
+                if tip_y < base_y:
+                    extended += 1
+        
+        return extended
+    
+    def detect_swipe(
+        self, 
+        landmarks: List[List[float]],
+        finger_count: Optional[int] = None
+    ) -> Optional[SwipeGesture]:
+        """Detect if a swipe gesture is being performed"""
+        if not landmarks or len(landmarks) < 21:
+            return None
+        
+        wrist = landmarks[0]
+        middle_finger_mcp = landmarks[9]
+        
+        current_pos = {
+            "x": middle_finger_mcp[0],
+            "y": middle_finger_mcp[1],
+            "z": middle_finger_mcp[2] if len(middle_finger_mcp) > 2 else 0,
+            "timestamp": datetime.now()
+        }
+        
+        self.position_history.append(current_pos)
+        if len(self.position_history) > self.max_history:
+            self.position_history.pop(0)
+        
+        if finger_count is None:
+            finger_count = self.count_extended_fingers(landmarks)
+        
+        if not self.swipe_start and len(self.position_history) >= 3:
+            recent = self.position_history[-3:]
+            movement_x = abs(recent[-1]["x"] - recent[0]["x"])
+            movement_y = abs(recent[-1]["y"] - recent[0]["y"])
+            
+            if movement_x > self.swipe_threshold or movement_y > self.swipe_threshold:
+                self.swipe_start = recent[0].copy()
+                self.swipe_in_progress = True
+        
+        if self.swipe_in_progress and self.swipe_start:
+            current = current_pos
+            delta_x = current["x"] - self.swipe_start["x"]
+            delta_y = current["y"] - self.swipe_start["y"]
+            
+            time_delta = (current["timestamp"] - self.swipe_start["timestamp"]).total_seconds()
+            
+            if time_delta > 0:
+                speed = np.sqrt(delta_x**2 + delta_y**2) / time_delta
+            else:
+                speed = 0
+            
+            if speed > self.min_swipe_speed and len(self.position_history) >= 10:
+                recent = self.position_history[-10:]
+                recent_movement_x = recent[-1]["x"] - recent[0]["x"]
+                recent_movement_y = recent[-1]["y"] - recent[0]["y"]
+                
+                if abs(recent_movement_x) < 0.01 and abs(recent_movement_y) < 0.01:
+                    direction = self._get_swipe_direction(
+                        self.swipe_start["x"] - current["x"],
+                        self.swipe_start["y"] - current["y"]
+                    )
+                    
+                    swipe = SwipeGesture(
+                        direction=direction,
+                        speed=speed,
+                        start_x=self.swipe_start["x"],
+                        start_y=self.swipe_start["y"],
+                        end_x=current["x"],
+                        end_y=current["y"],
+                        finger_count=finger_count
+                    )
+                    
+                    self._reset()
+                    return swipe
+        
+        if self.swipe_in_progress and len(self.position_history) > 60:
+            self._reset()
+        
+        return None
+    
+    def _get_swipe_direction(self, dx: float, dy: float) -> str:
+        """Determine swipe direction from deltas"""
+        abs_dx = abs(dx)
+        abs_dy = abs(dy)
+        
+        if abs_dx > abs_dy:
+            return "right" if dx > 0 else "left"
+        else:
+            return "down" if dy > 0 else "up"
+    
+    def detect_pinch(
+        self, 
+        landmarks: List[List[float]]
+    ) -> Optional[PinchGesture]:
+        """Detect pinch gesture (thumb + index finger)"""
+        if not landmarks or len(landmarks) < 21:
+            return None
+        
+        thumb_tip = landmarks[4]
+        index_tip = landmarks[8]
+        
+        distance = np.sqrt(
+            (thumb_tip[0] - index_tip[0])**2 +
+            (thumb_tip[1] - index_tip[1])**2 +
+            (thumb_tip[2] - index_tip[2])**2
+        )
+        
+        if distance < 0.08:
+            gesture_type = "grab"
+        elif distance < 0.12:
+            gesture_type = "zoom_in"
+        elif distance > 0.2:
+            gesture_type = "zoom_out"
+        else:
+            return None
+        
+        return PinchGesture(
+            thumb_tip=thumb_tip,
+            index_tip=index_tip,
+            distance=distance,
+            gesture_type=gesture_type
+        )
+
+
+class GestureActionMapper:
+    """
+    Maps detected gestures to LLM actions.
+    
+    Gesture-to-Action Rules:
+    - 2 fingers + swipe right: Query ChatGPT + Gemini
+    - 2 fingers + swipe left: Query previous context
+    - 1 finger + tap: Trigger RL learning loop
+    - Pinch: Capture and analyze content
+    - Open palm: Pause/stop
+    """
+    
+    def __init__(self):
+        self.action_rules: Dict[GestureAction, Dict] = {
+            GestureAction.QUERY_MULTI_LLM: {
+                "trigger": {"finger_count": 2, "swipe": "right"},
+                "llm_providers": ["chatgpt", "gemini"],
+                "prompt_template": "Analyze this learning content and explain key concepts: {content}",
+                "auto_submit": True
+            },
+            GestureAction.QUERY_CHATGPT: {
+                "trigger": {"finger_count": 2, "swipe": "left"},
+                "llm_providers": ["chatgpt"],
+                "prompt_template": "Explain this topic in simple terms: {topic}",
+                "auto_submit": True
+            },
+            GestureAction.TRIGGER_RL_LOOP: {
+                "trigger": {"finger_count": 1, "gesture": "tap"},
+                "rl_mode": True,
+                "feedback_enabled": True
+            },
+            GestureAction.CAPTURE_CONTENT: {
+                "trigger": {"gesture": "pinch"},
+                "capture_mode": True
+            },
+            GestureAction.PAUSE_SESSION: {
+                "trigger": {"gesture": "open_palm", "duration": 1.0},
+                "action": "pause"
+            },
+            GestureAction.RESUME_SESSION: {
+                "trigger": {"gesture": "open_palm", "duration": 1.0},
+                "action": "resume"
+            }
+        }
+        
+        self.custom_mappings: Dict[str, Dict] = {}
+        self.action_callbacks: Dict[GestureAction, List[Callable]] = {}
+        
+        self.swipe_detector = SwipeDetector()
+        self.gesture_buffer: List[Dict] = []
+    
+    def register_action_callback(
+        self, 
+        action: GestureAction, 
+        callback: Callable[[GestureActionEvent], None]
+    ):
+        """Register a callback for when an action is triggered"""
+        if action not in self.action_callbacks:
+            self.action_callbacks[action] = []
+        self.action_callbacks[action].append(callback)
+    
+    def add_custom_mapping(
+        self,
+        name: str,
+        gesture_pattern: Dict,
+        action: GestureAction,
+        parameters: Dict
+    ):
+        """Add a custom gesture-to-action mapping"""
+        self.custom_mappings[name] = {
+            "pattern": gesture_pattern,
+            "action": action,
+            "parameters": parameters
+        }
+    
+    def process_landmarks(
+        self, 
+        landmarks: List[List[float]],
+        context: Optional[Dict] = None
+    ) -> List[GestureActionEvent]:
+        """Process hand landmarks and return triggered actions"""
+        events = []
+        
+        if not landmarks:
+            return events
+        
+        finger_count = self.swipe_detector.count_extended_fingers(landmarks)
+        
+        swipe = self.swipe_detector.detect_swipe(landmarks, finger_count)
+        if swipe:
+            event = self._create_swipe_event(swipe, context)
+            if event:
+                events.append(event)
+        
+        pinch = self.swipe_detector.detect_pinch(landmarks)
+        if pinch:
+            event = self._create_pinch_event(pinch, context)
+            if event:
+                events.append(event)
+        
+        if finger_count >= 4:
+            event = self._create_open_palm_event(context)
+            if event:
+                events.append(event)
+        
+        self.gesture_buffer.append({
+            "finger_count": finger_count,
+            "landmarks": landmarks,
+            "timestamp": datetime.now()
+        })
+        
+        if len(self.gesture_buffer) > 100:
+            self.gesture_buffer = self.gesture_buffer[-50:]
+        
+        return events
+    
+    def _create_swipe_event(
+        self, 
+        swipe: SwipeGesture, 
+        context: Optional[Dict]
+    ) -> Optional[GestureActionEvent]:
+        """Create an action event from a swipe gesture"""
+        
+        if swipe.finger_count == 2:
+            if swipe.direction == "right":
+                return GestureActionEvent(
+                    action=GestureAction.QUERY_MULTI_LLM,
+                    gesture_name=f"2_finger_swipe_{swipe.direction}",
+                    parameters={
+                        "content": context.get("current_content", "") if context else "",
+                        "topic": context.get("topic", "") if context else "",
+                        "swipe_speed": swipe.speed
+                    },
+                    confidence=min(swipe.speed * 100, 1.0)
+                )
+            elif swipe.direction == "left":
+                return GestureActionEvent(
+                    action=GestureAction.SWIPE_LEFT,
+                    gesture_name=f"2_finger_swipe_{swipe.direction}",
+                    parameters={
+                        "topic": context.get("previous_topic", "") if context else "",
+                        "swipe_speed": swipe.speed
+                    },
+                    confidence=min(swipe.speed * 100, 1.0)
+                )
+        
+        elif swipe.finger_count == 1:
+            return GestureActionEvent(
+                action=GestureAction.TRIGGER_RL_LOOP,
+                gesture_name=f"1_finger_swipe_{swipe.direction}",
+                parameters={
+                    "direction": swipe.direction,
+                    "topic": context.get("topic", "") if context else ""
+                },
+                confidence=min(swipe.speed * 100, 1.0)
+            )
+        
+        elif swipe.finger_count == 3:
+            if swipe.direction == "up":
+                return GestureActionEvent(
+                    action=GestureAction.SAVE_PROGRESS,
+                    gesture_name=f"3_finger_swipe_{swipe.direction}",
+                    parameters={"auto": True},
+                    confidence=min(swipe.speed * 100, 1.0)
+                )
+        
+        return None
+    
+    def _create_pinch_event(
+        self, 
+        pinch: PinchGesture, 
+        context: Optional[Dict]
+    ) -> Optional[GestureActionEvent]:
+        """Create an action event from a pinch gesture"""
+        return GestureActionEvent(
+            action=GestureAction.CAPTURE_CONTENT,
+            gesture_name=f"pinch_{pinch.gesture_type}",
+            parameters={
+                "gesture_type": pinch.gesture_type,
+                "distance": pinch.distance,
+                "capture_area": context.get("selected_area") if context else None
+            },
+            confidence=0.9
+        )
+    
+    def _create_open_palm_event(
+        self, 
+        context: Optional[Dict]
+    ) -> Optional[GestureActionEvent]:
+        """Create an action event from open palm gesture"""
+        recent_gestures = [
+            g for g in self.gesture_buffer[-10:] 
+            if g["finger_count"] >= 4
+        ]
+        
+        if len(recent_gestures) >= 5:
+            return GestureActionEvent(
+                action=GestureAction.PAUSE_SESSION,
+                gesture_name="open_palm",
+                parameters={"duration": 2.0},
+                confidence=0.8
+            )
+        
+        return None
+    
+    def execute_action(
+        self, 
+        event: GestureActionEvent,
+        llm_orchestrator=None,
+        rl_loop=None
+    ) -> GestureActionEvent:
+        """Execute a gesture action"""
+        
+        if event.action == GestureAction.QUERY_MULTI_LLM and llm_orchestrator:
+            event = self._execute_llm_query(event, llm_orchestrator)
+        
+        elif event.action == GestureAction.TRIGGER_RL_LOOP and rl_loop:
+            event = self._execute_rl_loop(event, rl_loop)
+        
+        elif event.action == GestureAction.CAPTURE_CONTENT:
+            event = self._execute_capture(event)
+        
+        for callback in self.action_callbacks.get(event.action, []):
+            try:
+                callback(event)
+            except Exception as e:
+                logger.error(f"Callback error: {e}")
+        
+        return event
+    
+    def _execute_llm_query(
+        self, 
+        event: GestureActionEvent,
+        orchestrator
+    ) -> GestureActionEvent:
+        """Execute LLM query from gesture"""
+        import asyncio
+        
+        async def query():
+            from .llm_orchestrator_agent import LLMRequest, LLMProvider
+            
+            content = event.parameters.get("content", "")
+            topic = event.parameters.get("topic", "")
+            
+            if content:
+                prompt = f"Analyze this content and explain key concepts:\n\n{content}"
+            elif topic:
+                prompt = f"Explain '{topic}' in simple terms for learning purposes."
+            else:
+                prompt = "What are the main concepts I should understand about this topic?"
+            
+            request = LLMRequest(
+                prompt=prompt,
+                system_prompt="You are a helpful learning assistant that explains concepts clearly.",
+                providers=[LLMProvider.CHATGPT, LLMProvider.GEMINI]
+            )
+            
+            responses = await orchestrator.query_parallel(request)
+            
+            event.llm_responses = [
+                {
+                    "provider": r.provider.value,
+                    "content": r.content,
+                    "success": r.success
+                }
+                for r in responses
+            ]
+            
+            return event
+        
+        try:
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                asyncio.create_task(query())
+            else:
+                event = loop.run_until_complete(query())
+        except Exception as e:
+            logger.error(f"LLM query error: {e}")
+            event.llm_responses = [{"error": str(e)}]
+        
+        return event
+    
+    def _execute_rl_loop(
+        self, 
+        event: GestureActionEvent,
+        rl_loop
+    ) -> GestureActionEvent:
+        """Execute RL learning loop (OpenClaw-RL style)"""
+        
+        event.rl_feedback = {
+            "mode": "rl_optimization",
+            "action": event.action.value,
+            "topic": event.parameters.get("topic", ""),
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        return event
+    
+    def _execute_capture(
+        self, 
+        event: GestureActionEvent
+    ) -> GestureActionEvent:
+        """Execute content capture"""
+        event.parameters["captured"] = True
+        event.parameters["capture_timestamp"] = datetime.now().isoformat()
+        return event
+    
+    def get_available_actions(self) -> List[Dict]:
+        """Get list of available gesture actions"""
+        actions = []
+        
+        for action, rules in self.action_rules.items():
+            trigger = rules.get("trigger", {})
+            actions.append({
+                "action": action.value,
+                "gesture": trigger.get("gesture", f"{trigger.get('finger_count', '?')}_finger_swipe_{trigger.get('swipe', '')}"),
+                "description": self._get_action_description(action)
+            })
+        
+        return actions
+    
+    def _get_action_description(self, action: GestureAction) -> str:
+        """Get description for an action"""
+        descriptions = {
+            GestureAction.QUERY_MULTI_LLM: "Query multiple AI models simultaneously for the best answer",
+            GestureAction.QUERY_CHATGPT: "Query ChatGPT for an answer",
+            GestureAction.QUERY_GEMINI: "Query Gemini for an answer",
+            GestureAction.CUSTOM: "Custom action based on your gesture"
+        }
+        # Also check for string values
+        if hasattr(action, 'value'):
+            val = action.value
+            if val == 'trigger_rl_loop':
+                return "Start the RL learning loop to optimize responses"
+            elif val == 'capture_content':
+                return "Capture and save the current content or selection"
+            elif val == 'pause_session':
+                return "Pause the current learning session"
+            elif val == 'resume_session':
+                return "Resume the paused learning session"
+        return descriptions.get(action, "Action description not available")
+
+
+class RLLearningLoop:
+    """
+    OpenClaw-RL inspired learning loop.
+    
+    Features:
+    - Next-state feedback
+    - Binary reward signals
+    - Personal agent optimization
+    - User preference learning
+    """
+    
+    def __init__(self, user_id: str):
+        self.user_id = user_id
+        
+        self.conversation_history: List[Dict] = []
+        self.feedback_history: List[Dict] = []
+        self.preference_weights: Dict[str, float] = {}
+        
+        self.is_active = False
+        self.current_context = None
+        
+        self.reward_history: List[float] = []
+        self.policy_updates: List[Dict] = []
+    
+    def start_loop(self, context: Dict):
+        """Start the RL learning loop"""
+        self.is_active = True
+        self.current_context = context
+        self.conversation_history = []
+    
+    def end_loop(self):
+        """End the RL learning loop"""
+        self.is_active = False
+        self._compute_rewards()
+    
+    def add_interaction(
+        self, 
+        action: str, 
+        response: str,
+        context: Optional[Dict] = None
+    ):
+        """Add an interaction to the learning loop"""
+        interaction = {
+            "action": action,
+            "response": response,
+            "context": context or self.current_context,
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        self.conversation_history.append(interaction)
+        
+        if len(self.conversation_history) >= 3:
+            self._update_preferences()
+    
+    def add_feedback(self, quality: int, comment: Optional[str] = None):
+        """
+        Add user feedback (OpenClaw-RL style).
+        
+        Quality scale:
+        - 1: Very bad
+        - 2: Bad
+        - 3: Neutral
+        - 4: Good
+        - 5: Very good
+        """
+        feedback = {
+            "quality": quality,
+            "comment": comment,
+            "timestamp": datetime.now().isoformat(),
+            "interaction_index": len(self.conversation_history) - 1
+        }
+        
+        self.feedback_history.append(feedback)
+        
+        self._process_feedback(feedback)
+    
+    def _process_feedback(self, feedback: Dict):
+        """Process feedback for RL (binary reward + OPD)"""
+        
+        reward = 1.0 if feedback["quality"] >= 4 else -0.5
+        
+        self.reward_history.append(reward)
+        
+        if len(self.conversation_history) > 0:
+            last_interaction = self.conversation_history[-1]
+            last_interaction["reward"] = reward
+            last_interaction["feedback_quality"] = feedback["quality"]
+        
+        if feedback["quality"] >= 4 and feedback["comment"]:
+            self._apply_opd(feedback["comment"])
+    
+    def _apply_opd(self, corrective_hint: str):
+        """
+        Apply On-Policy Distillation from corrective hints.
+        
+        OpenClaw-RL OPD extracts textual hints from next state
+        and constructs enhanced teacher context.
+        """
+        
+        opd_update = {
+            "corrective_hint": corrective_hint,
+            "timestamp": datetime.now().isoformat(),
+            "policy_version": len(self.policy_updates) + 1
+        }
+        
+        self.policy_updates.append(opd_update)
+        
+        for word in corrective_hint.lower().split():
+            if word not in self.preference_weights:
+                self.preference_weights[word] = 0.5
+            self.preference_weights[word] += 0.1
+    
+    def _compute_rewards(self):
+        """Compute aggregate rewards from feedback"""
+        if not self.reward_history:
+            return
+        
+        avg_reward = sum(self.reward_history) / len(self.reward_history)
+        
+        self.policy_updates.append({
+            "type": "session_summary",
+            "total_interactions": len(self.conversation_history),
+            "total_feedback": len(self.feedback_history),
+            "average_reward": avg_reward,
+            "timestamp": datetime.now().isoformat()
+        })
+    
+    def _update_preferences(self):
+        """Update preference weights based on recent interactions"""
+        recent = self.conversation_history[-3:]
+        
+        for interaction in recent:
+            response = interaction.get("response", "").lower()
+            
+            positive_words = ["helpful", "clear", "understand", "good", "thanks"]
+            negative_words = ["confusing", "wrong", "don't", "no", "bad"]
+            
+            for word in positive_words:
+                if word in response:
+                    self.preference_weights[word] = self.preference_weights.get(word, 0.5) + 0.05
+            
+            for word in negative_words:
+                if word in response:
+                    self.preference_weights[word] = max(0.1, self.preference_weights.get(word, 0.5) - 0.05)
+    
+    def get_status(self) -> Dict:
+        """Get current RL loop status"""
+        return {
+            "is_active": self.is_active,
+            "total_interactions": len(self.conversation_history),
+            "total_feedback": len(self.feedback_history),
+            "average_reward": (
+                sum(self.reward_history) / len(self.reward_history) 
+                if self.reward_history else 0
+            ),
+            "policy_updates": len(self.policy_updates),
+            "top_preferences": sorted(
+                self.preference_weights.items(),
+                key=lambda x: x[1],
+                reverse=True
+            )[:10]
+        }
+    
+    def generate_personalized_prompt(self, base_prompt: str) -> str:
+        """Generate a personalized prompt based on learned preferences"""
+        
+        top_prefs = self.get_status()["top_preferences"]
+        
+        if not top_prefs:
+            return base_prompt
+        
+        style_guide = " ".join([word for word, _ in top_prefs[:5]])
+        
+        enhanced_prompt = f"{base_prompt}\n\nStyle guidance: {style_guide}"
+        
+        return enhanced_prompt