Data Platform - Agentic Use-Case

  @DI.Injectable()
  class LLMWithMCP {

      @DI.Inject(TYPES.Dependencies.Logger)
      private _logger: Dependencies.Logger;

      private openai: OpenAI;
      private mcpClients: Map<string, Client> = new Map();
      private availableTools: OpenAI.Chat.Completions.ChatCompletionTool[] = [];
      private isStarted: boolean = false;
  }

  async connectMCPServer(serverConfig: MCPServerConfig): Promise<void> {
      const transport = new StreamableHTTPClientTransport(
          new URL(serverConfig.url),
          { requestInit: { headers: serverConfig.headers } }
      );

      const client = new Client(
          { name: `openai-client-${serverConfig.name}`, version: "1.0.0" },
          { capabilities: {} }
      );

      await client.connect(transport);
      this.mcpClients.set(serverConfig.name, client);

      const { tools } = await client.listTools();

      for (const tool of tools) {
          this.availableTools.push({
              type: "function",
              function: {
                  name: `${serverConfig.name}__${tool.name}`,
                  description: tool.description || "",
                  parameters: tool.inputSchema as Record<string, unknown>,
              },
          });
      }
  }

  private async callMCPTool(toolName: string, args: Record<string, unknown>): Promise<string> {
      const [serverName, ...toolParts] = toolName.split("__");
      const actualToolName = toolParts.join("__");

      const client = this.mcpClients.get(serverName);
      if (!client) {
          throw new Error(`MCP server not found: ${serverName}`);
      }

      const result = await client.callTool({ name: actualToolName, arguments: args });
      // ...
  }

  while (iterations < maxIterations) {
      iterations++;

      const completion = await this.openai.chat.completions.create({
          model,
          messages: currentMessages,
          temperature: options.temperature,
          tools: this.availableTools.length > 0 ? this.availableTools : undefined,
      });

      const message = completion.choices[0].message;
      currentMessages.push(message);

      // No tool calls — the model has a final answer
      if (!message.tool_calls || message.tool_calls.length === 0) {
          return message.content || "";
      }

      // Execute all tool calls in parallel
      const toolResults = await Promise.allSettled(
          message.tool_calls.map(async (toolCall) => {
              const args = JSON.parse(toolCall.function.arguments);
              const result = await this.callMCPTool(toolCall.function.name, args);
              return { role: "tool" as const, tool_call_id: toolCall.id, content: result };
          })
      );

      for (const result of toolResults) {
          if (result.status === "fulfilled") {
              currentMessages.push(result.value);
          }
      }
  }

@Task.TaskModule({
    task: AgenticTask,
    providers: [
        {
            class: LLMWithMCP,
            isSingleton: true,
            symbol: "GlobalLLMHandler",
            scope: Task.ProviderScope.Controller,
        }
    ]
})
export class AgenticModule { }

public override async onChanges(changes: Task.Changes): Promise<void> {
    if (changes["activate"]) {
        this.activate = undefined;

        try {
            await this._globalLLMHandler.init(this.apiKey);

            const chatResult = await this._globalLLMHandler.chat(
                [{ role: "user", content: this.prompt }],
                {
                    model: this.model,
                    temperature: this.temperature,
                    systemPrompt: this.systemPrompt,
                }
            );

            this.result.emit(chatResult);
            this.success.emit(true);
        } catch (e) {
            this.logAndEmitError((e as Error).message);
        }

        if (this.destroySessionOnSuccess) {
            await this._globalLLMHandler.disconnectAll();
        }
    }
}

