_check_and_compact() — anthropic-sdk-python Function Reference

Source Code

src/anthropic/lib/tools/_beta_runner.py lines 158–240

    def _check_and_compact(self) -> bool:
        """
        Check token usage and compact messages if threshold exceeded.
        Returns True if compaction was performed, False otherwise.
        """
        if self._compaction_control is None or not self._compaction_control["enabled"]:
            return False

        message = self._get_last_message()
        tokens_used = 0
        if message is not None:
            total_input_tokens = (
                message.usage.input_tokens
                + (message.usage.cache_creation_input_tokens or 0)
                + (message.usage.cache_read_input_tokens or 0)
            )
            tokens_used = total_input_tokens + message.usage.output_tokens

        threshold = self._compaction_control.get("context_token_threshold", DEFAULT_THRESHOLD)

        if tokens_used < threshold:
            return False

        # Perform compaction
        log.info(f"Token usage {tokens_used} has exceeded the threshold of {threshold}. Performing compaction.")

        model = self._compaction_control.get("model", self._params["model"])

        messages = list(self._params["messages"])

        if messages[-1]["role"] == "assistant":
            # Remove tool_use blocks from the last message to avoid 400 error
            # (tool_use requires tool_result, which we don't have yet)
            non_tool_blocks = [
                block
                for block in messages[-1]["content"]
                if isinstance(block, dict) and block.get("type") != "tool_use"
            ]

            if non_tool_blocks:
                messages[-1]["content"] = non_tool_blocks
            else:
                messages.pop()

        messages = [
            *messages,
            BetaMessageParam(
                role="user",
                content=self._compaction_control.get("summary_prompt", DEFAULT_SUMMARY_PROMPT),
            ),
        ]

        response = self._client.beta.messages.create(
            model=model,
            messages=messages,
            max_tokens=self._params["max_tokens"],
            extra_headers={"X-Stainless-Helper": "compaction"},
        )

        log.info(f"Compaction complete. New token usage: {response.usage.output_tokens}")

        first_content = list(response.content)[0]

        if first_content.type != "text":
            raise ValueError("Compaction response content is not of type 'text'")

        self.set_messages_params(
            lambda params: {
                **params,
                "messages": [
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "text",
                                "text": first_content.text,
                            }
                        ],
                    }
                ],
            }