412 lines
13 KiB
Python
412 lines
13 KiB
Python
"""GOAP-based AI decision system for agents.
|
|
|
|
This module provides the main interface for GOAP-based decision making
|
|
using Goal-Oriented Action Planning.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, TYPE_CHECKING
|
|
|
|
from backend.domain.action import ActionType
|
|
from backend.domain.resources import ResourceType
|
|
from backend.domain.personality import get_trade_price_modifier
|
|
|
|
from .world_state import WorldState, create_world_state
|
|
from .goal import Goal
|
|
from .action import GOAPAction
|
|
from .planner import GOAPPlanner, ReactivePlanner, Plan
|
|
from .goals import get_all_goals
|
|
from .actions import get_all_actions
|
|
|
|
if TYPE_CHECKING:
|
|
from backend.domain.agent import Agent
|
|
from backend.core.market import OrderBook
|
|
from backend.core.ai import AIDecision, TradeItem
|
|
|
|
|
|
@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 = ""
|
|
trade_items: list[TradeItem] = field(default_factory=list)
|
|
adjust_order_id: Optional[str] = None
|
|
new_price: Optional[int] = None
|
|
|
|
# GOAP-specific fields
|
|
goal_name: str = ""
|
|
plan_length: int = 0
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
class GOAPAgentAI:
|
|
"""GOAP-based AI decision maker for agents.
|
|
|
|
This uses goal-oriented action planning to select actions:
|
|
1. Build world state from agent and market
|
|
2. Evaluate all goals and their priorities
|
|
3. Use planner to find action sequence for best goal
|
|
4. Return the first action as the decision
|
|
|
|
Falls back to reactive planning for simple decisions.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
agent: "Agent",
|
|
market: "OrderBook",
|
|
step_in_day: int = 1,
|
|
day_steps: int = 10,
|
|
current_turn: int = 0,
|
|
is_night: bool = False,
|
|
):
|
|
self.agent = agent
|
|
self.market = market
|
|
self.step_in_day = step_in_day
|
|
self.day_steps = day_steps
|
|
self.current_turn = current_turn
|
|
self.is_night = is_night
|
|
|
|
# Build world state
|
|
self.state = create_world_state(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
is_night=is_night,
|
|
)
|
|
|
|
# Initialize planners
|
|
self.planner = GOAPPlanner(max_iterations=50)
|
|
self.reactive_planner = ReactivePlanner()
|
|
|
|
# Get available goals and actions
|
|
self.goals = get_all_goals()
|
|
self.actions = get_all_actions()
|
|
|
|
# Personality shortcuts
|
|
self.p = agent.personality
|
|
self.skills = agent.skills
|
|
|
|
def decide(self) -> AIDecision:
|
|
"""Make a decision using GOAP planning.
|
|
|
|
Decision flow:
|
|
1. Force sleep if night
|
|
2. Try to find a plan for the highest priority goal
|
|
3. If no plan found, use reactive selection
|
|
4. Convert GOAP action to AIDecision with proper parameters
|
|
"""
|
|
# Night time - mandatory sleep
|
|
if self.is_night:
|
|
return AIDecision(
|
|
action=ActionType.SLEEP,
|
|
reason="Night time: sleeping",
|
|
goal_name="Sleep",
|
|
)
|
|
|
|
# Try GOAP planning
|
|
plan = self.planner.plan_for_goals(
|
|
initial_state=self.state,
|
|
goals=self.goals,
|
|
available_actions=self.actions,
|
|
)
|
|
|
|
if plan and not plan.is_empty:
|
|
# We have a plan - execute first action
|
|
goap_action = plan.first_action
|
|
return self._convert_to_decision(
|
|
goap_action=goap_action,
|
|
goal=plan.goal,
|
|
plan=plan,
|
|
)
|
|
|
|
# Fallback to reactive selection
|
|
best_action = self.reactive_planner.select_best_action(
|
|
state=self.state,
|
|
goals=self.goals,
|
|
available_actions=self.actions,
|
|
)
|
|
|
|
if best_action:
|
|
return self._convert_to_decision(
|
|
goap_action=best_action,
|
|
goal=None,
|
|
plan=None,
|
|
)
|
|
|
|
# Ultimate fallback - rest
|
|
return AIDecision(
|
|
action=ActionType.REST,
|
|
reason="No valid action found, resting",
|
|
)
|
|
|
|
def _convert_to_decision(
|
|
self,
|
|
goap_action: GOAPAction,
|
|
goal: Optional[Goal],
|
|
plan: Optional[Plan],
|
|
) -> AIDecision:
|
|
"""Convert a GOAP action to an AIDecision with proper parameters.
|
|
|
|
This handles the translation from abstract GOAP actions to
|
|
concrete decisions with order IDs, prices, etc.
|
|
"""
|
|
action_type = goap_action.action_type
|
|
target_resource = goap_action.target_resource
|
|
|
|
# Build reason string
|
|
if goal:
|
|
reason = f"{goal.name}: {goap_action.name}"
|
|
else:
|
|
reason = f"Reactive: {goap_action.name}"
|
|
|
|
# Handle different action types
|
|
if action_type == ActionType.CONSUME:
|
|
return AIDecision(
|
|
action=action_type,
|
|
target_resource=target_resource,
|
|
reason=reason,
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
elif action_type == ActionType.TRADE:
|
|
return self._create_trade_decision(goap_action, goal, plan, reason)
|
|
|
|
elif action_type in [ActionType.HUNT, ActionType.GATHER, ActionType.CHOP_WOOD,
|
|
ActionType.GET_WATER, ActionType.WEAVE]:
|
|
return AIDecision(
|
|
action=action_type,
|
|
target_resource=target_resource,
|
|
reason=reason,
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
elif action_type == ActionType.BUILD_FIRE:
|
|
return AIDecision(
|
|
action=action_type,
|
|
target_resource=ResourceType.WOOD,
|
|
reason=reason,
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
elif action_type in [ActionType.REST, ActionType.SLEEP]:
|
|
return AIDecision(
|
|
action=action_type,
|
|
reason=reason,
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
# Default case
|
|
return AIDecision(
|
|
action=action_type,
|
|
target_resource=target_resource,
|
|
reason=reason,
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
def _create_trade_decision(
|
|
self,
|
|
goap_action: GOAPAction,
|
|
goal: Optional[Goal],
|
|
plan: Optional[Plan],
|
|
reason: str,
|
|
) -> AIDecision:
|
|
"""Create a trade decision with actual market parameters.
|
|
|
|
This translates abstract "Buy X" or "Sell X" actions into
|
|
concrete decisions with order IDs, prices, and quantities.
|
|
"""
|
|
target_resource = goap_action.target_resource
|
|
action_name = goap_action.name.lower()
|
|
|
|
if "buy" in action_name:
|
|
# Find the best order to buy from
|
|
order = self.market.get_cheapest_order(target_resource)
|
|
|
|
if order and order.seller_id != self.agent.id:
|
|
# Calculate quantity to buy
|
|
can_afford = self.agent.money // max(1, order.price_per_unit)
|
|
space = self.agent.inventory_space()
|
|
quantity = min(2, can_afford, space, order.quantity)
|
|
|
|
if quantity > 0:
|
|
return AIDecision(
|
|
action=ActionType.TRADE,
|
|
target_resource=target_resource,
|
|
order_id=order.id,
|
|
quantity=quantity,
|
|
price=order.price_per_unit,
|
|
reason=f"{reason} @ {order.price_per_unit}c",
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
# Can't buy - fallback to gathering
|
|
return self._create_gather_fallback(target_resource, reason, goal, plan)
|
|
|
|
elif "sell" in action_name:
|
|
# Create a sell order
|
|
quantity_available = self.agent.get_resource_count(target_resource)
|
|
|
|
# Calculate minimum to keep
|
|
min_keep = self._get_min_keep(target_resource)
|
|
quantity_to_sell = min(3, quantity_available - min_keep)
|
|
|
|
if quantity_to_sell > 0:
|
|
price = self._calculate_sell_price(target_resource)
|
|
|
|
return AIDecision(
|
|
action=ActionType.TRADE,
|
|
target_resource=target_resource,
|
|
quantity=quantity_to_sell,
|
|
price=price,
|
|
reason=f"{reason} @ {price}c",
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
# Invalid trade action - rest
|
|
return AIDecision(
|
|
action=ActionType.REST,
|
|
reason="Trade not possible",
|
|
)
|
|
|
|
def _create_gather_fallback(
|
|
self,
|
|
resource_type: ResourceType,
|
|
reason: str,
|
|
goal: Optional[Goal],
|
|
plan: Optional[Plan],
|
|
) -> AIDecision:
|
|
"""Create a gather action as fallback when buying isn't possible."""
|
|
# Map resource to gather action
|
|
action_map = {
|
|
ResourceType.WATER: ActionType.GET_WATER,
|
|
ResourceType.BERRIES: ActionType.GATHER,
|
|
ResourceType.MEAT: ActionType.HUNT,
|
|
ResourceType.WOOD: ActionType.CHOP_WOOD,
|
|
}
|
|
|
|
action = action_map.get(resource_type, ActionType.GATHER)
|
|
|
|
return AIDecision(
|
|
action=action,
|
|
target_resource=resource_type,
|
|
reason=f"{reason} (gathering instead)",
|
|
goal_name=goal.name if goal else "",
|
|
plan_length=len(plan.actions) if plan else 0,
|
|
)
|
|
|
|
def _get_min_keep(self, resource_type: ResourceType) -> int:
|
|
"""Get minimum quantity to keep for survival."""
|
|
# Adjusted by hoarding rate
|
|
hoarding_mult = 0.5 + self.p.hoarding_rate
|
|
|
|
base_min = {
|
|
ResourceType.WATER: 2,
|
|
ResourceType.MEAT: 1,
|
|
ResourceType.BERRIES: 2,
|
|
ResourceType.WOOD: 1,
|
|
ResourceType.HIDE: 0,
|
|
}
|
|
|
|
return int(base_min.get(resource_type, 1) * hoarding_mult)
|
|
|
|
def _calculate_sell_price(self, resource_type: ResourceType) -> int:
|
|
"""Calculate sell price based on fair value and market conditions."""
|
|
# Get energy cost to produce
|
|
from backend.core.ai import get_energy_cost
|
|
from backend.config import get_config
|
|
|
|
config = get_config()
|
|
economy = getattr(config, 'economy', None)
|
|
energy_to_money_ratio = getattr(economy, 'energy_to_money_ratio', 150) if economy else 150
|
|
min_price = getattr(economy, 'min_price', 100) if economy else 100
|
|
|
|
energy_cost = get_energy_cost(resource_type)
|
|
fair_value = max(min_price, int(energy_cost * energy_to_money_ratio))
|
|
|
|
# Apply trading skill
|
|
sell_modifier = get_trade_price_modifier(self.skills.trading, is_buying=False)
|
|
|
|
# Get market signal
|
|
signal = self.market.get_market_signal(resource_type)
|
|
|
|
if signal == "sell": # Scarcity
|
|
price = int(fair_value * 1.3 * sell_modifier)
|
|
elif signal == "hold":
|
|
price = int(fair_value * sell_modifier)
|
|
else: # Surplus
|
|
cheapest = self.market.get_cheapest_order(resource_type)
|
|
if cheapest and cheapest.seller_id != self.agent.id:
|
|
price = max(min_price, cheapest.price_per_unit - 1)
|
|
else:
|
|
price = int(fair_value * 0.8 * sell_modifier)
|
|
|
|
return max(min_price, price)
|
|
|
|
|
|
def get_goap_decision(
|
|
agent: "Agent",
|
|
market: "OrderBook",
|
|
step_in_day: int = 1,
|
|
day_steps: int = 10,
|
|
current_turn: int = 0,
|
|
is_night: bool = False,
|
|
) -> AIDecision:
|
|
"""Convenience function to get a GOAP-based AI decision for an agent.
|
|
|
|
This is the main entry point for the GOAP AI system.
|
|
"""
|
|
ai = GOAPAgentAI(
|
|
agent=agent,
|
|
market=market,
|
|
step_in_day=step_in_day,
|
|
day_steps=day_steps,
|
|
current_turn=current_turn,
|
|
is_night=is_night,
|
|
)
|
|
return ai.decide()
|
|
|