Base Classes

NegologNegotiatorWrapper

`negmas_negolog.NegologNegotiatorWrapper`

Bases: SAONegotiator, ABC

Base wrapper class that bridges NegoLog agents to NegMAS SAONegotiator.

This wrapper translates between the two frameworks: - Converts NegMAS state/offers to NegoLog format - Converts NegoLog actions to NegMAS responses - Manages the lifecycle of the wrapped NegoLog agent

Subclasses should set the negolog_agent_class class attribute to the NegoLog agent class they wrap.

Source code in src/negmas_negolog/common.py

class NegologNegotiatorWrapper(SAONegotiator, ABC):
    """
    Base wrapper class that bridges NegoLog agents to NegMAS SAONegotiator.

    This wrapper translates between the two frameworks:
    - Converts NegMAS state/offers to NegoLog format
    - Converts NegoLog actions to NegMAS responses
    - Manages the lifecycle of the wrapped NegoLog agent

    Subclasses should set the `negolog_agent_class` class attribute to the
    NegoLog agent class they wrap.
    """

    # Subclasses must set this to the NegoLog agent class
    negolog_agent_class: Type[AbstractAgent]

    def __init__(
        self,
        preferences: BaseUtilityFunction | None = None,
        ufun: BaseUtilityFunction | None = None,
        name: str | None = None,
        parent: Controller | None = None,
        owner: Agent | None = None,
        id: str | None = None,
        type_name: str | None = None,
        session_time: int = 180,  # Default 3 minutes
        **kwargs,
    ):
        """
        Initialize the wrapper.

        Args:
            preferences: NegMAS preferences/utility function
            ufun: Utility function (overrides preferences if given)
            name: Negotiator name
            parent: Parent controller
            owner: Agent that owns this negotiator
            id: Unique identifier
            type_name: Type name for serialization
            session_time: Session time in seconds for NegoLog agent
            **kwargs: Additional arguments passed to parent
        """
        super().__init__(
            preferences=preferences,
            ufun=ufun,
            name=name,
            parent=parent,
            owner=owner,
            id=id,
            type_name=type_name,
            **kwargs,
        )

        self._session_time = session_time
        self._negolog_agent: Optional[AbstractAgent] = None
        self._preference_adapter: Optional[NegologPreferenceAdapter] = None
        self._issues: List[Issue] = []
        self._issue_names: List[str] = []
        self._initialized = False
        # Track the current negotiation step and cache act() results
        # This prevents calling act() multiple times per step (which corrupts
        # agent state for agents like AgentBuyog that increment round counters in act())
        self._current_step: int = -1
        self._cached_action = None

    def on_negotiation_start(self, state: SAOState) -> None:
        """
        Called when negotiation starts. Initialize the NegoLog agent.

        Args:
            state: Initial negotiation state
        """
        super().on_negotiation_start(state)
        self._initialize_negolog_agent()

    def _initialize_negolog_agent(self) -> None:
        """Initialize the wrapped NegoLog agent with the current negotiation context."""
        if self._initialized:
            return

        if not self.ufun:
            raise ValueError("Utility function must be set before negotiation starts")

        if not self.nmi:
            raise ValueError("NMI must be available before negotiation starts")

        # Build NegoLog Issues from NegMAS outcome space
        outcome_space = self.nmi.outcome_space
        if hasattr(outcome_space, "issues"):
            negmas_issues = outcome_space.issues
        else:
            raise ValueError("Outcome space must have issues defined")

        self._issues = []
        self._issue_names = []

        for i, negmas_issue in enumerate(negmas_issues):
            issue_name = getattr(negmas_issue, "name", f"issue_{i}")
            self._issue_names.append(issue_name)

            # Get all possible values for this issue
            if hasattr(negmas_issue, "all"):
                values = list(negmas_issue.all)
            elif hasattr(negmas_issue, "values"):
                values = list(negmas_issue.values)
            else:
                # Try to enumerate
                values = list(negmas_issue)

            # Convert values to strings if needed
            values = [str(v) if not isinstance(v, str) else v for v in values]

            negolog_issue = Issue(issue_name, values)
            self._issues.append(negolog_issue)

        # Create preference adapter
        reservation_value = getattr(self.ufun, "reserved_value", 0.0)
        if reservation_value == float("-inf"):
            reservation_value = 0.0

        self._preference_adapter = NegologPreferenceAdapter(
            ufun=self.ufun,
            issues=self._issues,
            issue_names=self._issue_names,
            reservation_value=reservation_value,
        )

        # Create the NegoLog agent
        self._negolog_agent = self.negolog_agent_class(
            preference=self._preference_adapter,
            session_time=self._session_time,
            estimators=[],  # No opponent models by default
        )

        # Initialize the agent
        self._negolog_agent.initiate(opponent_name=None)
        self._initialized = True
        self._current_step = -1
        self._cached_action = None

    def _get_relative_time(self, state: SAOState) -> float:
        """
        Get the relative time (0 to 1) for the NegoLog agent.

        Args:
            state: Current negotiation state

        Returns:
            Relative time between 0 and 1
        """
        return state.relative_time

    def _get_action_for_step(self, state: SAOState):
        """
        Get the NegoLog action for the current step, with caching.

        This method ensures act() is only called once per negotiation step,
        which is critical because some NegoLog agents (like AgentBuyog) maintain
        internal round counters that increment on each act() call.

        Args:
            state: Current negotiation state

        Returns:
            The cached or newly computed action from the NegoLog agent
        """
        current_step = state.step
        if current_step != self._current_step:
            # New step - call act() and cache the result
            self._current_step = current_step
            t = self._get_relative_time(state)
            self._cached_action = self._negolog_agent.act(t)
        return self._cached_action

    def _outcome_to_bid(self, outcome: Outcome) -> Bid:
        """Convert a NegMAS Outcome to a NegoLog Bid."""
        if self._preference_adapter is None:
            raise ValueError("Preference adapter not initialized")
        return self._preference_adapter._outcome_to_bid(outcome)

    def _bid_to_outcome(self, bid: Bid) -> Outcome:
        """Convert a NegoLog Bid to a NegMAS Outcome."""
        if self._preference_adapter is None:
            raise ValueError("Preference adapter not initialized")
        return self._preference_adapter._bid_to_outcome(bid)

    def propose(self, state: SAOState, dest: str | None = None) -> Outcome | None:
        """
        Generate a proposal using the wrapped NegoLog agent.

        Args:
            state: Current negotiation state
            dest: Destination negotiator ID (ignored)

        Returns:
            Outcome tuple to propose, or None
        """
        if not self._initialized:
            self._initialize_negolog_agent()

        if self._negolog_agent is None:
            return None

        # Get action (cached per step to avoid multiple act() calls)
        action = self._get_action_for_step(state)

        if action is None:
            return None

        # Convert NegoLog bid to NegMAS outcome
        return self._bid_to_outcome(action.bid)

    def respond(self, state: SAOState, source: str | None = None) -> ResponseType:
        """
        Respond to an offer using the wrapped NegoLog agent.

        Args:
            state: Current negotiation state (access offer via state.current_offer)
            source: ID of negotiator who made the offer

        Returns:
            ResponseType indicating acceptance/rejection
        """
        if not self._initialized:
            self._initialize_negolog_agent()

        if self._negolog_agent is None:
            return ResponseType.REJECT_OFFER

        offer = state.current_offer
        if offer is None:
            return ResponseType.REJECT_OFFER

        t = self._get_relative_time(state)

        # Convert offer to NegoLog bid and notify agent
        bid = self._outcome_to_bid(offer)
        self._negolog_agent.receive_bid(bid, t)

        # Invalidate the cached action since we received a new bid
        # The agent may now decide differently (e.g., to accept)
        self._current_step = -1
        self._cached_action = None

        # Get action from NegoLog agent (will be cached for this step)
        action = self._get_action_for_step(state)

        if action is None:
            return ResponseType.REJECT_OFFER

        # Check if action is Accept
        if isinstance(action, Accept):
            return ResponseType.ACCEPT_OFFER

        return ResponseType.REJECT_OFFER

    def on_negotiation_end(self, state: SAOState) -> None:
        """
        Called when negotiation ends. Clean up the NegoLog agent.

        Args:
            state: Final negotiation state
        """
        super().on_negotiation_end(state)

        if self._negolog_agent is not None:
            is_accept = state.agreement is not None
            t = self._get_relative_time(state)
            self._negolog_agent.terminate(is_accept, "opponent", t)

        # Reset state
        self._negolog_agent = None
        self._preference_adapter = None
        self._initialized = False

