DEV Community

shashank ms
shashank ms

Posted on

LLMs in Manufacturing: A Guide to Implementation

We are going to build a manufacturing root-cause analysis agent that turns unstructured operator notes and machine alerts into structured troubleshooting steps. It helps plant engineers and line supervisors reduce downtime by surfacing likely causes and corrective actions in seconds. I will walk through the exact Python service I deployed on our internal floor network.

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. I use Oxlo.ai because the flat per-request pricing does not balloon when I paste in lengthy error logs, which makes budgeting for a plant-floor tool predictable. See https://oxlo.ai/pricing for details.
  • A sample incident record. I have provided one below.

Step 1: Initialize the Oxlo.ai client

First, I import the SDK and instantiate the client against Oxlo.ai. I keep the API key in an environment variable so it does not end up in git.

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")
)

response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[
        {"role": "user", "content": "ping"}
    ],
)
print(response.choices[0].message.content)

Step 2: Define the system prompt

The system prompt is the only place where I encode the agent's behavior. I keep it strict: JSON output only, no prose, and always include confidence scores.

SYSTEM_PROMPT = """You are a manufacturing line troubleshooting assistant.
You receive an incident description containing operator notes and machine telemetry.
Respond with a single JSON object containing these keys:
- root_cause: a string describing the most likely cause.
- confidence: an integer from 1 to 10.
- corrective_actions: a list of strings, ordered by priority.
- required_downtime_minutes: an integer estimate.
Do not include markdown formatting, explanations, or text outside the JSON object."""

Step 3: Format the incident input

I found that feeding the model a consistent text template reduces hallucination. This function takes a raw incident dictionary and returns a flat string.

def format_incident(incident: dict) -> str:
    lines = [
        f"Machine ID: {incident['machine_id']}",
        f"Timestamp: {incident['timestamp']}",
        f"Error Codes: {', '.join(incident['error_codes'])}",
        f"Operator Notes: {incident['operator_notes']}",
        f"Sensor Snapshot: {incident['sensor_snapshot']}",
    ]
    return "\n".join(lines)

sample_incident = {
    "machine_id": "CNC-Mill-04",
    "timestamp": "2024-05-17T14:32:00Z",
    "error_codes": ["E-2301", "E-4402"],
    "operator_notes": "Finish pass started fine, then spindle load jumped to 95%. Chatter marks visible on titanium bracket. Coolant flow looks normal.",
    "sensor_snapshot": {
        "spindle_load_pct": 95,
        "vibration_rms": 4.8,
        "coolant_flow_lpm": 12.3,
        "feed_rate_mmpm": 800
    }
}

user_message = format_incident(sample_incident)
print(user_message)

Step 4: Call the model and parse output

Now I wire the formatted incident to the Oxlo.ai chat endpoint. I use Llama 3.3 70B because it follows structured instructions reliably on this hardware. After the call, I strip any markdown fences and parse the JSON.

import json

def analyze_incident(incident: dict) -> dict:
    user_message = format_incident(incident)

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

    raw = response.choices[0].message.content.strip()
    if raw.startswith("

```"):
        raw = raw.split("\n", 1)[1].rsplit("```

", 1)[0].strip()
    return json.loads(raw)

result = analyze_incident(sample_incident)
print(json.dumps(result, indent=2))

Step 5: Batch process a shift report

On the floor, incidents arrive in batches. This loop processes a list and accumulates results for a supervisor dashboard.

shift_incidents = [
    sample_incident,
    {
        "machine_id": "Robo-Weld-09",
        "timestamp": "2024-05-17T15:10:00Z",
        "error_codes": ["E-1120"],
        "operator_notes": "Arc lost twice during seam three. Tip shows excessive spatter buildup. Gas pressure reads 18 CFH.",
        "sensor_snapshot": {
            "arc_voltage": 22.1,
            "wire_speed_mpm": 8.5,
            "gas_pressure_cfh": 18.0
        }
    }
]

dashboard = []
for inc in shift_incidents:
    try:
        out = analyze_incident(inc)
        out["machine_id"] = inc["machine_id"]
        dashboard.append(out)
    except Exception as e:
        dashboard.append({"machine_id": inc["machine_id"], "error": str(e)})

print(json.dumps(dashboard, indent=2))

Run it

Running the script against the sample CNC mill incident produces the following output. The model correctly flags tool wear as the root cause because of the high spindle load and chatter correlation.

{
  "root_cause": "Progressive tool flank wear leading to increased cutting forces and chatter in titanium",
  "confidence": 8,
  "corrective_actions": [
    "Index or replace the current end-mill insert",
    "Reduce feed rate to 600 mmpm for titanium grade",
    "Verify tool length offset after change"
  ],
  "required_downtime_minutes": 25
}

Next steps

Wire this agent into your MQTT broker or SCADA alert stream so it triggers automatically on new error codes. You can also add a feedback loop where confirmed fixes are written to a vector store, then retrieve them with Oxlo.ai on recurring failures to build a living knowledge base for your plant.

Top comments (0)