"""
Implements the base classes for all agents that can join a `SCML2020OneShotWorld`.
Remarks:
- You can access all of the negotiators associated with the agent using
`self.negotiators` which is a dictionary mapping the `negotiator_id` to
a tuple of two values: The `SAONegotiator` object and a key-value context
dictionary. In 2021, the context will always be empty.
- The `negotiator_id` associated with a negotiation with some partner will
be the same as the agent ID of that partner. This means that all negotiators
engaged with some partner over all simulation steps will have the same ID
which is useful if you are keeping information about past negotiations and
partner behavior.
"""
from __future__ import annotations
import warnings
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
from negmas import (
ConstUtilityFunction,
Contract,
ControlledNegotiator,
ControlledSAONegotiator,
Entity,
Issue,
Outcome,
ResponseType,
SAOController,
SAOResponse,
SAOSingleAgreementController,
SAOState,
SAOSyncController,
)
from negmas.helpers import get_class
from negmas.preferences import UtilityFunction
from negmas.sao import SAONMI, SAONegotiator
from negmas.situated import RunningNegotiationInfo
if TYPE_CHECKING:
from scml.oneshot import OneShotAWI, OneShotUFun
__all__ = [
"OneShotAgent",
"OneShotSyncAgent",
"OneShotSingleAgreementAgent",
"OneShotIndNegotiatorsAgent",
"EndingNegotiator",
]
[docs]
class OneShotAgent(SAOController, Entity, ABC):
"""
Base class for all agents in the One-Shot game.
Remarks:
- You can access all of the negotiators associated with the agent using
`self.negotiators` which is a dictionary mapping the `negotiator_id` to
a tuple of two values: The `SAONegotiator` object and a key-value context
dictionary. In 2021, the context will always be empty.
- The `negotiator_id` associated with a negotiation with some partner will
be the same as the agent ID of that partner. This means that all negotiators
engaged with some partner over all simulation steps will have the same ID
which is useful if you are keeping information about past negotiations and
partner behavior.
"""
def __init__(self, owner=None, ufun: OneShotUFun | None = None, name=None):
super().__init__(
default_negotiator_type=ControlledSAONegotiator,
default_negotiator_params=None,
auto_kill=False,
name=name,
preferences=ufun,
)
self._awi = owner._awi if owner else None
self._owner = owner
self.set_preferences(owner.ufun if owner else None)
@property
[docs]
def awi(self) -> OneShotAWI:
"""Returns a `OneShotAWI` object for accessing the simulation."""
if not self._awi:
raise ValueError("The AWI is not assigned yet.")
return self._awi
@property
[docs]
def running_negotiations(self) -> list[RunningNegotiationInfo]:
"""The negotiations currently requested by the agent.
Returns:
A list of negotiation information objects (`RunningNegotiationInfo`)
"""
assert self._owner is not None
return self._owner.running_negotiations
@property
[docs]
def unsigned_contracts(self) -> list[Contract]:
"""
All contracts that are not yet signed.
"""
assert self._owner is not None
return self._owner.unsigned_contracts
[docs]
def init(self):
"""
Called once after the AWI is set.
Remarks:
- Use this for any proactive initialization code.
"""
[docs]
def make_ufun(self, add_exogenous=False):
"""
Creates a utility function for the agent.
Args:
add_exogenous: If `True` then the exogenous contracts of the agent
will be automatically added whenever the ufun is
evaluated for any set of contracts, offers or otherwise.
Remarks:
- You can always as assume that self.ufun returns the ufun for your.
You will not need to directly use this method in most cases.
"""
assert self._owner is not None
return self._owner.make_ufun(add_exogenous)
[docs]
def before_step(self):
"""
Called at the beginning of every step.
Remarks:
- Use this for any proactive code that needs to be done every
simulation step.
"""
pass
[docs]
def step(self):
"""
Called at the end of every step.
Remarks:
- Use this for any proactive code that needs to be done every
simulation step.
"""
pass
[docs]
def connect_to_oneshot_adapter(self, owner):
"""Connects the agent to its adapter (used internally)"""
self._owner = owner
self._awi = owner._awi
self.utility_function = owner.ufun
[docs]
def connect_to_2021_adapter(self, owner):
"""Connects the agent to its adapter (used internally)"""
self._owner = owner
self._awi = owner._awi
self.utility_function = owner.ufun
@abstractmethod
[docs]
def propose(self, negotiator_id: str, state: SAOState) -> Outcome | None:
"""
Proposes an offer to one of the partners.
Args:
negotiator_id: ID of the negotiator (and partner)
state: Mechanism state including current step
Returns:
an outcome to offer.
"""
[docs]
def respond(self, negotiator_id: str, state: SAOState, source=None) -> ResponseType:
"""
Responds to an offer from one of the partners.
Args:
negotiator_id: ID of the negotiator (and partner)
state: Mechanism state including current step
Returns:
A response type which can either be reject, accept, or end negotiation.
Remarks:
default behavior is to accept only if the current offer is the same
or has a higher utility compared with what the agent would have
proposed in the given state and reject otherwise
"""
offer = state.current_offer # type: ignore
myoffer = self.propose(negotiator_id, state)
if myoffer == offer:
return ResponseType.ACCEPT_OFFER
return ResponseType.REJECT_OFFER
@property
[docs]
def internal_state(self) -> dict[str, Any]:
"""
Returns the internal state of the agent for debugging purposes.
Remarks:
- In your agent, you can add any key-value pair to this dict and
then use agent_log_* methods to log this information at any point.
"""
return {}
[docs]
def on_negotiation_failure(
self,
partners: list[str],
annotation: dict[str, Any],
mechanism: SAONMI,
state: SAOState,
) -> None:
"""
Called whenever a negotiation ends without agreement.
Args:
partners: List of the partner IDs consisting from self and the opponent.
annotation: The annotation of the negotiation including the seller ID,
buyer ID, and the product.
mechanism: The `NegotiatorMechanismInterface` instance containing all information
about the negotiation.
state: The final state of the negotiation of the type `SAOState`
including the agreement if any.
"""
[docs]
def on_negotiation_success(self, contract: Contract, mechanism: SAONMI) -> None:
"""
Called whenever a negotiation ends with agreement.
Args:
contract: The `Contract` agreed upon.
mechanism: The `NegotiatorMechanismInterface` instance containing all information
about the negotiation that led to the `Contract` if any.
"""
[docs]
def sign_all_contracts(self, contracts: list[Contract]) -> list[str | None]:
"""Signs all contracts (used internally)"""
return [self.id] * len(contracts)
[docs]
def on_contract_executed(self, contract) -> None:
pass
[docs]
def on_contract_breached(self, contract, breaches, resolution) -> None:
pass
[docs]
def get_negotiator(self, partner_id: str) -> SAONegotiator:
"""
Returns the negotiator corresponding to the given partner ID.
Remarks:
- Note that the negotiator ID and the partner ID are always the same.
"""
return self.negotiators[partner_id][0]
[docs]
def get_ami(self, partner_id: str) -> SAONMI:
"""
Returns the `SAONMI` (Agent Mechanism Interface) connecting the agent
to the negotiation mechanism for the given partner.
"""
warnings.warn("get_ami is depricated. Use get_nmi")
return self.negotiators[partner_id][0].nmi
[docs]
def get_nmi(self, partner_id: str) -> SAONMI:
"""
Returns the `SAONMI` (Agent Mechanism Interface) connecting the agent
to the negotiation mechanism for the given partner.
"""
return self.negotiators[partner_id][0].nmi
[docs]
class OneShotSyncAgent(SAOSyncController, OneShotAgent, ABC): # type: ignore
"""
An agent that automatically accumulate offers from opponents and allows
you to control all negotiations centrally in the `counter_all` method.
"""
def __init__(self, *args, **kwargs):
kwargs["global_ufun"] = True
super().__init__(*args, **kwargs)
@abstractmethod
[docs]
def counter_all(
self, offers: dict[str, Outcome | None], states: dict[str, SAOState]
) -> dict[str, SAOResponse]:
"""Calculate a response to all offers from all negotiators
(negotiator ID is the key).
Args:
offers: Maps negotiator IDs to offers
states: Maps negotiator IDs to offers AT the time the offers were made.
Returns:
A dictionary mapping negotiator ID to an `SAOResponse`. The response
per agent consist of a tuple. In case of acceptance or ending the
negotiation the second item of the tuple should be None. In case of
rejection, the second item should be the counter offer.
Remarks:
- The response type CANNOT be WAIT.
- If the system determines that a loop is formed, the agent may
receive this call for a subset of negotiations not all of them.
"""
@abstractmethod
[docs]
def first_proposals(self) -> dict[str, Outcome | None]:
"""
Gets a set of proposals to use for initializing the negotiation.
Returns:
A dictionary mapping each negotiator (in self.negotiators dict) to
an outcome to be used as the first proposal if the agent is to start
a negotiation.
"""
return super().first_proposals()
[docs]
def sign_all_contracts(self, contracts: list[Contract]) -> list[str | None]:
"""Signs all contracts (used internally)"""
return [self.id] * len(contracts)
[docs]
class OneShotSingleAgreementAgent(SAOSingleAgreementController, OneShotSyncAgent): # type: ignore
"""
A synchronized agent that tries to get no more than one agreement.
This controller manages a set of negotiations from which only a single one
-- at most -- is likely to result in an agreement.
To guarantee a single agreement, pass `strict=True`
The general algorithm for this controller is something like this:
- Receive offers from all partners.
- Find the best offer among them by calling the abstract `best_offer`
method.
- Check if this best offer is acceptable using the abstract `is_acceptable`
method.
- If the best offer is acceptable, accept it and end all other negotiations.
- If the best offer is still not acceptable, then all offers are rejected
and with the partner who sent it receiving the result of `best_outcome`
while the rest of the partners receive the result of `make_outcome`.
- The default behavior of `best_outcome` is to return the outcome with
maximum utility.
- The default behavior of `make_outcome` is to return the best offer
received in this round if it is valid for the respective negotiation
and the result of `best_outcome` otherwise.
Args:
strict: If True the controller is **guaranteed** to get a single
agreement but it will have to send no-response repeatedly so
there is a higher chance of never getting an agreement when
two of those controllers negotiate with each other
"""
def __init__(self, *args, strict: bool = False, **kwargs):
super().__init__(*args, strict=strict, **kwargs)
@abstractmethod
[docs]
def is_acceptable(self, offer: Outcome, source: str, state: SAOState) -> bool:
"""Should decide if the given offer is acceptable
Args:
offer: The offer being tested
source: The ID of the negotiator that received this offer
state: The state of the negotiation handled by that negotiator
Remarks:
- If True is returned, this offer will be accepted and all other
negotiations will be ended.
"""
@abstractmethod
[docs]
def best_offer(self, offers: dict[str, Outcome]) -> str | None:
"""
Return the ID of the negotiator with the best offer
Args:
offers: A mapping from negotiator ID to the offer it received
Returns:
The ID of the negotiator with best offer. Ties should be broken.
Return None only if there is no way to calculate the best offer.
"""
@abstractmethod
[docs]
def is_better(
self, a: Outcome | None, b: Outcome | None, negotiator: str, state: SAOState
) -> bool:
"""Compares two outcomes of the same negotiation
Args:
a: "Outcome"
b: "Outcome"
negotiator: The negotiator for which the comparison is to be made
state: Current state of the negotiation
Returns:
True if utility(a) > utility(b)
"""
[docs]
class EndingNegotiator(SAONegotiator, ControlledNegotiator):
[docs]
def propose(self, state):
return None
[docs]
def respond(self, state, source=None):
return ResponseType.END_NEGOTIATION
[docs]
class OneShotIndNegotiatorsAgent(OneShotAgent):
"""
A one-shot agent that deligates all of its decisions to a set of independent
negotiators (one per partner per day).
Args:
default_negotiator_type: An `SAONegotiator` descendent to be used for
creating all negotiators. It can be passed either
as a class object or a string with the full class
name (e.g. "negmas.sao.AspirationNegotiator").
default_negotiator_type: A dict specifying the paratmers used to create
negotiators.
normalize_ufuns: If true, all utility functions will be normalized to have
a maximum of 1.0 (the minimum value may be negative).
set_reservation: If given, the reserved value of all ufuns will be
guaranteed to be between the minimum and maximum of
the ufun. This is needed to avoid failures of some
GeniusNegotiators.
Remarks:
- To use this class, you need to override `generate_ufuns`. If you
want to change the negotiator type used depending on the partner, you
can also override `generate_negotiator`.
- If you are using a `GeniusNegotiator` you must guarantee the following:
- All ufuns are of the type `LinearAdditiveUtilityFunction`.
- All ufuns are normalized with a maximum value of 1.0. You can
use `normalize_ufuns=True` to gruarantee that.
- All ufuns have a finite reserved value and at least one outcome is
above it. You can guarantee that by using `set_reservation=True`.
- All weights of the `LinearAdditiveUtilityFunction` must be between
zero and one and the weights must sum to one.
"""
def __init__(
self,
*args,
default_negotiator_type="negmas.sao.AspirationNegotiator",
default_negotiator_params=None,
normalize_ufuns=False,
set_reservation=False,
**kwargs,
):
super().__init__(*args, **kwargs)
self._default_negotiator_type = get_class(default_negotiator_type)
self._default_negotiator_params = (
dict() if not default_negotiator_params else default_negotiator_params
)
self._ufuns = dict()
self._normalize = normalize_ufuns
self._set_reservation = set_reservation
@abstractmethod
[docs]
def generate_ufuns(self) -> dict[str, UtilityFunction]:
"""
Returns a utility function for each partner. All ufuns **MUST** be of
type `LinearAdditiveUtilityFunction` if a genius negotiator is used.
"""
[docs]
def generate_negotiator(self, partner_id: str) -> SAONegotiator:
"""
Returns a negotiator to be used with some partner.
Remarks:
The default implementation will use the `default_negotiator_type`
and `default_negotiator_params`.
"""
return self._default_negotiator_type(**self._default_negotiator_params)
[docs]
def _urange(self, u: UtilityFunction, issues: tuple[Issue, ...]):
return u.minmax(issues=tuple(issues))
# if not isinstance(u, LinearAdditiveUtilityFunction) and not isinstance(
# u, LinearUtilityFunction
# ):
# mn = mx = 0.0
# for w, issue in zip(u.weights, issues):
# values = list(issue.values)
# mnv, mxv = min(values), max(values)
# if w > 0:
# mn += mnv * w
# mx += mxv * w
# else:
# mn += mxv * w
# mx += mnv * w
# return mn, mx
[docs]
def _unorm(self, u: UtilityFunction, mn, mx):
if mn == 0:
return u.scale_max(mx)
return u.normalize((mn, mx))
# if not isinstance(u, LinearAdditiveUtilityFunction) and not isinstance(
# u, LinearUtilityFunction
# ):
# # _, mx = self._urange(u, issues)
# if mx < 0:
# return None
# u.weights = {k: _ / mx for k, _ in u.weights.items()}
# return u
[docs]
def _get_ufuns(self):
"""
Internal method that makes sure the reservation value is set to a
meaningful value and that the ufun is normalized if needed
"""
ufuns = self.generate_ufuns()
if not self._normalize and not self._set_reservation:
return ufuns
if not self.awi:
raise ValueError("AWI is not found!!!")
for partner_id, u in ufuns.items():
if self.awi.is_system(partner_id):
continue
issues = (
self.awi.current_input_issues
if partner_id in self.awi.my_suppliers
else self.awi.current_output_issues
)
mn, mx = self._urange(u, tuple(issues))
if self._normalize:
u = self._unorm(u, mn, mx)
if u is None:
continue
if not self._set_reservation:
continue
if (
u.reserved_value is None
or u.reserved_value == float("-inf")
or u.reserved_value == float("nan")
):
u.reserved_value = mn - 1e-5
u.reserved_value = u.reserved_value / mx
if u.reserved_value > mx:
ufuns[partner_id] = ConstUtilityFunction(0.0, reserved_value=0.0)
return ufuns
[docs]
def init(self):
super().init()
self._ufuns = self._get_ufuns()
[docs]
def step(self):
super().step()
self._ufuns = self._get_ufuns()
[docs]
def make_negotiator(
self,
negotiator_type=None,
name: str | None = None,
**kwargs,
) -> ControlledSAONegotiator:
"""
Creates a negotiator but does not add it to the controller. Call
`add_negotiator` to add it.
Args:
negotiator_type: Type of the negotiator to be created.
name: negotiator name
**kwargs: any key-value pairs to be passed to the negotiator constructor
Returns:
The negotiator to be controlled. None for failure
Remarks:
If you would like not to negotiate, just return `EndingNegotiator()`
instead of None. The value None should only be returned if an exception
is to be thrown.
"""
if name is None:
return EndingNegotiator() # type: ignore
ufun = self._ufuns[name]
if ufun is None:
return EndingNegotiator() # type: ignore
negotiator = self.generate_negotiator(name)
negotiator.id = name
negotiator.name = name
negotiator.ufun = ufun
return negotiator # type: ignore
[docs]
def propose(self, negotiator_id, state):
raise ValueError(
"propose should never be called directly on OneShotIndNegotiatorsAgent"
)