Base context

Abstract base class for context management in PyAgenity agent graphs.

This module provides BaseContextManager, which defines the interface for trimming and managing message context in agent state objects.

Classes:

Name	Description
BaseContextManager	Abstract base class for context management in AI interactions.

Attributes:

Name	Type	Description
S
logger

Attributes

S module-attribute

S = TypeVar('S', bound=AgentState)

logger module-attribute

logger = getLogger(__name__)

Classes

BaseContextManager

Bases: ABC

Abstract base class for context management in AI interactions.

Subclasses should implement trim_context as either a synchronous or asynchronous method. Generic over AgentState or its subclasses.

Methods:

Name	Description
atrim_context	Trim context based on message count asynchronously.
trim_context	Trim context based on message count. Can be sync or async.

Source code in pyagenity/state/base_context.py

class BaseContextManager[S](ABC):
    """
    Abstract base class for context management in AI interactions.

    Subclasses should implement `trim_context` as either a synchronous or asynchronous method.
    Generic over AgentState or its subclasses.
    """

    @abstractmethod
    def trim_context(self, state: S) -> S:
        """
        Trim context based on message count. Can be sync or async.

        Subclasses may implement as either a synchronous or asynchronous method.

        Args:
            state: The state containing context to be trimmed.

        Returns:
            The state with trimmed context, either directly or as an awaitable.
        """
        raise NotImplementedError("Subclasses must implement this method (sync or async)")

    @abstractmethod
    async def atrim_context(self, state: S) -> S:
        """
        Trim context based on message count asynchronously.

        Args:
            state: The state containing context to be trimmed.

        Returns:
            The state with trimmed context.
        """
        raise NotImplementedError("Subclasses must implement this method")

Functions

atrim_context abstractmethod async

atrim_context(state)

Trim context based on message count asynchronously.

Parameters:

Name	Type	Description	Default
state	S	The state containing context to be trimmed.	required

Returns:

Type	Description
S	The state with trimmed context.

Source code in pyagenity/state/base_context.py

@abstractmethod
async def atrim_context(self, state: S) -> S:
    """
    Trim context based on message count asynchronously.

    Args:
        state: The state containing context to be trimmed.

    Returns:
        The state with trimmed context.
    """
    raise NotImplementedError("Subclasses must implement this method")

trim_context abstractmethod

trim_context(state)

Trim context based on message count. Can be sync or async.

Subclasses may implement as either a synchronous or asynchronous method.

Parameters:

Name	Type	Description	Default
state	S	The state containing context to be trimmed.	required

Returns:

Type	Description
S	The state with trimmed context, either directly or as an awaitable.

Source code in pyagenity/state/base_context.py

@abstractmethod
def trim_context(self, state: S) -> S:
    """
    Trim context based on message count. Can be sync or async.

    Subclasses may implement as either a synchronous or asynchronous method.

    Args:
        state: The state containing context to be trimmed.

    Returns:
        The state with trimmed context, either directly or as an awaitable.
    """
    raise NotImplementedError("Subclasses must implement this method (sync or async)")