290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""Intention System for BDI agents.
|
|
|
|
Intentions represent committed plans that the agent is executing.
|
|
Unlike desires (motivations) or goals (targets), intentions are
|
|
concrete action sequences the agent has decided to pursue.
|
|
|
|
Key concepts:
|
|
- Intention persistence: agents stick to plans unless interrupted
|
|
- Commitment strategies: different levels of plan commitment
|
|
- Plan monitoring: detecting when a plan becomes invalid
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Optional
|
|
|
|
if TYPE_CHECKING:
|
|
from backend.core.bdi.belief import BeliefBase
|
|
from backend.core.bdi.desire import DesireManager
|
|
from backend.core.goap.goal import Goal
|
|
from backend.core.goap.action import GOAPAction
|
|
from backend.core.goap.planner import Plan
|
|
|
|
|
|
class CommitmentStrategy(Enum):
|
|
"""How strongly an agent commits to their current intention."""
|
|
REACTIVE = "reactive" # Replan every turn (no commitment)
|
|
CAUTIOUS = "cautious" # Replan if priorities shift significantly
|
|
DETERMINED = "determined" # Stick to plan unless it becomes impossible
|
|
STUBBORN = "stubborn" # Only abandon for critical interrupts
|
|
|
|
|
|
@dataclass
|
|
class Intention:
|
|
"""A committed plan that the agent is executing.
|
|
|
|
Tracks the goal, plan, and execution progress.
|
|
"""
|
|
goal: "Goal" # The goal we're pursuing
|
|
plan: "Plan" # The plan to achieve it
|
|
start_turn: int # When we committed
|
|
actions_completed: int = 0 # How many actions done
|
|
last_action_success: bool = True # Did last action succeed?
|
|
consecutive_failures: int = 0 # Failures in a row
|
|
|
|
@property
|
|
def current_action(self) -> Optional["GOAPAction"]:
|
|
"""Get the next action to execute."""
|
|
if self.plan.is_empty:
|
|
return None
|
|
remaining = self.plan.actions[self.actions_completed:]
|
|
return remaining[0] if remaining else None
|
|
|
|
@property
|
|
def is_complete(self) -> bool:
|
|
"""Check if the plan has been fully executed."""
|
|
return self.actions_completed >= len(self.plan.actions)
|
|
|
|
@property
|
|
def remaining_actions(self) -> int:
|
|
"""Number of actions left in the plan."""
|
|
return max(0, len(self.plan.actions) - self.actions_completed)
|
|
|
|
def advance(self, success: bool) -> None:
|
|
"""Mark current action as executed and advance."""
|
|
self.last_action_success = success
|
|
if success:
|
|
self.actions_completed += 1
|
|
self.consecutive_failures = 0
|
|
else:
|
|
self.consecutive_failures += 1
|
|
|
|
|
|
class IntentionManager:
|
|
"""Manages the agent's current intention and commitment.
|
|
|
|
Responsibilities:
|
|
- Maintain the current intention (goal + plan)
|
|
- Decide when to continue vs. replan
|
|
- Handle plan failure and recovery
|
|
"""
|
|
|
|
def __init__(self, commitment_strategy: CommitmentStrategy = CommitmentStrategy.CAUTIOUS):
|
|
self.current_intention: Optional[Intention] = None
|
|
self.commitment_strategy = commitment_strategy
|
|
|
|
# Tracking
|
|
self.intentions_completed: int = 0
|
|
self.intentions_abandoned: int = 0
|
|
self.total_actions_executed: int = 0
|
|
|
|
# Thresholds for reconsideration
|
|
self.max_consecutive_failures: int = 2
|
|
self.priority_switch_threshold: float = 1.5 # New goal must be 1.5x priority
|
|
|
|
@classmethod
|
|
def from_personality(cls, personality) -> "IntentionManager":
|
|
"""Create an IntentionManager with commitment strategy based on personality."""
|
|
# Derive commitment from personality traits
|
|
# High hoarding + low risk tolerance = stubborn
|
|
commitment_score = (
|
|
personality.hoarding_rate * 0.4 +
|
|
(1.0 - personality.risk_tolerance) * 0.4 +
|
|
(1.0 - personality.market_affinity) * 0.2
|
|
)
|
|
|
|
if commitment_score > 0.7:
|
|
strategy = CommitmentStrategy.STUBBORN
|
|
elif commitment_score > 0.5:
|
|
strategy = CommitmentStrategy.DETERMINED
|
|
elif commitment_score > 0.3:
|
|
strategy = CommitmentStrategy.CAUTIOUS
|
|
else:
|
|
strategy = CommitmentStrategy.REACTIVE
|
|
|
|
return cls(commitment_strategy=strategy)
|
|
|
|
def has_intention(self) -> bool:
|
|
"""Check if we have an active intention."""
|
|
return (
|
|
self.current_intention is not None and
|
|
not self.current_intention.is_complete
|
|
)
|
|
|
|
def get_next_action(self) -> Optional["GOAPAction"]:
|
|
"""Get the next action from current intention."""
|
|
if not self.has_intention():
|
|
return None
|
|
return self.current_intention.current_action
|
|
|
|
def should_reconsider(
|
|
self,
|
|
beliefs: "BeliefBase",
|
|
desire_manager: "DesireManager",
|
|
available_goals: list["Goal"],
|
|
) -> bool:
|
|
"""Determine if we should reconsider our current intention.
|
|
|
|
This implements the commitment strategy logic.
|
|
"""
|
|
# No intention = definitely need to plan
|
|
if not self.has_intention():
|
|
return True
|
|
|
|
intention = self.current_intention
|
|
|
|
# Check for critical interrupts (always reconsider)
|
|
if beliefs.has_critical_need():
|
|
# But only if current intention isn't already addressing it
|
|
urgent_need = beliefs.get_most_urgent_need()
|
|
if not self._intention_addresses_need(intention, urgent_need):
|
|
return True
|
|
|
|
# Check for too many failures
|
|
if intention.consecutive_failures >= self.max_consecutive_failures:
|
|
return True
|
|
|
|
# Strategy-specific logic
|
|
if self.commitment_strategy == CommitmentStrategy.REACTIVE:
|
|
return True # Always replan
|
|
|
|
elif self.commitment_strategy == CommitmentStrategy.CAUTIOUS:
|
|
# Reconsider if a significantly better goal is available
|
|
return self._better_goal_available(
|
|
beliefs, desire_manager, available_goals,
|
|
threshold=self.priority_switch_threshold
|
|
)
|
|
|
|
elif self.commitment_strategy == CommitmentStrategy.DETERMINED:
|
|
# Only reconsider for much better goals or if plan is failing
|
|
return (
|
|
intention.consecutive_failures > 0 or
|
|
self._better_goal_available(
|
|
beliefs, desire_manager, available_goals,
|
|
threshold=2.0 # Need 2x priority to switch
|
|
)
|
|
)
|
|
|
|
elif self.commitment_strategy == CommitmentStrategy.STUBBORN:
|
|
# Only abandon for critical needs or impossible plans
|
|
return (
|
|
beliefs.has_critical_need() or
|
|
intention.consecutive_failures >= self.max_consecutive_failures
|
|
)
|
|
|
|
return False
|
|
|
|
def _intention_addresses_need(self, intention: Intention, need: str) -> bool:
|
|
"""Check if current intention addresses a vital need."""
|
|
if not intention or not need:
|
|
return False
|
|
|
|
goal_name = intention.goal.name.lower()
|
|
need_map = {
|
|
"thirst": "thirst",
|
|
"hunger": "hunger",
|
|
"heat": "heat",
|
|
"energy": "energy",
|
|
}
|
|
|
|
return need_map.get(need, "") in goal_name
|
|
|
|
def _better_goal_available(
|
|
self,
|
|
beliefs: "BeliefBase",
|
|
desire_manager: "DesireManager",
|
|
available_goals: list["Goal"],
|
|
threshold: float,
|
|
) -> bool:
|
|
"""Check if there's a significantly better goal available."""
|
|
if not self.has_intention():
|
|
return True
|
|
|
|
current_goal = self.current_intention.goal
|
|
world_state = beliefs.to_world_state()
|
|
current_priority = current_goal.get_priority(world_state)
|
|
|
|
# Get desire-filtered goals
|
|
filtered_goals = desire_manager.filter_goals_by_desire(available_goals, beliefs)
|
|
|
|
for goal in filtered_goals:
|
|
if goal.name == current_goal.name:
|
|
continue
|
|
|
|
goal_priority = goal.get_priority(world_state)
|
|
if goal_priority > current_priority * threshold:
|
|
return True
|
|
|
|
return False
|
|
|
|
def commit_to_plan(self, goal: "Goal", plan: "Plan", current_turn: int) -> None:
|
|
"""Commit to a new intention."""
|
|
# Track abandoned intention
|
|
if self.has_intention():
|
|
self.intentions_abandoned += 1
|
|
|
|
self.current_intention = Intention(
|
|
goal=goal,
|
|
plan=plan,
|
|
start_turn=current_turn,
|
|
)
|
|
|
|
def advance_intention(self, success: bool) -> None:
|
|
"""Record action execution and advance the intention."""
|
|
if not self.has_intention():
|
|
return
|
|
|
|
self.current_intention.advance(success)
|
|
self.total_actions_executed += 1
|
|
|
|
# Check if intention is complete
|
|
if self.current_intention.is_complete:
|
|
self.intentions_completed += 1
|
|
self.current_intention = None
|
|
|
|
def abandon_intention(self, reason: str = "unknown") -> None:
|
|
"""Abandon the current intention."""
|
|
if self.has_intention():
|
|
self.intentions_abandoned += 1
|
|
self.current_intention = None
|
|
|
|
def get_plan_progress(self) -> dict:
|
|
"""Get current plan execution progress."""
|
|
if not self.has_intention():
|
|
return {"has_intention": False}
|
|
|
|
intention = self.current_intention
|
|
return {
|
|
"has_intention": True,
|
|
"goal": intention.goal.name,
|
|
"actions_completed": intention.actions_completed,
|
|
"remaining_actions": intention.remaining_actions,
|
|
"consecutive_failures": intention.consecutive_failures,
|
|
"turns_active": 0, # Would need current_turn to calculate
|
|
}
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Convert to dictionary for debugging/logging."""
|
|
return {
|
|
"commitment_strategy": self.commitment_strategy.value,
|
|
"has_intention": self.has_intention(),
|
|
"current_goal": self.current_intention.goal.name if self.has_intention() else None,
|
|
"plan_progress": self.get_plan_progress(),
|
|
"stats": {
|
|
"intentions_completed": self.intentions_completed,
|
|
"intentions_abandoned": self.intentions_abandoned,
|
|
"total_actions": self.total_actions_executed,
|
|
},
|
|
}
|