🟢 TECHNICIAN TRACK • CAPSTONE PROJECT

Tutorial #10: Conveyor Jam Diagnosis from Alarm Logs

An Extensible Advisory Agent with Multiple Read-Only Tools

✅ CORE MISSION OF THIS TUTORIAL

By the end of this tutorial, the reader will be able to:

✅ Build a professional advisory diagnostic agent with a clear, auditable pipeline
✅ Analyze historical conveyor alarm logs using a clean, consistent data model
✅ Correlate alarm sequences against known fault patterns using read-only reference data
✅ Produce structured diagnostic hypotheses in JSON (schema-first, validation-first)
✅ Understand the complete agent pipeline: Load → Prompt → Run → Validate → Human Review

This capstone integrates the Technician Track into a single, safe diagnostic workflow you can reuse on real lines.

⚠️

⚠️ SAFETY BOUNDARY REMINDER

This tutorial provides recommendations only.

It must never be connected to:

Live PLCs
Production deployment pipelines
Safety-rated controllers
Motion or power systems

> All outputs are advisory-only and always require explicit human approval before any real-world action.

🌍 VENDOR-AGNOSTIC ENGINEERING NOTE

This tutorial uses:

▸ CSV exports from SCADA/HMI alarm history (timestamp + tag/alarm name)
▸ Rockwell FactoryTalk Alarm/Event logs (exported to text/CSV)
▸ Siemens WinCC / TIA Portal alarm archives (exported)
▸ Ignition alarm journal exports (CSV)
▸ Beckhoff TwinCAT alarm logs (exported)
▸ Any “TIMESTAMP | ALARM_NAME” plaintext log (post-cleaning)

No vendor-specific APIs required. This tutorial uses read-only files you can export from most HMIs/SCADAs.

TIME

100 min

EST. API COST

~$0.12

DIFFICULTY

Beginner

Prerequisites: Technician Track tutorials T1–T9 (foundations: safe prompting, tool use, structured outputs).

Cost note: the estimate assumes a handful of short runs using gpt-4o-mini while you iterate. Your cost depends on log length and how many times you re-run the agent.

🎯 Capstone Overview — Why This Exists

Packaging and conveyor systems frequently experience jam-related interruptions. Experienced engineers diagnose these issues by correlating alarm sequences, timing, and known failure modes.

This capstone builds an advisory diagnostic agent that:

✓ Analyzes historical conveyor alarm logs
✓ Correlates alarm sequences against known jam patterns (read-only reference)
✓ Produces structured diagnostic hypotheses with confidence scores (JSON)
✓ Requires human review before any operational action is taken

Key Principle: Humans keep authority. The agent proposes; operators decide.

1️⃣ System Boundary & Constraints

📥 Inputs

• Alarm log file (historical, cleaned)
• Fault reference file (known patterns)

📤 Outputs

• Structured diagnostic hypotheses (JSON)

⚠️ Hard Constraints

🔒 Read-only

🔒 Advisory-only

🔒 Stateless

🔒 No live PLC or fieldbus access

2️⃣ Architecture at a Glance

graph LR
    A[Alarm Log File] --> C[Reasoning Agent]
    B[Fault Reference File] --> C
    C --> D[Structured Diagnosis JSON]
    D --> E[Human Review]

    style A fill:#1a1a1e,stroke:#1f75ff,stroke-width:2px,color:#1f75ff
    style B fill:#1a1a1e,stroke:#1f75ff,stroke-width:2px,color:#1f75ff
    style C fill:#1a1a1e,stroke:#9e4aff,stroke-width:2px,color:#9e4aff
    style D fill:#1a1a1e,stroke:#04d9ff,stroke-width:2px,color:#04d9ff
    style E fill:#1a1a1e,stroke:#00ff7f,stroke-width:2px,color:#00ff7f

Pipeline flow (what happens every run):

▸ Load two read-only inputs: alarms + fault reference
▸ Prompt the agent with rules + schema + data
▸ Run an LLM call constrained to JSON output
▸ Validate + Review schema-check the output, then hand to a human

Perception–Decision–Action lens: Perception = load/exported alarm history + reference patterns. Decision = form hypotheses from sequences. Action = produce advisory JSON. Final step is always human review (no autonomy).

3️⃣ Data Model — Alarm Logs

Example alarm log format (timestamped sequence):

08:12:01 | PHOTOEYE_BLOCKED
08:12:03 | CONVEYOR_MOTOR_RUNNING
08:12:07 | JAM_TIMEOUT
08:12:08 | CONVEYOR_STOPPED
08:12:10 | OPERATOR_RESET
08:12:12 | PHOTOEYE_BLOCKED
08:12:15 | JAM_TIMEOUT

Format: TIMESTAMP | ALARM_NAME

