1
2
3
4
5
6
7
8
9

SAFE_STOPWORDS = {
    "the", "a", "an", "is", "are", "was", "were",
    "have", "has", "had", "will", "would", "could"
}

PROTECTED_WORDS = {
    "not", "don't", "doesn't", "never", "no", "none",
    "must", "required", "important", "critical"
}

1
2
3
4
5
6
7

PHRASE_ABBREVIATIONS = {
    "for example": "e.g.",
    "in order to": "to",
    "due to the fact that": "because",
    "at this point in time": "now",
    "prior to": "before"
}

1
2
3
4
5
6

FILLER_PHRASES = [
    "basically", "essentially", "actually",
    "it is worth noting that",
    "needless to say",
    "as a matter of fact"
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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

class LexicalCompressor(CompressionStrategy):
    name = "lexical"
    description = "Fast rule-based compression"
    supports_streaming = True

    def compress(
        self,
        text: str,
        config: CompressionConfig,
        content_type: str = None
    ) -> str:
        # Handle preserved sections
        prefix, middle, suffix = self._split_preserved_sections(
            text, config
        )

        if not middle:
            return text

        result = middle

        # Apply transformations in order
        result = self._normalize_whitespace(result)
        result = self._abbreviate_phrases(result)
        result = self._remove_fillers(result)

        # Only remove stopwords if needed for target ratio
        current_ratio = len(result) / len(middle)
        if config.aggressive_mode or current_ratio > config.target_ratio:
            result = self._remove_stopwords(result, config)

        result = self._compress_punctuation(result)
        result = self._normalize_whitespace(result)

        # Reconstruct with preserved sections
        parts = [p for p in [prefix, result, suffix] if p]
        return " ".join(parts)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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

def _score_sentences(self, sentences: list[str]) -> list[tuple[int, float]]:
    all_words = []
    sentence_words = []

    for sentence in sentences:
        words = self._tokenize(sentence)
        sentence_words.append(words)
        all_words.extend(words)

    # Document term frequency
    doc_tf = Counter(all_words)
    total_words = len(all_words)

    scored = []
    for idx, words in enumerate(sentence_words):
        score = 0.0
        for word in words:
            # TF component
            tf = doc_tf[word] / total_words

            # IDF component (pre-computed for common words)
            idf = self.COMMON_WORD_IDF.get(word, 2.0)

            # Self-information
            self_info = -math.log(tf + 0.001) if tf > 0 else 5.0

            score += tf * idf * self_info

        # Normalize by length
        score /= len(words)

        # Position bias: boost first and last
        if idx == 0:
            score *= 1.5
        elif idx == len(sentences) - 1:
            score *= 1.2

        scored.append((idx, score))

    return sorted(scored, key=lambda x: x[1], reverse=True)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

def _compress_sentences(
    self,
    sentences: list[str],
    config: CompressionConfig
) -> str:
    scored = self._score_sentences(sentences)

    # Calculate target character count
    total_chars = sum(len(s) for s in sentences)
    target_chars = int(total_chars * config.target_ratio)

    # Greedily select by importance
    selected_indices = set()
    current_chars = 0

    for idx, score in scored:
        if current_chars >= target_chars:
            break
        selected_indices.add(idx)
        current_chars += len(sentences[idx])

    # Always include first sentence
    if 0 not in selected_indices:
        selected_indices.add(0)

    # Return in original order
    result = [
        sentences[i] for i in range(len(sentences))
        if i in selected_indices
    ]

    return ' '.join(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

def _estimate_redundancy(self, sentences: list[str]) -> float:
    all_ngrams = []
    for sentence in sentences:
        words = self._tokenize(sentence)
        # Generate 2-grams and 3-grams
        for n in [2, 3]:
            for i in range(len(words) - n + 1):
                ngram = tuple(words[i:i+n])
                all_ngrams.append(ngram)

    ngram_counts = Counter(all_ngrams)
    repeated = sum(1 for count in ngram_counts.values() if count > 1)

    return repeated / len(ngram_counts) if ngram_counts else 0.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

@dataclass
class CompressionConfig:
    target_ratio: float = 0.5  # 50% compression
    quality_threshold: float = 0.85  # Minimum quality
    preserve_patterns: list[str] = field(default_factory=list)
    preserve_first_n_tokens: int = 0
    preserve_last_n_tokens: int = 0
    aggressive_mode: bool = False
    max_iterations: int = 5

    def __post_init__(self):
        if not 0.1 <= self.target_ratio <= 1.0:
            raise ValueError("target_ratio must be 0.1-1.0")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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

class AdaptiveCompressor:
    def __init__(self):
        self.history = []
        self.optimal_ratios = {}

    def compress_with_feedback(
        self,
        text: str,
        content_type: str,
        quality_metric: float
    ):
        # Start conservative
        if content_type not in self.optimal_ratios:
            ratio = 0.8
        else:
            ratio = self.optimal_ratios[content_type]

        config = CompressionConfig(target_ratio=ratio)
        compressed = self.compressor.compress(text, config)

        # Track results
        self.history.append({
            "type": content_type,
            "ratio": ratio,
            "quality": quality_metric
        })

        # Adjust for next time
        if quality_metric > 0.9:
            # Can compress more aggressively
            self.optimal_ratios[content_type] = max(0.4, ratio - 0.05)
        elif quality_metric < 0.8:
            # Too aggressive, back off
            self.optimal_ratios[content_type] = min(0.95, ratio + 0.1)

        return compressed