LLM Concepts for Industrial
Automation and Robotics

import torch
import torch.nn as nn
import torch.nn.functional as F

class TransformerBlock(nn.Module):
    """Single Transformer block for industrial sequence modeling."""
    def __init__(self, d_model=512, n_heads=8, d_ff=2048, dropout=0.1):
        super().__init__()
        self.attn    = nn.MultiheadAttention(d_model, n_heads, dropout=dropout, batch_first=True)
        self.ff      = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.GELU(),
            nn.Linear(d_ff, d_model),
            nn.Dropout(dropout)
        )
        self.norm1   = nn.LayerNorm(d_model)
        self.norm2   = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # Self-Attention + Residual
        attn_out, _ = self.attn(x, x, x, attn_mask=mask)
        x = self.norm1(x + self.dropout(attn_out))
        # Feed-Forward + Residual
        x = self.norm2(x + self.ff(x))
        return x

# Example: model robot joint torque sequences (batch=4, seq_len=32, features=512)
model = TransformerBlock(d_model=512, n_heads=8)
robot_seq = torch.randn(4, 32, 512)      # 4 robots, 32 timesteps, 512 features
output    = model(robot_seq)             # Shape: [4, 32, 512]
print(f"Output shape: {output.shape}")  # torch.Size([4, 32, 512])

from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset

# Load base model
model_name = "meta-llama/Llama-3.1-8B-Instruct"
tokenizer  = AutoTokenizer.from_pretrained(model_name)
model      = AutoModelForCausalLM.from_pretrained(model_name, load_in_4bit=True)

# LoRA configuration for industrial fine-tuning
lora_config = LoraConfig(
    task_type    = TaskType.CAUSAL_LM,
    r            = 16,          # Rank — higher = more capacity
    lora_alpha   = 32,          # Scaling factor
    lora_dropout = 0.05,
    target_modules = ["q_proj", "v_proj"]  # Attention layers only
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# Output: trainable params: 8,388,608 || all params: 8,038,572,032 || trainable%: 0.10%

# Load your industrial dataset
dataset = load_dataset("json", data_files={
    "train": "data/scada_fault_train.jsonl",
    "eval":  "data/scada_fault_eval.jsonl"
})
# Format: {"prompt": "ALARM: Motor_03 Overcurrent", "completion": "Root cause: ..."}

training_args = TrainingArguments(
    output_dir          = "./cobot-llm-lora",
    num_train_epochs    = 3,
    per_device_train_batch_size = 4,
    gradient_accumulation_steps = 4,
    learning_rate       = 2e-4,
    fp16                = True,
    logging_steps       = 50,
    save_steps          = 200,
    evaluation_strategy = "steps",
    eval_steps          = 200,
)

trainer = Trainer(
    model           = model,
    args            = training_args,
    train_dataset   = dataset["train"],
    eval_dataset    = dataset["eval"],
)
trainer.train()
model.save_pretrained("./cobot-llm-finetuned")

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings  import HuggingFaceEmbeddings
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.llms import Ollama

# 1. Load industrial documentation (PLC manuals, maintenance logs, ISO standards)
loader   = DirectoryLoader("./industrial_docs/", glob="**/*.pdf")
docs     = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
chunks   = splitter.split_documents(docs)

# 2. Embed and store in vector database
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vectordb   = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_plc_db")

# 3. Build RAG chain with local LLM (runs on-premises — no cloud data leakage)
llm = Ollama(model="llama3.1:8b", temperature=0.1)  # Low temp for factual answers

qa_chain = RetrievalQA.from_chain_type(
    llm       = llm,
    retriever = vectordb.as_retriever(search_kwargs={"k": 5}),
    chain_type = "stuff",
    return_source_documents = True
)

# 4. Query the system
result = qa_chain.invoke({
    "query": "What is the troubleshooting procedure for Allen-Bradley E-Stop fault code F07?"
})
print(result["result"])
print("\nSources:", [d.metadata["source"] for d in result["source_documents"]])

from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain import hub

# Define industrial tools the LLM agent can call
@tool
def read_plc_tag(tag_name: str) -> str:
    """Read a real-time value from PLC tag via OPC-UA. Args: tag_name (str)"""
    # Connect to Kepware OPC-UA server
    from opcua import Client
    client = Client("opc.tcp://kepware-server:4840")
    client.connect()
    node  = client.get_node(f"ns=2;s=PLC.{tag_name}")
    value = node.get_value()
    client.disconnect()
    return f"{tag_name} = {value}"

@tool
def trigger_alarm(alarm_id: str, message: str) -> str:
    """Trigger a SCADA alarm with given ID and message."""
    # POST to SCADA REST API (e.g., Ignition Gateway)
    import requests
    resp = requests.post("http://ignition-gateway/api/alarm",
                         json={"id": alarm_id, "message": message, "priority": "High"})
    return f"Alarm {alarm_id} triggered: {resp.status_code}"

@tool
def get_robot_joint_state(robot_id: str) -> str:
    """Get current joint positions and velocities of a collaborative robot."""
    import requests
    resp = requests.get(f"http://robot-api/v1/robots/{robot_id}/joints")
    return resp.json()

# Build the LLM agent
llm   = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [read_plc_tag, trigger_alarm, get_robot_joint_state]
agent = create_react_agent(llm, tools, hub.pull("hwchase17/react"))

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

# Run autonomous industrial task
result = agent_executor.invoke({
    "input": "Check if Robot_01 joint 3 torque is safe. If over 80 Nm, trigger alarm FAULT_J3 and log the current state."
})
print(result["output"])