(...)
    def merge(self, alarms: list[dict], resource_state: dict) -> dict:
        calendar = resource_state.get("Calendar", {})
        shift = calendar.get("Shift", {})
        employee = resource_state.get("Employee", {})
        enterprise = resource_state.get("Enterprise", {})
        site = resource_state.get("Site", {})
        facility = resource_state.get("Facility", {})
        area = resource_state.get("Area", {})
        resource = resource_state.get("Resource", {})

        state_resource_uid = resource.get("UID")

        import json
        
        self.logger.debug(f'Merge: resource- {json.dumps(resource, sort_keys=True)}')
        self.logger.debug(f'Merge: calendar- {json.dumps(calendar, sort_keys=True)}')
        self.logger.debug(f'Merge: shift- {json.dumps(shift, sort_keys=True)}')
        self.logger.debug(f'Merge: state_resource_uid- {json.dumps(state_resource_uid, sort_keys=True)}')

        joined_alarms = []
        for alarm in alarms:
            self.logger.debug(f'UID: alarm - {json.dumps(alarm.get("Resource_UID"), sort_keys=True)}')
            if alarm.get("Resource_UID") != state_resource_uid:
                continue

            joined_alarms.append({
                "alarm_code": alarm.get("Alarm_Code"),
                "alarm_type": alarm.get("Alarm_Type"),
                "alarm_severity": alarm.get("Alarm_Severity"),
                "alarm_cause": alarm.get("Alarm_Cause"),
                "alarm_description": alarm.get("Alarm_Description"),
                "alarm_timestamp": alarm.get("Timestamp"),
                "alarm_action": alarm.get("Action"),
            })

        return {
            "resource": {
                "uid": state_resource_uid,
                "name": resource.get("Name"),
                "enterprise": enterprise.get("Name"),
                "site": site.get("Name"),
                "facility": facility.get("Name"),
                "area": area.get("Name"),
            },
            "resource_state": {
                "system_state": resource.get("SystemState"),
                "previous_system_state": resource.get("PreviousSystemState"),
                "control_state": resource.get("ControlState"),
                "semie10_state": resource.get("SEMIE10State"),
                "modified_by": resource.get("ModifiedBy"),
                "modified_on": resource.get("ModifiedOn"),
            },
            "operator": {
                "name": employee.get("Name"),
                "number": employee.get("Number"),
            },
            "shift": {
                "name": shift.get("Name"),
                "start": shift.get("StartTime"),
                "end": shift.get("EndTime"),
                "date": calendar.get("CD"),
                "timezone": calendar.get("Timezone"),
            },
            "alarms": joined_alarms,
        }

    async def main(self, inputs, outputs):
        import json
        
        self.logger.debug(f'Starting python script: alarms- {json.dumps(inputs['alarms'], sort_keys=True)} resource_state_change- {json.dumps(inputs['resource_state_change'], sort_keys=True)} times')
        dataset = self.merge(inputs['alarms'], inputs['resource_state_change'])
        outputs.join_datasets.emit(dataset)
        self.logger.debug("Finished python script")
        pass

You are a senior machine engineer with deep expertise in industrial systems, diagnostics, maintenance, and failure analysis.

Your role is to analyze machine alarms, logs, and sensor data and produce accurate technical reports that explain what happened, why it happened, and what actions should be taken.

You must prioritize clarity, structure, and visual readability over strict markup formats.

---

# OUTPUT STYLE RULES (MANDATORY)

## 1. Visual Structure Rules

You MUST structure all responses using:

- Clear section headers using ALL CAPS
- Indentation for hierarchy using 2 spaces
- Labels using COLON FORMAT
- Emphasis using ALL CAPS or [BRACKETS]
- Output must be in vanilla HTML5

---

## 2. Formatting Conventions

### Emphasis Rules
- CRITICAL ITEMS → ALL CAPS
- Important labels → Label:
- Optional emphasis → [LIKE THIS]

### Alarms
Always format alarms like:
• ALARM: ALARM_123
• ALARM: ALARM_ABC

Never paraphrase alarm codes. Always breakline between alarms.

---

## 3. Required Report Structure (STRICT ORDER)

Every response MUST follow this structure:

SUMMARY
KEY ALARMS & OBSERVATIONS
LIKELY ROOT CAUSE
IMPACT ON MACHINE
RECOMMENDED ACTIONS
CONFIDENCE LEVEL

---

## 4. Section Rules

### SUMMARY
- 2–4 sentences max
- Plain readable explanation
- No bullet overload

---

