AgentDescriptionGuard

autogen.AgentDescriptionGuard #

AgentDescriptionGuard(agent)

Manages soft-signal description mutation for LLM-based speaker selection.

Wraps an agent via composition and toggles an [UNAVAILABLE] prefix on its description so that LLM-based auto-selection is less likely to choose it when it is unavailable (e.g. circuit breaker open). Restore on recovery.

Warning

This guard owns the [UNAVAILABLE] prefix on agent.description. External code may modify the description between mark_unavailable and mark_available calls -- those changes are preserved. However, if external code adds or removes the [UNAVAILABLE] prefix itself, the guard's state becomes inconsistent.

Source code in autogen/agentchat/eligibility_policy.py

def __init__(self, agent: Agent) -> None:
    self._agent = agent
    self._lock = threading.Lock()
    self._original_description: str | None | object = self._UNSET

mark_unavailable #

mark_unavailable()

Prepend [UNAVAILABLE] to agent.description (idempotent, thread-safe).

Note

If agent.description is None, it is treated as "" for the prefix but the original None value is remembered and restored by :meth:mark_available.

Source code in autogen/agentchat/eligibility_policy.py

def mark_unavailable(self) -> None:
    """Prepend [UNAVAILABLE] to agent.description (idempotent, thread-safe).

    Note:
        If ``agent.description`` is ``None``, it is treated as ``""`` for the
        prefix but the original ``None`` value is remembered and restored by
        :meth:`mark_available`.
    """
    with self._lock:
        current = self._agent.description or ""
        if current.startswith(_UNAVAILABLE_PREFIX):
            return
        self._original_description = self._agent.description
        self._agent.description = _UNAVAILABLE_PREFIX + current

mark_available #

mark_available()

Strip [UNAVAILABLE] prefix from description (thread-safe).

If the current agent.description does not start with the [UNAVAILABLE] prefix this is a no-op. Otherwise, the prefix is stripped and the description is restored to the value it held at the time :meth:mark_unavailable was called. Any modifications made to the suffix after the prefix (e.g. by external code appending text) are preserved: the prefix is stripped from the live value, not replaced wholesale.

Note

The check is structural (prefix present or absent) rather than based on internal guard state. If another source added the [UNAVAILABLE] prefix without going through this guard instance, calling this method will strip it -- see the class-level Warning for details on multi-guard scenarios.

Source code in autogen/agentchat/eligibility_policy.py

def mark_available(self) -> None:
    """Strip [UNAVAILABLE] prefix from description (thread-safe).

    If the current ``agent.description`` does not start with the
    ``[UNAVAILABLE]`` prefix this is a no-op.  Otherwise, the prefix is
    stripped and the description is restored to the value it held at the
    time :meth:`mark_unavailable` was called.  Any modifications made to
    the suffix after the prefix (e.g. by external code appending text) are
    preserved: the prefix is stripped from the live value, not replaced
    wholesale.

    Note:
        The check is structural (prefix present or absent) rather than
        based on internal guard state.  If another source added the
        ``[UNAVAILABLE]`` prefix without going through this guard instance,
        calling this method will strip it -- see the class-level Warning
        for details on multi-guard scenarios.
    """
    with self._lock:
        current = self._agent.description or ""
        if not current.startswith(_UNAVAILABLE_PREFIX):
            # Clear stale state -- external code may have removed the prefix.
            self._original_description = self._UNSET
            return
        # Only restore original if THIS guard placed the prefix.
        if self._original_description is self._UNSET:
            # Another source added the prefix; strip it but don't guess
            # what the original value was -- just remove the prefix text.
            self._agent.description = current[len(_UNAVAILABLE_PREFIX) :]
        elif self._original_description is None:
            # Original was None and no suffix was appended -> restore None.
            suffix = current[len(_UNAVAILABLE_PREFIX) :]
            self._agent.description = None if suffix == "" else suffix
        else:
            self._agent.description = current[len(_UNAVAILABLE_PREFIX) :]
        self._original_description = self._UNSET