diff --git a/pyproject.toml b/pyproject.toml index 2ac981a..cd1f445 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "tqdm>=4.65.0", "litellm>=1.83.0", "tokenizers>=0.20.0", + "tiktoken", ] [project.scripts] @@ -89,6 +90,10 @@ ignore_missing_imports = true module = "litellm.*" ignore_missing_imports = true +[[tool.mypy.overrides]] +module = "tiktoken.*" +ignore_missing_imports = true + [tool.ruff] line-length = 100 target-version = "py310" diff --git a/src/cordon/cli.py b/src/cordon/cli.py index 27065bb..a05cde2 100644 --- a/src/cordon/cli.py +++ b/src/cordon/cli.py @@ -128,6 +128,30 @@ def parse_args() -> argparse.Namespace: default=None, help="Batch size for k-NN scoring queries (default: auto-detect based on GPU memory)", ) + config_group.add_argument( + "--token-budget", + type=int, + default=None, + help="Maximum token budget for output; dynamically adjusts percentile to fit (overrides --anomaly-percentile)", + ) + config_group.add_argument( + "--tokenizer-encoding", + type=str, + default="cl100k_base", + help="tiktoken encoding for token counting (default: cl100k_base)", + ) + config_group.add_argument( + "--max-blocks", + type=int, + default=None, + help="Maximum number of anomaly blocks to output (keeps highest scoring)", + ) + config_group.add_argument( + "--min-score", + type=float, + default=None, + help="Minimum anomaly score threshold for output blocks", + ) # output options output_group = parser.add_argument_group("output options") @@ -160,9 +184,8 @@ def parse_args() -> argparse.Namespace: "--quiet", "-q", action="store_true", - help="Suppress progress bars (useful for CI or library usage)", + help="Suppress all human-readable banners and progress bars, keeping only formatted output on stdout", ) - return parser.parse_args() @@ -212,6 +235,7 @@ def _display_results( detailed: bool, output_path: Path | None, force: bool, + quiet: bool, ) -> None: """Display analysis results, optionally with detailed statistics. @@ -220,8 +244,9 @@ def _display_results( detailed: Whether to print detailed statistics before the output. output_path: Optional path to save anomalous blocks (None prints to stdout). force: If True, overwrite an existing output file. + quiet: If True, suppress human-readable banners and stats. """ - if detailed: + if detailed and not quiet: print(f"Total lines: {result.total_lines:,}") print("\nAnalysis Statistics:") print(f" Total windows created: {result.total_windows:,}") @@ -243,7 +268,8 @@ def _display_results( else: print(result.output) - print() + if not quiet: + print() def analyze_file( @@ -252,6 +278,7 @@ def analyze_file( detailed: bool, output_path: Path | None = None, force: bool = False, + quiet: bool = False, ) -> None: """Analyze a single log file and print results. @@ -261,13 +288,15 @@ def analyze_file( detailed: Whether to show detailed statistics. output_path: Optional path to save anomalous blocks (None prints to stdout). force: If True, overwrite an existing output file. + quiet: If True, suppress human-readable banners and stats. """ if not _validate_file(log_path): return - print("=" * 80) - print(f"Analyzing: {log_path}") - print("=" * 80) + if not quiet: + print("=" * 80) + print(f"Analyzing: {log_path}") + print("=" * 80) try: result = analyzer.analyze_file_detailed(log_path) @@ -275,7 +304,7 @@ def analyze_file( print(f"Error analyzing {log_path}: {error}", file=sys.stderr) return - _display_results(result, detailed, output_path, force) + _display_results(result, detailed, output_path, force, quiet) def analyze_stdin( @@ -283,6 +312,7 @@ def analyze_stdin( detailed: bool, output_path: Path | None = None, force: bool = False, + quiet: bool = False, ) -> None: """Analyze log data from stdin and print results. @@ -291,12 +321,14 @@ def analyze_stdin( detailed: Whether to show detailed statistics. output_path: Optional path to save anomalous blocks (None prints to stdout). force: If True, overwrite an existing output file. + quiet: If True, suppress human-readable banners and stats. """ text = sys.stdin.read() - print("=" * 80) - print("Analyzing: ") - print("=" * 80) + if not quiet: + print("=" * 80) + print("Analyzing: ") + print("=" * 80) try: result = analyzer.analyze_text_detailed(text) @@ -304,7 +336,7 @@ def analyze_stdin( print(f"Error analyzing : {error}", file=sys.stderr) return - _display_results(result, detailed, output_path, force) + _display_results(result, detailed, output_path, force, quiet) def _print_backend_info(config: AnalysisConfig) -> None: @@ -355,6 +387,12 @@ def _main_impl() -> None: file=sys.stderr, ) + if args.token_budget is not None and not isclose(args.anomaly_percentile, 0.1): + print( + "Warning: --anomaly-percentile is overridden by --token-budget", + file=sys.stderr, + ) + # create configuration from arguments try: config = AnalysisConfig( @@ -376,17 +414,21 @@ def _main_impl() -> None: api_key=args.api_key, endpoint=args.endpoint, show_progress=not args.quiet, + token_budget=args.token_budget, + tokenizer_encoding=args.tokenizer_encoding, output_format=args.output_format, + max_blocks=args.max_blocks, + min_score=args.min_score, ) except ValueError as error: print(f"Configuration error: {error}", file=sys.stderr) sys.exit(1) - # create analyzer - print("Initializing analyzer...") - _print_backend_info(config) - _print_filtering_mode(config) - print() + if not args.quiet: + print("Initializing analyzer...") + _print_backend_info(config) + _print_filtering_mode(config) + print() try: analyzer = SemanticLogAnalyzer(config) @@ -399,14 +441,18 @@ def _main_impl() -> None: except Exception as error: print(f"Initialization error: {error}", file=sys.stderr) sys.exit(1) - print() + + if not args.quiet: + print() # analyze each log file for log_path in args.logfiles: if str(log_path) == "-": - analyze_stdin(analyzer, args.detailed, args.output, args.force) + analyze_stdin(analyzer, args.detailed, args.output, args.force, quiet=args.quiet) else: - analyze_file(log_path, analyzer, args.detailed, args.output, args.force) + analyze_file( + log_path, analyzer, args.detailed, args.output, args.force, quiet=args.quiet + ) def main() -> None: diff --git a/src/cordon/core/config.py b/src/cordon/core/config.py index ffaf00e..5fc7504 100644 --- a/src/cordon/core/config.py +++ b/src/cordon/core/config.py @@ -69,7 +69,16 @@ class AnalysisConfig: request_timeout: HTTP request timeout in seconds (remote backend). show_progress: Whether to display tqdm progress bars during embedding and scoring. Set to False for CI or library use. + token_budget: Maximum token count for output. When set, dynamically + computes anomaly_percentile from the input size to fit + the output within this budget. Overrides anomaly_percentile. + tokenizer_encoding: tiktoken encoding name for token counting + (default: cl100k_base, used by GPT-4/GPT-3.5). output_format: Output format for anomaly blocks. + max_blocks: Maximum number of anomaly blocks to include in output. + Keeps the highest-scoring blocks. None disables the limit. + min_score: Minimum anomaly score threshold. Blocks below this + score are excluded from output. None disables the threshold. """ window_size: int = 4 @@ -91,7 +100,11 @@ class AnalysisConfig: endpoint: str | None = None request_timeout: float = 60.0 show_progress: bool = True + token_budget: int | None = None + tokenizer_encoding: str = "cl100k_base" output_format: Literal["xml", "json"] = "xml" + max_blocks: int | None = None + min_score: float | None = None def __post_init__(self) -> None: """Validate configuration parameters.""" @@ -122,6 +135,12 @@ def _validate_core_params(self) -> None: raise ValueError( f"output_format must be one of {_ALLOWED_FORMATS}, got {self.output_format!r}" ) + if self.token_budget is not None and self.token_budget < 1: + raise ValueError("token_budget must be >= 1 if set") + if self.max_blocks is not None and self.max_blocks < 1: + raise ValueError("max_blocks must be >= 1 if set") + if self.min_score is not None and self.min_score < 0: + raise ValueError("min_score must be >= 0 if set") def _validate_anomaly_range(self) -> None: """Validate anomaly range parameters.""" diff --git a/src/cordon/pipeline.py b/src/cordon/pipeline.py index caba6a6..6d7c88b 100644 --- a/src/cordon/pipeline.py +++ b/src/cordon/pipeline.py @@ -1,5 +1,7 @@ +import logging import time from collections.abc import Sequence +from dataclasses import replace from pathlib import Path import numpy as np @@ -23,6 +25,8 @@ from cordon.postprocess.merger import IntervalMerger from cordon.segmentation.windower import SlidingWindowSegmenter +logger = logging.getLogger(__name__) + class SemanticLogAnalyzer: """High-level API for semantic log analysis. @@ -153,14 +157,46 @@ def _analyze_lines(self, lines_list: list[tuple[int, str]]) -> AnalysisResult: del embedded # stage 5: thresholding - significant = self._thresholder.select_significant(scored, self.config) + thresholding_config = self.config + if self.config.token_budget is not None: + import tiktoken + + enc = tiktoken.get_encoding(self.config.tokenizer_encoding) + total_tokens = sum(len(enc.encode(text)) for _, text in lines_list) + + if total_tokens > 0: + budget_percentile = min(self.config.token_budget / total_tokens, 1.0) + else: + budget_percentile = 1.0 + + thresholding_config = replace(self.config, anomaly_percentile=budget_percentile) + + logger.info( + "Token budget: %d/%d tokens (%.1f%% percentile)", + self.config.token_budget, + total_tokens, + budget_percentile * 100, + ) + + significant = self._thresholder.select_significant(scored, thresholding_config) significant_windows = len(significant) # stage 6: merging merged = self._merger.merge_windows(significant) - merged_blocks_count = len(merged) del significant + # stage 6b: post-merge filters + if self.config.min_score is not None: + merged = [b for b in merged if b.max_score >= self.config.min_score] + + if self.config.max_blocks is not None and len(merged) > self.config.max_blocks: + merged = sorted(merged, key=lambda b: b.max_score, reverse=True)[ + : self.config.max_blocks + ] + merged = sorted(merged, key=lambda b: b.start_line) + + merged_blocks_count = len(merged) + # stage 7: formatting output = self._formatter.format_blocks(merged, lines_list) diff --git a/tests/test_cli.py b/tests/test_cli.py index 9236070..59b06ff 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,7 +6,7 @@ import pytest -from cordon.cli import analyze_file, parse_args +from cordon.cli import analyze_file, analyze_stdin, parse_args class TestParseArgs: @@ -90,6 +90,32 @@ def test_stdin_argument(self, monkeypatch: pytest.MonkeyPatch) -> None: args = parse_args() assert str(args.logfiles[0]) == "-" + def test_token_budget(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test --token-budget parsing.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--token-budget", "500", "test.log"]) + args = parse_args() + assert args.token_budget == 500 + + def test_token_budget_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --token-budget defaults to None.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.token_budget is None + + def test_tokenizer_encoding(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test --tokenizer-encoding parsing.""" + monkeypatch.setattr( + sys, "argv", ["cordon", "--tokenizer-encoding", "p50k_base", "test.log"] + ) + args = parse_args() + assert args.tokenizer_encoding == "p50k_base" + + def test_tokenizer_encoding_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --tokenizer-encoding defaults to cl100k_base.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.tokenizer_encoding == "cl100k_base" + class TestAnalyzeFile: """Tests for analyze_file function.""" @@ -174,8 +200,10 @@ def test_overwrite_protection(self, capsys: pytest.CaptureFixture[str], tmp_path analyzer = MagicMock() mock_result = MagicMock() - mock_result.output = "" + mock_result.output = "" + mock_result.merged_blocks = 0 analyzer.analyze_file_detailed.return_value = mock_result + analyze_file( log_file, analyzer, @@ -199,6 +227,7 @@ def test_force_overwrite(self, tmp_path: Path) -> None: analyzer = MagicMock() mock_result = MagicMock() mock_result.output = "new" + mock_result.merged_blocks = 1 analyzer.analyze_file_detailed.return_value = mock_result analyze_file( @@ -218,6 +247,7 @@ def test_stdout_output(self, capsys: pytest.CaptureFixture[str], tmp_path: Path) analyzer = MagicMock() mock_result = MagicMock() mock_result.output = "results" + mock_result.merged_blocks = 0 analyzer.analyze_file_detailed.return_value = mock_result analyze_file(log_file, analyzer, detailed=False) @@ -255,6 +285,77 @@ def test_detailed_output(self, capsys: pytest.CaptureFixture[str], tmp_path: Pat assert "Score Distribution:" in captured.out +class TestQuietMode: + """Tests for --quiet banner suppression.""" + + def test_quiet_suppresses_banners( + self, capsys: pytest.CaptureFixture[str], tmp_path: Path + ) -> None: + """Test that --quiet suppresses all human-readable banners.""" + log_file = tmp_path / "test.log" + log_file.write_text("line 1\n") + + analyzer = MagicMock() + mock_result = MagicMock() + mock_result.output = "" + mock_result.merged_blocks = 0 + analyzer.analyze_file_detailed.return_value = mock_result + + analyze_file(log_file, analyzer, detailed=True, quiet=True) + captured = capsys.readouterr() + assert "Analyzing:" not in captured.out + assert "=" * 80 not in captured.out + assert "Total lines" not in captured.out + assert "Score Distribution" not in captured.out + assert "" in captured.out + + def test_quiet_suppresses_stdin_banners( + self, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch + ) -> None: + """Test that --quiet suppresses banners for stdin analysis.""" + monkeypatch.setattr("sys.stdin", MagicMock(read=MagicMock(return_value="line 1\n"))) + + analyzer = MagicMock() + mock_result = MagicMock() + mock_result.output = "" + mock_result.merged_blocks = 0 + analyzer.analyze_text_detailed.return_value = mock_result + + analyze_stdin(analyzer, detailed=True, quiet=True) + captured = capsys.readouterr() + assert "Analyzing:" not in captured.out + assert "=" * 80 not in captured.out + assert "" in captured.out + + +class TestNewFlags: + """Tests for --max-blocks and --min-score flags.""" + + def test_max_blocks_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --max-blocks is parsed correctly.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--max-blocks", "10", "test.log"]) + args = parse_args() + assert args.max_blocks == 10 + + def test_max_blocks_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --max-blocks defaults to None.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.max_blocks is None + + def test_min_score_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --min-score is parsed correctly.""" + monkeypatch.setattr(sys, "argv", ["cordon", "--min-score", "0.5", "test.log"]) + args = parse_args() + assert args.min_score == 0.5 + + def test_min_score_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that --min-score defaults to None.""" + monkeypatch.setattr(sys, "argv", ["cordon", "test.log"]) + args = parse_args() + assert args.min_score is None + + class TestMainEntryPoint: """Tests for the main() entry point.""" diff --git a/tests/test_core.py b/tests/test_core.py index 8e07112..0b9cad7 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -178,6 +178,37 @@ def test_output_format_json(self) -> None: config = AnalysisConfig(output_format="json") assert config.output_format == "json" + def test_token_budget_validation(self) -> None: + """Test that invalid token_budget values are rejected.""" + with pytest.raises(ValueError, match="token_budget"): + AnalysisConfig(token_budget=0) + with pytest.raises(ValueError, match="token_budget"): + AnalysisConfig(token_budget=-1) + config = AnalysisConfig(token_budget=1000) + assert config.token_budget == 1000 + + def test_max_blocks_validation(self) -> None: + """Test that invalid max_blocks values are rejected.""" + with pytest.raises(ValueError, match="max_blocks"): + AnalysisConfig(max_blocks=0) + with pytest.raises(ValueError, match="max_blocks"): + AnalysisConfig(max_blocks=-1) + config = AnalysisConfig(max_blocks=5) + assert config.max_blocks == 5 + config_none = AnalysisConfig(max_blocks=None) + assert config_none.max_blocks is None + + def test_min_score_validation(self) -> None: + """Test that invalid min_score values are rejected.""" + with pytest.raises(ValueError, match="min_score"): + AnalysisConfig(min_score=-0.1) + config = AnalysisConfig(min_score=0.0) + assert config.min_score == 0.0 + config_pos = AnalysisConfig(min_score=0.5) + assert config_pos.min_score == 0.5 + config_none = AnalysisConfig(min_score=None) + assert config_none.min_score is None + class TestAnalysisResult: """Tests for AnalysisResult dataclass.""" diff --git a/tests/test_pipeline_unit.py b/tests/test_pipeline_unit.py index 4b15bd4..f0272e0 100644 --- a/tests/test_pipeline_unit.py +++ b/tests/test_pipeline_unit.py @@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch from cordon.core.config import AnalysisConfig +from cordon.core.types import MergedBlock, ScoredWindow, TextWindow from cordon.pipeline import SemanticLogAnalyzer @@ -184,3 +185,159 @@ def test_analyze_lines(self, mock_create: MagicMock) -> None: lines: list[tuple[int, str]] = [(1, "hello"), (2, "world")] result = analyzer.analyze_lines(lines) assert result.total_lines == 2 + + +class TestTokenBudget: + """Tests for token budget mode.""" + + @patch("cordon.pipeline.create_embedder") + def test_token_budget_computes_percentile(self, mock_create: MagicMock) -> None: + """Test that token budget dynamically adjusts percentile.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + config = AnalysisConfig(device="cpu", token_budget=100) + analyzer = SemanticLogAnalyzer(config) + result = analyzer.analyze_text_detailed("word " * 100) + assert result.total_lines >= 1 + + @patch("cordon.pipeline.create_embedder") + def test_token_budget_larger_than_file(self, mock_create: MagicMock) -> None: + """Test that budget larger than file results in percentile capped at 1.0.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + config = AnalysisConfig(device="cpu", token_budget=999999) + analyzer = SemanticLogAnalyzer(config) + result = analyzer.analyze_text_detailed("short text") + assert result.total_lines >= 1 + + @patch("cordon.pipeline.create_embedder") + def test_token_budget_empty_input(self, mock_create: MagicMock) -> None: + """Test token budget with empty input.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + config = AnalysisConfig(device="cpu", token_budget=100) + analyzer = SemanticLogAnalyzer(config) + result = analyzer.analyze_text_detailed("") + assert result.total_lines == 0 + + +class TestPostMergeFiltering: + """Tests for max_blocks and min_score post-merge filtering.""" + + @patch("cordon.pipeline.create_embedder") + def test_max_blocks_filtering(self, mock_create: MagicMock) -> None: + """Test that max_blocks keeps only the top N highest-scoring blocks.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + mock_reader = MagicMock() + mock_reader.read_lines.return_value = iter([(i, f"line {i}") for i in range(1, 21)]) + + blocks = [ + MergedBlock(start_line=1, end_line=3, original_windows=(0,), max_score=0.3), + MergedBlock(start_line=5, end_line=7, original_windows=(1,), max_score=0.9), + MergedBlock(start_line=10, end_line=12, original_windows=(2,), max_score=0.6), + MergedBlock(start_line=15, end_line=17, original_windows=(3,), max_score=0.8), + ] + mock_merger = MagicMock() + mock_merger.merge_windows.return_value = blocks + + mock_thresholder = MagicMock() + window = TextWindow(content="test", start_line=1, end_line=3, window_id=0) + mock_thresholder.select_significant.return_value = [ScoredWindow(window=window, score=0.5)] + + config = AnalysisConfig(device="cpu", max_blocks=2) + analyzer = SemanticLogAnalyzer( + config, + reader=mock_reader, + merger=mock_merger, + thresholder=mock_thresholder, + ) + result = analyzer.analyze_file_detailed(Path("dummy.log")) + + assert result.merged_blocks == 2 + assert len(result.blocks) == 2 + scores = [b.max_score for b in result.blocks] + assert 0.9 in scores + assert 0.8 in scores + assert result.blocks[0].start_line < result.blocks[1].start_line + + @patch("cordon.pipeline.create_embedder") + def test_min_score_filtering(self, mock_create: MagicMock) -> None: + """Test that min_score drops blocks below the threshold.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + mock_reader = MagicMock() + mock_reader.read_lines.return_value = iter([(i, f"line {i}") for i in range(1, 21)]) + + blocks = [ + MergedBlock(start_line=1, end_line=3, original_windows=(0,), max_score=0.2), + MergedBlock(start_line=5, end_line=7, original_windows=(1,), max_score=0.5), + MergedBlock(start_line=10, end_line=12, original_windows=(2,), max_score=0.8), + ] + mock_merger = MagicMock() + mock_merger.merge_windows.return_value = blocks + + mock_thresholder = MagicMock() + window = TextWindow(content="test", start_line=1, end_line=3, window_id=0) + mock_thresholder.select_significant.return_value = [ScoredWindow(window=window, score=0.5)] + + config = AnalysisConfig(device="cpu", min_score=0.5) + analyzer = SemanticLogAnalyzer( + config, + reader=mock_reader, + merger=mock_merger, + thresholder=mock_thresholder, + ) + result = analyzer.analyze_file_detailed(Path("dummy.log")) + + assert result.merged_blocks == 2 + assert len(result.blocks) == 2 + assert all(b.max_score >= 0.5 for b in result.blocks) + + @patch("cordon.pipeline.create_embedder") + def test_max_blocks_and_min_score_combined(self, mock_create: MagicMock) -> None: + """Test that min_score is applied before max_blocks.""" + mock_embedder = MagicMock() + mock_create.return_value = mock_embedder + mock_embedder.embed_windows.return_value = iter([]) + + mock_reader = MagicMock() + mock_reader.read_lines.return_value = iter([(i, f"line {i}") for i in range(1, 21)]) + + blocks = [ + MergedBlock(start_line=1, end_line=3, original_windows=(0,), max_score=0.2), + MergedBlock(start_line=5, end_line=7, original_windows=(1,), max_score=0.5), + MergedBlock(start_line=10, end_line=12, original_windows=(2,), max_score=0.7), + MergedBlock(start_line=15, end_line=17, original_windows=(3,), max_score=0.9), + ] + mock_merger = MagicMock() + mock_merger.merge_windows.return_value = blocks + + mock_thresholder = MagicMock() + window = TextWindow(content="test", start_line=1, end_line=3, window_id=0) + mock_thresholder.select_significant.return_value = [ScoredWindow(window=window, score=0.5)] + + config = AnalysisConfig(device="cpu", min_score=0.5, max_blocks=2) + analyzer = SemanticLogAnalyzer( + config, + reader=mock_reader, + merger=mock_merger, + thresholder=mock_thresholder, + ) + result = analyzer.analyze_file_detailed(Path("dummy.log")) + + assert result.merged_blocks == 2 + assert len(result.blocks) == 2 + scores = [b.max_score for b in result.blocks] + assert 0.9 in scores + assert 0.7 in scores