`init(preferences=None, ufun=None, name=None, parent=None, owner=None, id=None, type_name=None, session_time=180, **kwargs)`

Initialize the wrapper.

Parameters:

Name	Type	Description	Default
`preferences`	`BaseUtilityFunction \| None`	NegMAS preferences/utility function	`None`
`ufun`	`BaseUtilityFunction \| None`	Utility function (overrides preferences if given)	`None`
`name`	`str \| None`	Negotiator name	`None`
`parent`	`Controller \| None`	Parent controller	`None`
`owner`	`Agent \| None`	Agent that owns this negotiator	`None`
`id`	`str \| None`	Unique identifier	`None`
`type_name`	`str \| None`	Type name for serialization	`None`
`session_time`	`int`	Session time in seconds for NegoLog agent	`180`
`**kwargs`		Additional arguments passed to parent	`{}`

Source code in src/negmas_negolog/common.py

def __init__(
    self,
    preferences: BaseUtilityFunction | None = None,
    ufun: BaseUtilityFunction | None = None,
    name: str | None = None,
    parent: Controller | None = None,
    owner: Agent | None = None,
    id: str | None = None,
    type_name: str | None = None,
    session_time: int = 180,  # Default 3 minutes
    **kwargs,
):
    """
    Initialize the wrapper.

    Args:
        preferences: NegMAS preferences/utility function
        ufun: Utility function (overrides preferences if given)
        name: Negotiator name
        parent: Parent controller
        owner: Agent that owns this negotiator
        id: Unique identifier
        type_name: Type name for serialization
        session_time: Session time in seconds for NegoLog agent
        **kwargs: Additional arguments passed to parent
    """
    super().__init__(
        preferences=preferences,
        ufun=ufun,
        name=name,
        parent=parent,
        owner=owner,
        id=id,
        type_name=type_name,
        **kwargs,
    )

    self._session_time = session_time
    self._negolog_agent: Optional[AbstractAgent] = None
    self._preference_adapter: Optional[NegologPreferenceAdapter] = None
    self._issues: List[Issue] = []
    self._issue_names: List[str] = []
    self._initialized = False
    # Track the current negotiation step and cache act() results
    # This prevents calling act() multiple times per step (which corrupts
    # agent state for agents like AgentBuyog that increment round counters in act())
    self._current_step: int = -1
    self._cached_action = None

