Manufacturing floors generate a constant stream of unstructured defect reports, shift notes, and sensor alerts. I built a lightweight LLM pipeline that reads raw text from the floor, classifies defects by type and severity, and routes tickets to the correct maintenance team. It runs entirely on Oxlo.ai, so I pay a flat rate per request instead of burning through tokens on long equipment logs. See Oxlo.ai pricing for details.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai - A few sample defect reports (I provide them below)
Step 1: Configure the Oxlo.ai client
I start by instantiating the OpenAI-compatible client pointing at Oxlo.ai. I use Llama 3.3 70B because it handles structured instruction following well.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def test_connection():
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[{"role": "user", "content": "Say OK"}],
)
return response.choices[0].message.content
print(test_connection())
Step 2: Write the system prompt
The system prompt is the contract. It tells the model exactly what fields to return and how to score severity. I keep it strict so the output is predictable.
SYSTEM_PROMPT = """You are a manufacturing defect classifier. The user will paste an unstructured defect report from the factory floor.
Analyze the report and produce a JSON object with exactly these keys:
- defect_type: one of [Mechanical, Electrical, Software, Hydraulic, Quality]
- severity: one of [Critical, High, Medium, Low]
- confidence: integer 1-10
- summary: a one-sentence summary of the defect
- corrective_action: a concise, specific fix
- assigned_team: one of [Maintenance_Mech, Maintenance_Elec, IT_Operations, Quality_Assurance, Production_Supervisor]
Rules:
- If the report mentions sparks, smoke, or safety risk, severity must be Critical.
- If a CNC machine is named, assigned_team is Maintenance_Mech.
- Respond with ONLY the JSON object, no markdown fences."""
Step 3: Build the classifier function
Now I wrap the API call in a function that sends the report and parses the JSON response. I set temperature low because consistency matters more than creativity on the factory floor.
import json
def classify_defect(report_text: str):
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": report_text},
],
temperature=0.1,
)
content = response.choices[0].message.content.strip()
# Clean up accidental markdown fences
if content.startswith("
```"):
content = content.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(content)
Step 4: Add batch processing and validation
Reports arrive in clusters after each shift. I process a list of reports and validate that every output contains the required keys. If a key is missing, I log the error instead of crashing the line.
REQUIRED_KEYS = {"defect_type", "severity", "confidence", "summary", "corrective_action", "assigned_team"}
def process_shift_reports(reports: list[str]):
results = []
for report in reports:
try:
parsed = classify_defect(report)
if not REQUIRED_KEYS.issubset(parsed.keys()):
missing = REQUIRED_KEYS - parsed.keys()
raise ValueError(f"Missing keys: {missing}")
results.append(parsed)
except Exception as e:
results.append({"error": str(e), "raw_report": report[:200]})
return results
Step 5: Build a routing notifier
Classification is useless if the ticket sits in a queue. I add a function that prints routing instructions. In production, this POSTs to a Slack webhook or your CMMS API.
def route_ticket(classification: dict):
if "error" in classification:
print(f"ESCALATION: failed to classify report: {classification['error']}")
return
team = classification["assigned_team"]
severity = classification["severity"]
# Swap this print statement for a real webhook in production
print(f"[{severity}] Ticket for {team}: {classification['summary']}")
print(f"Suggested fix: {classification['corrective_action']}\n")
Step 6: Run the full pipeline
I tie the pieces together in a main block that simulates a shift handoff. Because Oxlo.ai charges per request, not per token, feeding it a multi-paragraph maintenance log costs the same as a one-line prompt. That makes this pipeline predictable even when operators paste long sensor dumps.
if __name__ == "__main__":
shift_reports = [
"Line 3 CNC mill making grinding noise since 06:00. Operator Smith noted vibration level 4 out of 10. Tool change did not help. Oil pressure gauge normal.",
"SMOKE observed near panel B2 electrical cabinet at 14:23. Emergency stop triggered. No injuries. Needs immediate attention.",
"Barcode scanner on packing station 7 failing intermittently. Reads 1 in 5 boxes. Cable looks fine.",
]
classified = process_shift_reports(shift_reports)
for ticket in classified:
route_ticket(ticket)
Run it
I run the script with my Oxlo.ai key set. Here is the output I see against the sample reports.
[Medium] Ticket for Maintenance_Mech: CNC mill on Line 3 producing abnormal grinding noise and vibration after tool change.
Suggested fix: Inspect spindle bearings and belt tension; schedule mechanical maintenance if vibration persists.
[Critical] Ticket for Maintenance_Elec: Smoke detected near electrical cabinet B2 triggering emergency stop.
Suggested fix: Isolate power to B2, inspect for short circuits or overheating components, and replace damaged breakers before restart.
[Low] Ticket for IT_Operations: Barcode scanner on packing station 7 experiencing intermittent read failures.
Suggested fix: Re-seat USB connection, update scanner firmware, and test with a fresh calibration sheet.
Wrap-up
Two concrete ways to extend this. First, persist the structured output to SQLite or POST it directly to your CMMS work-order API so the loop closes automatically. Second, add a vision step using Oxlo.ai's kimi-k2.6 or gemma-3-27b to read photos of the defect from an operator tablet, merging image and text into a single ticket.
Top comments (0)