Source code for scml.oneshot.rl.action

Defines ways to encode and decode actions.
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Iterable

import numpy as np
from attr import define
from gymnasium import Space, spaces
from import ResponseType, field
from negmas.helpers import distribute_integer_randomly
from negmas.outcomes.issue_ops import itertools
from negmas.sao.common import SAOResponse

from scml.oneshot.awi import OneShotAWI
from scml.oneshot.context import BaseContext
from scml.scml2019.common import QUANTITY
from .helpers import recover_offers, encode_given_offers

__all__ = [

[docs] class ActionManager(ABC): """ Manges actions of an agent in an RL environment. """
[docs] context: BaseContext
[docs] continuous: bool = False
[docs] n_suppliers: int = field(init=False, default=8)
[docs] n_consumers: int = field(init=False, default=8)
[docs] n_partners: int = field(init=False, default=16)
[docs] def make_space(self) -> Space: """Creates the action space""" ...
[docs] def decode(self, awi: OneShotAWI, action: np.ndarray) -> dict[str, SAOResponse]: """Decodes an action from an array to a `PurchaseOrder` and a `CounterMessage`.""" ...
[docs] def encode(self, awi: OneShotAWI, responses: dict[str, SAOResponse]) -> np.ndarray: """Encodes an action as an array. This is only used for testing so it is optional""" _ = awi, responses raise NotImplementedError( f"{self.__class__.__name__} does not implement `encode`." )
def safemin(x: Iterable | int | float | str): if isinstance(x, Iterable): return min(x) return x @define(frozen=True)
[docs] class FlexibleActionManager(ActionManager): """ An action manager that matches any context. Args: n_prices: Number of distinct prices allowed in the action. max_quantity: Maximum allowed quantity to offer in any negotiation. The number of quantities is one plus that because zero is allowed to model ending negotiation. n_partners: Maximum of partners allowed in the action. Remarks: - This action manager will always generate offers that are within the price and quantity limits given in its parameters. Wen decoding them, it will scale them up so that the maximum corresponds to the actual value in the world it finds itself. For example, if `n_prices` is 10 and the world has only two prices currently in the price issue, it will use any value less than 5 as the minimum price and any value above 5 as the maximum price. If on the other hand the current price issue has 20 values, then it will scale by multiplying the number given in the encoded action (ranging from 0 to 9) by 19/9 which makes it range from 0 to 19 which is what is expected by the world. - This action manager will adjust offers for different number of partners as follows: - If the true number of partners is larger than `n_partners` used by this action manager, it will simply use `n_partners` of them and always end negotiations with the rest of them. - If the true number of partners is smaller than `n_partners`, it will use the first `n_partners` values in the encoded action and increase the quantities of any counter offers (i.e. ones in which the response is REJECT_OFFER) by the amount missing from the ignored partners in the encoded action up to the maximum quantities allowed by the current negotiation context. For example, if `n_partneers` is 4 and we have only 2 partners in reality, and the received quantities from partners were [4, 3] while the maximum quantity allowed is 10 and the encoded action was [2, *, 3, *, 2, *, 1, *] (where we ignored prices), then the encoded action will be converted to [(Reject, 5, *), (Accept, 3, *)] where the 3 extra units that were supposed to be offered to the last two partners are moved to the first partner. If the maximum quantity allowed was 4 in that example, the result will be [(Reject, 4, *), (Accept, 3, *)]. """
[docs] capacity_multiplier: int = 1
[docs] n_prices: int = 2
[docs] max_group_size: int = 2
[docs] reduce_space_size: bool = True
[docs] extra_checks: bool = False
[docs] max_quantity: int = field(init=False, default=10)
[docs] def __attrs_post_init__(self): p = self.context.extract_context_params(self.reduce_space_size) if p.nlines: object.__setattr__( self, "max_quantity", self.capacity_multiplier * p.nlines ) object.__setattr__(self, "n_consumers", p.nconsumers) object.__setattr__(self, "n_suppliers", p.nsuppliers) object.__setattr__(self, "n_partners", self.n_suppliers + self.n_consumers)
[docs] def make_space(self) -> spaces.MultiDiscrete | spaces.Box: """Creates the action space""" return ( spaces.MultiDiscrete( np.asarray( [self.max_quantity + 1, self.n_prices] * self.n_partners ).flatten() ) if not self.continuous else spaces.Box(0.0, 1.0, shape=(self.n_partners * 2,)) )
[docs] def decode(self, awi: OneShotAWI, action: np.ndarray) -> dict[str, SAOResponse]: """ Generates offers to all partners from an encoded action. Default is to return the action as it is assuming it is a `dict[str, SAOResponse]` """ action = action.reshape((action.size // 2, 2)) if not (len(action) == self.n_partners): raise AssertionError( f"{len(action)=}, {self.n_partners=} ({self.n_suppliers=}, {self.n_consumers=})" ) offers = recover_offers( action, awi, self.n_suppliers, self.n_consumers, self.max_group_size, self.continuous, self.n_prices, ) separated_offers, responses = dict(), dict() nmis = awi.current_nmis for k, v in offers.items(): if "+" not in k: separated_offers[k] = tuple(int(_) for _ in v) if v else v continue partners = k.split("+") if v is None: separated_offers |= dict(zip(partners, itertools.repeat(None))) continue q = v[QUANTITY] dist = distribute_integer_randomly(q, len(partners)) separated_offers |= dict(zip(partners, ((_, v[1], v[-1]) for _ in dist))) for k, v in separated_offers.items(): nmi = nmis.get(k, None) if nmi is None: continue if v is None: responses[k] = SAOResponse(ResponseType.END_NEGOTIATION, None) continue partner_offer = nmi.state.current_offer # type: ignore if v == partner_offer: responses[k] = SAOResponse(ResponseType.ACCEPT_OFFER, partner_offer) continue responses[k] = SAOResponse(ResponseType.REJECT_OFFER, v) return responses
[docs] def encode(self, awi: OneShotAWI, responses: dict[str, SAOResponse]) -> np.ndarray: """ Receives offers for all partners and generates the corresponding action. Used mostly for debugging and testing. """ offers = dict() for k, v in responses.items(): if v.response == ResponseType.END_NEGOTIATION: offers[k] = None continue offers[k] = v.outcome encoded = encode_given_offers( offers, awi, self.n_suppliers, self.n_consumers, self.max_group_size, self.continuous, ) return np.asarray(encoded, dtype=np.float32 if self.continuous else np.int32)
[docs] DefaultActionManager = FlexibleActionManager
"""The default action manager"""