`on_negotiation_end(state)`

Called when negotiation ends. Clean up the NegoLog agent.

Parameters:

Name	Type	Description	Default
`state`	`SAOState`	Final negotiation state	required

Source code in src/negmas_negolog/common.py

def on_negotiation_end(self, state: SAOState) -> None:
    """
    Called when negotiation ends. Clean up the NegoLog agent.

    Args:
        state: Final negotiation state
    """
    super().on_negotiation_end(state)

    if self._negolog_agent is not None:
        is_accept = state.agreement is not None
        t = self._get_relative_time(state)
        self._negolog_agent.terminate(is_accept, "opponent", t)

    # Reset state
    self._negolog_agent = None
    self._preference_adapter = None
    self._initialized = False

`on_negotiation_start(state)`

Called when negotiation starts. Initialize the NegoLog agent.

Parameters:

Name	Type	Description	Default
`state`	`SAOState`	Initial negotiation state	required

Source code in src/negmas_negolog/common.py

def on_negotiation_start(self, state: SAOState) -> None:
    """
    Called when negotiation starts. Initialize the NegoLog agent.

    Args:
        state: Initial negotiation state
    """
    super().on_negotiation_start(state)
    self._initialize_negolog_agent()

`propose(state, dest=None)`

Generate a proposal using the wrapped NegoLog agent.

Parameters:

Name	Type	Description	Default
`state`	`SAOState`	Current negotiation state	required
`dest`	`str \| None`	Destination negotiator ID (ignored)	`None`

Returns:

Type	Description
`Outcome \| None`	Outcome tuple to propose, or None

Source code in src/negmas_negolog/common.py