Each line represents a discrete alarm event in chronological order.

4️⃣ Tool 1 — Alarm Log Loader (Read-Only)

Simple file reader that loads alarm log as list of strings:

Python

def load_alarm_log(file_path: str) -> list[str]:
    """
    Loads alarm log file and returns list of alarm lines.

    Args:
        file_path: Path to alarm log file

    Returns:
        List of alarm strings (stripped, non-empty)
    """
    with open(file_path, "r") as f:
        return [line.strip() for line in f if line.strip()]

Read-Only Guarantee: This tool opens files in read mode only ("r"). No write operations possible.

5️⃣ Tool 2 — Fault Pattern Reference Loader (Read-Only)

Loads known fault patterns from JSON reference file:

Python

def load_fault_reference(file_path: str) -> dict:
    """
    Loads fault reference patterns from JSON file.

    Args:
        file_path: Path to fault_reference.json

    Returns:
        Dictionary of known fault patterns
    """
    import json
    with open(file_path, "r") as f:
        return json.load(f)

Example Fault Reference Structure:

{
  "recurring_jam": {
    "signature": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"],
    "description": "Repeated blockage at same photoeye"
  },
  "motor_stall": {
    "signature": ["CONVEYOR_MOTOR_RUNNING", "JAM_TIMEOUT", "CONVEYOR_STOPPED"],
    "description": "Motor running but jam timeout triggered"
  }
}

6️⃣ Output Contract — Structured Diagnosis Schema

All agent outputs must conform to this JSON schema:

JSON

{
	  "detected_patterns": [],
	  "possible_faults": [
	    {
	      "fault_name": "",
	      "confidence": 0.0,
	      "matched_alarms": [],
	      "reasoning_summary": "",
	      "evidence": {
	        "reference_key": "",
	        "signature_matched": []
	      }
	    }
	  ],
	  "notes": []
	}

detected_patterns

Names/keys of patterns detected from the alarm log

possible_faults

Hypotheses with confidence + evidence (what matched and why)

reasoning_summary

Short, auditable explanation (no long chain-of-thought)

notes

Caveats, data quality warnings, and human review reminders

7️⃣ Prompt Construction

Build system prompt that embeds rules, schema, data:

Python

def build_diagnostic_prompt(alarms: list[str], fault_reference: dict) -> str:
    """
    Constructs diagnostic prompt with embedded rules and data.

    Args:
        alarms: List of alarm log lines
        fault_reference: Known fault patterns dictionary

    Returns:
        Formatted prompt string
    """
    return f"""
You are an industrial diagnostic analysis agent.

	Rules:
	- Advisory only
	- No control decisions
	- No maintenance actions
	- Output ONLY valid JSON
	- Follow the schema exactly
	- Your output must be auditable: include evidence + a short reasoning summary per hypothesis
	- If data is ambiguous, reduce confidence and add notes
	
	Schema:
	{{
	  "detected_patterns": [],
	  "possible_faults": [
	    {{
	      "fault_name": "",
	      "confidence": 0.0,
	      "matched_alarms": [],
	      "reasoning_summary": "",
	      "evidence": {{
	        "reference_key": "",
	        "signature_matched": []
	      }}
	    }}
	  ],
	  "notes": []
	}}

	Few-shot example (format only — do NOT copy values blindly):
	Input alarm log:
	08:12:01 | PHOTOEYE_BLOCKED
	08:12:07 | JAM_TIMEOUT

	Example output:
	{{
	  "detected_patterns": ["recurring_jam"],
	  "possible_faults": [
	    {{
	      "fault_name": "recurring_jam",
	      "confidence": 0.7,
	      "matched_alarms": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"],
	      "reasoning_summary": "The alarm sequence matches the recurring_jam signature in the reference file.",
	      "evidence": {{
	        "reference_key": "recurring_jam",
	        "signature_matched": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"]
	      }}
	    }}
	  ],
	  "notes": ["Advisory only — verify mechanically before acting."]
	}}
	
	Known fault patterns:
	{fault_reference}
	
Alarm log:
{alarms}
"""

8️⃣ Agent Execution Pipeline

Execute agent with loaded data and parse JSON response:

Python

from openai import OpenAI
import json

client = OpenAI()

# Continuing from previous sections:
# load_alarm_log, load_fault_reference, and build_diagnostic_prompt

# Load data using tools
alarms = load_alarm_log("alarms.log")
fault_reference = load_fault_reference("fault_reference.json")

# Execute agent
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "user",
            "content": build_diagnostic_prompt(alarms, fault_reference)
        }
    ],
    response_format={"type": "json_object"}
)

# Parse structured output
diagnosis = json.loads(response.choices[0].message.content)

Note: Using response_format={"type": "json_object"} enforces JSON output from the model.

