agentchat.contrib.capabilities.transforms
MessageTransform
class MessageTransform(Protocol)
Defines a contract for message transformation.
Classes implementing this protocol should provide an apply_transform
method
that takes a list of messages and returns the transformed list.
apply_transform
def apply_transform(messages: List[Dict]) -> List[Dict]
Applies a transformation to a list of messages.
Arguments:
messages
- A list of dictionaries representing messages.
Returns:
A new list of dictionaries containing the transformed messages.
get_logs
def get_logs(pre_transform_messages: List[Dict],
post_transform_messages: List[Dict]) -> Tuple[str, bool]
Creates the string including the logs of the transformation
Alongside the string, it returns a boolean indicating whether the transformation had an effect or not.
Arguments:
pre_transform_messages
- A list of dictionaries representing messages before the transformation.post_transform_messages
- A list of dictionaries representig messages after the transformation.
Returns:
A tuple with a string with the logs and a flag indicating whether the transformation had an effect or not.
MessageHistoryLimiter
class MessageHistoryLimiter()
Limits the number of messages considered by an agent for response generation.
This transform keeps only the most recent messages up to the specified maximum number of messages (max_messages). It trims the conversation history by removing older messages, retaining only the most recent messages.
__init__
def __init__(max_messages: Optional[int] = None,
keep_first_message: bool = False)
Arguments:
max_messages Optional[int]: Maximum number of messages to keep in the context. Must be greater than 0 if not None. keep_first_message bool: Whether to keep the original first message in the conversation history. Defaults to False.
apply_transform
def apply_transform(messages: List[Dict]) -> List[Dict]
Truncates the conversation history to the specified maximum number of messages.
This method returns a new list containing the most recent messages up to the specified maximum number of messages (max_messages). If max_messages is None, it returns the original list of messages unmodified.
Arguments:
messages
List[Dict] - The list of messages representing the conversation history.
Returns:
List[Dict]
- A new list containing the most recent messages up to the specified maximum.
MessageTokenLimiter
class MessageTokenLimiter()
Truncates messages to meet token limits for efficient processing and response generation.
This transformation applies two levels of truncation to the conversation history:
- Truncates each individual message to the maximum number of tokens specified by max_tokens_per_message.
- Truncates the overall conversation history to the maximum number of tokens specified by max_tokens.
NOTE: Tokens are counted using the encoder for the specified model. Different models may yield different token counts for the same text.
NOTE: For multimodal LLMs, the token count may be inaccurate as it does not account for the non-text input (e.g images).
The truncation process follows these steps in order:
- The minimum tokens threshold (
min_tokens
) is checked (0 by default). If the total number of tokens in messages are less than this threshold, then the messages are returned as is. In other case, the following process is applied. - Messages are processed in reverse order (newest to oldest).
- Individual messages are truncated based on max_tokens_per_message. For multimodal messages containing both text and other types of content, only the text content is truncated.
- The overall conversation history is truncated based on the max_tokens limit. Once the accumulated token count exceeds this limit, the current message being processed get truncated to meet the total token count and any remaining messages get discarded.
- The truncated conversation history is reconstructed by prepending the messages to a new list to preserve the original message order.
__init__
def __init__(max_tokens_per_message: Optional[int] = None,
max_tokens: Optional[int] = None,
min_tokens: Optional[int] = None,
model: str = "gpt-3.5-turbo-0613",
filter_dict: Optional[Dict] = None,
exclude_filter: bool = True)
Arguments:
max_tokens_per_message
None or int - Maximum number of tokens to keep in each message. Must be greater than or equal to 0 if not None.max_tokens
Optional[int] - Maximum number of tokens to keep in the chat history. Must be greater than or equal to 0 if not None.min_tokens
Optional[int] - Minimum number of tokens in messages to apply the transformation. Must be greater than or equal to 0 if not None.model
str - The target OpenAI model for tokenization alignment.filter_dict
None or dict - A dictionary to filter out messages that you want/don't want to compress. If None, no filters will be applied.exclude_filter
bool - If exclude filter is True (the default value), messages that match the filter will be excluded from token truncation. If False, messages that match the filter will be truncated.
apply_transform
def apply_transform(messages: List[Dict]) -> List[Dict]
Applies token truncation to the conversation history.
Arguments:
messages
List[Dict] - The list of messages representing the conversation history.
Returns:
List[Dict]
- A new list containing the truncated messages up to the specified token limits.
TextMessageCompressor
class TextMessageCompressor()
A transform for compressing text messages in a conversation history.
It uses a specified text compression method to reduce the token count of messages, which can lead to more efficient processing and response generation by downstream models.
__init__
def __init__(text_compressor: Optional[TextCompressor] = None,
min_tokens: Optional[int] = None,
compression_params: Dict = dict(),
cache: Optional[AbstractCache] = None,
filter_dict: Optional[Dict] = None,
exclude_filter: bool = True)
Arguments:
text_compressor
TextCompressor or None - An instance of a class that implements the TextCompressor protocol. If None, it defaults to LLMLingua.min_tokens
int or None - Minimum number of tokens in messages to apply the transformation. Must be greater than or equal to 0 if not None. If None, no threshold-based compression is applied.compression_args
dict - A dictionary of arguments for the compression method. Defaults to an empty dictionary.cache
None or AbstractCache - The cache client to use to store and retrieve previously compressed messages. If None, no caching will be used.filter_dict
None or dict - A dictionary to filter out messages that you want/don't want to compress. If None, no filters will be applied.exclude_filter
bool - If exclude filter is True (the default value), messages that match the filter will be excluded from compression. If False, messages that match the filter will be compressed.
apply_transform
def apply_transform(messages: List[Dict]) -> List[Dict]
Applies compression to messages in a conversation history based on the specified configuration.
The function processes each message according to the compression_args
and min_tokens
settings, applying
the specified compression configuration and returning a new list of messages with reduced token counts
where possible.
Arguments:
messages
List[Dict] - A list of message dictionaries to be compressed.
Returns:
List[Dict]
- A list of dictionaries with the message content compressed according to the configured method and scope.
TextMessageContentName
class TextMessageContentName()
A transform for including the agent's name in the content of a message.
__init__
def __init__(position: str = "start",
format_string: str = "{name}:\n",
deduplicate: bool = True,
filter_dict: Optional[Dict] = None,
exclude_filter: bool = True)
Arguments:
position
str - The position to add the name to the content. The possible options are 'start' or 'end'. Defaults to 'start'.format_string
str - The f-string to format the message name with. Use '{name}' as a placeholder for the agent's name. Defaults to '{name}: ' and must contain '{name}'.deduplicate
bool - Whether to deduplicate the formatted string so it doesn't appear twice (sometimes the LLM will add it to new messages itself). Defaults to True.filter_dict
None or dict - A dictionary to filter out messages that you want/don't want to compress. If None, no filters will be applied.exclude_filter
bool - If exclude filter is True (the default value), messages that match the filter will be excluded from compression. If False, messages that match the filter will be compressed.
apply_transform
def apply_transform(messages: List[Dict]) -> List[Dict]
Applies the name change to the message based on the position and format string.
Arguments:
messages
List[Dict] - A list of message dictionaries.
Returns:
List[Dict]
- A list of dictionaries with the message content updated with names.