### KEY ALARMS & OBSERVATIONS
- Bullet list only
- Must include timestamps if available
- Must preserve alarm codes exactly

Example:
• ALARM: OVERCURRENT_01 triggered at 10:32:14
• EVENT: MOTOR_TORQUE_SPIKE detected 2 seconds later

---

### LIKELY ROOT CAUSE

Must include:

Who was the operator:
• ...

Primary Cause:
• ...

Contributing Factors:
• ...

Assumptions:
• If unknown, explicitly state "INSUFFICIENT DATA"

---

### IMPACT ON MACHINE

Must include:

Affected Systems:
• ...

Severity:
• LOW / MEDIUM / HIGH / CRITICAL

Risks:
• Equipment damage
• Production downtime
• Safety risk (if applicable)

---

### RECOMMENDED ACTIONS

Must be split into:

IMMEDIATE:
• Actions required now to prevent damage

SHORT-TERM:
• Repairs, inspections, replacements

LONG-TERM:
• Preventive improvements and tuning

---

### CONFIDENCE LEVEL

Format:

Confidence: HIGH / MEDIUM / LOW
Reason:
• Short justification based on data completeness and consistency

---

## 5. ENGINEERING PRINCIPLES

- Prefer simplest explanation that fits all alarms
- Do not hallucinate machine types, models, or missing signals
- If data is insufficient, explicitly state uncertainty
- Rank multiple hypotheses by likelihood when needed
- Avoid generic maintenance advice unless it ties to observed alarms

---

## 6. SANITY CHECKS (MANDATORY BEFORE FINAL ANSWER)

Before responding, validate:

• Does the alarm sequence make causal sense?
• Are time intervals physically plausible?
• Are multiple alarms being incorrectly treated as independent?
• Are values being summed incorrectly instead of compared?

If inconsistencies exist:
→ Explicitly mention them in ASSUMPTIONS

---

## 7. VISUAL REPORTING GOAL

Your output should behave like a:
"well-structured industrial maintenance report displayed in a SCADA dashboard"

Priorities:
1. Readability
2. Deterministic structure
3. Fast scanning by engineers
4. No formatting fragility

...

(...)
  resource = new ComplexLogResourceEventInput()
  {
      Resource = resource,
      Reason = "Broken Coil",
      StateModel = stateModel,
      StateModelTransition = stateModel.StateTransitions.Find(sm => sm.Name == $"{resource.CurrentMainState.CurrentState.Name.Replace(" ", "")} to UnscheduledDown"),
      IgnoreLastServiceId = true
  }.ComplexLogResourceEventSync().Resource;

  var selectedScenario = AlarmScenarios[_random.Next(AlarmScenarios.Length)];
  var correlationId = Guid.NewGuid().ToString();

  var events = selectedScenario.Alarms.Select(alarm =>
      new PostEventInput()
      {
          AppProperties = new Cmf.Foundation.BusinessOrchestration.DataPlatform.Domain.AppProperties()
          {
              ApplicationName = "AlarmGenerator",
              EventDefinition = "\\IoTEventDefinitions\\CDM\\Resource\\ResourceAlarm",
              EventTime = DateTime.UtcNow,
          },
          IgnoreLastServiceId = true,
          NumberOfRetries = 3,
          Data = Newtonsoft.Json.Linq.JObject.FromObject(
              BuildAlarmData(resource, area, facility, site, enterprise, (alarm.Code, alarm.Type, alarm.Severity, alarm.Description, alarm.Cause),
              (correlationId, "MachineAlarm", "AlarmGenerator", "11.2.4", "MachineAlarm")))
      }).ToList();

  Console.WriteLine($"Sending scenario '{selectedScenario.Name}' [{events.Count} alarms, correlationId={correlationId}] on {resource.Name}");
  foreach (var e in events)
      Console.WriteLine($"  -> {((string?)e.Data?["Alarm"]?["Code"]) ?? "?"} [{(string?)e.Data?["Alarm"]?["Severity"]}]");

  var result = new PostEventsInput() { IoTEvents = events }.PostMultipleIoTEventsSync();