302 lines
9.1 KiB
Python
302 lines
9.1 KiB
Python
"""AI decision system for agents in the Village Simulation.
|
|
|
|
This module provides the main entry point for AI decisions, supporting
|
|
both GOAP (Goal-Oriented Action Planning) and BDI (Belief-Desire-Intention)
|
|
reasoning systems.
|
|
|
|
AI System Options:
|
|
- GOAP: Fast, reactive planning with goal prioritization
|
|
- BDI: Persistent beliefs, long-term desires, plan commitment
|
|
|
|
Configure via config.json "ai.use_bdi" (default: false for backward compatibility)
|
|
|
|
Major features:
|
|
- Each agent has unique personality traits affecting all decisions
|
|
- Emergent professions: Hunters, Gatherers, Traders, Generalists
|
|
- Class inequality through varied strategies and skills
|
|
- Traders focus on arbitrage (buy low, sell high)
|
|
- Personality affects: risk tolerance, hoarding, market participation
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, TYPE_CHECKING
|
|
|
|
from backend.domain.agent import Agent
|
|
from backend.domain.action import ActionType, ACTION_CONFIG
|
|
from backend.domain.resources import ResourceType
|
|
|
|
if TYPE_CHECKING:
|
|
from backend.core.market import OrderBook
|
|
|
|
|
|
@dataclass
|
|
class TradeItem:
|
|
"""A single item to buy/sell in a trade."""
|
|
order_id: str
|
|
resource_type: ResourceType
|
|
quantity: int
|
|
price_per_unit: int
|
|
|
|
|
|
@dataclass
|
|
class AIDecision:
|
|
"""A decision made by the AI for an agent."""
|
|
action: ActionType
|
|
target_resource: Optional[ResourceType] = None
|
|
order_id: Optional[str] = None
|
|
quantity: int = 1
|
|
price: int = 0
|
|
reason: str = ""
|
|
# For multi-item trades
|
|
trade_items: list[TradeItem] = field(default_factory=list)
|
|
# For price adjustments
|
|
adjust_order_id: Optional[str] = None
|
|
new_price: Optional[int] = None
|
|
# GOAP/BDI fields
|
|
goal_name: str = ""
|
|
plan_length: int = 0
|
|
bdi_info: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"action": self.action.value,
|
|
"target_resource": self.target_resource.value if self.target_resource else None,
|
|
"order_id": self.order_id,
|
|
"quantity": self.quantity,
|
|
"price": self.price,
|
|
"reason": self.reason,
|
|
"trade_items": [
|
|
{
|
|
"order_id": t.order_id,
|
|
"resource_type": t.resource_type.value,
|
|
"quantity": t.quantity,
|
|
"price_per_unit": t.price_per_unit,
|
|
}
|
|
for t in self.trade_items
|
|
],
|
|
"adjust_order_id": self.adjust_order_id,
|
|
"new_price": self.new_price,
|
|
"goal_name": self.goal_name,
|
|
"plan_length": self.plan_length,
|
|
"bdi_info": self.bdi_info,
|
|
}
|
|
|
|
|
|
# Resource to action for gathering
|
|
RESOURCE_ACTIONS: dict[ResourceType, ActionType] = {
|
|
ResourceType.MEAT: ActionType.HUNT,
|
|
ResourceType.BERRIES: ActionType.GATHER,
|
|
ResourceType.WATER: ActionType.GET_WATER,
|
|
ResourceType.WOOD: ActionType.CHOP_WOOD,
|
|
ResourceType.HIDE: ActionType.HUNT,
|
|
ResourceType.CLOTHES: ActionType.WEAVE,
|
|
}
|
|
|
|
# Energy cost to gather each resource (used for efficiency calculations)
|
|
def get_energy_cost(resource_type: ResourceType) -> int:
|
|
"""Get the energy cost to produce one unit of a resource."""
|
|
action = RESOURCE_ACTIONS.get(resource_type)
|
|
if not action:
|
|
return 10
|
|
config = ACTION_CONFIG.get(action)
|
|
if not config:
|
|
return 10
|
|
energy_cost = abs(config.energy_cost)
|
|
avg_output = max(1, (config.min_output + config.max_output) / 2) if config.output_resource else 1
|
|
return int(energy_cost / avg_output)
|
|
|
|
|
|
# Cached config values to avoid repeated lookups
|
|
_cached_ai_config = None
|
|
_cached_economy_config = None
|
|
_cached_use_bdi = None
|
|
|
|
|
|
def _get_ai_config():
|
|
"""Get AI-relevant configuration values (cached)."""
|
|
global _cached_ai_config
|
|
if _cached_ai_config is None:
|
|
from backend.config import get_config
|
|
_cached_ai_config = get_config().agent_stats
|
|
return _cached_ai_config
|
|
|
|
|
|
def _get_economy_config():
|
|
"""Get economy/market configuration values (cached)."""
|
|
global _cached_economy_config
|
|
if _cached_economy_config is None:
|
|
from backend.config import get_config
|
|
_cached_economy_config = getattr(get_config(), 'economy', None)
|
|
return _cached_economy_config
|
|
|
|
|
|
def _should_use_bdi() -> bool:
|
|
"""Check if BDI should be used (cached)."""
|
|
global _cached_use_bdi
|
|
if _cached_use_bdi is None:
|
|
from backend.config import get_config
|
|
config = get_config()
|
|
ai_config = getattr(config, 'ai', None)
|
|
_cached_use_bdi = getattr(ai_config, 'use_bdi', False) if ai_config else False
|
|
return _cached_use_bdi
|
|
|
|
|
|
def reset_ai_config_cache():
|
|
"""Reset the cached config values (call after config reload)."""
|
|
global _cached_ai_config, _cached_economy_config, _cached_use_bdi
|
|
_cached_ai_config = None
|
|
_cached_economy_config = None
|
|
_cached_use_bdi = None
|
|
|
|
# Also reset BDI state if it was being used
|
|
try:
|
|
from backend.core.bdi import reset_bdi_state
|
|
reset_bdi_state()
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
def get_ai_decision(
|
|
agent: Agent,
|
|
market: "OrderBook",
|
|
step_in_day: int = 1,
|
|
day_steps: int = 10,
|
|
current_turn: int = 0,
|
|
is_night: bool = False,
|
|
) -> AIDecision:
|
|
"""Get an AI decision for an agent.
|
|
|
|
Uses either GOAP or BDI based on config setting "ai.use_bdi".
|
|
|
|
Args:
|
|
agent: The agent to make a decision for
|
|
market: The market order book
|
|
step_in_day: Current step within the day
|
|
day_steps: Total steps per day
|
|
current_turn: Current simulation turn
|
|
is_night: Whether it's currently night time
|
|
|
|
Returns:
|
|
AIDecision with the chosen action and parameters
|
|
"""
|
|
if _should_use_bdi():
|
|
return _get_bdi_decision(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
current_turn=current_turn,
|
|
is_night=is_night,
|
|
)
|
|
else:
|
|
return _get_goap_decision(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
current_turn=current_turn,
|
|
is_night=is_night,
|
|
)
|
|
|
|
|
|
def _get_goap_decision(
|
|
agent: Agent,
|
|
market: "OrderBook",
|
|
step_in_day: int,
|
|
day_steps: int,
|
|
current_turn: int,
|
|
is_night: bool,
|
|
) -> AIDecision:
|
|
"""Get an AI decision using GOAP (Goal-Oriented Action Planning)."""
|
|
from backend.core.goap.goap_ai import get_goap_decision
|
|
|
|
goap_decision = get_goap_decision(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
current_turn=current_turn,
|
|
is_night=is_night,
|
|
)
|
|
|
|
# Convert GOAP AIDecision to our AIDecision (they should be compatible)
|
|
return AIDecision(
|
|
action=goap_decision.action,
|
|
target_resource=goap_decision.target_resource,
|
|
order_id=goap_decision.order_id,
|
|
quantity=goap_decision.quantity,
|
|
price=goap_decision.price,
|
|
reason=goap_decision.reason,
|
|
trade_items=[
|
|
TradeItem(
|
|
order_id=t.order_id,
|
|
resource_type=t.resource_type,
|
|
quantity=t.quantity,
|
|
price_per_unit=t.price_per_unit,
|
|
)
|
|
for t in goap_decision.trade_items
|
|
],
|
|
adjust_order_id=goap_decision.adjust_order_id,
|
|
new_price=goap_decision.new_price,
|
|
goal_name=goap_decision.goal_name,
|
|
plan_length=goap_decision.plan_length,
|
|
)
|
|
|
|
|
|
def _get_bdi_decision(
|
|
agent: Agent,
|
|
market: "OrderBook",
|
|
step_in_day: int,
|
|
day_steps: int,
|
|
current_turn: int,
|
|
is_night: bool,
|
|
) -> AIDecision:
|
|
"""Get an AI decision using BDI (Belief-Desire-Intention)."""
|
|
from backend.core.bdi import get_bdi_decision
|
|
|
|
bdi_decision = get_bdi_decision(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
current_turn=current_turn,
|
|
is_night=is_night,
|
|
)
|
|
|
|
# Convert BDI AIDecision to our AIDecision
|
|
return AIDecision(
|
|
action=bdi_decision.action,
|
|
target_resource=bdi_decision.target_resource,
|
|
order_id=bdi_decision.order_id,
|
|
quantity=bdi_decision.quantity,
|
|
price=bdi_decision.price,
|
|
reason=bdi_decision.reason,
|
|
trade_items=[
|
|
TradeItem(
|
|
order_id=t.order_id,
|
|
resource_type=t.resource_type,
|
|
quantity=t.quantity,
|
|
price_per_unit=t.price_per_unit,
|
|
)
|
|
for t in bdi_decision.trade_items
|
|
],
|
|
adjust_order_id=bdi_decision.adjust_order_id,
|
|
new_price=bdi_decision.new_price,
|
|
goal_name=bdi_decision.goal_name,
|
|
plan_length=bdi_decision.plan_length,
|
|
bdi_info=bdi_decision.bdi_info,
|
|
)
|
|
|
|
|
|
def on_agent_death(agent_id: str) -> None:
|
|
"""Clean up AI state when an agent dies.
|
|
|
|
Call this from the engine when an agent is removed.
|
|
"""
|
|
if _should_use_bdi():
|
|
try:
|
|
from backend.core.bdi import remove_agent_bdi_state
|
|
remove_agent_bdi_state(agent_id)
|
|
except ImportError:
|
|
pass
|