LLM Applications in Streaming Analytics: Use Cases and Benefits

#learnai #oxlo #ai

We are building a real-time log stream analyzer that reads server logs in fixed windows, flags anomalies, and tracks recurring issues across batches. It helps DevOps teams surface problems immediately without standing up a separate vector database or complex stream processor. Because Oxlo.ai charges a flat rate per request, expanding the log window to capture more context does not inflate cost the way token-based pricing would.

What you'll need

Before starting, grab the following:

Python 3.10 or newer
The OpenAI SDK: pip install openai
An Oxlo.ai API key from https://portal.oxlo.ai

Step 1: Mock the log stream and init the Oxlo.ai client

I will generate fake nginx-style logs and point the OpenAI-compatible client at Oxlo.ai.

import time
import random
from datetime import datetime
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key="YOUR_OXLO_API_KEY"
)

LOG_LEVELS = ["INFO", "WARN", "ERROR", "INFO", "INFO"]
MESSAGES = [
    "GET /api/v1/users 200",
    "POST /api/v1/login 200",
    "Connection timeout to db-primary",
    "Disk usage above 90 percent",
    "GET /healthz 503",
]

def log_stream():
    while True:
        ts = datetime.utcnow().isoformat()
        level = random.choice(LOG_LEVELS)
        msg = random.choice(MESSAGES)
        yield f"{ts} [{level}] {msg}"
        time.sleep(0.5)

if __name__ == "__main__":
    for _ in range(3):
        print(next(log_stream()))

Step 2: Define the system prompt

The prompt forces JSON output so downstream code can act on severity scores without regex parsing.

SYSTEM_PROMPT = """You are a streaming log analyst.
Analyze the provided log window and return a JSON object with exactly these keys:
- summary: one sentence describing what happened in this window
- anomalies: list of unusual or error-level events
- severity: one of low, medium, high, critical
- recurring_pattern: if the previous summary mentions a similar issue, name it, otherwise write null

Respond with valid JSON only. Do not wrap the output in markdown."""

Step 3: Build the batch processor

This function takes a window of log lines, formats them, and sends the batch to Oxlo.ai using Llama 3.3 70B.

import json

def analyze_window(log_lines, previous_summary=None):
    batch = "\n".join(log_lines)
    context = ""
    if previous_summary:
        context = f"\nPrevious window summary: {previous_summary}\n"

    user_message = f"{context}Analyze these logs:\n{batch}"

    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.1,
        max_tokens=512,
    )

    raw = response.choices[0].message.content.strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {
            "summary": raw,
            "anomalies": [],
            "severity": "low",
            "recurring_pattern": None
        }

Step 4: Maintain state across windows

Streaming analytics falls apart without continuity. We pass the previous window's summary into the next request so the model can detect recurring patterns like repeated database timeouts.

def run_analyzer(window_size=10, max_windows=5):
    stream = log_stream()
    previous_summary = None

    for i in range(max_windows):
        window = [next(stream) for _ in range(window_size)]
        result = analyze_window(window, previous_summary)

        print(f"--- Window {i + 1} ---")
        print(json.dumps(result, indent=2))

        previous_summary = result.get("summary")
        time.sleep(1)

    print("Processing complete.")

Run it

Wire the pieces together and process five windows.

if __name__ == "__main__":
    run_analyzer(window_size=10, max_windows=5)

Example output:

--- Window 1 ---
{
  "summary": "Mostly healthy traffic with one database timeout.",
  "anomalies": ["Connection timeout to db-primary"],
  "severity": "medium",
  "recurring_pattern": null
}
--- Window 2 ---
{
  "summary": "Continued healthy traffic plus a disk space warning.",
  "anomalies": ["Disk usage above 90 percent"],
  "severity": "high",
  "recurring_pattern": "database timeout"
}
--- Window 3 ---
{
  "summary": "Healthy traffic with repeated database timeouts.",
  "anomalies": ["Connection timeout to db-primary"],
  "severity": "high",
  "recurring_pattern": "database timeout"
}
...

Wrap-up

Swap the mock generator for a Redis or Kafka consumer to run this against production logs. If you want to keep the prototype free, switch the model to deepseek-v3.2 on Oxlo.ai, which sits in the free tier. For pricing details, see https://oxlo.ai/pricing.

DEV Community