Building Streaming LLM Applications: Best Practices and Examples

#engineering #oxlo #ai

We are going to build a streaming DevOps log triage agent that reads raw server logs and emits a structured markdown report token by token. This cuts perceived latency during incidents because operators see the severity and summary as soon as the model generates them, not after a full round-trip wait.

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

Step 1: Initialize the Oxlo.ai client and verify streaming

I start by importing the SDK and pointing it at Oxlo.ai. I also run a one-word sanity check to confirm streaming works.

from openai import OpenAI

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

response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[
        {"role": "user", "content": "Say hello in one word."},
    ],
    stream=True,
)

for chunk in response:
    print(chunk.choices[0].delta.content or "", end="", flush=True)
print()

Step 2: Define the system prompt

The system prompt forces a consistent markdown structure so downstream tools can parse the report without extra formatting noise.

SYSTEM_PROMPT = """You are a DevOps triage agent. Analyze the provided server logs and produce a structured markdown report with exactly these sections:

## Severity
## Summary
## Likely Cause
## Remediation

Keep responses concise. Use bullet points where helpful. Do not ask clarifying questions."""

Step 3: Build the streaming consumer

Now I wrap the call in a function that prints each delta as it arrives. This keeps the terminal updating in real time instead of blocking until the full response is ready.

def triage_logs(log_text: str):
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Analyze these logs:\n\n{log_text}"},
        ],
        stream=True,
        temperature=0.2,
    )

    print("=== TRIAGE REPORT ===")
    for chunk in response:
        token = chunk.choices[0].delta.content or ""
        print(token, end="", flush=True)
    print("\n=====================")

Step 4: Add timing and error handling

Measuring time to first token is essential for SLA monitoring. I also catch API errors so a network blip does not crash the whole pipeline.

import time
import sys
from openai import APIError

def triage_logs(log_text: str):
    start = time.time()
    first_token_time = None

    try:
        response = client.chat.completions.create(
            model="llama-3.3-70b",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": f"Analyze these logs:\n\n{log_text}"},
            ],
            stream=True,
            temperature=0.2,
        )
    except APIError as e:
        print(f"Oxlo.ai API error: {e}", file=sys.stderr)
        sys.exit(1)

    print("Starting stream...")
    for chunk in response:
        if first_token_time is None:
            first_token_time = time.time() - start
            print(f"\n[Time to first token: {first_token_time:.2f}s]\n", flush=True)

        token = chunk.choices[0].delta.content or ""
        print(token, end="", flush=True)

    total = time.time() - start
    print(f"\n[Total elapsed: {total:.2f}s]")

Step 5: Wrap it in a CLI

Finally, I add a minimal argparse interface so the script can be dropped into a shell pipeline or called from a runbook.

import argparse

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Stream a triage report from Oxlo.ai")
    parser.add_argument("--logs", required=True, help="Raw server logs to analyze")
    args = parser.parse_args()

    triage_logs(args.logs)

Run it

Save the complete script as triage.py, set your key, and pass a sample nginx timeout log.

export OXLO_API_KEY="YOUR_OXLO_API_KEY"
python triage.py --logs "2024-05-20T14:32:01Z nginx: upstream timed out (110: Connection timed out) while connecting to upstream, client: 10.0.1.42, server: api.example.com, request: \"POST /v1/batch HTTP/1.1\", upstream: \"http://10.0.2.15:8080/v1/batch\", host: \"api.example.com\""

Typical streamed output looks like this:

Starting stream...

[Time to first token: 0.38s]

## Severity
High

## Summary
- Repeated upstream timeout errors from nginx to the internal batch service at 10.0.2.15:8080.
- Issue isolated to the /v1/batch endpoint.

## Likely Cause
- The upstream application server is either overloaded, crashed, or unreachable due to a network partition.

## Remediation
- Check CPU and memory on 10.0.2.15.
- Verify the batch service process is running and listening on port 8080.
- Review recent deployments to the batch service.
- Consider scaling the backend or increasing the nginx proxy_connect_timeout if the service is slow but healthy.

[Total elapsed: 2.08s]

Wrap-up and next steps

Because Oxlo.ai uses request-based pricing, feeding long log dumps into this agent does not inflate cost the way token-based providers do. You can see exact plan details at https://oxlo.ai/pricing.

Two concrete ways to extend this: wire the script into an alerting webhook so it triggers on PagerDuty incidents, or switch to the async AsyncOpenAI client to triage multiple log files in parallel without blocking.

DEV Community