« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Hybrid compression strategies."""
3from __future__ import annotations
5from datetime import UTC, datetime
7from lexigram.ai.rag.context_compression.base import AbstractCompressor
8from lexigram.ai.rag.context_compression.types import (
9 CompressionResult,
10 CompressionStrategy,
11)
14class HybridCompressor(AbstractCompressor):
15 """Combine multiple compression strategies.
17 Applies multiple compressors in sequence for maximum compression
18 while maintaining quality.
20 Example:
21 >>> compressor = HybridCompressor(
22 ... compressors=[
23 ... SemanticDeduplicationCompressor(),
24 ... ExtractiveSummaryCompressor(max_sentences=10),
25 ... TokenLimitCompressor(max_tokens=500),
26 ... ]
27 ... )
28 >>> result = await compressor.compress(very_long_context, query="...")
29 """
31 def __init__(self, compressors: list[AbstractCompressor]):
32 """Initialize hybrid compressor.
34 Args:
35 compressors: List of compressors to apply in sequence.
36 """
37 self.compressors = compressors
39 async def compress(
40 self,
41 context: str | list[str],
42 query: str | None = None,
43 **kwargs,
44 ) -> CompressionResult:
45 """Compress using multiple strategies in sequence."""
46 original_text = self._normalize_context(context)
47 original_tokens = self._estimate_tokens(original_text)
49 current_text = original_text
50 intermediate_results = []
52 # Apply each compressor in sequence
53 for _i, compressor in enumerate(self.compressors):
54 result = await compressor.compress(current_text, query=query, **kwargs)
55 current_text = result.compressed_text
56 intermediate_results.append(
57 {
58 "compressor": compressor.__class__.__name__,
59 "strategy": result.strategy.value,
60 "compression_ratio": result.compression_ratio,
61 "tokens": result.compressed_tokens,
62 },
63 )
65 compressed_text = current_text
66 compressed_tokens = self._estimate_tokens(compressed_text)
67 compression_ratio = (
68 compressed_tokens / original_tokens if original_tokens > 0 else 1.0
69 )
71 return CompressionResult(
72 original_text=original_text,
73 compressed_text=compressed_text,
74 original_tokens=original_tokens,
75 compressed_tokens=compressed_tokens,
76 compression_ratio=compression_ratio,
77 strategy=CompressionStrategy.HYBRID,
78 metadata={
79 "num_compressors": len(self.compressors),
80 "intermediate_results": intermediate_results,
81 "timestamp": datetime.now(UTC).isoformat(),
82 },
83 )