Production logs are noisy. A single deployment can generate thousands of lines of output — stack traces buried in INFO messages, warnings you can't parse in under 30 seconds, and errors that appear 200 lines before the actual failure. Most teams either ignore logs until something breaks, or spend 20 minutes hunting through them after an incident.
There's a better approach: feed your logs to a language model and get a structured, human-readable summary in seconds. This isn't a replacement for proper observability — keep your Grafana/Loki stack — but it's a fast triage layer that shaves time off every incident.
This article walks through building a working log summarizer in Python: one that ingests logs, extracts relevant chunks, and returns a concise incident summary.
Why Raw Logs Are Hard to Triage
Logs aren't designed for humans. They're optimized for machines to write and search. A typical 500 lines of output during an incident might contain:
- 400 lines of normal request handling
- 10 lines of retries and connection resets
- 2 lines of the actual error
- 88 lines of stack trace
A language model can parse all of this, understand the sequence of events, and produce: "Database connection pool exhausted after 3 retries, followed by 502 errors on /api/orders. The root connection failure started at 14:32:01."
That's the output you actually want at 3am.
Step 1: Log Preprocessing and Chunking
Language models have token limits. A 500MB log file won't fit in a single prompt. The solution is to preprocess: filter, deduplicate, and chunk the logs before sending them to the model.
import re
from dataclasses import dataclass
LOG_LEVEL_PATTERN = re.compile(
r"(?P<level>ERROR|WARN(?:ING)?|FATAL|CRITICAL|PANIC)", re.IGNORECASE
)
TIMESTAMP_PATTERN = re.compile(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}")
@dataclass
class LogChunk:
lines: list[str]
has_error: bool
start_ts: str | None
end_ts: str | None
def extract_chunks(
log_lines: list[str],
window: int = 50,
max_chunks: int = 8,
) -> list[LogChunk]:
error_indices = [
i for i, line in enumerate(log_lines)
if LOG_LEVEL_PATTERN.search(line)
]
seen: set[int] = set()
chunks: list[LogChunk] = []
for idx in error_indices:
start = max(0, idx - window // 2)
end = min(len(log_lines), idx + window // 2)
key = (start // window, end // window)
if key in seen:
continue
seen.add(key)
window_lines = log_lines[start:end]
timestamps = [
m.group() for line in window_lines
if (m := TIMESTAMP_PATTERN.search(line))
]
chunks.append(LogChunk(
lines=window_lines,
has_error=True,
start_ts=timestamps[0] if timestamps else None,
end_ts=timestamps[-1] if timestamps else None,
))
if len(chunks) >= max_chunks:
break
return chunks
The key idea: instead of summarizing every log line, we find windows around ERROR/FATAL lines, deduplicate overlapping windows, and pass only those to the model. For a typical incident, this produces 3–6 focused excerpts totalling under 3,000 tokens.
Step 2: Building the Summarization Prompt
Prompt engineering matters here. A vague “summarize these logs” prompt gets vague output. You want structured answers: what failed, when, what the impact was, and what to investigate next.
import json
SYSTEM_PROMPT = (
"You are an expert SRE analyzing application logs.\n"
"Given log excerpts, return a JSON object with these exact fields:\n"
"- summary: one-sentence description of the incident\n"
"- severity: one of [critical, high, medium, low]\n"
"- root_cause: your best hypothesis based on the evidence\n"
"- impacted_services: list of service names you can identify\n"
"- timeline: list of {timestamp, event} objects, chronological\n"
"- recommended_action: the single most important next diagnostic step\n"
"Return only valid JSON. No markdown fences, no explanation."
)
def build_prompt(chunks: list) -> str:
sections = []
for i, chunk in enumerate(chunks, 1):
header = f"--- Excerpt {i}"
if chunk.start_ts:
header += f" ({chunk.start_ts} to {chunk.end_ts})"
sections.append(header + "\n" + "\n".join(chunk.lines))
return "\n\n".join(sections)
def summarize_logs(log_text: str, client) -> dict:
lines = log_text.splitlines()
chunks = extract_chunks(lines)
if not chunks:
chunks = [LogChunk(
lines=lines[-100:],
has_error=False,
start_ts=None,
end_ts=None,
)]
response = client.messages.create(
model="your-model-here",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": build_prompt(chunks)}],
)
return json.loads(response.content[0].text)
The client parameter is your LLM SDK instance. The function is model-agnostic by design — swap in any provider that supports a system prompt.
Step 3: CLI and Pipeline Integration
A summarizer is only useful if it fits into existing workflows. Here's a minimal CLI wrapper that reads from stdin or a file:
#!/usr/bin/env python3
import argparse
import json
import sys
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--file", help="Log file path (default: stdin)")
parser.add_argument("--json", action="store_true", help="Raw JSON output")
args = parser.parse_args()
log_text = open(args.file).read() if args.file else sys.stdin.read()
if not log_text.strip():
print("No log input.", file=sys.stderr)
sys.exit(1)
client = init_llm_client() # initialize your SDK client here
result = summarize_logs(log_text, client)
if args.json:
print(json.dumps(result, indent=2))
return
print(f"\n[{result['severity'].upper()}] {result['summary']}")
print(f"Root cause: {result['root_cause']}")
print(f"Impacted: {', '.join(result['impacted_services'])}")
print(f"Next step: {result['recommended_action']}")
if result.get("timeline"):
print("\nTimeline:")
for event in result["timeline"]:
print(f" {event['timestamp']} {event['event']}")
if __name__ == "__main__":
main()
Pipe this with --json into a Slack webhook or PagerDuty API and you get machine-readable summaries that auto-populate incident tickets. It integrates cleanly with Loki alert webhooks, AWS Lambda triggers on CloudWatch filters, or any alerting system that can execute a shell command.
Practical Considerations
Cost. You're sending compressed log excerpts, not full files. A typical incident window is under 3,000 tokens. At current API rates, that's fractions of a cent per summary — negligible compared to the time saved.
Accuracy. The model will occasionally misidentify service names or get timestamps slightly wrong. Treat the summary as a starting hypothesis, not ground truth. It directs your attention; your judgment closes the loop.
Privacy and security. Logs often contain IP addresses, user IDs, session tokens, and internal hostnames. Before sending logs to any external API, either redact PII with a preprocessing pass, or run a local model. If you're deploying this in a regulated environment, review the security hardening checklists relevant to your stack before going live.
Local models. Running a quantized local model (7B–13B parameters) via Ollama works well for this use case. Log summarization doesn't require frontier-model reasoning — it needs pattern recognition and structured output, which smaller models handle reliably.
The Takeaway
This pattern won't replace structured logging or distributed tracing. What it does is cut the first 15 minutes of "what happened?" from every incident. Feed the relevant log window to a language model, get a structured summary, start fixing things.
The full pipeline is about 150 lines of Python. The hardest part is the prompt — spend time getting the output schema right, and everything else is plumbing.
I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.
Top comments (0)