​
ScoreBasedContextCreator

class ScoreBasedContextCreator(BaseContextCreator):

A default implementation of context creation strategy, which inherits from :obj:BaseContextCreator.

This class provides a strategy to generate a conversational context from a list of chat history records while ensuring the total token count of the context does not exceed a specified limit. It prunes messages based on their score if the total token count exceeds the limit.

Parameters:

  • token_counter (BaseTokenCounter): An instance responsible for counting tokens in a message.
  • token_limit (int): The maximum number of tokens allowed in the generated context.

​
init

def __init__(self, token_counter: BaseTokenCounter, token_limit: int):

​
token_counter

def token_counter(self):

​
token_limit

def token_limit(self):

​
create_context

def create_context(self, records: List[ContextRecord]):

Constructs conversation context from chat history while respecting token limits.

Key strategies:

  1. System message is always prioritized and preserved
  2. Truncation removes low-score messages first
  3. Final output maintains chronological order and in history memory, the score of each message decreases according to keep_rate. The newer the message, the higher the score.

Parameters:

  • records (List[ContextRecord]): List of context records with scores and timestamps.

Returns:

Tuple[List[OpenAIMessage], int]:

  • Ordered list of OpenAI messages
  • Total token count of the final context

​
_extract_system_message

def _extract_system_message(self, records: List[ContextRecord]):

Extracts the system message from records and validates it.

Parameters:

  • records (List[ContextRecord]): List of context records representing conversation history.

Returns:

Tuple[Optional[_ContextUnit], List[_ContextUnit]]: containing:

  • The system message as a _ContextUnit, if valid; otherwise, None.
  • An empty list, serving as the initial container for regular messages.

​
_truncation_sort_key

def _truncation_sort_key(self, unit: _ContextUnit):

Defines the sorting key for the truncation phase.

Sorting priority:

  • Primary: Sort by score in descending order (higher scores first).
  • Secondary: Sort by timestamp in ascending order (older messages first when scores are equal).

Parameters:

  • unit (_ContextUnit): A _ContextUnit representing a conversation record.

Returns:

Tuple[float, float]:

  • Negative score for descending order sorting.
  • Timestamp for ascending order sorting.

​
_conversation_sort_key

def _conversation_sort_key(self, unit: _ContextUnit):

Defines the sorting key for assembling the final output.

Sorting priority:

  • Primary: Sort by timestamp in ascending order (chronological order).
  • Secondary: Sort by score in descending order (higher scores first when timestamps are equal).

Parameters:

  • unit (_ContextUnit): A _ContextUnit representing a conversation record.

Returns:

Tuple[float, float]:

  • Timestamp for chronological sorting.
  • Negative score for descending order sorting.

​
_assemble_output

def _assemble_output(
    self,
    context_units: List[_ContextUnit],
    system_unit: Optional[_ContextUnit]
):

Assembles final message list with proper ordering and token count.

Parameters:

  • context_units (List[_ContextUnit]): Sorted list of regular message units.
  • system_unit (Optional[_ContextUnit]): System message unit (if present).

Returns:

Tuple[List[OpenAIMessage], int]: Tuple of (ordered messages, total tokens)