Camel.memories.context creators.score based

​

ScoreBasedContextCreator

class ScoreBasedContextCreator(BaseContextCreator):

• 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]):

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.
4. Tool calls and their responses are kept together to maintain API compatibility

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

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

​

_group_tool_calls_and_responses

def _group_tool_calls_and_responses(self, units: List[_ContextUnit]):

• units (List[_ContextUnit]): List of context units to analyze.

​

_truncate_with_tool_call_awareness

def _truncate_with_tool_call_awareness(
    self,
    regular_units: List[_ContextUnit],
    tool_call_groups: Dict[str, List[_ContextUnit]],
    system_tokens: int
):

1. It treats tool call groups (request + responses) and standalone messages as individual items to be included.
2. It sorts all items by score and greedily adds them to the context.
3. Partial Truncation: If a complete tool group is too large to fit,it attempts to add the request message and as many of the most recent response chunks as the token budget allows.

• regular_units (List[_ContextUnit]): All regular message units.
• tool_call_groups (Dict[str, List[_ContextUnit]]): Grouped tool calls.
• system_tokens (int): Tokens used by the system message.

​

_extract_system_message

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

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

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

​

_conversation_sort_key

def _conversation_sort_key(self, unit: _ContextUnit):

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

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

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

​

_assemble_output

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

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