def propose(self, state: SAOState, dest: str | None = None) -> Outcome | None:
    """
    Generate a proposal using the wrapped NegoLog agent.

    Args:
        state: Current negotiation state
        dest: Destination negotiator ID (ignored)

    Returns:
        Outcome tuple to propose, or None
    """
    if not self._initialized:
        self._initialize_negolog_agent()

    if self._negolog_agent is None:
        return None

    # Get action (cached per step to avoid multiple act() calls)
    action = self._get_action_for_step(state)

    if action is None:
        return None

    # Convert NegoLog bid to NegMAS outcome
    return self._bid_to_outcome(action.bid)

`respond(state, source=None)`

Respond to an offer using the wrapped NegoLog agent.

Parameters:

Name	Type	Description	Default
`state`	`SAOState`	Current negotiation state (access offer via state.current_offer)	required
`source`	`str \| None`	ID of negotiator who made the offer	`None`

Returns:

Type	Description
`ResponseType`	ResponseType indicating acceptance/rejection

Source code in src/negmas_negolog/common.py

def respond(self, state: SAOState, source: str | None = None) -> ResponseType:
    """
    Respond to an offer using the wrapped NegoLog agent.

    Args:
        state: Current negotiation state (access offer via state.current_offer)
        source: ID of negotiator who made the offer

    Returns:
        ResponseType indicating acceptance/rejection
    """
    if not self._initialized:
        self._initialize_negolog_agent()

    if self._negolog_agent is None:
        return ResponseType.REJECT_OFFER

    offer = state.current_offer
    if offer is None:
        return ResponseType.REJECT_OFFER

    t = self._get_relative_time(state)

    # Convert offer to NegoLog bid and notify agent
    bid = self._outcome_to_bid(offer)
    self._negolog_agent.receive_bid(bid, t)

    # Invalidate the cached action since we received a new bid
    # The agent may now decide differently (e.g., to accept)
    self._current_step = -1
    self._cached_action = None

    # Get action from NegoLog agent (will be cached for this step)
    action = self._get_action_for_step(state)

    if action is None:
        return ResponseType.REJECT_OFFER

    # Check if action is Accept
    if isinstance(action, Accept):
        return ResponseType.ACCEPT_OFFER

    return ResponseType.REJECT_OFFER

The base class for all NegoLog agent wrappers. Inherits from negmas.sao.SAONegotiator.

Usage

To create a custom wrapper for a NegoLog agent:

from negmas_negolog import NegologNegotiatorWrapper

class MyCustomAgent(NegologNegotiatorWrapper):
    """Wrapper for a custom NegoLog agent."""
    negolog_agent_class = MyNegoLogAgent

Constructor Parameters

Parameter	Type	Default	Description
`preferences`	`BaseUtilityFunction`	`None`	NegMAS preferences/utility function
`ufun`	`BaseUtilityFunction`	`None`	Utility function (overrides preferences)
`name`	`str`	`None`	Negotiator name
`session_time`	`int`	`180`	Session time in seconds for NegoLog agent

Key Methods

`propose(state, dest=None)`

Generate a proposal using the wrapped NegoLog agent.

Parameters: - state: Current SAOState - dest: Destination negotiator ID (optional)

Returns: Outcome tuple or None

`respond(state, source=None)`

Respond to an offer using the wrapped NegoLog agent.

Parameters: - state: Current SAOState (access offer via state.current_offer) - source: ID of negotiator who made the offer

Returns: ResponseType (ACCEPT_OFFER or REJECT_OFFER)

NegologPreferenceAdapter

`negmas_negolog.NegologPreferenceAdapter`

Bases: Preference

Adapter that wraps a NegMAS utility function to provide NegoLog Preference interface.

This allows NegoLog agents to use NegMAS utility functions transparently.

Source code in src/negmas_negolog/common.py

