shashank ms

Posted on Sep 4

Best Practices for Optimizing LLMs in Production Environments

#engineering #oxlo #ai

We are going to build a production incident triage agent that reads raw application logs, classifies severity, and drafts remediation steps. If you run on-call rotations and want to cut noise without missing real outages, this tool gives you a concrete starting point.

What you'll need

Python 3.10 or newer
The OpenAI SDK: pip install openai
An Oxlo.ai API key from https://portal.oxlo.ai
Optional: a Slack webhook URL exported as SLACK_WEBHOOK_URL if you want live paging

I will use llama-3.3-70b as the workhorse because it handles structured tool calls reliably. Oxlo.ai serves it with no cold starts and flat per-request pricing, so feeding it large log dumps does not inflate cost the way token-based billing does.

Step 1: Configure the client

First, instantiate the OpenAI-compatible client pointing at Oxlo.ai. I keep the API key in an environment variable so it does not leak into source control.

import os
from openai import OpenAI

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

Step 2: Define the tool schema and system prompt

The model needs strict instructions and a predictable output schema. I use native function calling so the response is always machine-readable JSON.

SYSTEM_PROMPT = """You are a site reliability engineer triaging production logs.
Analyze the provided log snippet and submit a triage report.
- P0 means customer-facing outage or data loss.
- P1 means degraded performance or failed background jobs affecting revenue.
- P2 means isolated errors with retry logic handling them.
- P3 means warnings or noise.
Set page_oncall to true only for P0 incidents."""

TRIAGE_TOOL = {
    "type": "function",
    "function": {
        "name": "submit_triage_report",
        "description": "Submit a structured triage report for production logs.",
        "parameters": {
            "type": "object",
            "properties": {
                "severity": {"type": "string", "enum": ["P0", "P1", "P2", "P3"]},
                "summary": {"type": "string"},
                "remediation": {"type": "array", "items": {"type": "string"}},
                "page_oncall": {"type": "boolean"}
            },
            "required": ["severity", "summary", "remediation", "page_oncall"]
        }
    }
}

Step 3: Sanitize and chunk logs

Production logs can be huge. Instead of blindly truncating, I keep the most recent lines and deduplicate repetitive stack traces to preserve signal while staying inside context limits.

def prepare_log_context(raw_logs: str, max_lines: int = 200) -> str:
    lines = raw_logs.strip().splitlines()
    if len(lines) > max_lines:
        lines = lines[-max_lines:]
    deduped = []
    prev = None
    for line in lines:
        if line != prev:
            deduped.append(line)
            prev = line
    return "\n".join(deduped)

Because Oxlo.ai bills per request rather than per token, sending a fuller context window does not increase the cost per call. That makes long-context triage practical.

Step 4: Triage with tool calling and retries

In production, transient network errors happen. I wrap the Oxlo.ai call in a small retry loop and force the model to emit a tool call so parsing never surprises me.

import json
import time

def triage_logs(raw_logs: str, retries: int = 3) -> dict:
    context = prepare_log_context(raw_logs)
    for attempt in range(1, retries + 1):
        try:
            response = client.chat.completions.create(
                model="llama-3.3-70b",
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": f"Logs:\n{context}"},
                ],
                tools=[TRIAGE_TOOL],
                tool_choice={"type": "function", "function": {"name": "submit_triage_report"}},
                temperature=0.1,
            )
            tool_call = response.choices[0].message.tool_calls[0]
            return json.loads(tool_call.function.arguments)
        except Exception:
            if attempt == retries:
                raise
            time.sleep(1.5 * attempt)
    return {}

Step 5: Wire up the escalation webhook

If the report marks page_oncall, I POST to a Slack webhook and print to stdout so a pager bridge can pick it up. This closes the loop between detection and response.

import json
import os
import urllib.request

def maybe_escalate(report: dict) -> None:
    if not report.get("page_oncall"):
        return

    text = (f":rotating_light: *{report['severity']}* incident detected\n"
            f"*Summary:* {report['summary']}\n"
            f"*Steps:* {', '.join(report['remediation'])}")

    payload = json.dumps({"text": text}).encode("utf-8")
    webhook_url = os.getenv("SLACK_WEBHOOK_URL")

    if webhook_url:
        req = urllib.request.Request(
            webhook_url,
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            resp.read()

    print(f"ESCALATION TRIGGERED: {report['severity']} - {report['summary']}")

Step 6: Instrument latency and errors

In production I track how long each triage takes and how many fail after retries. Because Oxlo.ai charges a flat rate per request, my cost per run is fixed regardless of log length, but latency still tells me if I need a faster model or smaller chunks.

import time
from datetime import datetime
from functools import wraps

def instrument(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        status = "ok"
        try:
            return func(*args, **kwargs)
        except Exception:
            status = "error"
            raise
        finally:
            elapsed = time.time() - start
            print(f"[{datetime.utcnow().isoformat()}] {func.__name__} "
                  f"status={status} latency_ms={int(elapsed * 1000)}")
    return wrapper

@instrument
def run_triage(raw_logs: str) -> dict:
    report = triage_logs(raw_logs)
    maybe_escalate(report)
    return report

Run it

Here is a sample log stream from a fake payment worker. I pipe it through the agent and print the structured report.

if __name__ == "__main__":
    SAMPLE_LOGS = """
2024-05-21T14:32:10Z [INFO] payment-worker started batch id=9821
2024-05-21T14:32:11Z [ERROR] payment-worker database connection timeout after 30s
2024-05-21T14:32:12Z [ERROR] payment-worker retry 1/3 failed
2024-05-21T14:32:13Z [ERROR] payment-worker retry 2/3 failed
2024-05-21T14:32:14Z [FATAL] payment-worker all retries exhausted, dropping 412 jobs
2024-05-21T14:32:15Z [WARN] payment-worker health check returning 503
"""

    report = run_triage(SAMPLE_LOGS)
    print(json.dumps(report, indent=2))

Expected output:

[2024-05-21T14:32:20.123456] run_triage status=ok latency_ms=890
ESCALATION TRIGGERED: P0 - Payment worker exhausted all retries and dropped 412 jobs
{
  "severity": "P0",
  "summary": "Payment worker exhausted all retries and dropped 412 jobs",
  "remediation": [
    "Restart the payment-worker and verify database connectivity",
    "Replay dropped jobs from the dead-letter queue",
    "Scale database connection pool temporarily"
  ],
  "page_oncall": true
}

Next steps

Swap llama-3.3-70b for qwen-3-32b if your logs contain multilingual stack traces, or try kimi-k2.6 when you need vision support for dashboard screenshots alongside text logs. If you want to experiment without cost concerns, deepseek-v3.2 sits on Oxlo.ai's free tier and handles coding-heavy traces well. Check https://oxlo.ai/pricing to see how request-based billing keeps long-context triage predictable.

DEV Community