ccproxy.llms.utils.context_truncation

ccproxy.llms.utils.context_truncation

Context window truncation utilities.

truncate_to_fit

truncate_to_fit(
    request_data,
    max_input_tokens,
    preserve_recent=10,
    safety_margin=0.9,
)

Truncate request to fit within token limit.

Strategy: 1. Always preserve system prompt and tools 2. Try to preserve the last N messages (preserve_recent) 3. Remove oldest messages first 4. If too few messages to truncate, reduce preserve_recent dynamically 5. As a last resort, truncate large content blocks within messages 6. Add a truncation notice when content is removed

Parameters:

Name	Type	Description	Default
request_data	dict[str, Any]	The request payload	required
max_input_tokens	int	Model's max input token limit	required
preserve_recent	int	Number of recent messages to always keep	10
safety_margin	float	Target this fraction of max to allow for estimation error	0.9

Returns:

Type	Description
tuple[dict[str, Any], bool]	Tuple of (modified_request_data, was_truncated)

Source code in ccproxy/llms/utils/context_truncation.py

def truncate_to_fit(
    request_data: dict[str, Any],
    max_input_tokens: int,
    preserve_recent: int = 10,
    safety_margin: float = 0.9,
) -> tuple[dict[str, Any], bool]:
    """Truncate request to fit within token limit.

    Strategy:
    1. Always preserve system prompt and tools
    2. Try to preserve the last N messages (preserve_recent)
    3. Remove oldest messages first
    4. If too few messages to truncate, reduce preserve_recent dynamically
    5. As a last resort, truncate large content blocks within messages
    6. Add a truncation notice when content is removed

    Args:
        request_data: The request payload
        max_input_tokens: Model's max input token limit
        preserve_recent: Number of recent messages to always keep
        safety_margin: Target this fraction of max to allow for estimation error

    Returns:
        Tuple of (modified_request_data, was_truncated)
    """
    target_tokens = int(max_input_tokens * safety_margin)

    current_tokens = estimate_request_tokens(request_data)
    if current_tokens <= target_tokens:
        return request_data, False

    # Work on a copy
    modified = copy.deepcopy(request_data)
    messages = modified.get("messages", [])

    # If we have fewer messages than preserve_recent, reduce preserve_recent
    # We need at least 1 message to be truncatable for this strategy to work
    effective_preserve = min(preserve_recent, len(messages) - 1)

    # If we have 0 or 1 messages, we can't do message-level truncation
    # Skip to content-level truncation
    if effective_preserve < 0:
        effective_preserve = 0

    # Split into truncatable and preserved messages
    if effective_preserve > 0:
        truncatable = messages[:-effective_preserve]
        preserved = messages[-effective_preserve:]
    else:
        truncatable = list(messages)
        preserved = []

    # Remove oldest messages until we're under the limit
    removed_count = 0
    while truncatable and estimate_request_tokens(modified) > target_tokens:
        truncatable.pop(0)
        removed_count += 1
        modified["messages"] = truncatable + preserved

    # Check if we're still over the limit after removing all truncatable messages
    if estimate_request_tokens(modified) > target_tokens:
        logger.info(
            "context_truncation_message_level_insufficient",
            reason="still_over_limit_after_message_truncation",
            message_count=len(modified.get("messages", [])),
            current_tokens=estimate_request_tokens(modified),
            target_tokens=target_tokens,
            category="context_management",
        )

        # Fallback: truncate large content blocks within remaining messages
        truncated_messages, blocks_truncated = _truncate_large_content_blocks(
            modified.get("messages", [])
        )
        modified["messages"] = truncated_messages

        if blocks_truncated > 0:
            logger.info(
                "context_truncation_content_level",
                blocks_truncated=blocks_truncated,
                current_tokens=estimate_request_tokens(modified),
                target_tokens=target_tokens,
                category="context_management",
            )

        # If still over limit after content truncation, log error
        final_tokens = estimate_request_tokens(modified)
        if final_tokens > target_tokens:
            logger.error(
                "context_truncation_failed",
                reason="still_over_limit_after_all_truncation",
                final_tokens=final_tokens,
                target_tokens=target_tokens,
                messages_removed=removed_count,
                blocks_truncated=blocks_truncated,
                category="context_management",
            )
            # Still return the truncated version - it's better than nothing
            # The API will return an error, but at least we tried

    # Add truncation notice as first user message if we removed content
    if removed_count > 0:
        notice = {
            "role": "user",
            "content": f"[Context truncated - {removed_count} earlier messages removed to fit context window]",
        }
        modified["messages"] = [notice] + modified["messages"]

    final_tokens = estimate_request_tokens(modified)

    logger.info(
        "context_truncated",
        original_tokens=current_tokens,
        final_tokens=final_tokens,
        messages_removed=removed_count,
        target_tokens=target_tokens,
        effective_preserve_recent=effective_preserve,
        category="context_management",
    )

    return modified, True