class NegologPreferenceAdapter(Preference):
    """
    Adapter that wraps a NegMAS utility function to provide NegoLog Preference interface.

    This allows NegoLog agents to use NegMAS utility functions transparently.
    """

    def __init__(
        self,
        ufun: BaseUtilityFunction,
        issues: List[Issue],
        issue_names: List[str],
        reservation_value: float = 0.0,
    ):
        """
        Initialize the preference adapter.

        Args:
            ufun: NegMAS utility function
            issues: List of NegoLog Issue objects
            issue_names: List of issue names (for mapping)
            reservation_value: Reservation value (utility if negotiation fails)
        """
        # Initialize parent without loading from JSON
        super().__init__(profile_json_path=None, generate_bids=False)

        # IMPORTANT: Keep profile_json_path as None (not empty string) so that
        # EstimatedPreference and opponent models don't try to open a file.
        # The parent constructor sets it to "" but we override it here.
        self.profile_json_path = None

        self._ufun = ufun
        self._issues = issues
        self._reservation_value = reservation_value
        self._issue_names = issue_names

        # Build issue weights from the NegMAS ufun if it's a LinearAdditive type
        # This is needed for opponent models that use these weights
        self._issue_weights = {}
        self._value_weights = {}

        # Try to extract weights from the NegMAS ufun
        if hasattr(ufun, "weights") and hasattr(ufun, "values"):
            ufun_weights = ufun.weights
            ufun_values = ufun.values
            for i, issue in enumerate(issues):
                self._issue_weights[issue] = float(ufun_weights[i])
                self._value_weights[issue] = {}
                # ufun_values[i] is a TableFun with a 'mapping' attribute
                val_fun = ufun_values[i]
                if hasattr(val_fun, "mapping"):
                    mapping = val_fun.mapping
                elif callable(val_fun):
                    # Try to call it for each value
                    mapping = {v: val_fun(v) for v in issue.values}
                else:
                    mapping = {}
                for value in issue.values:
                    val_weight = mapping.get(value, 0.5)
                    self._value_weights[issue][value] = float(val_weight)
        else:
            # Fall back to equal weights
            for issue in issues:
                self._issue_weights[issue] = 1.0 / len(issues)
                self._value_weights[issue] = {}
                for value in issue.values:
                    self._value_weights[issue][value] = 0.5

    def get_utility(self, bid: Bid) -> float:
        """
        Calculate utility using the NegMAS ufun.

        Args:
            bid: NegoLog Bid object

        Returns:
            Utility value from the NegMAS ufun
        """
        # Convert NegoLog Bid to NegMAS Outcome (tuple)
        outcome = self._bid_to_outcome(bid)
        return float(self._ufun(outcome))

    def _bid_to_outcome(self, bid: Bid) -> Outcome:
        """Convert a NegoLog Bid to a NegMAS Outcome tuple."""
        values = []
        for issue_name in self._issue_names:
            # Find the issue by name
            for issue in self._issues:
                if issue.name == issue_name:
                    values.append(bid[issue])
                    break
        return tuple(values)

    def _outcome_to_bid(self, outcome: Outcome) -> Bid:
        """Convert a NegMAS Outcome tuple to a NegoLog Bid."""
        content = {}
        for i, issue in enumerate(self._issues):
            content[issue] = outcome[i]
        bid = Bid(content)
        bid.utility = self.get_utility(bid)
        return bid

    @property
    def bids(self) -> List[Bid]:
        """
        Generate all possible bids lazily.

        Returns:
            Sorted list of all possible bids (descending by utility)
        """
        if len(self._bids) > 0:
            return self._bids

        # Generate all bid combinations
        bids = [Bid({}, -1)]

        for issue in self._issues:
            new_bids = []
            for value_name in issue.values:
                for bid in bids:
                    _bid = bid.copy()
                    _bid[issue] = value_name
                    new_bids.append(_bid)
            bids = new_bids

        # Assign utilities and sort
        for bid in bids:
            bid.utility = self.get_utility(bid)

        bids = sorted(bids, reverse=True)
        self._bids = bids

        return bids

`bids` `property`

Generate all possible bids lazily.

Returns:

Type	Description
`List[Bid]`	Sorted list of all possible bids (descending by utility)

`init(ufun, issues, issue_names, reservation_value=0.0)`

Initialize the preference adapter.

Parameters:

Name	Type	Description	Default
`ufun`	`BaseUtilityFunction`	NegMAS utility function	required
`issues`	`List[Issue]`	List of NegoLog Issue objects	required
`issue_names`	`List[str]`	List of issue names (for mapping)	required
`reservation_value`	`float`	Reservation value (utility if negotiation fails)	`0.0`

Source code in src/negmas_negolog/common.py

