LLM Cognitive Planning for VLA Systems
Learning Objectives
By the end of this chapter, you will be able to:
- Understand the role of Large Language Models in cognitive planning for robotics
- Implement LLM-based planning for Vision-Language-Action systems
- Design effective prompts for robotic task planning
- Integrate LLM planning with vision and action systems
- Handle uncertainty and context in LLM-based robotic planning
Introduction to LLM Cognitive Planning
Large Language Models (LLMs) have emerged as powerful tools for cognitive planning in robotics, bridging the gap between high-level human commands and low-level robot actions. In Vision-Language-Action (VLA) systems, LLMs serve as the cognitive layer that interprets natural language commands, reasons about the environment, and generates executable action plans.
The Role of LLMs in Robotics
LLMs bring several key capabilities to robotic planning:
- Natural Language Understanding: Interpreting human commands in natural language
- World Modeling: Reasoning about objects, relationships, and physics
- Task Decomposition: Breaking complex tasks into executable steps
- Context Awareness: Understanding the current state and constraints
- Common Sense Reasoning: Applying general knowledge to specific situations
Cognitive Planning Architecture
# LLM-based cognitive planning architecture
"""
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Human │ │ LLM Cognitive │ │ Robot │
│ Command │───▶│ Planner │───▶│ Execution │
│ (Natural │ │ (Task │ │ (Low-level │
│ Language) │ │ Decomposition) │ │ Actions) │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
▼
┌──────────────────┐
│ Environment │
│ Context │
│ (Vision, │
│ State, etc.) │
└──────────────────┘
"""
Setting Up LLM Integration
Required Libraries and Setup
import openai
import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
class Action(Enum):
"""Enumeration of possible robot actions"""
MOVE_TO = "move_to"
GRASP = "grasp"
RELEASE = "release"
NAVIGATE = "navigate"
DETECT_OBJECT = "detect_object"
FOLLOW_PATH = "follow_path"
WAIT = "wait"
SPEAK = "speak"
@dataclass
class RobotAction:
"""Represents a single robot action"""
action_type: Action
parameters: Dict[str, Any]
description: str
duration_estimate: float = 1.0 # seconds
@dataclass
class PlanStep:
"""A step in the robot execution plan"""
step_id: int
action: RobotAction
preconditions: List[str]
effects: List[str]
next_step_ids: List[int]
class LLMBasedPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
Initialize LLM-based cognitive planner
Args:
api_key (str): OpenAI API key
model (str): LLM model to use for planning
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.conversation_history = []
def plan_task(self,
command: str,
environment_context: Dict[str, Any],
robot_capabilities: List[str] = None) -> List[RobotAction]:
"""
Generate a plan for a given command in the environment context
Args:
command (str): Natural language command
environment_context (dict): Current environment state
robot_capabilities (list): List of robot capabilities
Returns:
List[RobotAction]: Sequence of actions to execute
"""
if robot_capabilities is None:
robot_capabilities = [
"navigation", "grasping", "object detection",
"manipulation", "speech"
]
# Create a detailed prompt for the LLM
prompt = self._create_planning_prompt(
command,
environment_context,
robot_capabilities
)
# Get plan from LLM
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self._get_system_prompt()},
{"role": "user", "content": prompt}
],
temperature=0.3, # Lower temperature for more consistent planning
response_format={"type": "json_object"} # Request JSON response
)
# Parse and validate the response
plan_json = json.loads(response.choices[0].message.content)
actions = self._parse_plan_actions(plan_json)
# Add to conversation history
self.conversation_history.append({
"command": command,
"plan": actions,
"timestamp": time.time()
})
return actions
def _create_planning_prompt(self, command: str,
environment_context: Dict[str, Any],
robot_capabilities: List[str]) -> str:
"""Create a detailed prompt for task planning"""
prompt = f"""
You are a cognitive planner for a robot. Given a natural language command and the current environment context, generate a detailed plan of actions for the robot to execute.
Command: "{command}"
Environment Context:
{json.dumps(environment_context, indent=2)}
Robot Capabilities:
{', '.join(robot_capabilities)}
Please generate a plan as a JSON array of actions. Each action should have:
- type: The action type (one of: {', '.join([action.value for action in Action])})
- parameters: Parameters needed for the action
- description: Human-readable description of the action
- duration_estimate: Estimated time in seconds to complete the action
Example response format:
{{
"actions": [
{{
"type": "move_to",
"parameters": {{"location": [1.0, 2.0, 0.0]}},
"description": "Move to the kitchen counter",
"duration_estimate": 5.0
}},
{{
"type": "grasp",
"parameters": {{"object": "cup", "position": [1.2, 2.1, 0.8]}},
"description": "Grasp the red cup",
"duration_estimate": 3.0
}}
]
}}
The plan should be:
1. Feasible given the robot's capabilities
2. Safe and appropriate for the environment
3. Efficient in terms of the number of steps
4. Detailed enough for the robot to execute
"""
return prompt
def _get_system_prompt(self) -> str:
"""Get the system prompt for the LLM"""
return """
You are an expert cognitive planner for robotics applications. Your role is to interpret natural language commands and generate detailed, executable plans for robots.
Guidelines:
1. Always consider the physical constraints of the environment
2. Take into account the robot's capabilities and limitations
3. Generate safe and feasible plans
4. Break complex tasks into simple, executable steps
5. Consider the current state of the environment when planning
6. Use precise coordinates and object descriptions when possible
7. Include error handling and verification steps when appropriate
"""
def _parse_plan_actions(self, plan_json: Dict[str, Any]) -> List[RobotAction]:
"""Parse the JSON plan into RobotAction objects"""
actions = []
for action_data in plan_json.get("actions", []):
action_type = Action(action_data["type"])
parameters = action_data.get("parameters", {})
description = action_data.get("description", "")
duration_estimate = action_data.get("duration_estimate", 1.0)
action = RobotAction(
action_type=action_type,
parameters=parameters,
description=description,
duration_estimate=duration_estimate
)
actions.append(action)
return actions
Advanced Planning Techniques
Context-Aware Planning
class ContextAwarePlanner(LLMBasedPlanner):
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
super().__init__(api_key, model)
self.long_term_memory = []
self.current_plan = None
self.plan_execution_state = {}
def plan_with_context(self,
command: str,
environment_context: Dict[str, Any],
previous_actions: List[RobotAction] = None,
failed_attempts: List[str] = None) -> List[RobotAction]:
"""
Generate a plan considering context from previous interactions
Args:
command (str): Natural language command
environment_context (dict): Current environment state
previous_actions (list): Actions already taken in this task
failed_attempts (list): Previous failed attempts at similar tasks
Returns:
List[RobotAction]: Sequence of actions to execute
"""
# Build comprehensive context
full_context = {
"current_command": command,
"environment": environment_context,
"previous_actions": previous_actions or [],
"failed_attempts": failed_attempts or [],
"long_term_memory": self.long_term_memory[-10:] # Last 10 interactions
}
prompt = self._create_contextual_planning_prompt(full_context)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self._get_contextual_system_prompt()},
{"role": "user", "content": prompt}
],
temperature=0.2, # Even lower for consistency with context
response_format={"type": "json_object"}
)
plan_json = json.loads(response.choices[0].message.content)
actions = self._parse_plan_actions(plan_json)
# Update internal state
self.current_plan = actions
self.plan_execution_state = {"step": 0, "completed": []}
return actions
def _create_contextual_planning_prompt(self, context: Dict[str, Any]) -> str:
"""Create a prompt that includes historical context"""
return f"""
You are a cognitive planner for a robot. Generate a plan considering all available context.
Current Command: "{context['current_command']}"
Current Environment:
{json.dumps(context['environment'], indent=2)}
Previous Actions in This Task:
{json.dumps(context['previous_actions'], indent=2)}
Failed Attempts (to avoid repeating):
{json.dumps(context['failed_attempts'], indent=2)}
Long-term Memory (relevant past interactions):
{json.dumps(context['long_term_memory'], indent=2)}
Generate a JSON plan with actions that:
1. Address the current command
2. Learn from previous actions and failures
3. Are consistent with the robot's state
4. Consider the environment's current state
5. Include verification steps where appropriate
Return the plan in the standard JSON format with actions array.
"""
def _get_contextual_system_prompt(self) -> str:
"""System prompt for context-aware planning"""
return """
You are an advanced cognitive planner for robotics with memory capabilities.
Consider all provided context when generating plans:
1. Account for the robot's current state based on previous actions
2. Learn from failed attempts to avoid repeating mistakes
3. Use long-term memory to inform planning decisions
4. Maintain consistency with the task's progress
5. Include error checking and verification steps when appropriate
6. Adapt your planning based on the robot's learning history
"""
def update_memory(self, command: str, plan: List[RobotAction],
outcome: Dict[str, Any], success: bool):
"""Update the planner's memory with the results of plan execution"""
memory_entry = {
"command": command,
"plan": [action.description for action in plan],
"outcome": outcome,
"success": success,
"timestamp": time.time()
}
self.long_term_memory.append(memory_entry)
# Keep only recent memory to manage storage
if len(self.long_term_memory) > 100:
self.long_term_memory = self.long_term_memory[-100:]
Hierarchical Task Planning
Multi-Level Planning Architecture
class HierarchicalPlanner:
def __init__(self, llm_planner: LLMBasedPlanner):
"""
Hierarchical planner that can plan at different levels of abstraction
Args:
llm_planner: Base LLM planner for low-level planning
"""
self.llm_planner = llm_planner
self.task_library = self._initialize_task_library()
def _initialize_task_library(self) -> Dict[str, List[RobotAction]]:
"""Initialize a library of common tasks"""
return {
"pick_and_place": [
RobotAction(Action.DETECT_OBJECT, {"object_type": "any"}, "Detect target object"),
RobotAction(Action.MOVE_TO, {"location": "object_location"}, "Move to object"),
RobotAction(Action.GRASP, {"object": "target"}, "Grasp the object"),
RobotAction(Action.MOVE_TO, {"location": "destination"}, "Move to destination"),
RobotAction(Action.RELEASE, {"object": "target"}, "Release the object")
],
"room_navigation": [
RobotAction(Action.NAVIGATE, {"waypoints": []}, "Navigate through room"),
RobotAction(Action.DETECT_OBSTACLES, {"range": 2.0}, "Detect obstacles"),
RobotAction(Action.AVOID_OBSTACLES, {}, "Avoid detected obstacles")
],
"object_search": [
RobotAction(Action.MOVE_TO, {"location": "search_pattern_start"}, "Start search pattern"),
RobotAction(Action.DETECT_OBJECT, {"object_type": "target"}, "Look for object"),
RobotAction(Action.MOVE_TO, {"location": "next_search_point"}, "Move to next search point")
]
}
def plan_high_level_task(self, task_name: str,
parameters: Dict[str, Any],
environment_context: Dict[str, Any]) -> List[RobotAction]:
"""
Plan a high-level task using the task library
Args:
task_name (str): Name of the high-level task
parameters (dict): Parameters for the task
environment_context (dict): Current environment state
Returns:
List[RobotAction]: Detailed sequence of low-level actions
"""
if task_name not in self.task_library:
# Fall back to LLM planning for unknown tasks
command = f"Perform {task_name} with parameters {parameters}"
return self.llm_planner.plan_task(command, environment_context)
# Get the template for the high-level task
template_actions = self.task_library[task_name].copy()
# Specialize the template with specific parameters
specialized_actions = self._specialize_actions(
template_actions,
parameters,
environment_context
)
return specialized_actions
def _specialize_actions(self,
template_actions: List[RobotAction],
parameters: Dict[str, Any],
environment_context: Dict[str, Any]) -> List[RobotAction]:
"""Specialize template actions with specific parameters"""
specialized = []
for action in template_actions:
specialized_action = self._specialize_action(
action,
parameters,
environment_context
)
specialized.append(specialized_action)
return specialized
def _specialize_action(self,
template_action: RobotAction,
parameters: Dict[str, Any],
environment_context: Dict[str, Any]) -> RobotAction:
"""Specialize a single template action"""
# Create a new action with updated parameters
new_parameters = template_action.parameters.copy()
new_description = template_action.description
# Replace placeholders in parameters
for key, value in parameters.items():
if key in new_parameters:
new_parameters[key] = value
# Also check if the description contains the key
if f"{{{key}}}" in new_description:
new_description = new_description.replace(f"{{{key}}}", str(value))
# For location-based actions, we might need to look up actual coordinates
if "location" in new_parameters and isinstance(new_parameters["location"], str):
actual_location = self._resolve_location(
new_parameters["location"],
environment_context
)
new_parameters["location"] = actual_location
return RobotAction(
action_type=template_action.action_type,
parameters=new_parameters,
description=new_description,
duration_estimate=template_action.duration_estimate
)
def _resolve_location(self, location_name: str,
environment_context: Dict[str, Any]) -> List[float]:
"""Resolve a named location to actual coordinates"""
# Look up named locations in environment context
locations = environment_context.get("named_locations", {})
if location_name in locations:
return locations[location_name]
# If not found, return a default location or ask LLM to infer
# For now, return a default location
return [0.0, 0.0, 0.0]
Planning with Uncertainty and Feedback
Adaptive Planning System
class AdaptivePlanner:
def __init__(self, base_planner: LLMBasedPlanner):
"""
Planner that can adapt to uncertainty and feedback
Args:
base_planner: Base planner to use for initial planning
"""
self.base_planner = base_planner
self.current_plan = []
self.execution_history = []
self.uncertainty_model = UncertaintyModel()
def execute_plan_with_monitoring(self,
plan: List[RobotAction],
environment_context: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute a plan while monitoring for deviations and adapting as needed
Args:
plan (List[RobotAction]): Plan to execute
environment_context (dict): Current environment state
Returns:
Dict: Execution results including any adaptations made
"""
results = {
"executed_actions": [],
"successful": True,
"adaptations": [],
"metrics": {}
}
self.current_plan = plan
start_time = time.time()
for i, action in enumerate(plan):
# Check if the environment has changed significantly
updated_context = self._get_updated_context(environment_context)
# Assess if the current action is still appropriate
if not self._is_action_appropriate(action, updated_context):
adaptation = self._adapt_current_plan(i, updated_context)
results["adaptations"].append(adaptation)
# If adaptation failed, stop execution
if not adaptation["success"]:
results["successful"] = False
break
# Execute the action
action_result = self._execute_single_action(action, updated_context)
results["executed_actions"].append({
"action": action,
"result": action_result,
"success": action_result.get("success", False)
})
# Record execution for learning
self.execution_history.append({
"action": action,
"context": updated_context,
"result": action_result,
"timestamp": time.time()
})
# If action failed, try to recover
if not action_result.get("success", False):
recovery_result = self._attempt_recovery(action, updated_context)
if recovery_result["success"]:
results["adaptations"].append({
"type": "recovery",
"original_action": action,
"recovery_action": recovery_result["action"],
"result": recovery_result
})
else:
results["successful"] = False
break
# Record metrics
results["metrics"] = {
"execution_time": time.time() - start_time,
"actions_completed": len(results["executed_actions"]),
"success_rate": sum(1 for a in results["executed_actions"]
if a["success"]) / len(results["executed_actions"]) if results["executed_actions"] else 0
}
return results
def _is_action_appropriate(self, action: RobotAction,
context: Dict[str, Any]) -> bool:
"""Check if an action is still appropriate given the current context"""
# This is a simplified check - in practice, this would be more sophisticated
preconditions = self._get_action_preconditions(action)
for precondition in preconditions:
if not self._check_precondition(precondition, context):
return False
return True
def _get_action_preconditions(self, action: RobotAction) -> List[str]:
"""Get preconditions for an action"""
# Define preconditions for different action types
preconditions = {
Action.MOVE_TO: ["robot_is_operational", "path_is_clear"],
Action.GRASP: ["object_is_reachable", "gripper_is_free", "object_is_stable"],
Action.NAVIGATE: ["robot_is_operational", "map_is_known"],
Action.DETECT_OBJECT: ["camera_is_operational", "lighting_is_sufficient"]
}
return preconditions.get(action.action_type, [])
def _check_precondition(self, precondition: str, context: Dict[str, Any]) -> bool:
"""Check if a precondition is satisfied in the current context"""
# This would interface with the robot's sensors and state
# For now, return True for all preconditions
return True
def _adapt_current_plan(self, current_step: int,
context: Dict[str, Any]) -> Dict[str, Any]:
"""Adapt the current plan based on changed context"""
remaining_plan = self.current_plan[current_step:]
# Replan the remaining portion of the task
# This is a simplified approach - in practice, you might do partial replanning
if remaining_plan:
# Create a new command based on the remaining actions
remaining_command = self._describe_remaining_task(remaining_plan, context)
try:
new_plan = self.base_planner.plan_task(remaining_command, context)
# Replace the remaining portion of the plan
self.current_plan = self.current_plan[:current_step] + new_plan
return {
"success": True,
"type": "replanning",
"original_remaining": remaining_plan,
"new_plan": new_plan
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
return {"success": True, "type": "no_adaptation_needed"}
def _describe_remaining_task(self, remaining_actions: List[RobotAction],
context: Dict[str, Any]) -> str:
"""Describe the remaining task in natural language for replanning"""
descriptions = [action.description for action in remaining_actions]
return f"Continue with the remaining task: {', '.join(descriptions)}. " \
f"Consider the updated environment context: {json.dumps(context)}"
def _execute_single_action(self, action: RobotAction,
context: Dict[str, Any]) -> Dict[str, Any]:
"""Execute a single action (simulated in this example)"""
# In a real system, this would interface with the robot's control system
# For simulation, we'll return a success result
time.sleep(action.duration_estimate * 0.1) # Simulate execution time
return {
"success": True,
"execution_time": action.duration_estimate * 0.1,
"actual_parameters": action.parameters
}
def _attempt_recovery(self, failed_action: RobotAction,
context: Dict[str, Any]) -> Dict[str, Any]:
"""Attempt to recover from a failed action"""
recovery_strategies = {
Action.MOVE_TO: [Action.WAIT, Action.DETECT_OBSTACLES],
Action.GRASP: [Action.DETECT_OBJECT, Action.MOVE_TO],
Action.NAVIGATE: [Action.DETECT_OBSTACLES, Action.WAIT]
}
strategies = recovery_strategies.get(failed_action.action_type, [])
for strategy in strategies:
# Try to execute a recovery action
recovery_action = RobotAction(
action_type=strategy,
parameters={"context": context},
description=f"Recovery action for failed {failed_action.action_type.value}",
duration_estimate=2.0
)
recovery_result = self._execute_single_action(recovery_action, context)
if recovery_result["success"]:
return {
"success": True,
"action": recovery_action,
"result": recovery_result
}
return {"success": False, "reason": "No recovery strategy succeeded"}
def _get_updated_context(self, base_context: Dict[str, Any]) -> Dict[str, Any]:
"""Get the most up-to-date environment context"""
# In practice, this would query sensors and the robot's state
# For now, return the base context
return base_context
Integration with Vision Systems
Vision-Grounded Planning
class VisionGroundedPlanner:
def __init__(self, llm_planner: LLMBasedPlanner, vision_system=None):
"""
Planner that incorporates visual information into planning
Args:
llm_planner: Base LLM planner
vision_system: Vision system interface (optional)
"""
self.llm_planner = llm_planner
self.vision_system = vision_system
self.object_database = {} # Store known object properties
def plan_with_vision(self, command: str,
initial_context: Dict[str, Any] = None) -> List[RobotAction]:
"""
Plan a task incorporating real-time visual information
Args:
command (str): Natural language command
initial_context (dict): Initial environment context
Returns:
List[RobotAction]: Plan incorporating visual information
"""
if initial_context is None:
initial_context = {}
# Get current visual information
visual_context = self._get_visual_context()
# Merge visual context with initial context
full_context = self._merge_contexts(initial_context, visual_context)
# Plan using the combined context
plan = self.llm_planner.plan_task(command, full_context)
# Validate the plan against visual information
validated_plan = self._validate_plan_with_vision(plan, visual_context)
return validated_plan
def _get_visual_context(self) -> Dict[str, Any]:
"""Get current context from vision system"""
if self.vision_system:
# In a real system, this would call the vision system
# For now, we'll simulate visual perception
return {
"objects_detected": [
{"name": "red_cup", "type": "cup", "position": [1.0, 0.5, 0.0], "color": "red"},
{"name": "blue_bottle", "type": "bottle", "position": [1.2, 0.7, 0.0], "color": "blue"},
{"name": "wooden_table", "type": "table", "position": [1.0, 0.6, 0.0], "material": "wood"}
],
"surfaces": [
{"name": "kitchen_counter", "type": "counter", "position": [1.0, 0.6, 0.8]}
],
"obstacles": [
{"type": "chair", "position": [0.5, 1.0, 0.0]}
],
"navigation_map": "partial_map_data"
}
else:
# Return empty context if no vision system
return {
"objects_detected": [],
"surfaces": [],
"obstacles": [],
"navigation_map": "empty"
}
def _merge_contexts(self, initial: Dict[str, Any],
visual: Dict[str, Any]) -> Dict[str, Any]:
"""Merge initial context with visual context"""
merged = initial.copy()
# Add or update visual information
for key, value in visual.items():
merged[key] = value
# Add derived information
merged["object_names"] = [obj["name"] for obj in visual.get("objects_detected", [])]
merged["object_types"] = list(set(obj["type"] for obj in visual.get("objects_detected", [])))
return merged
def _validate_plan_with_vision(self, plan: List[RobotAction],
visual_context: Dict[str, Any]) -> List[RobotAction]:
"""Validate and potentially adjust plan based on visual information"""
validated_plan = []
for action in plan:
if action.action_type == Action.GRASP:
# Validate grasp action with visual information
validated_action = self._validate_grasp_action(action, visual_context)
elif action.action_type == Action.MOVE_TO:
# Validate move action with visual information
validated_action = self._validate_move_action(action, visual_context)
else:
validated_action = action
validated_plan.append(validated_action)
return validated_plan
def _validate_grasp_action(self, action: RobotAction,
visual_context: Dict[str, Any]) -> RobotAction:
"""Validate a grasp action against visual information"""
# Check if the object to be grasped exists in the visual context
target_object = action.parameters.get("object")
if target_object:
# Look for the object in detected objects
detected_objects = visual_context.get("objects_detected", [])
matching_objects = [obj for obj in detected_objects
if target_object.lower() in obj["name"].lower() or
target_object.lower() in obj["type"].lower()]
if not matching_objects:
# Object not found, try to update the action
print(f"Warning: Object '{target_object}' not found in visual context")
# In a real system, you might want to add a detection step
# or ask the user for clarification
return action
elif len(matching_objects) == 1:
# Found exactly one matching object, update position if needed
matched_obj = matching_objects[0]
if "position" in matched_obj:
action.parameters["position"] = matched_obj["position"]
action.description = f"Grasp the {matched_obj['name']} at {matched_obj['position']}"
return action
def _validate_move_action(self, action: RobotAction,
visual_context: Dict[str, Any]) -> RobotAction:
"""Validate a move action against visual information"""
target_location = action.parameters.get("location")
if isinstance(target_location, str):
# If location is a name, try to resolve it using visual context
resolved_location = self._resolve_named_location(target_location, visual_context)
if resolved_location:
action.parameters["location"] = resolved_location
action.description = f"Move to {target_location} at {resolved_location}"
return action
def _resolve_named_location(self, location_name: str,
visual_context: Dict[str, Any]) -> Optional[List[float]]:
"""Resolve a named location using visual context"""
# Look for named locations in visual context
surfaces = visual_context.get("surfaces", [])
for surface in surfaces:
if location_name.lower() in surface["name"].lower():
return surface["position"]
# Look for objects that might serve as landmarks
objects = visual_context.get("objects_detected", [])
for obj in objects:
if location_name.lower() in obj["name"].lower():
# Return position, possibly adjusted for interaction
pos = obj["position"].copy()
pos[2] += 0.2 # Raise slightly above object
return pos
return None
Planning with Multiple Modalities
Multi-Modal Cognitive Architecture
class MultiModalCognitivePlanner:
def __init__(self,
llm_planner: LLMBasedPlanner,
vision_planner: VisionGroundedPlanner,
speech_planner: LLMBasedPlanner = None):
"""
Multi-modal cognitive planner integrating vision, language, and action
Args:
llm_planner: Base LLM planner
vision_planner: Vision-grounded planner
speech_planner: Optional specialized planner for speech-related tasks
"""
self.llm_planner = llm_planner
self.vision_planner = vision_planner
self.speech_planner = speech_planner or llm_planner
self.modality_weights = {
"language": 0.4,
"vision": 0.4,
"context": 0.2
}
def plan_multimodal_task(self,
command: str,
language_context: Dict[str, Any] = None,
vision_context: Dict[str, Any] = None,
action_context: Dict[str, Any] = None) -> List[RobotAction]:
"""
Plan a task using multiple modalities
Args:
command (str): Natural language command
language_context (dict): Language-specific context
vision_context (dict): Vision-specific context
action_context (dict): Action/physical context
Returns:
List[RobotAction]: Integrated plan
"""
# Integrate all modalities into a unified context
unified_context = self._integrate_modalities(
language_context,
vision_context,
action_context
)
# Determine the best planning approach based on the command
planning_approach = self._determine_planning_approach(command)
if planning_approach == "vision_heavy":
return self.vision_planner.plan_with_vision(command, unified_context)
elif planning_approach == "language_heavy":
return self.llm_planner.plan_task(command, unified_context)
else:
# Use integrated approach
return self._integrated_planning(command, unified_context)
def _integrate_modalities(self,
language_ctx: Dict[str, Any],
vision_ctx: Dict[str, Any],
action_ctx: Dict[str, Any]) -> Dict[str, Any]:
"""Integrate information from multiple modalities"""
integrated = {}
# Add language context with appropriate weighting
if language_ctx:
integrated.update({
"language_context": language_ctx,
"command_semantics": self._extract_semantics(language_ctx.get("command", ""))
})
# Add vision context with appropriate weighting
if vision_ctx:
integrated.update(vision_ctx)
integrated["visual_confidence"] = vision_ctx.get("confidence", 1.0)
# Add action context
if action_ctx:
integrated.update(action_ctx)
# Add cross-modal relationships
integrated["cross_modal_relations"] = self._find_cross_modal_relations(
language_ctx, vision_ctx, action_ctx
)
return integrated
def _extract_semantics(self, command: str) -> Dict[str, Any]:
"""Extract semantic information from a command"""
# This would typically use NLP techniques
# For now, we'll do simple keyword extraction
semantics = {
"action_verb": self._extract_action_verb(command),
"target_object": self._extract_target_object(command),
"spatial_relations": self._extract_spatial_relations(command),
"temporal_constraints": self._extract_temporal_constraints(command)
}
return semantics
def _extract_action_verb(self, command: str) -> str:
"""Extract the main action verb from a command"""
# Simple keyword matching - in practice, use proper NLP
command_lower = command.lower()
action_verbs = ["pick", "grasp", "move", "go", "navigate", "place", "put", "take"]
for verb in action_verbs:
if verb in command_lower:
return verb
return "unknown"
def _extract_target_object(self, command: str) -> str:
"""Extract the target object from a command"""
# Simple keyword matching - in practice, use proper NLP
command_lower = command.lower()
common_objects = ["cup", "bottle", "book", "box", "chair", "table", "apple", "water"]
for obj in common_objects:
if obj in command_lower:
return obj
return "unknown_object"
def _extract_spatial_relations(self, command: str) -> List[str]:
"""Extract spatial relations from a command"""
command_lower = command.lower()
spatial_relations = []
spatial_keywords = ["to", "from", "near", "on", "under", "above", "behind", "in front of"]
for keyword in spatial_keywords:
if keyword in command_lower:
spatial_relations.append(keyword)
return spatial_relations
def _extract_temporal_constraints(self, command: str) -> List[str]:
"""Extract temporal constraints from a command"""
command_lower = command.lower()
temporal_keywords = ["now", "immediately", "slowly", "quickly", "carefully", "fast"]
return [word for word in temporal_keywords if word in command_lower]
def _find_cross_modal_relations(self,
language_ctx: Dict[str, Any],
vision_ctx: Dict[str, Any],
action_ctx: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Find relationships between different modalities"""
relations = []
if vision_ctx and language_ctx:
# Find connections between language targets and visual objects
target_object = self._extract_target_object(language_ctx.get("command", ""))
detected_objects = vision_ctx.get("objects_detected", [])
for obj in detected_objects:
if target_object.lower() in obj["name"].lower() or target_object.lower() in obj["type"].lower():
relations.append({
"type": "language_vision_match",
"language_target": target_object,
"visual_object": obj["name"],
"confidence": 0.9
})
return relations
def _determine_planning_approach(self, command: str) -> str:
"""Determine the best planning approach based on the command"""
command_lower = command.lower()
vision_heavy_keywords = ["find", "locate", "search", "look for", "identify", "recognize"]
language_heavy_keywords = ["tell", "explain", "describe", "report"]
for keyword in vision_heavy_keywords:
if keyword in command_lower:
return "vision_heavy"
for keyword in language_heavy_keywords:
if keyword in command_lower:
return "language_heavy"
return "integrated"
def _integrated_planning(self, command: str, context: Dict[str, Any]) -> List[RobotAction]:
"""Perform integrated planning using all modalities"""
# Create a comprehensive prompt that incorporates all modalities
prompt = f"""
You are a multi-modal cognitive planner integrating language, vision, and action capabilities.
Natural Language Command: "{command}"
Language Context: {json.dumps(self._extract_semantics(command), indent=2)}
Vision Context: {json.dumps(context.get('vision_context', {}), indent=2)}
Action Context: {json.dumps(context.get('action_context', {}), indent=2)}
Cross-Modal Relations: {json.dumps(context.get('cross_modal_relations', []), indent=2)}
Generate a plan that:
1. Understands the natural language command
2. Incorporates visual information about the environment
3. Considers the robot's physical capabilities and constraints
4. Creates a coherent sequence of actions
5. Handles potential ambiguities by leveraging multiple modalities
Return the plan as a JSON array of actions.
"""
response = self.llm_planner.client.chat.completions.create(
model=self.llm_planner.model,
messages=[
{"role": "system", "content": self._get_multimodal_system_prompt()},
{"role": "user", "content": prompt}
],
temperature=0.3,
response_format={"type": "json_object"}
)
plan_json = json.loads(response.choices[0].message.content)
return self.llm_planner._parse_plan_actions(plan_json)
def _get_multimodal_system_prompt(self) -> str:
"""Get system prompt for multi-modal planning"""
return """
You are a multi-modal cognitive planner for robotics, integrating language understanding,
visual perception, and physical action planning. Consider information from all modalities
when generating plans:
1. Language: Interpret the natural language command and extract semantic meaning
2. Vision: Use visual information about objects, locations, and obstacles
3. Action: Consider the robot's physical capabilities and environmental constraints
4. Integration: Create coherent plans that leverage multiple sources of information
Always generate safe, feasible, and efficient plans that accomplish the user's intent.
"""
Best Practices for LLM-Based Planning
1. Prompt Engineering for Robotics
- Be Specific: Include detailed environmental context and robot capabilities
- Use Examples: Provide examples of well-formed plans for few-shot learning
- Include Constraints: Explicitly mention safety, efficiency, and feasibility constraints
- Structure Responses: Request structured JSON responses for easy parsing
2. Error Handling and Validation
- Validate Plans: Check that generated plans are physically feasible
- Handle Ambiguity: Implement disambiguation strategies for unclear commands
- Provide Feedback: Give users clear feedback when plans cannot be executed
- Maintain State: Keep track of execution state for adaptive planning
3. Performance Optimization
- Caching: Cache results for common command patterns
- Hierarchical Planning: Use high-level templates for common tasks
- Parallel Processing: Process multiple planning requests concurrently
- Model Selection: Choose appropriate model size for real-time requirements
Troubleshooting Common Issues
1. Hallucination Problems
- Provide more specific context and constraints
- Use fact-checking steps in the planning process
- Implement validation steps before execution
- Use lower temperature settings for more consistent output
2. Planning Inconsistency
- Use structured prompts with clear formatting requirements
- Implement plan validation steps
- Maintain consistent context across planning sessions
- Use system prompts that emphasize consistency
3. Performance Issues
- Optimize prompts to reduce token usage
- Use smaller models for simple tasks
- Implement caching for common scenarios
- Consider edge deployment options for low-latency requirements
Exercises
- Implement a simple LLM-based planner for a specific robotic task
- Create a hierarchical planning system for complex multi-step tasks
- Develop an adaptive planning system that responds to environmental changes
- Integrate vision feedback into the planning process for object manipulation
Summary
LLM cognitive planning provides a powerful approach to bridging natural language commands with robotic action execution. By properly structuring prompts, maintaining context, and integrating with other modalities like vision, LLMs can serve as sophisticated cognitive planners that adapt to changing environments and handle complex, ambiguous commands. The key to success lies in careful prompt engineering, robust error handling, and seamless integration with the robot's perception and action systems.