LLM Planning Code Examples
1. Basic LLM Cognitive Planner
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
class BasicLLMPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
Initialize basic LLM planner
Args:
api_key (str): OpenAI API key
model (str): LLM model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
def plan_task(self, command: str, environment_context: Dict[str, Any]) -> List[RobotAction]:
"""
Plan a task based on natural language command and environment context
Args:
command (str): Natural language command
environment_context (dict): Current environment state
Returns:
List[RobotAction]: Sequence of actions to execute
"""
prompt = self._create_planning_prompt(command, environment_context)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self._get_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._parse_plan_actions(plan_json)
def _create_planning_prompt(self, command: str, environment_context: Dict[str, Any]) -> str:
"""Create a detailed prompt for task planning"""
return 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)}
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
}}
]
}}
"""
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
# Example usage
# planner = BasicLLMPlanner("your-api-key")
# environment = {
# "objects": [
# {"name": "red_cup", "type": "cup", "position": [1.0, 0.5, 0.0]},
# {"name": "wooden_table", "type": "table", "position": [1.0, 0.6, 0.0]}
# ],
# "robot_position": [0.0, 0.0, 0.0]
# }
# plan = planner.plan_task("Pick up the red cup and place it on the table", environment)
# for action in plan:
# print(f"- {action.description}")
2. Context-Aware LLM Planner
class ContextAwareLLMPlanner(BasicLLMPlanner):
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,
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:]
# Example usage
# context_planner = ContextAwareLLMPlanner("your-api-key")
# environment = {
# "objects": [{"name": "blue_bottle", "type": "bottle", "position": [1.5, 0.8, 0.0]}],
# "robot_position": [0.5, 0.5, 0.0]
# }
# previous = [RobotAction(Action.MOVE_TO, {"location": [1.0, 0.5, 0.0]}, "Moved to object")]
# plan = context_planner.plan_with_context(
# "Grasp the blue bottle",
# environment,
# previous_actions=previous
# )
3. Hierarchical LLM Planner
class HierarchicalLLMPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
Hierarchical planner that can plan at different levels of abstraction
Args:
api_key (str): OpenAI API key
model (str): LLM model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
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_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
return [0.0, 0.0, 0.0]
def _llm_plan_task(self, command: str, environment_context: Dict[str, Any]) -> List[RobotAction]:
"""Use LLM to plan an unknown task"""
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)}
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
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are an expert cognitive planner for robotics applications."},
{"role": "user", "content": prompt}
],
temperature=0.3,
response_format={"type": "json_object"}
)
plan_json = json.loads(response.choices[0].message.content)
return self._parse_plan_actions(plan_json)
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
# Example usage
# hierarchical_planner = HierarchicalLLMPlanner("your-api-key")
# environment = {
# "named_locations": {"kitchen_counter": [1.0, 0.5, 0.8]},
# "objects": [{"name": "coffee_mug", "type": "mug", "position": [1.0, 0.5, 0.8]}]
# }
# plan = hierarchical_planner.plan_high_level_task(
# "pick_and_place",
# {"target_object": "coffee_mug", "destination": "kitchen_counter"},
# environment
# )
4. Adaptive LLM Planner
class AdaptiveLLMPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
Planner that can adapt to uncertainty and feedback
Args:
api_key (str): OpenAI API key
model (str): LLM model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.current_plan = []
self.execution_history = []
self.uncertainty_model = self._initialize_uncertainty_model()
def _initialize_uncertainty_model(self):
"""Initialize model for handling uncertainty"""
# In practice, this would be a more sophisticated uncertainty quantification model
return {
"action_success_probabilities": {},
"environment_change_probabilities": {}
}
def execute_plan_with_adaptation(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
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._llm_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=getattr(Action, strategy) if isinstance(strategy, str) else 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
def _llm_plan_task(self, command: str, environment_context: Dict[str, Any]) -> List[RobotAction]:
"""Use LLM to plan a task"""
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)}
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
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are an expert cognitive planner for robotics applications."},
{"role": "user", "content": prompt}
],
temperature=0.3,
response_format={"type": "json_object"}
)
plan_json = json.loads(response.choices[0].message.content)
return self._parse_plan_actions(plan_json)
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
# Example usage
# adaptive_planner = AdaptiveLLMPlanner("your-api-key")
# environment = {
# "objects": [{"name": "red_ball", "type": "ball", "position": [0.5, 0.5, 0.0]}],
# "robot_position": [0.0, 0.0, 0.0]
# }
# plan = adaptive_planner._llm_plan_task("Pick up the red ball", environment)
# results = adaptive_planner.execute_plan_with_adaptation(plan, environment)
5. Multi-Modal LLM Planner
class MultiModalLLMPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
Multi-modal planner that integrates language, vision, and action planning
Args:
api_key (str): OpenAI API key
model (str): LLM model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
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_heavy_planning(command, unified_context)
elif planning_approach == "language_heavy":
return self._language_heavy_planning(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 _vision_heavy_planning(self, command: str, context: Dict[str, Any]) -> List[RobotAction]:
"""Plan using vision-heavy approach"""
# Emphasize visual information in the prompt
prompt = f"""
You are a vision-focused cognitive planner for a robot. Given a natural language command and rich visual context, generate a detailed plan.
Command: "{command}"
Visual Context: {json.dumps(context.get('vision_context', {}), indent=2)}
Language Context: {json.dumps(context.get('language_context', {}), indent=2)}
Generate a plan that prioritizes visual information while respecting the command.
"""
return self._generate_plan_from_prompt(prompt)
def _language_heavy_planning(self, command: str, context: Dict[str, Any]) -> List[RobotAction]:
"""Plan using language-heavy approach"""
# Emphasize language understanding in the prompt
prompt = f"""
You are a language-focused cognitive planner for a robot. Given a natural language command and context, generate a detailed plan.
Command: "{command}"
Language Context: {json.dumps(context.get('language_context', {}), indent=2)}
Visual Context: {json.dumps(context.get('vision_context', {}), indent=2)}
Generate a plan that prioritizes understanding the command's intent while considering the visual context.
"""
return self._generate_plan_from_prompt(prompt)
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.
"""
return self._generate_plan_from_prompt(prompt)
def _generate_plan_from_prompt(self, prompt: str) -> List[RobotAction]:
"""Generate a plan from a prompt"""
response = self.client.chat.completions.create(
model=self.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._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.
"""
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
# Example usage
# multimodal_planner = MultiModalLLMPlanner("your-api-key")
# vision_context = {
# "objects_detected": [
# {"name": "blue_cup", "type": "cup", "position": [1.0, 0.5, 0.8]},
# {"name": "wooden_table", "type": "table", "position": [1.0, 0.6, 0.0]}
# ]
# }
# language_context = {"command": "Pick up the blue cup"}
# plan = multimodal_planner.plan_multimodal_task(
# "Pick up the blue cup and place it on the table",
# language_context=language_context,
# vision_context=vision_context
# )
6. LLM Planning with Feedback and Learning
class LearningLLMPlanner:
def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
"""
LLM planner with learning capabilities from execution feedback
Args:
api_key (str): OpenAI API key
model (str): LLM model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.experience_memory = []
self.performance_metrics = {
"success_rate": 0.0,
"average_execution_time": 0.0,
"common_failures": {}
}
def plan_with_experience(self, command: str, environment_context: Dict[str, Any]) -> List[RobotAction]:
"""
Plan a task using accumulated experience
Args:
command (str): Natural language command
environment_context (dict): Current environment state
Returns:
List[RobotAction]: Sequence of actions to execute
"""
# Find similar past experiences
similar_experiences = self._find_similar_experiences(command, environment_context)
# Create prompt with experience context
prompt = self._create_experience_enhanced_prompt(
command,
environment_context,
similar_experiences
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self._get_learning_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._parse_plan_actions(plan_json)
def _find_similar_experiences(self, command: str, environment_context: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Find similar experiences from memory"""
# This is a simplified similarity check - in practice, use embeddings or semantic search
similar = []
for experience in self.experience_memory[-20:]: # Check last 20 experiences
if self._is_command_similar(command, experience["command"]):
similar.append(experience)
return similar
def _is_command_similar(self, cmd1: str, cmd2: str) -> bool:
"""Check if two commands are similar"""
# Simple keyword-based similarity
cmd1_lower = cmd1.lower()
cmd2_lower = cmd2.lower()
# Check for common action verbs
action_verbs = ["pick", "grasp", "move", "go", "navigate", "place", "put", "take"]
for verb in action_verbs:
if verb in cmd1_lower and verb in cmd2_lower:
return True
return False
def _create_experience_enhanced_prompt(self, command: str,
environment_context: Dict[str, Any],
similar_experiences: List[Dict[str, Any]]) -> str:
"""Create a prompt enhanced with experience information"""
experience_text = ""
if similar_experiences:
experience_text = "Similar past experiences:\n"
for i, exp in enumerate(similar_experiences[-3:]): # Use last 3 similar experiences
experience_text += f"Experience {i+1}: Command '{exp['command']}' -> "
experience_text += f"Success: {exp['success']}, Actions: {len(exp['plan'])}\n"
return f"""
You are a cognitive planner for a robot with learning capabilities. Use past experiences to inform your planning.
Current Command: "{command}"
Environment Context:
{json.dumps(environment_context, indent=2)}
{experience_text}
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
Consider what has worked well in similar situations in the past.
"""
def _get_learning_system_prompt(self) -> str:
"""System prompt for learning-enhanced planning"""
return """
You are an expert cognitive planner for robotics with learning capabilities.
Use past experiences to inform your planning decisions:
1. When you see similar commands, consider what worked well before
2. Adapt plans based on environmental similarities
3. Learn from past failures to avoid repeating mistakes
4. Consider the success patterns in your experience base
5. Generate plans that incorporate lessons learned from execution feedback
"""
def record_execution_result(self, command: str, plan: List[RobotAction],
execution_result: Dict[str, Any], success: bool):
"""Record the result of plan execution for learning"""
experience = {
"command": command,
"plan": [action.description for action in plan],
"plan_actions": plan, # Keep the full action objects
"execution_result": execution_result,
"success": success,
"timestamp": time.time(),
"environment_context": execution_result.get("environment_context", {})
}
self.experience_memory.append(experience)
# Update performance metrics
self._update_performance_metrics(experience)
# Keep only recent experiences to manage memory
if len(self.experience_memory) > 1000:
self.experience_memory = self.experience_memory[-1000:]
def _update_performance_metrics(self, experience: Dict[str, Any]):
"""Update performance metrics based on experience"""
# Update success rate
successful_experiences = [exp for exp in self.experience_memory if exp["success"]]
self.performance_metrics["success_rate"] = len(successful_experiences) / len(self.experience_memory) if self.experience_memory else 0
# Update common failures
if not experience["success"]:
failure_reason = experience["execution_result"].get("error", "unknown")
if failure_reason in self.performance_metrics["common_failures"]:
self.performance_metrics["common_failures"][failure_reason] += 1
else:
self.performance_metrics["common_failures"][failure_reason] = 1
def get_performance_report(self) -> Dict[str, Any]:
"""Get a report on planner performance"""
return {
"total_experiences": len(self.experience_memory),
"success_rate": self.performance_metrics["success_rate"],
"common_failures": self.performance_metrics["common_failures"],
"learning_effectiveness": self._calculate_learning_effectiveness()
}
def _calculate_learning_effectiveness(self) -> float:
"""Calculate how effective the learning is"""
if len(self.experience_memory) < 2:
return 0.0
# Compare success rate of recent experiences vs early experiences
mid_point = len(self.experience_memory) // 2
early_experiences = self.experience_memory[:mid_point]
recent_experiences = self.experience_memory[mid_point:]
early_success_rate = sum(1 for exp in early_experiences if exp["success"]) / len(early_experiences) if early_experiences else 0
recent_success_rate = sum(1 for exp in recent_experiences if exp["success"]) / len(recent_experiences) if recent_experiences else 0
# Effectiveness is improvement in success rate
return max(0, recent_success_rate - early_success_rate)
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
# Example usage
# learning_planner = LearningLLMPlanner("your-api-key")
# environment = {"objects": [{"name": "green_bottle", "type": "bottle", "position": [1.2, 0.7, 0.0]}]}
# plan = learning_planner.plan_with_experience("Pick up the green bottle", environment)
# Simulate execution and record result
# execution_result = {"success": True, "actions_completed": len(plan)}
# learning_planner.record_execution_result("Pick up the green bottle", plan, execution_result, True)
# Get performance report
# report = learning_planner.get_performance_report()
# print(f"Success rate: {report['success_rate']:.2f}")
# print(f"Learning effectiveness: {report['learning_effectiveness']:.2f}")
These examples demonstrate various approaches to LLM-based planning for robotics:
- Basic LLM Planner: Simple task planning with natural language understanding
- Context-Aware Planner: Planning that considers historical context and memory
- Hierarchical Planner: Planning using predefined task templates and abstraction levels
- Adaptive Planner: Planning that can adjust to changing conditions and failures
- Multi-Modal Planner: Planning that integrates information from multiple sources (language, vision, etc.)
- Learning Planner: Planning that improves over time based on execution feedback
Each example builds on the previous ones, showing how to incorporate more sophisticated capabilities like context awareness, adaptation, multi-modal integration, and learning from experience.