192 lines
7.1 KiB
Python
192 lines
7.1 KiB
Python
"""Bannerlord feudal hierarchy data models.
|
|
|
|
All inter-agent communication uses typed Pydantic models. No raw dicts
|
|
cross agent boundaries — every message is validated at construction time.
|
|
|
|
Design: Ahilan & Dayan (2019) Feudal Multi-Agent Hierarchies.
|
|
Refs: #1097, #1099.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
# ── Subgoal vocabulary ────────────────────────────────────────────────────────
|
|
|
|
SUBGOAL_TOKENS = frozenset(
|
|
{
|
|
"EXPAND_TERRITORY", # Take or secure a fief — War Vassal
|
|
"RAID_ECONOMY", # Raid enemy villages for denars — War Vassal
|
|
"FORTIFY", # Upgrade or repair a settlement — Economy Vassal
|
|
"RECRUIT", # Fill party to capacity — Logistics Companion
|
|
"TRADE", # Execute profitable trade route — Caravan Companion
|
|
"ALLY", # Pursue non-aggression / alliance — Diplomacy Vassal
|
|
"SPY", # Gain information on target faction — Scout Companion
|
|
"HEAL", # Rest party until wounds recovered — Logistics Companion
|
|
"CONSOLIDATE", # Hold territory, no expansion — Economy Vassal
|
|
"TRAIN", # Level troops via auto-resolve bandits — War Vassal
|
|
}
|
|
)
|
|
|
|
|
|
# ── King subgoal ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
class KingSubgoal(BaseModel):
|
|
"""Strategic directive issued by the King agent to vassals.
|
|
|
|
The King operates on campaign-map timescale (days to weeks of in-game
|
|
time). His sole output is one subgoal token plus optional parameters.
|
|
He never micro-manages primitives.
|
|
"""
|
|
|
|
token: str = Field(..., description="One of SUBGOAL_TOKENS")
|
|
target: str | None = Field(None, description="Named target (settlement, lord, faction)")
|
|
quantity: int | None = Field(None, description="For RECRUIT, TRADE tokens", ge=1)
|
|
priority: float = Field(1.0, ge=0.0, le=2.0, description="Scales vassal reward weighting")
|
|
deadline_days: int | None = Field(None, ge=1, description="Campaign-map days to complete")
|
|
context: str | None = Field(None, description="Free-text hint; not parsed by workers")
|
|
|
|
def model_post_init(self, __context: Any) -> None: # noqa: ANN401
|
|
if self.token not in SUBGOAL_TOKENS:
|
|
raise ValueError(
|
|
f"Unknown subgoal token {self.token!r}. Must be one of: {sorted(SUBGOAL_TOKENS)}"
|
|
)
|
|
|
|
|
|
# ── Inter-agent messages ──────────────────────────────────────────────────────
|
|
|
|
|
|
class SubgoalMessage(BaseModel):
|
|
"""King → Vassal direction."""
|
|
|
|
msg_type: Literal["subgoal"] = "subgoal"
|
|
from_agent: Literal["king"] = "king"
|
|
to_agent: str = Field(..., description="e.g. 'war_vassal', 'economy_vassal'")
|
|
subgoal: KingSubgoal
|
|
issued_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class TaskMessage(BaseModel):
|
|
"""Vassal → Companion direction."""
|
|
|
|
msg_type: Literal["task"] = "task"
|
|
from_agent: str = Field(..., description="e.g. 'war_vassal'")
|
|
to_agent: str = Field(..., description="e.g. 'logistics_companion'")
|
|
primitive: str = Field(..., description="One of the companion primitives")
|
|
args: dict[str, Any] = Field(default_factory=dict)
|
|
priority: float = Field(1.0, ge=0.0, le=2.0)
|
|
issued_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class ResultMessage(BaseModel):
|
|
"""Companion / Vassal → Parent direction."""
|
|
|
|
msg_type: Literal["result"] = "result"
|
|
from_agent: str
|
|
to_agent: str
|
|
success: bool
|
|
outcome: dict[str, Any] = Field(default_factory=dict, description="Primitive-specific result")
|
|
reward_delta: float = Field(0.0, description="Computed reward contribution")
|
|
completed_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class StateUpdateMessage(BaseModel):
|
|
"""GABS → All agents (broadcast).
|
|
|
|
Sent every campaign tick. Agents consume at their own cadence.
|
|
"""
|
|
|
|
msg_type: Literal["state"] = "state"
|
|
game_state: dict[str, Any] = Field(..., description="Full GABS state snapshot")
|
|
tick: int = Field(..., ge=0)
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
# ── Reward snapshots ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class WarReward(BaseModel):
|
|
"""Computed reward for the War Vassal at a given tick."""
|
|
|
|
territory_delta: float = 0.0
|
|
army_strength_ratio: float = 1.0
|
|
casualty_cost: float = 0.0
|
|
supply_cost: float = 0.0
|
|
subgoal_bonus: float = 0.0
|
|
|
|
@property
|
|
def total(self) -> float:
|
|
w1, w2, w3, w4, w5 = 0.40, 0.25, 0.20, 0.10, 0.05
|
|
return (
|
|
w1 * self.territory_delta
|
|
+ w2 * self.army_strength_ratio
|
|
- w3 * self.casualty_cost
|
|
- w4 * self.supply_cost
|
|
+ w5 * self.subgoal_bonus
|
|
)
|
|
|
|
|
|
class EconomyReward(BaseModel):
|
|
"""Computed reward for the Economy Vassal at a given tick."""
|
|
|
|
daily_denars_income: float = 0.0
|
|
food_stock_buffer: float = 0.0
|
|
loyalty_average: float = 50.0
|
|
construction_queue_length: int = 0
|
|
subgoal_bonus: float = 0.0
|
|
|
|
@property
|
|
def total(self) -> float:
|
|
w1, w2, w3, w4, w5 = 0.35, 0.25, 0.20, 0.15, 0.05
|
|
return (
|
|
w1 * self.daily_denars_income
|
|
+ w2 * self.food_stock_buffer
|
|
+ w3 * self.loyalty_average
|
|
- w4 * self.construction_queue_length
|
|
+ w5 * self.subgoal_bonus
|
|
)
|
|
|
|
|
|
class DiplomacyReward(BaseModel):
|
|
"""Computed reward for the Diplomacy Vassal at a given tick."""
|
|
|
|
allies_count: int = 0
|
|
truce_duration_value: float = 0.0
|
|
relations_score_weighted: float = 0.0
|
|
active_wars_front: int = 0
|
|
subgoal_bonus: float = 0.0
|
|
|
|
@property
|
|
def total(self) -> float:
|
|
w1, w2, w3, w4, w5 = 0.30, 0.25, 0.25, 0.15, 0.05
|
|
return (
|
|
w1 * self.allies_count
|
|
+ w2 * self.truce_duration_value
|
|
+ w3 * self.relations_score_weighted
|
|
- w4 * self.active_wars_front
|
|
+ w5 * self.subgoal_bonus
|
|
)
|
|
|
|
|
|
# ── Victory condition ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class VictoryCondition(BaseModel):
|
|
"""Sovereign Victory (M5) — evaluated each campaign tick."""
|
|
|
|
holds_king_title: bool = False
|
|
territory_control_pct: float = Field(
|
|
0.0, ge=0.0, le=100.0, description="% of Calradia fiefs held"
|
|
)
|
|
majority_threshold: float = Field(
|
|
51.0, ge=0.0, le=100.0, description="Required % for majority control"
|
|
)
|
|
|
|
@property
|
|
def achieved(self) -> bool:
|
|
return self.holds_king_title and self.territory_control_pct >= self.majority_threshold
|