from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from typing import Any, Generic, TypeVar
from negmas.common import MechanismState, NegotiatorMechanismInterface
from negmas.events import Event, EventSink, EventSource, Notifier
from negmas.helpers.inout import ConfigReader
from negmas.mechanisms import Mechanism
from negmas.negotiators import Negotiator
from negmas.outcomes import Issue
from negmas.preferences import Preferences, UtilityFunction
from negmas.types import Rational
from .breaches import Breach
from .common import NegotiationRequestInfo, RenegotiationRequest, RunningNegotiationInfo
from .contract import Contract
from .entity import Entity
from .awi import AgentWorldInterface
__all__ = ["Agent"]
TAWI = TypeVar("TAWI", bound=AgentWorldInterface)
[docs]
class Agent(Entity, EventSink, ConfigReader, Notifier, Rational, Generic[TAWI], ABC):
"""Base class for all agents that can run within a `World` and engage in situated negotiations"""
# def __getstate__(self):
# return self.name, self.awi
#
# def __setstate__(self, state):
# name, awi = state
# super().__init__(name=name)
# self._awi = awi
def __init__(
self,
name: str | None = None,
type_postfix: str = "",
preferences: Preferences | None = None,
ufun: UtilityFunction | None = None,
):
super().__init__(type_postfix=type_postfix)
Rational.__init__(self, name=name, preferences=preferences, ufun=ufun)
self._running_negotiations: dict[str, RunningNegotiationInfo] = {}
self._requested_negotiations: dict[str, NegotiationRequestInfo] = {}
self._accepted_requests: dict[str, NegotiationRequestInfo] = {}
self.contracts: list[Contract] = []
self._unsigned_contracts: set[Contract] = set()
self._awi: TAWI | None = None
# def to_dict(self) -> Dict[str, Any]:
# """Converts the agent into dict for storage purposes.
#
# The agent need not be recoverable from this representation.
#
# """
# try:
# d = to_dict(vars(dict), deep=False, keep_private=False, add_type_field=False)
# # _ = json.dumps(d)
# return d
# except Exception:
# return {"id": self.id, "name": self.name}
@property
def initialized(self) -> bool:
"""Was the agent initialized (i.e. was init_() called)"""
return self._initialized
@property
def unsigned_contracts(self) -> list[Contract]:
"""
All contracts that are not yet signed.
"""
return list(self._unsigned_contracts)
@property
def requested_negotiations(self) -> list[NegotiationRequestInfo]:
"""The negotiations currently requested by the agent.
Returns:
A list of negotiation request information objects (`NegotiationRequestInfo`)
"""
return list(self._requested_negotiations.values())
@property
def accepted_negotiation_requests(self) -> list[NegotiationRequestInfo]:
"""
A list of negotiation requests sent to this agent that are already accepted by it.
Remarks:
- These negotiations did not start yet as they are still not accepted by all partners.
Once that happens, they will be moved to `running_negotiations`
"""
return list(self._accepted_requests.values())
@property
def negotiation_requests(self) -> list[NegotiationRequestInfo]:
"""A list of the negotiation requests sent by this agent that are not yet accepted or rejected.
Remarks:
- These negotiations did not start yet as they are still not accepted by all partners.
Once that happens, they will be moved to `running_negotiations`
"""
return list(self._requested_negotiations.values())
@property
def running_negotiations(self) -> list[RunningNegotiationInfo]:
"""The negotiations currently requested by the agent.
Returns:
A list of negotiation information objects (`RunningNegotiationInfo`)
"""
return list(self._running_negotiations.values())
@property
def awi(self) -> TAWI:
"""Gets the Agent-world interface."""
return self._awi # type: ignore
@awi.setter
def awi(self, awi: TAWI):
"""Sets the Agent-world interface. Should only be called by the world."""
self._awi = awi
[docs]
def create_negotiation_request(
self,
issues: list[Issue] | tuple[Issue, ...],
partners: list[str],
annotation: dict[str, Any] | None,
negotiator: Negotiator | None,
extra: dict[str, Any] | None,
) -> str:
"""
Creates a new `NegotiationRequestInfo` record and returns itS ID
Args:
issues: negotiation issues
partners: partners
annotation: annotation
negotiator: the negotiator to use
extra: any extra information
Returns:
A unique identifier for this negotiation info structure
"""
req_id = str(uuid.uuid4())
self._requested_negotiations[req_id] = NegotiationRequestInfo(
issues=issues,
partners=partners,
annotation=annotation,
negotiator=negotiator,
extra=extra,
requested=True,
uuid=req_id,
)
return req_id
def _request_negotiation(
self,
issues: tuple[Issue, ...],
partners: list[str],
roles: list[str] | None = None,
annotation: dict[str, Any] | None = None,
mechanism_name: str | None = None,
mechanism_params: dict[str, Any] | None = None,
negotiator: Negotiator | None = None,
extra: dict[str, Any] | None = None,
group: str | None = None,
) -> bool:
"""
Requests to start a negotiation with some other agents
Args:
issues: Negotiation issues
annotation: Extra information to be passed to the `partners` when asking them to join the negotiation
partners: A list of partners to participate in the negotiation.
Note that the caller itself may not be in this list which
makes it possible for an agent to request a negotaition
that it does not participate in. If that is not to be
allowed in some world, override this method and explicitly
check for these kinds of negotiations and return False.
If partners is passed as a single string or as a list
containing a single string, then he caller will be added
at the beginning of the list. This will only be done if
`roles` was passed as None.
roles: The roles of different partners. If None then each role for each partner will be None
mechanism_name: Name of the mechanism to use. It must be one of the mechanism_names that are supported by the
`World` or None which means that the `World` should select the mechanism. If None, then `roles` and `my_role`
must also be None
mechanism_params: A dict of parameters used to initialize the mechanism object
negotiator: My negotiator to use in this negotiation. Can be none
extra: Any extra information I would like to keep to myself for this negotiation
group: The negotiation group
Returns:
List["Agent"] the list of partners who rejected the negotiation if any. If None then the negotiation was
accepted. If empty then the negotiation was not started from the world manager
Remarks:
- The function will create a request ID that will be used in callbacks `on_neg_request_accepted` and
`on_neg_request_rejected`.
- This function is a private function as the name implies and should not be called directly in any world.
- World designers extending this class for their worlds, should define a way to start negotiations that
calls this function. The simplest way is to just define a `request_negotiation` function that calls this
private version directly with the same parameters.
"""
if roles is None:
if isinstance(partners, str) or isinstance(partners, Agent):
partners = [partners] # type: ignore
if len(partners) == 1 and partners[0] != self.id:
partners = [self.id, partners[0]]
req_id = self.create_negotiation_request(
issues=issues,
partners=partners,
annotation=annotation,
negotiator=negotiator,
extra=extra,
)
return self.awi.request_negotiation_about(
issues=issues,
partners=partners,
req_id=req_id,
roles=roles,
annotation=annotation,
group=group,
mechanism_name=mechanism_name,
mechanism_params=mechanism_params,
)
[docs]
def on_event(self, event: Event, sender: EventSource):
if not isinstance(sender, Mechanism) and not isinstance(sender, Mechanism):
raise ValueError(
f"Sender of the negotiation end event is of type {sender.__class__.__name__} "
f"not Mechanism!!"
)
[docs]
@abstractmethod
def on_neg_request_rejected(self, req_id: str, by: list[str] | None):
"""Called when a requested negotiation is rejected
Args:
req_id: The request ID passed to _request_negotiation
by: A list of agents that refused to participate or None if the failure was for another reason
"""
# if event.type == "negotiation_end":
# # will be sent by the World once a negotiation in which this agent is involved is completed l
# mechanism_id = sender.id
# self._running_negotiations.pop(mechanism_id, None)
# ------------------------------------------------------------------
# EVENT CALLBACKS (Called by the `World` when certain events happen)
# ------------------------------------------------------------------
[docs]
def on_neg_request_rejected_(self, req_id: str, by: list[str] | None):
"""Called when a requested negotiation is rejected
Args:
req_id: The request ID passed to _request_negotiation
by: A list of agents that refused to participate or None if the failure was for another reason
"""
self.on_neg_request_rejected(req_id, by)
self._requested_negotiations.pop(req_id, None)
[docs]
@abstractmethod
def on_neg_request_accepted(
self, req_id: str, mechanism: NegotiatorMechanismInterface
):
"""Called when a requested negotiation is accepted"""
[docs]
def on_neg_request_accepted_(
self, req_id: str, mechanism: NegotiatorMechanismInterface
):
"""Called when a requested negotiation is accepted"""
my_request = req_id is not None
_request_dict = self._requested_negotiations
if req_id is None:
# I am not the requesting agent
req_id = mechanism.id
_request_dict = self._accepted_requests
neg = _request_dict.get(req_id, None)
if neg is None:
return
if my_request:
self.on_neg_request_accepted(req_id, mechanism)
if (
mechanism.id not in self._running_negotiations.keys()
or neg.negotiator is not None
):
self._running_negotiations[mechanism.id] = RunningNegotiationInfo(
extra=_request_dict[req_id].extra,
negotiator=neg.negotiator,
annotation=_request_dict[req_id].annotation,
uuid=req_id,
my_request=my_request,
)
_request_dict.pop(req_id, None)
[docs]
@abstractmethod
def on_negotiation_failure(
self,
partners: list[str],
annotation: dict[str, Any],
mechanism: NegotiatorMechanismInterface,
state: MechanismState,
) -> None:
"""Called whenever a negotiation ends without agreement"""
[docs]
def on_negotiation_failure_(
self,
partners: list[str],
annotation: dict[str, Any],
mechanism: NegotiatorMechanismInterface,
state: MechanismState,
) -> None:
"""Called whenever a negotiation ends without agreement"""
self.on_negotiation_failure(partners, annotation, mechanism, state)
self._running_negotiations.pop(mechanism.id, None)
[docs]
@abstractmethod
def on_negotiation_success(
self, contract: Contract, mechanism: NegotiatorMechanismInterface
) -> None:
"""Called whenever a negotiation ends with agreement"""
[docs]
def on_negotiation_success_(
self, contract: Contract, mechanism: NegotiatorMechanismInterface
) -> None:
"""Called whenever a negotiation ends with agreement"""
self.on_negotiation_success(contract, mechanism)
self._unsigned_contracts.add(contract)
self._running_negotiations.pop(mechanism.id, None)
[docs]
def on_contract_signed(self, contract: Contract) -> None:
"""Called whenever a contract is signed by all partners"""
[docs]
def on_contract_signed_(self, contract: Contract) -> None:
"""Called whenever a contract is signed by all partners"""
self.on_contract_signed(contract)
if contract in self._unsigned_contracts:
self._unsigned_contracts.remove(contract)
self.contracts.append(contract)
[docs]
def on_contract_cancelled(self, contract: Contract, rejectors: list[str]) -> None:
"""Called whenever at least a partner did not sign the contract"""
[docs]
def on_contract_cancelled_(self, contract: Contract, rejectors: list[str]) -> None:
"""Called whenever at least a partner did not sign the contract"""
self.on_contract_cancelled(contract, rejectors)
if contract in self._unsigned_contracts:
self._unsigned_contracts.remove(contract)
@abstractmethod
def _respond_to_negotiation_request(
self,
initiator: str,
partners: list[str],
issues: list[Issue],
annotation: dict[str, Any],
mechanism: NegotiatorMechanismInterface,
role: str | None,
req_id: str | None,
) -> Negotiator | None:
"""
Called by the mechanism to ask for joining a negotiation. The agent can refuse by returning a None
Args:
initiator: The ID of the agent that initiated the negotiation request
partners: The partner list (will include this agent)
issues: The list of issues
annotation: Any annotation specific to this negotiation.
mechanism: The mechanism that started the negotiation
role: The role of this agent in the negotiation
req_id: The req_id passed to the AWI when starting the negotiation (only to the initiator).
Returns:
None to refuse the negotiation or a `Negotiator` object appropriate to the given mechanism to accept it.
Remarks:
- It is expected that world designers will introduce a better way to respond and override this function to
call it
"""
[docs]
def respond_to_negotiation_request_(
self,
initiator: str,
partners: list[str],
issues: list[Issue],
annotation: dict[str, Any],
mechanism: NegotiatorMechanismInterface,
role: str | None,
req_id: str | None,
) -> Negotiator | None:
"""Called when a negotiation request is received"""
if req_id is not None:
# I am the one who requested this negotiation
info = self._requested_negotiations.get(req_id, None)
if info and info.negotiator is not None:
if (
mechanism.id not in self._running_negotiations.keys()
or info.negotiator is not None
):
self._running_negotiations[mechanism.id] = RunningNegotiationInfo(
extra=None,
negotiator=info.negotiator,
annotation=annotation,
uuid=req_id,
my_request=initiator == self.id,
)
return info.negotiator
negotiator = self._respond_to_negotiation_request(
initiator=initiator,
partners=partners,
issues=issues,
annotation=annotation,
mechanism=mechanism,
role=role,
req_id=req_id,
)
if negotiator is not None:
self._accepted_requests[mechanism.id] = NegotiationRequestInfo(
partners,
issues,
annotation,
uuid,
negotiator,
extra={"my_request": False},
requested=False,
)
if (
mechanism.id not in self._running_negotiations.keys()
or negotiator is not None
):
self._running_negotiations[mechanism.id] = RunningNegotiationInfo(
extra=None,
negotiator=negotiator,
annotation=annotation,
uuid=req_id,
my_request=initiator == self.id,
)
return negotiator
[docs]
def on_simulation_step_ended(self):
"""Will be called at the end of the simulation step after everything else"""
[docs]
def on_simulation_step_started(self):
"""Will be called at the beginning of the simulation step before everything else (except init)"""
[docs]
@abstractmethod
def step(self):
"""Called by the simulator at every simulation step"""
[docs]
@abstractmethod
def init(self):
"""Called to initialize the agent **after** the world is initialized. the AWI is accessible at this point."""
[docs]
def on_contracts_finalized(
self,
signed: list[Contract],
cancelled: list[Contract],
rejectors: list[list[str]],
) -> None:
"""
Called for all contracts in a single step to inform the agent about which were finally signed
and which were rejected by any agents (including itself)
Args:
signed: A list of signed contracts. These are binding
cancelled: A list of cancelled contracts. These are not binding
rejectors: A list of lists where each of the internal lists gives the rejectors of one of the
cancelled contracts. Notice that it is possible that this list is empty which
means that the contract other than being rejected by any agents (if that was possible in
the specific world).
Remarks:
The default implementation is to call `on_contract_signed` for singed contracts and `on_contract_cancelled`
for cancelled contracts
"""
for contract in signed:
self.on_contract_signed_(contract)
for contract, r in zip(cancelled, rejectors):
self.on_contract_cancelled_(contract, r)
[docs]
@abstractmethod
def set_renegotiation_agenda(
self, contract: Contract, breaches: list[Breach]
) -> RenegotiationRequest | None:
"""
Received by partners in ascending order of their total breach levels in order to set the
renegotiation agenda when contract execution fails
Args:
contract: The contract being breached
breaches: All breaches on `contract`
Returns:
Renegotiation agenda (issues to negotiate about to avoid reporting the breaches).
"""
[docs]
@abstractmethod
def respond_to_renegotiation_request(
self, contract: Contract, breaches: list[Breach], agenda: RenegotiationRequest
) -> Negotiator | None:
"""
Called to respond to a renegotiation request
Args:
agenda:
contract:
breaches:
Returns:
"""
[docs]
def sign_contract(self, contract: Contract) -> str | None:
"""Called after the signing delay from contract conclusion to sign the contract. Contracts become binding
only after they are signed."""
return self.id
[docs]
def sign_all_contracts(
self, contracts: list[Contract]
) -> None | str | dict[str, str | None] | list[str | None]:
"""Called to sign all contracts concluded in a single step by this agent
Args:
contracts: A list of contracts to sign/ refuse to sign
Return:
You can return any of the following:
- `None` to indicate refusing to sign all contracts.
- `str` (specifically, the agent ID) to indicate signing ALL contracts.
- `List[Optional[str]]` A list with a value for each input contract where `None` means refusal to sign that
contract and a string (agent ID) indicates acceptance to sign it. Note that in this case, the number of
values in the returned list must match that of the contacts (and they should obviously correspond to the
contracts).
- `Dict[str, Optional[str]]` A mapping from contract ID to either a `None` for rejection to sign or a string
(for acceptance to sign). Contracts with IDs not in the keys will assumed not to be signed.
Remarks:
- default implementation calls `sign_contract` for each contract returning the results
"""
return [self.sign_contract(contract) for contract in contracts]
[docs]
@abstractmethod
def on_contract_executed(self, contract: Contract) -> None:
"""
Called after successful contract execution for which the agent is one of the partners.
"""
[docs]
@abstractmethod
def on_contract_breached(
self, contract: Contract, breaches: list[Breach], resolution: Contract | None
) -> None:
"""
Called after complete processing of a contract that involved a breach.
Args:
contract: The contract
breaches: All breaches committed (even if they were resolved)
resolution: The resolution contract if re-negotiation was successful. None if not.
"""
def __str__(self):
return f"{self.name}"
__repr__ = __str__