DEV Community

shashank ms
shashank ms

Posted on

Best Practices for LLM Deployment in Scientific Computing

#engineering #oxlo #ai

We are building a simulation log analyzer that ingests long stdout files from HPC jobs, extracts convergence metrics, and decides whether to archive or flag the run. It is meant for computational scientists who need to review dozens of nightly simulations without opening every log by hand.

What you'll need

Step 1: Configure the Oxlo.ai client with production timeouts

I always start by setting a generous timeout and enabling retries at the transport layer. Oxlo.ai serves models warm, so you will not hit cold starts, but network blips still happen in clustered environments.

from openai import OpenAI

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

Step 2: Lock down the system prompt

The system prompt is the contract. I keep it strict, forbid prose, and instruct the model to emit results only through the structured tool we will define next.

SYSTEM_PROMPT = """You are an HPC log monitor attached to a computational fluid dynamics solver. Read the full stdout log provided by the user and call the submit_analysis tool with exact findings. Be concise and deterministic. Do not add conversational text."""

Step 3: Define a forced function schema for structured extraction

Instead of parsing freeform JSON from a message body, I force a function call. This is the single most reliable way to get structured data out of an LLM in production.

tools = [
    {
        "type": "function",
        "function": {
            "name": "submit_analysis",
            "description": "Submit the structured analysis of one simulation log.",
            "parameters": {
                "type": "object",
                "properties": {
                    "status": {
                        "type": "string",
                        "enum": ["converged", "diverged", "incomplete", "error"]
                    },
                    "final_residual": {"type": "number"},
                    "iterations_completed": {"type": "integer"},
                    "anomalies": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "recommended_action": {
                        "type": "string",
                        "enum": ["archive", "alert_operator", "retry"]
                    }
                },
                "required": [
                    "status",
                    "final_residual",
                    "iterations_completed",
                    "anomalies",
                    "recommended_action"
                ]
            }
        }
    }
]

Step 4: Send long logs in one request

Simulation logs routinely exceed fifty thousand tokens. Because Oxlo.ai uses flat per-request pricing, I send the entire log in one shot rather than maintaining chunking logic. See https://oxlo.ai/pricing for current plans. Here is a short example log; in production this string is the full stdout file.

sample_log = """[INFO] Solver started: Reynolds-Averaged Navier-Stokes
[ITER 1] Residual: 1.000000e+00
[ITER 100] Residual: 3.450000e-02
[ITER 1000] Residual: 1.200000e-04
[ITER 4500] Residual: 9.800000e-07
[INFO] Convergence criterion met
[INFO] Writing field data to disk
[INFO] Solver finished normally
"""

response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": sample_log},
    ],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "submit_analysis"}},
)

Step 5: Wrap the pipeline in a resilient monitor

Wrapping the call in a small class gives me a single place to handle API errors, parse tool arguments, and enforce a consistent interface. I also keep the raw response available for audit trails.

import json
from openai import APIError

class SimulationMonitor:
    def __init__(self, client):
        self.client = client

    def analyze(self, log_text: str) -> dict:
        try:
            resp = self.client.chat.completions.create(
                model="llama-3.3-70b",
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": log_text},
                ],
                tools=tools,
                tool_choice={"type": "function", "function": {"name": "submit_analysis"}},
            )
            tool_call = resp.choices[0].message.tool_calls[0]
            return json.loads(tool_call.function.arguments)
        except APIError as e:
            return {"error": str(e), "status": "unknown"}

Run it

With the monitor defined, point it at a log string and print the structured report.

monitor = SimulationMonitor(client)
report = monitor.analyze(sample_log)
print(json.dumps(report, indent=2))

Example output:

{
  "status": "converged",
  "final_residual": 9.8e-07,
  "iterations_completed": 4500,
  "anomalies": [],
  "recommended_action": "archive"
}

Wrap-up

Wire this monitor into your job epilog script so every completed run is assessed automatically before data is moved to scratch. If you later need deeper reasoning on failure modes, swap the model string to deepseek-v3.2 or kimi-k2.6 without changing any other code.

Top comments (0)