"""Belief System for BDI agents. The BeliefBase maintains persistent state about the world from the agent's perspective, including: - Current sensory data (vitals, inventory, market) - Memory of past events (failed trades, good hunting spots) - Dirty flags for efficient updates This replaces the transient WorldState creation with a persistent belief system. """ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Optional from collections import deque if TYPE_CHECKING: from backend.domain.agent import Agent from backend.core.market import OrderBook @dataclass class MemoryEvent: """A remembered event that may influence future decisions.""" event_type: str # "trade_failed", "hunt_failed", "good_deal", etc. turn: int data: dict = field(default_factory=dict) relevance: float = 1.0 # Decays over time @dataclass class BeliefBase: """Persistent belief system for a BDI agent. Maintains both current perceptions and memories of past events. Uses dirty flags to avoid unnecessary recomputation. """ # Current perception state (cached WorldState fields) thirst_pct: float = 1.0 hunger_pct: float = 1.0 heat_pct: float = 1.0 energy_pct: float = 1.0 # Resource counts water_count: int = 0 food_count: int = 0 meat_count: int = 0 berries_count: int = 0 wood_count: int = 0 hide_count: int = 0 # Inventory state has_clothes: bool = False inventory_space: int = 0 inventory_full: bool = False # Economic state money: int = 0 is_wealthy: bool = False # Market beliefs (what we believe about market availability) can_buy_water: bool = False can_buy_food: bool = False can_buy_meat: bool = False can_buy_berries: bool = False can_buy_wood: bool = False water_market_price: int = 0 food_market_price: int = 0 wood_market_price: int = 0 # Time beliefs is_night: bool = False is_evening: bool = False step_in_day: int = 0 day_steps: int = 10 current_turn: int = 0 # Personality (cached from agent, rarely changes) wealth_desire: float = 0.5 hoarding_rate: float = 0.5 risk_tolerance: float = 0.5 market_affinity: float = 0.5 is_trader: bool = False gather_preference: float = 1.0 hunt_preference: float = 1.0 trade_preference: float = 1.0 # Skills hunting_skill: float = 1.0 gathering_skill: float = 1.0 trading_skill: float = 1.0 # Thresholds (from config) critical_threshold: float = 0.25 low_threshold: float = 0.45 # Calculated urgencies thirst_urgency: float = 0.0 hunger_urgency: float = 0.0 heat_urgency: float = 0.0 energy_urgency: float = 0.0 # === BDI Extensions === # Memory system - stores past events that influence decisions memories: deque = field(default_factory=lambda: deque(maxlen=50)) # Track failed actions for this resource type (turn -> count) failed_hunts: int = 0 failed_gathers: int = 0 failed_trades: int = 0 # Track successful trades with agents (agent_id -> positive count) trusted_traders: dict = field(default_factory=dict) # Track failed trades with agents (agent_id -> negative count) distrusted_traders: dict = field(default_factory=dict) # Dirty flags for optimization _vitals_dirty: bool = True _inventory_dirty: bool = True _market_dirty: bool = True _last_update_turn: int = -1 def update_from_sensors( self, agent: "Agent", market: "OrderBook", step_in_day: int = 1, day_steps: int = 10, current_turn: int = 0, is_night: bool = False, ) -> None: """Update beliefs from current agent and market state. Uses dirty flags to skip unchanged data. """ from backend.domain.resources import ResourceType from backend.config import get_config self.current_turn = current_turn self.step_in_day = step_in_day self.day_steps = day_steps self.is_night = is_night self.is_evening = step_in_day >= day_steps - 2 # Always update vitals (they change every turn) stats = agent.stats self.thirst_pct = stats.thirst / stats.MAX_THIRST self.hunger_pct = stats.hunger / stats.MAX_HUNGER self.heat_pct = stats.heat / stats.MAX_HEAT self.energy_pct = stats.energy / stats.MAX_ENERGY # Update inventory self.water_count = agent.get_resource_count(ResourceType.WATER) self.meat_count = agent.get_resource_count(ResourceType.MEAT) self.berries_count = agent.get_resource_count(ResourceType.BERRIES) self.wood_count = agent.get_resource_count(ResourceType.WOOD) self.hide_count = agent.get_resource_count(ResourceType.HIDE) self.food_count = self.meat_count + self.berries_count self.has_clothes = agent.has_clothes() self.inventory_space = agent.inventory_space() self.inventory_full = agent.inventory_full() self.money = agent.money # Update personality (cached, rarely changes) self.wealth_desire = agent.personality.wealth_desire self.hoarding_rate = agent.personality.hoarding_rate self.risk_tolerance = agent.personality.risk_tolerance self.market_affinity = agent.personality.market_affinity self.gather_preference = agent.personality.gather_preference self.hunt_preference = agent.personality.hunt_preference self.trade_preference = agent.personality.trade_preference # Skills self.hunting_skill = agent.skills.hunting self.gathering_skill = agent.skills.gathering self.trading_skill = agent.skills.trading # Market availability def get_market_info(resource_type: ResourceType) -> tuple[bool, int]: order = market.get_cheapest_order(resource_type) if order and order.seller_id != agent.id and agent.money >= order.price_per_unit: return True, order.price_per_unit return False, 0 self.can_buy_water, self.water_market_price = get_market_info(ResourceType.WATER) self.can_buy_meat, meat_price = get_market_info(ResourceType.MEAT) self.can_buy_berries, berries_price = get_market_info(ResourceType.BERRIES) self.can_buy_wood, self.wood_market_price = get_market_info(ResourceType.WOOD) self.can_buy_food = self.can_buy_meat or self.can_buy_berries food_price = min( meat_price if self.can_buy_meat else float('inf'), berries_price if self.can_buy_berries else float('inf') ) self.food_market_price = int(food_price) if food_price != float('inf') else 0 # Wealth calculation config = get_config() economy_config = getattr(config, 'economy', None) min_wealth_target = getattr(economy_config, 'min_wealth_target', 5000) if economy_config else 5000 wealth_target = int(min_wealth_target * (0.5 + self.wealth_desire)) self.is_wealthy = self.money >= wealth_target # Trader check self.is_trader = self.trade_preference > 1.3 and self.market_affinity > 0.5 # Config thresholds agent_config = config.agent_stats self.critical_threshold = agent_config.critical_threshold self.low_threshold = 0.45 # Calculate urgencies self._calculate_urgencies() # Decay old memories self._decay_memories() self._last_update_turn = current_turn def _calculate_urgencies(self) -> None: """Calculate urgency values for each vital stat.""" def calc_urgency(pct: float, critical: float, low: float) -> float: if pct >= low: return 0.0 elif pct >= critical: return 1.0 - (pct - critical) / (low - critical) else: return 1.0 + (critical - pct) / critical * 2.0 self.thirst_urgency = calc_urgency(self.thirst_pct, self.critical_threshold, self.low_threshold) self.hunger_urgency = calc_urgency(self.hunger_pct, self.critical_threshold, self.low_threshold) self.heat_urgency = calc_urgency(self.heat_pct, self.critical_threshold, self.low_threshold) if self.energy_pct < 0.25: self.energy_urgency = 2.0 elif self.energy_pct < 0.40: self.energy_urgency = 1.0 else: self.energy_urgency = 0.0 def _decay_memories(self) -> None: """Decay memory relevance over time.""" for memory in self.memories: age = self.current_turn - memory.turn # Memories decay exponentially with age memory.relevance = max(0.1, 1.0 / (1.0 + age * 0.1)) def add_memory(self, event_type: str, data: dict = None) -> None: """Add a new memory event.""" self.memories.append(MemoryEvent( event_type=event_type, turn=self.current_turn, data=data or {}, relevance=1.0, )) def record_failed_action(self, action_type: str) -> None: """Record a failed action for learning.""" if action_type == "hunt": self.failed_hunts += 1 self.add_memory("hunt_failed") elif action_type == "gather": self.failed_gathers += 1 self.add_memory("gather_failed") elif action_type == "trade": self.failed_trades += 1 self.add_memory("trade_failed") def record_successful_action(self, action_type: str) -> None: """Record a successful action, reducing failure counts.""" if action_type == "hunt": self.failed_hunts = max(0, self.failed_hunts - 1) elif action_type == "gather": self.failed_gathers = max(0, self.failed_gathers - 1) elif action_type == "trade": self.failed_trades = max(0, self.failed_trades - 1) def record_trade_partner(self, partner_id: str, success: bool) -> None: """Track trade relationship with another agent.""" if success: self.trusted_traders[partner_id] = self.trusted_traders.get(partner_id, 0) + 1 # Reduce distrust if they've been bad before if partner_id in self.distrusted_traders: self.distrusted_traders[partner_id] = max(0, self.distrusted_traders[partner_id] - 1) else: self.distrusted_traders[partner_id] = self.distrusted_traders.get(partner_id, 0) + 1 def get_trade_trust(self, partner_id: str) -> float: """Get trust level for a trade partner (-1 to 1).""" trust = self.trusted_traders.get(partner_id, 0) distrust = self.distrusted_traders.get(partner_id, 0) total = trust + distrust if total == 0: return 0.0 # Unknown partner return (trust - distrust) / total def has_critical_need(self) -> bool: """Check if any vital stat is critical (requires immediate attention).""" return ( self.thirst_urgency >= 2.0 or self.hunger_urgency >= 2.0 or self.heat_urgency >= 2.0 or self.energy_urgency >= 2.0 ) def get_most_urgent_need(self) -> Optional[str]: """Get the most urgent vital need, if any.""" urgencies = { "thirst": self.thirst_urgency, "hunger": self.hunger_urgency, "heat": self.heat_urgency, "energy": self.energy_urgency, } max_urgency = max(urgencies.values()) if max_urgency < 0.5: return None return max(urgencies, key=urgencies.get) def to_world_state(self): """Convert beliefs to a WorldState for GOAP planner compatibility.""" from backend.core.goap.world_state import WorldState return WorldState( thirst_pct=self.thirst_pct, hunger_pct=self.hunger_pct, heat_pct=self.heat_pct, energy_pct=self.energy_pct, water_count=self.water_count, food_count=self.food_count, meat_count=self.meat_count, berries_count=self.berries_count, wood_count=self.wood_count, hide_count=self.hide_count, has_clothes=self.has_clothes, inventory_space=self.inventory_space, inventory_full=self.inventory_full, money=self.money, is_wealthy=self.is_wealthy, can_buy_water=self.can_buy_water, can_buy_food=self.can_buy_food, can_buy_meat=self.can_buy_meat, can_buy_berries=self.can_buy_berries, can_buy_wood=self.can_buy_wood, water_market_price=self.water_market_price, food_market_price=self.food_market_price, wood_market_price=self.wood_market_price, is_night=self.is_night, is_evening=self.is_evening, step_in_day=self.step_in_day, day_steps=self.day_steps, wealth_desire=self.wealth_desire, hoarding_rate=self.hoarding_rate, risk_tolerance=self.risk_tolerance, market_affinity=self.market_affinity, is_trader=self.is_trader, gather_preference=self.gather_preference, hunt_preference=self.hunt_preference, trade_preference=self.trade_preference, hunting_skill=self.hunting_skill, gathering_skill=self.gathering_skill, trading_skill=self.trading_skill, critical_threshold=self.critical_threshold, low_threshold=self.low_threshold, ) def to_dict(self) -> dict: """Convert to dictionary for debugging/logging.""" return { "vitals": { "thirst": round(self.thirst_pct, 2), "hunger": round(self.hunger_pct, 2), "heat": round(self.heat_pct, 2), "energy": round(self.energy_pct, 2), }, "urgencies": { "thirst": round(self.thirst_urgency, 2), "hunger": round(self.hunger_urgency, 2), "heat": round(self.heat_urgency, 2), "energy": round(self.energy_urgency, 2), }, "inventory": { "water": self.water_count, "meat": self.meat_count, "berries": self.berries_count, "wood": self.wood_count, "hide": self.hide_count, "space": self.inventory_space, }, "economy": { "money": self.money, "is_wealthy": self.is_wealthy, }, "memory": { "failed_hunts": self.failed_hunts, "failed_gathers": self.failed_gathers, "failed_trades": self.failed_trades, "memory_count": len(self.memories), }, }