9️⃣ Output Validation

Validate agent output before accepting diagnosis:

Python

def validate_diagnosis(data: dict) -> bool:
    """
    Validates diagnostic output against required schema.

    Args:
        data: Parsed JSON output from agent

    Returns:
        True if valid, False otherwise
    """
    if not isinstance(data, dict):
        return False

    required_top_keys = ["detected_patterns", "possible_faults", "notes"]
    if any(key not in data for key in required_top_keys):
        return False

    if not isinstance(data["detected_patterns"], list) or not all(
        isinstance(item, str) for item in data["detected_patterns"]
    ):
        return False

    if not isinstance(data["notes"], list) or not all(isinstance(item, str) for item in data["notes"]):
        return False

    if not isinstance(data["possible_faults"], list):
        return False

    required_fault_keys = [
        "fault_name",
        "confidence",
        "matched_alarms",
        "reasoning_summary",
        "evidence",
    ]

    for fault in data["possible_faults"]:
        if not isinstance(fault, dict):
            return False
        if any(key not in fault for key in required_fault_keys):
            return False
        if not isinstance(fault["fault_name"], str):
            return False
        if not isinstance(fault["confidence"], (int, float)) or not (0.0 <= float(fault["confidence"]) <= 1.0):
            return False
        if not isinstance(fault["matched_alarms"], list) or not all(
            isinstance(item, str) for item in fault["matched_alarms"]
        ):
            return False
        if not isinstance(fault["reasoning_summary"], str):
            return False
        if not isinstance(fault["evidence"], dict):
            return False
        if "reference_key" not in fault["evidence"] or "signature_matched" not in fault["evidence"]:
            return False
        if not isinstance(fault["evidence"]["reference_key"], str):
            return False
        if not isinstance(fault["evidence"]["signature_matched"], list) or not all(
            isinstance(item, str) for item in fault["evidence"]["signature_matched"]
        ):
            return False

    return True


# Usage
if validate_diagnosis(diagnosis):
    print("✓ Valid diagnostic output")
else:
    raise ValueError("Invalid diagnosis schema")

Critical: Never accept unvalidated agent outputs. Validation acts as a schema contract enforcer.

🔟 Single Lab — Run Complete Pipeline

📋 Objective

Execute the complete diagnostic agent pipeline once and inspect the structured diagnostic output. Verify that all components integrate correctly.

🛠️ Setup Requirements

Create two files in your working directory:

1. alarms.log

08:12:01 | PHOTOEYE_BLOCKED
08:12:03 | CONVEYOR_MOTOR_RUNNING
08:12:07 | JAM_TIMEOUT
08:12:08 | CONVEYOR_STOPPED
08:12:10 | OPERATOR_RESET
08:12:12 | PHOTOEYE_BLOCKED
08:12:15 | JAM_TIMEOUT

2. fault_reference.json

{
  "recurring_jam": {
    "signature": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"],
    "description": "Repeated blockage at same location"
  },
  "motor_stall": {
    "signature": ["CONVEYOR_MOTOR_RUNNING", "JAM_TIMEOUT"],
    "description": "Motor running but jam timeout triggered"
  }
}

▶️ Execution Steps

1. Create the two data files above
2. Run the reference implementation (see Section 11 below)
3. Inspect the JSON diagnostic output
4. Verify schema compliance with validator

Expected Behavior: Agent should detect both "recurring_jam" (PHOTOEYE_BLOCKED appears twice) and "motor_stall" patterns in the alarm sequence. It should also include evidence (matched signature alarms) and a short reasoning summary for each hypothesis.

1️⃣1️⃣ Known Limitations

📊 Pattern-Based Diagnosis

Agent relies on pre-defined fault patterns. Novel failure modes may not be detected.

🧹 Clean Data Required

Assumes alarm logs are pre-processed and formatted consistently. No noise filtering.

👤 Human Review Mandatory

All diagnostic outputs are advisory. Operator must validate before acting.

1️⃣2️⃣ Reference Implementation (Complete Assembly)

Note: This complete script is here for verification and “copy-run” testing. The primary learning path is the step-by-step sections above.

conveyor_diagnostic_agent.py

# conveyor_diagnostic_agent.py

from openai import OpenAI
import json

def load_alarm_log(file_path: str) -> list[str]:
    """
    Loads alarm log file and returns list of alarm lines.

    Args:
        file_path: Path to alarm log file

    Returns:
        List of alarm strings (stripped, non-empty)
    """
    with open(file_path, "r") as f:
        return [line.strip() for line in f if line.strip()]

def load_fault_reference(file_path: str) -> dict:
    """
    Loads fault reference patterns from JSON file.

    Args:
        file_path: Path to fault_reference.json

    Returns:
        Dictionary of known fault patterns
    """
    with open(file_path, "r") as f:
        return json.load(f)

