Skip to content

ccproxy.llms.utils

ccproxy.llms.utils

LLM utility modules for token estimation and context management.

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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

estimate_messages_tokens 

estimate_messages_tokens(messages)

Estimate total tokens for a list of messages.

Parameters:

Name Type Description Default
messages list[dict[str, Any]]

List of message dicts with role and content

 required

Returns:

Type Description
int

Estimated total token count
Source code in ccproxy/llms/utils/token_estimation.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def estimate_messages_tokens(messages: list[dict[str, Any]]) -> int:
    """Estimate total tokens for a list of messages.

    Args:
        messages: List of message dicts with role and content

    Returns:
        Estimated total token count
    """
    total = 0
    for msg in messages:
        # Role contributes ~2 tokens
        total += 2
        total += estimate_tokens(msg.get("content"))
    return total

estimate_request_tokens 

estimate_request_tokens(request_data)

Estimate total input tokens for a request.

Includes messages, system prompt, and tool definitions.

Parameters:

Name Type Description Default
request_data dict[str, Any]

The request payload dictionary

 required

Returns:

Type Description
int

Estimated total input token count
Source code in ccproxy/llms/utils/token_estimation.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def estimate_request_tokens(request_data: dict[str, Any]) -> int:
    """Estimate total input tokens for a request.

    Includes messages, system prompt, and tool definitions.

    Args:
        request_data: The request payload dictionary

    Returns:
        Estimated total input token count
    """
    total = 0

    # Messages
    messages = request_data.get("messages", [])
    total += estimate_messages_tokens(messages)

    # System prompt
    system = request_data.get("system")
    if system:
        total += estimate_tokens(system)

    # Tools
    tools = request_data.get("tools", [])
    if tools:
        total += estimate_tokens(json.dumps(tools))

    return total

estimate_tokens 

estimate_tokens(content)

Estimate token count for content.

Uses ~3 characters per token heuristic for English text. This is a conservative estimate - actual may be lower.

Parameters:

Name Type Description Default
content Any

Message content (string, list of blocks, or dict)

 required

Returns:

Type Description
int

Estimated token count
Source code in ccproxy/llms/utils/token_estimation.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def estimate_tokens(content: Any) -> int:
    """Estimate token count for content.

    Uses ~3 characters per token heuristic for English text.
    This is a conservative estimate - actual may be lower.

    Args:
        content: Message content (string, list of blocks, or dict)

    Returns:
        Estimated token count
    """
    if content is None:
        return 0

    if isinstance(content, str):
        # ~3 chars per token for English, be conservative
        return max(1, len(content) // 3)

    if isinstance(content, list):
        total = 0
        for block in content:
            if isinstance(block, dict):
                block_type = block.get("type", "")
                if block_type == "text":
                    total += estimate_tokens(block.get("text", ""))
                elif block_type == "tool_use":
                    # Tool name + input
                    total += estimate_tokens(block.get("name", ""))
                    total += estimate_tokens(json.dumps(block.get("input", {})))
                elif block_type == "tool_result":
                    total += estimate_tokens(block.get("content", ""))
                elif block_type == "image":
                    # Images are ~1600 tokens for typical size
                    total += 1600
                else:
                    # Generic block - serialize and estimate
                    total += estimate_tokens(json.dumps(block))
            else:
                total += estimate_tokens(block)
        return total

    if isinstance(content, dict):
        return estimate_tokens(json.dumps(content))

    return estimate_tokens(str(content))

get_max_input_tokens 

get_max_input_tokens(model)

Get max input tokens for a model.

Supports pattern matching for model variants: - Exact match: "claude-opus-4-5-20251101" - Prefix match: "claude-opus-4-5-*" matches "claude-opus-4-5-20251101"

Parameters:

Name Type Description Default
model str

Model name or identifier

 required

Returns:

Type Description
int | None

Max input tokens if known, None otherwise
Source code in ccproxy/llms/utils/token_estimation.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def get_max_input_tokens(model: str) -> int | None:
    """Get max input tokens for a model.

    Supports pattern matching for model variants:
    - Exact match: "claude-opus-4-5-20251101"
    - Prefix match: "claude-opus-4-5-*" matches "claude-opus-4-5-20251101"

    Args:
        model: Model name or identifier

    Returns:
        Max input tokens if known, None otherwise
    """
    limits = _load_token_limits()

    # Try exact match first
    if model in limits:
        return limits[model]

    # Try prefix matching (for patterns like claude-opus-4-5-*)
    for pattern, max_tokens in limits.items():
        if pattern.endswith("*"):
            prefix = pattern[:-1]
            if model.startswith(prefix):
                return max_tokens

    # Try matching known model families
    model_lower = model.lower()
    for known_model, max_tokens in limits.items():
        if known_model.lower() in model_lower or model_lower in known_model.lower():
            return max_tokens

    return None