Incident Summariser

Field	Value
Path	2 — AI-Augmented Engineering
Position	7 of 12
Difficulty	🟢 Local only
Estimated time	4 hours
AWS cost	None

---

Agent Pickup Instructions

# 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 <log_file> 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: <log_file> [<log_file2> ...] [--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

# 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)

@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:

  {
    "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 <path> — 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 [<service>] {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

# 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

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:

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:

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 <file>" 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.