Spaces:
Sleeping
Sleeping
| # Copyright (c) Meta Platforms, Inc. and affiliates. | |
| # All rights reserved. | |
| # | |
| # This source code is licensed under the BSD-style license found in the | |
| # LICENSE file in the root directory of this source tree. | |
| """ | |
| Data models for the Store Manager Environment. | |
| The Store Manager environment simulates a retail store where an LLM agent | |
| acts as the store manager, applying discounts to products to maximize profit | |
| while managing inventory and handling product expiry. | |
| """ | |
| from __future__ import annotations | |
| from typing import Dict, List, Literal, Optional | |
| from openenv.core.env_server.types import Action, Observation, State | |
| from pydantic import BaseModel, Field | |
| class ProductState(BaseModel): | |
| """State of a single product in the store inventory.""" | |
| product_id: int = Field(..., description="Unique integer identifier for the product") | |
| name: str = Field(..., description="Human-readable product name") | |
| cost_price: float = Field(..., description="Cost to the store per unit") | |
| selling_price: float = Field(..., description="Current selling price (after discount)") | |
| base_selling_price: float = Field(..., description="Original selling price before any discount") | |
| quantity: int = Field(..., description="Units currently in stock") | |
| expiry_step: int = Field( | |
| ..., description="Episode step number at which this product expires" | |
| ) | |
| base_probability: float = Field( | |
| ..., description="Base pick-probability weight before discount" | |
| ) | |
| current_discount_pct: float = Field( | |
| default=0.0, description="Active discount percentage this step (0/10/20/50)" | |
| ) | |
| effective_pick_prob: float = Field( | |
| default=0.0, | |
| description="Normalised pick probability this step after applying discount (sums to 1 across all active products)", | |
| ) | |
| is_active: bool = Field( | |
| default=True, description="False if the product is sold out or expired" | |
| ) | |
| sales_history: List[Dict] = Field( | |
| default_factory=list, | |
| description="List of {step, discount_pct, units_sold} records", | |
| ) | |
| # Zone / Shelf Placement fields | |
| zone: int = Field( | |
| default=2, description="Current shelf zone (1=premium/high-traffic, 2=standard, 3=back/low-traffic)" | |
| ) | |
| # Restocking / Procurement fields | |
| holding_cost_rate: float = Field( | |
| default=0.02, description="Holding cost per unit per step as a fraction of cost_price" | |
| ) | |
| restock_cost_per_unit: float = Field( | |
| default=0.0, description="Cost per unit when reordering (may exceed cost_price due to rush ordering)" | |
| ) | |
| restock_lead_time: int = Field( | |
| default=2, description="Number of steps until a restock order is delivered" | |
| ) | |
| pending_restock_quantity: int = Field( | |
| default=0, description="Units currently on order (not yet delivered)" | |
| ) | |
| pending_restock_arrives_at_step: Optional[int] = Field( | |
| default=None, description="Step number when the pending restock order will arrive" | |
| ) | |
| # Compounding popularity effect | |
| popularity_multiplier: float = Field( | |
| default=1.0, | |
| description=( | |
| "Cumulative pick-probability multiplier earned through repeated discounting. " | |
| "Starts at 1.0 and compounds up each time a discount action is applied: " | |
| "popularity_multiplier *= (1 + discount_pct / 100). " | |
| "Persists across steps within an episode; resets to 1.0 on reset()." | |
| ), | |
| ) | |
| # Derived signal fields (recomputed each step — do not set manually) | |
| steps_until_expiry: int = Field( | |
| default=0, | |
| description=( | |
| "Steps remaining before this product expires (expiry_step - current_step). " | |
| "Near-expiry threshold: <= 3 steps. Below-cost discounting is permitted " | |
| "when steps_until_expiry <= 3; the price floor loosens progressively." | |
| ), | |
| ) | |
| margin_per_unit: float = Field( | |
| default=0.0, | |
| description=( | |
| "Current per-unit profit margin: selling_price - cost_price. " | |
| "Positive = profitable. Zero = break-even. " | |
| "Negative = selling at a loss (only possible when steps_until_expiry <= 3)." | |
| ), | |
| ) | |
| class StoremanagerAction(Action): | |
| """Agent action: discount a product, restock inventory, or move a product to a different zone.""" | |
| action_type: Literal["discount", "restock", "place"] = Field( | |
| default="discount", | |
| description="Type of action: 'discount' to apply a price discount, 'restock' to order more inventory, 'place' to move product to a different shelf zone", | |
| ) | |
| product_id: int = Field(..., description="ID of the product to act on") | |
| discount_pct: Literal[0, 10, 20, 50] = Field( | |
| default=0, description="Discount percentage (used when action_type='discount'): 0, 10, 20, or 50" | |
| ) | |
| restock_quantity: Optional[int] = Field( | |
| default=None, description="Number of units to order (used when action_type='restock'); must be > 0" | |
| ) | |
| target_zone: Optional[Literal[1, 2, 3]] = Field( | |
| default=None, description="Destination zone (used when action_type='place'): 1=premium, 2=standard, 3=back" | |
| ) | |
| class StoremanagerObservation(Observation): | |
| """Full observation returned to the agent after each step.""" | |
| inventory: List[ProductState] = Field( | |
| default_factory=list, | |
| description="Current state of all products (including inactive ones)", | |
| ) | |
| current_step: int = Field(default=0, description="Current step number (0-indexed)") | |
| max_steps: int = Field(default=30, description="Total steps in this episode") | |
| steps_remaining: int = Field( | |
| default=30, description="Steps remaining in the episode" | |
| ) | |
| cumulative_profit: float = Field( | |
| default=0.0, description="Total profit accumulated so far in the episode" | |
| ) | |
| last_step_profit: float = Field( | |
| default=0.0, description="Net profit from the most recent step" | |
| ) | |
| last_step_units_sold: int = Field( | |
| default=0, description="Total units sold across all products in the last step" | |
| ) | |
| last_expired_products: List[str] = Field( | |
| default_factory=list, | |
| description="Names of products that expired at the end of the last step", | |
| ) | |
| num_customers_per_step: int = Field( | |
| default=10, description="Fixed number of customers simulated per step" | |
| ) | |
| task_name: str = Field( | |
| default="custom", | |
| description="Active difficulty task: easy / medium / hard / expert / custom", | |
| ) | |
| error: Optional[str] = Field( | |
| default=None, description="Error message if the last action was invalid" | |
| ) | |
| # Zone information | |
| zone_capacity: Dict[int, int] = Field( | |
| default_factory=lambda: {1: 2, 2: 4, 3: 99}, | |
| description="Maximum number of products allowed per zone", | |
| ) | |
| zone_occupancy: Dict[int, int] = Field( | |
| default_factory=lambda: {1: 0, 2: 0, 3: 0}, | |
| description="Current number of active products in each zone", | |
| ) | |
| zone_multipliers: Dict[int, float] = Field( | |
| default_factory=lambda: {1: 1.5, 2: 1.0, 3: 0.6}, | |
| description="Pick-probability multiplier per zone", | |
| ) | |
| # Step cost breakdown | |
| holding_cost_this_step: float = Field( | |
| default=0.0, description="Total holding cost charged this step across all products" | |
| ) | |
| placement_cost_this_step: float = Field( | |
| default=0.0, description="Labor cost charged this step if a zone placement was made" | |
| ) | |
| restock_cost_this_step: float = Field( | |
| default=0.0, description="Ordering cost charged this step if a restock order was placed" | |
| ) | |
| unjustified_discount_penalty: float = Field( | |
| default=0.0, | |
| description=( | |
| "Extra penalty this step for selling below cost_price when steps_until_expiry >= 4. " | |
| "Equals the total per-unit loss on unjustified below-cost units sold. " | |
| "Zero when below-cost selling is justified by near-expiry urgency." | |
| ), | |
| ) | |
| class StoremanagerState(State): | |
| """Internal state snapshot for debugging and orchestration.""" | |
| num_products: int = Field(default=8, description="Number of products in this episode") | |
| max_steps: int = Field(default=30, description="Episode length in steps") | |
| cumulative_profit: float = Field(default=0.0, description="Total episode profit so far") | |
| last_reward: float = Field(default=0.0, description="Reward from the last step") | |
| active_product_count: int = Field( | |
| default=0, description="Number of currently active (non-expired, in-stock) products" | |
| ) | |
| inventory_snapshot: List[Dict] = Field( | |
| default_factory=list, | |
| description="Serialised inventory for state introspection", | |
| ) | |