Base converter

Classes:

Name	Description
BaseConverter	Abstract base class for all LLM response converters.
ConverterType	Enumeration of supported converter types for LLM responses.

Classes

BaseConverter

Bases: ABC

Abstract base class for all LLM response converters.

Subclasses should implement methods to convert standard and streaming LLM responses into PyAgenity's internal message/event formats.

Attributes:

Name	Type	Description
state	AgentState | None	Optional agent state for context during conversion.

Methods:

Name	Description
__init__	Initialize the converter.
convert_response	Convert a standard agent response to a Message.
convert_streaming_response	Convert a streaming agent response to an async generator of EventModel or Message.

Source code in pyagenity/adapters/llm/base_converter.py

class BaseConverter(ABC):
    """
    Abstract base class for all LLM response converters.

    Subclasses should implement methods to convert standard and streaming
    LLM responses into PyAgenity's internal message/event formats.

    Attributes:
        state (AgentState | None): Optional agent state for context during conversion.
    """

    def __init__(self, state: AgentState | None = None) -> None:
        """
        Initialize the converter.

        Args:
            state (AgentState | None): Optional agent state for context during conversion.
        """
        self.state = state

    @abstractmethod
    async def convert_response(self, response: Any) -> Message:
        """
        Convert a standard agent response to a Message.

        Args:
            response (Any): The raw response from the LLM or agent.

        Returns:
            Message: The converted message object.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Conversion not implemented for this converter")

    @abstractmethod
    async def convert_streaming_response(
        self,
        config: dict,
        node_name: str,
        response: Any,
        meta: dict | None = None,
    ) -> AsyncGenerator[EventModel | Message, None]:
        """
        Convert a streaming agent response to an async generator of EventModel or Message.

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            response (Any): The raw streaming response from the LLM or agent.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            EventModel | Message: Chunks of the converted streaming response.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Streaming not implemented for this converter")

Attributes

state instance-attribute

state = state

Functions

__init__

__init__(state=None)

Initialize the converter.

Parameters:

Name	Type	Description	Default
state	AgentState | None	Optional agent state for context during conversion.	None

Source code in pyagenity/adapters/llm/base_converter.py

def __init__(self, state: AgentState | None = None) -> None:
    """
    Initialize the converter.

    Args:
        state (AgentState | None): Optional agent state for context during conversion.
    """
    self.state = state

convert_response abstractmethod async

convert_response(response)

Convert a standard agent response to a Message.

Parameters:

Name	Type	Description	Default
response	Any	The raw response from the LLM or agent.	required

Returns:

Name	Type	Description
Message	Message	The converted message object.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_response(self, response: Any) -> Message:
    """
    Convert a standard agent response to a Message.

    Args:
        response (Any): The raw response from the LLM or agent.

    Returns:
        Message: The converted message object.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Conversion not implemented for this converter")

convert_streaming_response abstractmethod async

convert_streaming_response(config, node_name, response, meta=None)

Convert a streaming agent response to an async generator of EventModel or Message.

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters.	required
node_name	str	Name of the node processing the response.	required
response	Any	The raw streaming response from the LLM or agent.	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Type	Description
AsyncGenerator[EventModel | Message, None]	EventModel | Message: Chunks of the converted streaming response.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_streaming_response(
    self,
    config: dict,
    node_name: str,
    response: Any,
    meta: dict | None = None,
) -> AsyncGenerator[EventModel | Message, None]:
    """
    Convert a streaming agent response to an async generator of EventModel or Message.

    Args:
        config (dict): Node configuration parameters.
        node_name (str): Name of the node processing the response.
        response (Any): The raw streaming response from the LLM or agent.
        meta (dict | None): Optional metadata for conversion.

    Yields:
        EventModel | Message: Chunks of the converted streaming response.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Streaming not implemented for this converter")

ConverterType

Bases: Enum

Enumeration of supported converter types for LLM responses.

Attributes:

Name	Type	Description
ANTHROPIC
CUSTOM
GOOGLE
LITELLM
OPENAI

Source code in pyagenity/adapters/llm/base_converter.py

class ConverterType(Enum):
    """Enumeration of supported converter types for LLM responses."""

    OPENAI = "openai"
    LITELLM = "litellm"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    CUSTOM = "custom"

Attributes

ANTHROPIC class-attribute instance-attribute

ANTHROPIC = 'anthropic'

CUSTOM class-attribute instance-attribute

CUSTOM = 'custom'

GOOGLE class-attribute instance-attribute

GOOGLE = 'google'

LITELLM class-attribute instance-attribute

LITELLM = 'litellm'

OPENAI class-attribute instance-attribute

OPENAI = 'openai'