def build_diagnostic_prompt(alarms: list[str], fault_reference: dict) -> str:
    """
    Constructs diagnostic prompt with embedded rules and data.

    Args:
        alarms: List of alarm log lines
        fault_reference: Known fault patterns dictionary

    Returns:
        Formatted prompt string
    """
    return f"""
You are an industrial diagnostic analysis agent.

	Rules:
	- Advisory only
	- No control decisions
	- No maintenance actions
	- Output ONLY valid JSON
	- Follow the schema exactly
	- Your output must be auditable: include evidence + a short reasoning summary per hypothesis
	- If data is ambiguous, reduce confidence and add notes
	
	Schema:
	{{
	  "detected_patterns": [],
	  "possible_faults": [
	    {{
	      "fault_name": "",
	      "confidence": 0.0,
	      "matched_alarms": [],
	      "reasoning_summary": "",
	      "evidence": {{
	        "reference_key": "",
	        "signature_matched": []
	      }}
	    }}
	  ],
	  "notes": []
	}}

	Few-shot example (format only — do NOT copy values blindly):
	Input alarm log:
	08:12:01 | PHOTOEYE_BLOCKED
	08:12:07 | JAM_TIMEOUT

	Example output:
	{{
	  "detected_patterns": ["recurring_jam"],
	  "possible_faults": [
	    {{
	      "fault_name": "recurring_jam",
	      "confidence": 0.7,
	      "matched_alarms": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"],
	      "reasoning_summary": "The alarm sequence matches the recurring_jam signature in the reference file.",
	      "evidence": {{
	        "reference_key": "recurring_jam",
	        "signature_matched": ["PHOTOEYE_BLOCKED", "JAM_TIMEOUT"]
	      }}
	    }}
	  ],
	  "notes": ["Advisory only — verify mechanically before acting."]
	}}
	
	Known fault patterns:
	{fault_reference}
	
Alarm log:
{alarms}
"""

def validate_diagnosis(data: dict) -> bool:
    """
    Validates diagnostic output against required schema.

    Args:
        data: Parsed JSON output from agent

    Returns:
        True if valid, False otherwise
    """
    if not isinstance(data, dict):
        return False

    required_top_keys = ["detected_patterns", "possible_faults", "notes"]
    if any(key not in data for key in required_top_keys):
        return False

    if not isinstance(data["detected_patterns"], list) or not all(
        isinstance(item, str) for item in data["detected_patterns"]
    ):
        return False

    if not isinstance(data["notes"], list) or not all(isinstance(item, str) for item in data["notes"]):
        return False

    if not isinstance(data["possible_faults"], list):
        return False

    required_fault_keys = [
        "fault_name",
        "confidence",
        "matched_alarms",
        "reasoning_summary",
        "evidence",
    ]

    for fault in data["possible_faults"]:
        if not isinstance(fault, dict):
            return False
        if any(key not in fault for key in required_fault_keys):
            return False
        if not isinstance(fault["fault_name"], str):
            return False
        if not isinstance(fault["confidence"], (int, float)) or not (0.0 <= float(fault["confidence"]) <= 1.0):
            return False
        if not isinstance(fault["matched_alarms"], list) or not all(
            isinstance(item, str) for item in fault["matched_alarms"]
        ):
            return False
        if not isinstance(fault["reasoning_summary"], str):
            return False
        if not isinstance(fault["evidence"], dict):
            return False
        if "reference_key" not in fault["evidence"] or "signature_matched" not in fault["evidence"]:
            return False
        if not isinstance(fault["evidence"]["reference_key"], str):
            return False
        if not isinstance(fault["evidence"]["signature_matched"], list) or not all(
            isinstance(item, str) for item in fault["evidence"]["signature_matched"]
        ):
            return False

    return True

def run_agent() -> dict:
    alarms = load_alarm_log("alarms.log")
    faults = load_fault_reference("fault_reference.json")

    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "user",
                "content": build_diagnostic_prompt(alarms, faults)
            }
        ],
        response_format={"type": "json_object"}
    )

    diagnosis = json.loads(response.choices[0].message.content)

    if not validate_diagnosis(diagnosis):
        raise ValueError("Invalid diagnosis schema")

    return diagnosis

if __name__ == "__main__":
    print(json.dumps(run_agent(), indent=2))

🧭 Engineering Posture

This capstone reinforced:

✓ Read-only tools over data modification
✓ Structured outputs over unstructured text
✓ Schema validation over blind acceptance
✓ Advisory diagnostics over automated actions
✓ Human authority over machine autonomy

🎓

Capstone Complete!

You've completed the Technician Track and built a professional advisory diagnostic agent.

Next Step

Explore Developer Track →