feat: Add "Thinking" Behavior When Robot is Confused (#1518)

config/schema/single_mode_schema.json

@@ -89,6 +89,20 @@
                 }
             }
         },
+"thinking_behavior": {
+  "type": "object",
+  "description": "Configuration for thinking behavior when the robot is processing complex queries",
+  "properties": {
+    "enabled": { "type": "boolean", "default": false },
+    "face_action": { "type": "string", "default": "think" },
+    "move_action": { "type": "string", "default": "stand still" },
+    "trigger_delay": { "type": "number", "default": 1.0, "minimum": 0.1 },
+    "min_duration": { "type": "number", "default": 1.0, "minimum": 0.1 },
+    "max_duration": { "type": "number", "default": 3.0, "minimum": 0.5 }
+  },
+  "required": ["enabled"],
+  "additionalProperties": false
+}
         "backgrounds": {
             "type": "array",
             "items": {

config/spot.json5

@@ -57,4 +57,12 @@
       connector: "ros2",
     },
   ],
+  thinking_behavior: {
+    enabled: true,
+    face_action: "think",
+    move_action: "stand still",
+    trigger_delay: 1.0,
+    min_duration: 1.0,
+    max_duration: 3.0,
+  },
 }

src/llm/output_model.py

@@ -1,30 +1,15 @@
+from typing import Optional
 from pydantic import BaseModel, Field
 
 
 class Action(BaseModel):
-    """
-    Executable action with its argument.
-
-    Parameters
-    ----------
-    type : str
-        Type of action to execute, such as 'move' or 'speak'
-    value : str
-        The action argument, such as the magnitude of a movement or the sentence to speak
-    """
-
     type: str = Field(..., description="The specific type of action, such as 'move' or 'speak'")
     value: str = Field(..., description="The action argument")
 
 
 class CortexOutputModel(BaseModel):
-    """
-    Output model for the Cortex LLM responses.
-
-    Parameters
-    ----------
-    actions : list[Action]
-        List of actions to be executed
-    """
-
     actions: list[Action] = Field(..., description="List of actions to execute")
+    thinking_duration: Optional[float] = Field(
+        default=None,
+        description="Optional duration in seconds to show thinking pose before executing actions"
+    )

src/runtime/config.py

@@ -89,6 +89,14 @@ def validate_config_schema(raw_config: dict) -> None:
 
 
 @dataclass
+class ThinkingBehaviorConfig:
+    """Configuration for thinking behavior when robot is processing complex queries."""
+    enabled: bool = False
+    face_action: str = "think"
+    move_action: str = "stand still"
+    trigger_delay: float = 1.0
+    min_duration: float = 1.0
+    max_duration: float = 3.0
 class RuntimeConfig:
     """
     Runtime configuration for the agent.
@@ -157,6 +165,7 @@ class RuntimeConfig:
     action_dependencies: Optional[Dict[str, List[str]]] = None
     knowledge_base: Optional[Dict[str, Any]] = None
     mcp_servers: Optional[Any] = None
+    thinking_behavior: Optional[ThinkingBehaviorConfig] = None
 
 
 def add_meta(

src/runtime/cortex.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import time
+import json
 from typing import List, Optional, Union
 
 from actions.orchestrator import ActionOrchestrator
@@ -668,6 +669,56 @@ async def _tick(self, cortex_generation: int) -> None:
             logging.debug("No output from LLM")
             return
 
+
+    async def _trigger_thinking_pose(self) -> bool:
+        if not self.current_config or not self.current_config.thinking_behavior:
+            return False
+        tb_config = self.current_config.thinking_behavior
+        if not tb_config.enabled or not self.action_orchestrator:
+            return False
+        try:
+            from llm.output_model import Action
+            think_face = Action(type="emotion", value=json.dumps({"action": tb_config.face_action}))
+            think_move = Action(type="move", value=json.dumps({"action": tb_config.move_action}))
+            await self.action_orchestrator.promise([think_face, think_move])
+            logging.info(f"Thinking pose triggered: face={tb_config.face_action}, move={tb_config.move_action}")
+            return True
+        except Exception as e:
+            logging.warning(f"Failed to trigger thinking pose: {e}")
+            return False
+
+    async def _execute_with_thinking_behavior(self, prompt: str):
+        tb_config = self.current_config.thinking_behavior if self.current_config else None
+        thinking_triggered = False
+        thinking_task = None
+        if tb_config and tb_config.enabled:
+            async def delayed_thinking():
+                await asyncio.sleep(tb_config.trigger_delay)
+                return await self._trigger_thinking_pose()
+            thinking_task = asyncio.create_task(delayed_thinking())
+        try:
+            final_output = None
+            async for output in self.current_config.cortex_llm.ask_stream(prompt):
+                final_output = output
+            if thinking_task and not thinking_task.done():
+                thinking_task.cancel()
+                try:
+                    await thinking_task
+                except asyncio.CancelledError:
+                    pass
+            elif thinking_task and thinking_task.done():
+                thinking_triggered = thinking_task.result()
+            if thinking_triggered and final_output and getattr(final_output, 'thinking_duration', None):
+                duration = min(max(final_output.thinking_duration, tb_config.min_duration), tb_config.max_duration)
+                remaining = max(0, duration - tb_config.trigger_delay)
+                if remaining > 0:
+                    await asyncio.sleep(remaining)
+            return final_output
+        except Exception as e:
+            if thinking_task and not thinking_task.done():
+                thinking_task.cancel()
+            raise e
+
     def get_mode_info(self) -> dict:
         """
         Get information about the current mode and available transitions.