--- title: "Incident Summariser" description: "Incident response tooling sits squarely in platform/SRE-adjacent engineering — proof you can apply AI to operational, not just developer-facing, problems." source: "https://confidentprep.com/paths/ai-augmented-engineering/incident-summariser/" path: "AI-Augmented Engineering" repository: "https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/incident-summariser" token_estimate: 4224 --- # Incident Summariser ## Overview Converts noisy log files (up to 100MB) into structured incident reports with a timeline, root cause, and action items — using pre-filtering to cut token usage well before any LLM call happens. The pre-filter-before-you-prompt discipline here is a cost lesson that compounds with the observability project before it: cheaper input means a cheaper, faster pipeline in production. ## What to do | Field | Value | |---|---| | Path | 2 — AI-Augmented Engineering | | Position | 7 of 12 | | Difficulty | 🟢 Local only | | Estimated time | 4 hours | | AWS cost | None | --- ### Agent Pickup Instructions ```bash # 1. Enter project cd projects/p2-07-incident-summariser # 2. Create virtual environment python -m venv .venv && source .venv/bin/activate # 3. Install dependencies pip install -r requirements.txt # 4. Copy env file and add your API key cp .env.example .env # 5. Summarise a log file python src/main.py tests/fixtures/app.log # 6. Multi-service correlation python src/main.py tests/fixtures/app.log tests/fixtures/db.log --output report.md # 7. Run tests pytest tests/ -v ``` **Done when:** - [ ] `python src/main.py ` prints a structured incident report with timeline, root cause, and action items - [ ] Pre-filter stats logged to stdout: `Filtered N/M lines (X% reduction)` - [ ] `root_cause.confidence` is a float between 0 and 1 - [ ] Multi-file mode accepts up to 5 log files and merges by timestamp - [ ] All 4 tests pass --- ### What this project is An incident summarization pipeline that processes log files up to 100MB using streaming reads (never loading the full file into memory), extracts error lines and stack traces with regex pre-filtering before any LLM call, correlates multiple service logs by timestamp, and returns a structured incident report containing a timeline of events, a root cause hypothesis with confidence score, and prioritized action items. The pre-filtering step logs the token reduction percentage to prove that smart filtering — not raw file ingestion — is what makes large log analysis practical. --- ### What the learner achieves "I built an incident summarization pipeline that handles 100MB log files by streaming and pre-filtering with regex before any LLM call, reducing token usage by over 90% on noisy logs — and I can show the exact reduction percentage in my output." --- ### Folder structure ``` p2-07-incident-summariser/ ├── src/ │ ├── main.py # CLI: [ ...] [--output report.md] │ ├── llm.py # Provider-agnostic LLM wrapper │ ├── pre_filter.py # stream_filter_errors, calculate_reduction_pct │ ├── correlator.py # merge_multi_service_logs, sort_by_timestamp │ ├── summariser.py # build_incident_prompt, parse_incident_report │ └── reporter.py # format_markdown_report ├── tests/ │ ├── fixtures/ │ │ ├── app.log # 1000-line mixed log with errors │ │ ├── db.log # 500-line database service log │ │ └── clean.log # Log with zero ERROR lines │ ├── test_pre_filter.py │ ├── test_correlator.py │ └── test_summariser.py ├── .env.example ├── requirements.txt └── README.md ``` --- ### .env.example ```bash # LLM provider: anthropic | openai | ollama LLM_PROVIDER=anthropic # Model to use LLM_MODEL= # API key for Anthropic ANTHROPIC_API_KEY= # API key for OpenAI OPENAI_API_KEY= # Ollama base URL OLLAMA_BASE_URL=http://localhost:11434 # Maximum filtered lines to send to LLM (cap to avoid huge prompts) MAX_FILTERED_LINES=500 # Max log files to correlate in multi-service mode MAX_LOG_FILES=5 # Minimum confidence score to display root cause (0.0 to 1.0) MIN_CONFIDENCE=0.0 ``` --- ### requirements.txt ``` anthropic==0.30.0 openai==1.35.0 python-dotenv==1.0.1 rich==13.7.1 pytest==8.2.2 pytest-mock==3.14.0 ``` --- ### src/ — what to implement #### `src/llm.py` **`get_completion(prompt: str, system: str = "") -> str`** - Standard interface: reads `LLM_PROVIDER` and `LLM_MODEL` from env - Routes to Anthropic, OpenAI, or Ollama - Edge cases: raise `RuntimeError` if API key is missing **`get_json_completion(prompt: str, system: str = "") -> dict | list`** - Calls `get_completion`, strips markdown fences, parses JSON - Raises `ValueError` with raw text if parsing fails --- #### `src/pre_filter.py` **`stream_filter_errors(filepath: str) -> tuple[list[str], int]`** - Inputs: path to log file - Output: `(filtered_lines, total_lines)` where `filtered_lines` is a list of strings - Behavior: - Open file with `open(filepath, encoding="utf-8", errors="replace")` — streaming, not `readlines()` - For each line: check if it matches any ERROR/WARN pattern; also capture stack trace lines (indented lines following an ERROR match) - Error patterns: lines containing `ERROR`, `WARN`, `Exception`, `Traceback`, `FATAL`, `CRITICAL` - Stack trace capture: after an `ERROR`/`Exception`/`Traceback` line, include subsequent indented lines (starting with whitespace or `at `) until a non-indented line appears - Return the filtered list and the total line count - Edge cases: binary file content causes decode error → `errors="replace"` handles it; empty file → `([], 0)` **`calculate_reduction_pct(total_lines: int, filtered_lines: int) -> float`** - Inputs: total and filtered line counts - Output: percentage reduction as float, e.g. 94.3 - Behavior: `(1 - filtered_lines / total_lines) * 100` rounded to 1 decimal place - Edge cases: if `total_lines == 0`, return `0.0` **`cap_filtered_lines(filtered_lines: list[str], max_lines: int) -> list[str]`** - Inputs: filtered line list, max count - Output: first `max_lines` lines with a note appended if truncated: `["[... truncated to {max_lines} lines ...]"]` --- #### `src/correlator.py` **`TimestampedLine` (dataclass)** ```python @dataclass class TimestampedLine: raw: str timestamp: str | None # ISO or common log format service: str # Derived from filename (basename without extension) lineno: int ``` **`parse_timestamp(line: str) -> str | None`** - Inputs: a log line string - Output: timestamp string or `None` - Behavior: try common formats with regex: - ISO: `\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}` - Apache/nginx: `\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2}` - Unix epoch: `\d{10}\.\d+` - Return the first match found **`merge_multi_service_logs(filepaths: list[str], filtered_lines_per_file: list[list[str]]) -> list[TimestampedLine]`** - Inputs: list of file paths (for service names), list of filtered line lists per file - Output: flat sorted list of `TimestampedLine` by timestamp - Behavior: tag each line with its service name; parse timestamp; sort by timestamp string (lexicographic sort works for ISO format); lines without timestamps go to end --- #### `src/summariser.py` **`build_incident_prompt(lines: list[TimestampedLine | str], service_names: list[str]) -> str`** - Inputs: filtered log lines, list of service names involved - Output: formatted prompt string - Behavior: - Format each line as `[{service}] {timestamp} {raw}` or just `{raw}` if no timestamp - Build user prompt with: service names, log lines, instruction to return JSON **`parse_incident_report(raw_response: str) -> dict`** - Inputs: LLM response string (expected JSON) - Output: structured dict matching schema: ```json { "timeline": [{"timestamp": "...", "event": "..."}], "root_cause": { "hypothesis": "...", "confidence": 0.85, "evidence": ["...", "..."] }, "action_items": [ {"priority": "HIGH", "description": "..."} ] } ``` - Behavior: call `get_json_completion`; validate keys exist; ensure `confidence` is float in 0-1 range; ensure `action_items` each have `priority` and `description` - Edge cases: if `confidence` is outside 0-1, clamp to range; if JSON missing keys, return a minimal report with `"hypothesis": "Unable to determine"` and `"confidence": 0.0` **`summarise_logs(filepath_or_lines, service_names: list[str]) -> dict`** - Orchestrates the full pipeline for a set of pre-filtered lines --- #### `src/reporter.py` **`format_markdown_report(report: dict, service_names: list[str], reduction_stats: dict) -> str`** - Inputs: parsed incident report, service names, `{"total": int, "filtered": int, "reduction_pct": float}` - Output: formatted markdown string - Sections: - `# Incident Report` - `## Services Involved`: list of service names - `## Pre-filter Stats`: table showing total lines, filtered lines, reduction % - `## Timeline`: ordered list of timestamped events - `## Root Cause`: hypothesis, confidence as percentage, evidence list - `## Action Items`: ordered list with priority badges (HIGH/MEDIUM/LOW) --- #### `src/main.py` CLI entry point using `argparse`. Arguments: - Positional: `log_files` — 1 to 5 log file paths (nargs="+") - `--output ` — write markdown report to file; if omitted, print to stdout Behavior: 1. Load `.env` 2. Validate: at most `MAX_LOG_FILES` files 3. For each file: call `stream_filter_errors`; log `[] {total} lines → {filtered} filtered ({pct}% reduction)` 4. Call `merge_multi_service_logs` if multiple files 5. Call `cap_filtered_lines` 6. Call `build_incident_prompt` and `parse_incident_report` 7. Call `format_markdown_report` 8. Write or print --- ### tests/ — what to test #### Test 1 — Pre-filter extracts ERROR lines only (`test_pre_filter.py`) - Create a 10-line string: 7 INFO lines, 2 ERROR lines, 1 WARN line - Write to temp file; call `stream_filter_errors` - Assert `filtered_lines` has exactly 3 entries (2 ERROR + 1 WARN) - Assert total_lines == 10 #### Test 2 — Token reduction is measurable on 1000-line log (`test_pre_filter.py`) - Use `tests/fixtures/app.log` (1000 lines, mix of INFO/ERROR) - Call `stream_filter_errors` - Call `calculate_reduction_pct` - Assert reduction > 50% (the fixture has mostly INFO lines) #### Test 3 — Structured output has required keys (`test_summariser.py`) - Mock LLM to return a valid JSON incident report - Call `parse_incident_report` - Assert top-level keys: `timeline`, `root_cause`, `action_items` - Assert `root_cause["confidence"]` is a float between 0 and 1 #### Test 4 — Confidence is clamped to 0-1 range (`test_summariser.py`) - Mock LLM to return `{"root_cause": {"confidence": 1.5, ...}, ...}` - Call `parse_incident_report` - Assert `confidence == 1.0` (clamped) #### Test 5 — Multi-service merge sorts by timestamp (`test_correlator.py`) - Create two lists of lines with timestamps: list A has `2024-01-01 10:00:05`, list B has `2024-01-01 10:00:02` - Call `merge_multi_service_logs` - Assert the merged list has the `10:00:02` line before the `10:00:05` line #### Test 6 — Stack trace lines included after ERROR (`test_pre_filter.py`) - Write a temp file with: `ERROR something failed\n at module.py:42\n at main.py:10\nINFO next line` - Call `stream_filter_errors` - Assert the two `at ` lines are included in `filtered_lines` - Assert the `INFO` line is not included --- ### README.md content ```markdown # Incident Summariser Converts noisy log files (up to 100MB) into structured incident reports with timeline, root cause, and action items — using pre-filtering to reduce token usage before any LLM call. ## Quick start ```bash python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt cp .env.example .env # add your API key ## Single log file python src/main.py app.log ## Multi-service correlation (up to 5 files) python src/main.py app.log db.log auth.log --output report.md ``` ## Output sections 1. **Timeline** — ordered list of key events with timestamps 2. **Root Cause** — hypothesis with confidence score (0-1) and evidence 3. **Action Items** — prioritized list (HIGH/MEDIUM/LOW) 4. **Pre-filter Stats** — how many lines were filtered and token reduction % ## Pre-filtering strategy ``` 100MB log → stream read → regex filter (ERROR/WARN/Exception/Traceback) → stack trace capture → cap at MAX_FILTERED_LINES → LLM call (10-50x fewer tokens) ``` ## Running tests ```bash pytest tests/ -v ``` ``` --- ### GUIDE.md content ```markdown # Build Guide — Incident Summariser ## Step 1 — Stream, don't load Never use `f.readlines()` or `f.read()` on a large log file. Stream line by line: ```python def stream_filter_errors(filepath): total = 0 filtered = [] in_stack_trace = False with open(filepath, encoding="utf-8", errors="replace") as f: for line in f: # streams — no full load total += 1 is_error = any(p in line for p in ERROR_PATTERNS) is_stack_line = line.startswith((" ", "\t")) or line.strip().startswith("at ") if is_error: filtered.append(line.rstrip()) in_stack_trace = True elif in_stack_trace and is_stack_line: filtered.append(line.rstrip()) else: in_stack_trace = False return filtered, total ``` ## Step 2 — Always log the reduction Before calling the LLM, log: ```python pct = calculate_reduction_pct(total, len(filtered)) print(f"[{service}] {total} lines → {len(filtered)} filtered ({pct}% reduction)") ``` This is the key interview talking point: "I reduced token usage by 94% before the LLM ever sees the data." ## Step 3 — The incident prompt System prompt: ``` You are an SRE analyzing logs to produce an incident report. Return ONLY valid JSON with this schema: { "timeline": [{"timestamp": "...", "event": "..."}], "root_cause": {"hypothesis": "...", "confidence": 0.0-1.0, "evidence": []}, "action_items": [{"priority": "HIGH|MEDIUM|LOW", "description": "..."}] } ``` ## Step 4 — Multi-service correlation For each file, tag lines with the service name (basename without extension). After filtering, merge all tagged lines into one list and sort by timestamp. The LLM receives `[auth] 10:00:02 ERROR ...` lines alongside `[db] 10:00:05 ERROR ...` lines — cross-service correlation in one prompt. ## Step 5 — Validate confidence Always clamp confidence to 0-1: ```python confidence = max(0.0, min(1.0, float(raw["root_cause"]["confidence"]))) ``` ## How to talk about this in an interview **"How do you handle 100MB log files?"** > Streaming read — I never load the full file into memory. I filter line by line with regex, keeping only ERROR/WARN/exception lines and their stack traces. That reduces a 100MB file to a few hundred lines before any LLM call. **"What's the token reduction?"** > On our test fixture, 94% — 1000 lines down to 58. I log this explicitly so the number is visible in every run. **"How do you handle multiple services?"** > I tag each line with its service name, then merge and sort by timestamp. The LLM sees a unified timeline: `[auth] ERROR ...` followed by `[db] WARN ...`. Cross-service patterns emerge naturally. ``` --- ### Failure modes to handle | Failure | Where | How to handle | |---|---|---| | Log file is binary | `pre_filter.py` | `errors="replace"` in `open()` — replacement chars are benign in filtered output | | Log file is empty | `pre_filter.py` | Return `([], 0)`; `main.py` prints "No lines in " and skips it | | More than 5 log files provided | `main.py` | Print error and exit 1: "Maximum 5 log files supported" | | LLM returns confidence outside 0-1 | `summariser.py` | Clamp: `max(0.0, min(1.0, value))` | | No ERROR lines in any log | `main.py` | Print "No error lines found — generating summary from all lines (capped)"; use `cap_filtered_lines` on raw lines | | JSON parse fails | `summariser.py` | Return minimal report: `{timeline: [], root_cause: {hypothesis: "Unable to determine", confidence: 0.0, evidence: []}, action_items: []}` | --- ### The metric this project measures **What is measured:** Token reduction percentage (pre-filter effectiveness) and confidence score accuracy. **Format (stdout):** ``` [app] 1000 lines → 58 filtered (94.2% reduction) [db] 500 lines → 21 filtered (95.8% reduction) Root cause confidence: 0.82 ``` **Target:** On `tests/fixtures/app.log` (1000 lines, >80% INFO), the reduction must exceed 50%. `root_cause.confidence` must be a float in 0-1. These two numbers together prove that the pre-filtering pipeline and LLM output parsing are both working correctly. ### Model source currency - Anthropic model list: https://docs.anthropic.com/en/docs/about-claude/models/overview - OpenAI model list: https://platform.openai.com/docs/models - verifiedOn: 2026-08-07 - `LLM_MODEL` is required and has no implicit default; choose a supported model for the configured provider at setup time. ## Source code A reference implementation of Incident Summariser — README, guide, source and tests — is at https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/incident-summariser.