Skip to main content

agentchat.contrib.capabilities.transforms

MessageTransform​

class MessageTransform(Protocol)

定义了消息转换的契约。

实现该协议的类应该提供一个 apply_transform 方法，该方法接受一个消息列表并返回转换后的列表。

apply_transform​

def apply_transform(messages: List[Dict]) -> List[Dict]

对消息列表应用转换。

参数：

  • messages - 一个表示消息的字典列表。

返回值：

包含转换后消息的新字典列表。

get_logs​

def get_logs(pre_transform_messages: List[Dict],
             post_transform_messages: List[Dict]) -> Tuple[str, bool]

创建包含转换日志的字符串

除了字符串，它还返回一个布尔值，指示转换是否产生了效果。

参数：

  • pre_transform_messages - 转换前的消息列表。
  • post_transform_messages - 转换后的消息列表。

返回值：

包含日志的字符串和一个指示转换是否产生了效果的标志的元组。

MessageHistoryLimiter​

class MessageHistoryLimiter()

限制用于生成响应的消息数量。

该转换仅保留最近的消息，最多保留指定数量的消息（max_messages）。 它通过删除旧消息来修剪对话历史，只保留最近的消息。

__init__​

def __init__(max_messages: Optional[int] = None)

参数：

max_messages Optional[int]: 要保留在上下文中的最大消息数。如果不为 None，则必须大于 0。

apply_transform​

def apply_transform(messages: List[Dict]) -> List[Dict]

将对话历史截断为指定的最大消息数。

该方法返回一个新列表，其中包含最近的消息，最多为指定的最大消息数（max_messages）。 如果 max_messages 为 None，则返回未修改的原始消息列表。

参数：

  • messages List[Dict] - 表示对话历史的消息列表。

返回值：

  • List[Dict] - 包含最近的消息，最多为指定的最大消息数的新列表。

MessageTokenLimiter​

class MessageTokenLimiter()

将消息截断到满足令牌限制，以实现高效处理和生成响应。

该转换对对话历史应用两个级别的截断：

  1. 将每个单独的消息截断为 max_tokens_per_message 指定的最大令牌数。
  2. 将整个对话历史截断为 max_tokens 指定的最大令牌数。 注意：令牌计数是使用指定模型的编码器进行计算的。不同的模型可能对相同的文本产生不同的令牌计数。

注意：对于多模态语言模型，令牌计数可能不准确，因为它没有考虑非文本输入（例如图像）。

截断过程按以下顺序进行：

  1. 检查最小令牌阈值（默认为0）。如果消息中的总令牌数小于此阈值，则将消息按原样返回。否则，应用以下过程。
  2. 按照相反的顺序处理消息（从最新到最旧）。
  3. 根据每条消息的最大令牌数对各个消息进行截断。对于包含文本和其他类型内容的多模态消息，只截断文本内容。
  4. 根据最大令牌限制截断整个对话历史记录。一旦累积的令牌数超过此限制，当前正在处理的消息将被截断以满足总令牌数，并且任何剩余的消息将被丢弃。
  5. 通过将消息添加到新列表的开头来重构截断的对话历史记录，以保留原始消息顺序。

__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)

参数：

  • max_tokens_per_message None 或 int - 每条消息中要保留的最大令牌数。如果不为None，则必须大于等于0。
  • max_tokens Optional[int] - 要在聊天历史记录中保留的最大令牌数。如果不为None，则必须大于等于0。
  • min_tokens Optional[int] - 要应用转换的消息中的最小令牌数。如果不为None，则必须大于等于0。
  • model str - 用于令牌化对齐的目标OpenAI模型。
  • filter_dict None 或 dict - 用于过滤您想要/不想要压缩的消息的字典。如果为None，则不应用任何过滤器。
  • exclude_filter bool - 如果排除过滤器为True（默认值），则与过滤器匹配的消息将被排除在令牌截断之外。如果为False，则与过滤器匹配的消息将被截断。

apply_transform​

def apply_transform(messages: List[Dict]) -> List[Dict]

将令牌截断应用于对话历史记录。

参数：

  • messages List[Dict] - 表示对话历史记录的消息列表。

返回值：

  • List[Dict] - 包含截断消息的新列表，截断消息的数量不超过指定的令牌限制。

TextMessageCompressor​

class TextMessageCompressor()

用于压缩对话历史记录中文本消息的转换器。

它使用指定的文本压缩方法来减少消息的令牌计数，从而可以实现更高效的对话历史记录。

__init__​

def __init__(text_compressor: Optional[TextCompressor] = None,
             min_tokens: Optional[int] = None,
             compression_params: Dict = dict(),
             cache: Optional[AbstractCache] = Cache.disk(),
             filter_dict: Optional[Dict] = None,
             exclude_filter: bool = True)

参数:

  • text_compressor TextCompressor 或 None - 实现 TextCompressor 协议的类的实例。如果为 None，则默认为 LLMLingua。
  • min_tokens int 或 None - 应用转换的消息中的最小令牌数。如果不为 None，则必须大于等于 0。如果为 None，则不应用基于阈值的压缩。
  • compression_args dict - 压缩方法的参数字典。默认为空字典。
  • cache None 或 AbstractCache - 用于存储和检索先前压缩的消息的缓存客户端。如果为 None，则不使用缓存。
  • filter_dict None 或 dict - 用于过滤要压缩/不压缩的消息的字典。如果为 None，则不应用任何过滤器。
  • exclude_filter bool - 如果 exclude_filter 为 True（默认值），则匹配过滤器的消息将被排除在压缩之外。如果为 False，则匹配过滤器的消息将被压缩。

apply_transform​

def apply_transform(messages: List[Dict]) -> List[Dict]

根据指定的配置对会话历史中的消息应用压缩。

该函数根据 compression_args 和 min_tokens 设置处理每条消息，应用指定的压缩配置，并返回一个新的消息列表，其中尽可能减少令牌数量。

参数:

  • messages List[Dict] - 要压缩的消息字典列表。

返回值:

  • List[Dict] - 根据配置的方法和范围压缩消息内容的字典列表。