def __init__(
    self,
    ufun: BaseUtilityFunction,
    issues: List[Issue],
    issue_names: List[str],
    reservation_value: float = 0.0,
):
    """
    Initialize the preference adapter.

    Args:
        ufun: NegMAS utility function
        issues: List of NegoLog Issue objects
        issue_names: List of issue names (for mapping)
        reservation_value: Reservation value (utility if negotiation fails)
    """
    # Initialize parent without loading from JSON
    super().__init__(profile_json_path=None, generate_bids=False)

    # IMPORTANT: Keep profile_json_path as None (not empty string) so that
    # EstimatedPreference and opponent models don't try to open a file.
    # The parent constructor sets it to "" but we override it here.
    self.profile_json_path = None

    self._ufun = ufun
    self._issues = issues
    self._reservation_value = reservation_value
    self._issue_names = issue_names

    # Build issue weights from the NegMAS ufun if it's a LinearAdditive type
    # This is needed for opponent models that use these weights
    self._issue_weights = {}
    self._value_weights = {}

    # Try to extract weights from the NegMAS ufun
    if hasattr(ufun, "weights") and hasattr(ufun, "values"):
        ufun_weights = ufun.weights
        ufun_values = ufun.values
        for i, issue in enumerate(issues):
            self._issue_weights[issue] = float(ufun_weights[i])
            self._value_weights[issue] = {}
            # ufun_values[i] is a TableFun with a 'mapping' attribute
            val_fun = ufun_values[i]
            if hasattr(val_fun, "mapping"):
                mapping = val_fun.mapping
            elif callable(val_fun):
                # Try to call it for each value
                mapping = {v: val_fun(v) for v in issue.values}
            else:
                mapping = {}
            for value in issue.values:
                val_weight = mapping.get(value, 0.5)
                self._value_weights[issue][value] = float(val_weight)
    else:
        # Fall back to equal weights
        for issue in issues:
            self._issue_weights[issue] = 1.0 / len(issues)
            self._value_weights[issue] = {}
            for value in issue.values:
                self._value_weights[issue][value] = 0.5

`get_utility(bid)`

Calculate utility using the NegMAS ufun.

Parameters:

Name	Type	Description	Default
`bid`	`Bid`	NegoLog Bid object	required

Returns:

Type	Description
`float`	Utility value from the NegMAS ufun

Source code in src/negmas_negolog/common.py

def get_utility(self, bid: Bid) -> float:
    """
    Calculate utility using the NegMAS ufun.

    Args:
        bid: NegoLog Bid object

    Returns:
        Utility value from the NegMAS ufun
    """
    # Convert NegoLog Bid to NegMAS Outcome (tuple)
    outcome = self._bid_to_outcome(bid)
    return float(self._ufun(outcome))

Adapts NegMAS utility functions to NegoLog's Preference interface. Used internally by the wrappers.

How It Works

Takes a NegMAS BaseUtilityFunction
Creates NegoLog Issue objects from the NegMAS outcome space
Evaluates bids using the NegMAS utility function
Provides issue/value weights for opponent models

This adapter allows NegoLog agents to use NegMAS utility functions transparently, without requiring JSON preference files.

Base Classes

NegologNegotiatorWrapper

negmas_negolog.NegologNegotiatorWrapper

__init__(preferences=None, ufun=None, name=None, parent=None, owner=None, id=None, type_name=None, session_time=180, **kwargs)

on_negotiation_end(state)

on_negotiation_start(state)

propose(state, dest=None)

respond(state, source=None)

Usage

Constructor Parameters

Key Methods

propose(state, dest=None)

respond(state, source=None)

NegologPreferenceAdapter

negmas_negolog.NegologPreferenceAdapter

bids property

__init__(ufun, issues, issue_names, reservation_value=0.0)

get_utility(bid)

How It Works

`negmas_negolog.NegologNegotiatorWrapper`

`init(preferences=None, ufun=None, name=None, parent=None, owner=None, id=None, type_name=None, session_time=180, **kwargs)`

`on_negotiation_end(state)`

`on_negotiation_start(state)`

`propose(state, dest=None)`

`respond(state, source=None)`

`propose(state, dest=None)`

`respond(state, source=None)`

`negmas_negolog.NegologPreferenceAdapter`

`bids` `property`

`init(ufun, issues, issue_names, reservation_value=0.0)`

`get_utility(bid)`