# Agent-as-a-Service {% hint style="success" %} #### Overview This workshop introduces the architectural boundary between LLM-enriched ETL and a genuine AI agent. Previous workshops in this series called an LLM directly from a PDI REST Client step - one prompt in, one JSON response out. That is a pure function: no external state, no mid-reasoning decisions. This workshop is different. PDI calls a deployed Python agent that uses the LLM's intermediate output to decide what external data to retrieve before producing its final answer. The lookup target is not known until after the first LLM call completes. PDI cannot replicate this in the transformation canvas - the sequence requires a reasoning loop that only the agent can own. {% endhint %} The test for whether something is an agent is simple: can PDI replicate the behaviour by adding more steps to the canvas? Here's summary of the different approaches:

	Direct LLM Call / Multi-Stage	LangExtract	Agent as a Service
What it does	Sends one prompt per row, returns structured JSON for that row	Extracts named entities with char offsets from a single text; returns one row per entity	Reads current entry plus retrieved history, reasons across all texts, returns one assessment
Input to the service	Single text field from the PDI row	Single text field + prompt + few-shot examples	Current log text + N history entries (retrieved by PDI before calling the agent)
Output shape	One row in, one enriched row out	One row in, N entity rows out (one per extraction); pivoted by Row Denormaliser	One row in, one assessment row out
LLM calls per row	1 (or N fixed stages)	1 per extraction pass (typically 2 passes over chunked text)	1 - reads all texts together in a single context
Requires history from other records?	No - processes each row in isolation	No - processes each document in isolation	Yes - history retrieved by PDI via Database Join is essential to the assessment
Decision grounded in	LLM inference on current text only	LLM extraction from current text only; char offsets trace back to source	LLM reasoning across current text AND verified historical records from the database
Can PDI replicate it?	Yes - multi-stage MJV + REST Client pattern	Partially - regex/rules cover known entity formats; LangExtract handles novel/variable ones	No - cross-text pattern reasoning is not encodable as PDI steps
Use when	Classifying, enriching, or summarising individual records	Extracting typed fields from free-form text where regex rules are too brittle	Assessment depends on what previous records say, not just the current one

{% hint style="success" %} A maintenance team logs fault observations as free-text entries against industrial assets. Each entry is a single paragraph written by an engineer in the field — no fixed schema, no controlled vocabulary. PDI reads these entries from a database table and must produce a structured priority assessment for each one. A complete assessment requires three things: * What fault is being described in this entry? (classification) * What has happened to this asset previously? (history retrieval) * Does the current entry represent a new fault, an acceleration of a known pattern, or normal operating variation - given what has happened before? (pattern reasoning) {% endhint %}

{% tabs %} {% tab title="Start Agent" %} **Start the agent** **Windows (PowerShell)** ```powershell cd $env:USERPROFILE\LLM-PDI-Integration\agent-maintenance .\agent-venv\scripts\Activate.ps1 ``` ```powershell # Keep this terminal open uvicorn agent.agent:app --host 0.0.0.0 --port 8000 ``` **macOS / Linux** ```bash cd /opt/agent-maintenance source agent-venv/bin/activate ``` ```bash # Keep this terminal open uvicorn agent.agent:app --host 0.0.0.0 --port 8000 ``` ``` Expected output: INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) ... ``` {% hint style="info" %} Leave this terminal open while running the PDI pipeline. The agent process must remain running to serve assessment requests. {% endhint %} **Verify the setup** Run the checks that match your platform. **Windows (PowerShell)** ```powershell Test-Path .\data\asset_history.db curl.exe http://localhost:11434 curl.exe http://localhost:8000/docs ``` **macOS / Linux** ```bash test -f data/asset_history.db && echo "Database OK" curl http://localhost:11434 curl http://localhost:8000/docs ``` Success means: * `data/asset_history.db` exists * Ollama responds on `localhost:11434` * The agent starts without import errors * FastAPI responds on port `8000` ``` Windows PowerShell Copyright (C) Microsoft Corporation. All rights reserved. Install the latest PowerShell for new features and improvements! https://aka.ms/PSWindows PS C:\Users\jpore> curl http://localhost:8000/health Security Warning: Script Execution Risk Invoke-WebRequest parses the content of the web page. Script code in the web page might be run when the page is parsed. RECOMMENDED ACTION: Use the -UseBasicParsing switch to avoid script code execution. Do you want to continue? [Y] Yes [A] Yes to All [N] No [L] No to All [S] Suspend [?] Help (default is "N"): Y StatusCode : 200 StatusDescription : OK Content : {"status":"ok","model":"llama3.1:8b"} RawContent : HTTP/1.1 200 OK Content-Length: 37 Content-Type: application/json Date: Wed, 08 Apr 2026 13:52:08 GMT Server: uvicorn {"status":"ok","model":"llama3.1:8b"} Forms : {} Headers : {[Content-Length, 37], [Content-Type, application/json], [Date, Wed, 08 Apr 2026 13:52:08 GMT], [Server, uvicorn]} Images : {} InputFields : {} Links : {} ParsedHtml : mshtml.HTMLDocumentClass RawContentLength : 37 PS C:\Users\jpore> ``` {% endtab %} {% tab title="Boundary Argument" %} {% hint style="info" %} #### Boundary Argument PDI handles the complexity of structured data with remarkable flexibility. There are steps that mimic the actions of an AI agent - RegEx, Fuzzy matching, Rule Executor and so on. Together these provide a powerful framework that provides similarity but not semantic equivalence - inference - on unstructured data. {% endhint %}

Task	PDI Step(s)	Capability	Verdict
Extract structured asset ID (e.g. ASSET-0042)	Regex Evaluation	Extracts fixed-format patterns reliably	PDI handles this
Look up asset history rows by asset_id	Database Lookup Database Join	Exact key match, returns all history rows	PDI handles this
Classify fault type from known keywords ("bearing", "temperature")	Rule Executor (Drools) Regex Evaluation + MJV	IF/THEN rules on keyword presence	PDI handles known, enumerated faults
Fuzzy-match current description against a known fault library	Fuzzy Match step	Levenshtein / Jaro-Winkler on string pairs	PDI handles surface similarity; not semantic equivalence
Assign final priority once fault type is known	Filter Rows Switch/Case MJV	Pure deterministic routing on field values	PDI handles this completely

{% hint style="info" %} #### Why Regex fails "bearing noise getting worse" and "intermittent vibration on startup since maintenance" both indicate bearing degradation. They share no common n-gram, no common keyword sequence, and no pattern a regex can match. The semantic equivalence exists at the meaning level, not the surface level. Regex operates only on surface form. {% endhint %} {% hint style="info" %} #### Why Rule Executor (Drools) fails A rule can fire on the keyword "bearing" in the current entry. But the rule cannot say: "this entry, combined with two previous entries that mention vibration and one bearing replacement 18 months ago, indicates recurrence of the same root cause rather than a new fault." That conclusion requires reading and integrating four separate natural-language texts. A Drools rule operates on field values in the current row - it does not read and synthesise multiple text fields from a joined history set. You could try to flatten the history into a single concatenated field and write a rule against that. But the rule would still operate on keyword presence, not meaning. "No unusual noise but slight vibration" contains the word "vibration" - a keyword rule would flag it identically to "severe vibration on every startup". The negation and qualifier ("slight", "no unusual") change the meaning completely. Rules cannot parse that. {% endhint %} {% hint style="info" %} #### Why Fuzzy Match fails The Fuzzy Match step finds strings with high character-level similarity. "Bearing noise" and "audible roughness in drive shaft" have low character similarity but identical diagnostic meaning. Conversely, "bearing temperature normal" and "bearing temperature elevated" have very high character similarity but opposite meanings. Fuzzy matching on maintenance log text produces both false positives and false negatives at an unacceptable rate for safety-relevant prioritisation. {% endhint %} {% hint style="info" %} #### Why MJV (JavaScript) fails MJV can implement any algorithm expressible in JavaScript. A skilled developer could write a JavaScript function that: (1) concatenates history entries, (2) counts keyword occurrences, (3) applies weighted scoring. This is a rules system with extra steps. It fails on the same cases rules fail on - novel descriptions, combined symptoms, qualified negations, and cross-entry pattern inference. It does not understand language; it manipulates strings. {% endhint %} *** Given these four history entries for asset PUMP-017: ``` 2025-09-12: slight rumble on startup, clears after 2 minutes, logged for monitoring 2025-11-03: intermittent vibration under load, no temperature anomaly, bearing checked ok 2026-01-18: bearing replaced (scheduled maintenance) 2026-03-29: rougher than usual on startup, louder than before the January maintenance ``` {% hint style="info" %} The correct assessment is: RECURRENCE - bearing degradation has restarted 10 weeks post-replacement, which is abnormally fast and indicates either incorrect installation or an underlying cause not addressed in the January maintenance. Priority: HIGH. No PDI step produces this assessment. The Database Lookup returns the four history rows as data. MJV can concatenate them. No step can read them together and conclude that the current entry matches the pre-January pattern and that pattern recurrence within 10 weeks of a replacement is abnormal. {% endhint %} {% hint style="success" %} This cannot be replicated by any combination of PDI transformation steps. Read multiple unstructured natural-language history entries alongside a new unstructured entry and reason about whether they collectively describe a known degradation pattern, a recurrence, or a new fault mode. This is the task the agent owns. PDI retrieves the history as structured rows. The agent reads the history text and the current entry together and reasons about what they collectively indicate. PDI cannot do this. The agent can. That is the boundary. Everything else in the pipeline is PDI. {% endhint %} {% endtab %} {% tab title="Pipeline Test" %} {% hint style="info" %} #### Pipeline Overview The scenario uses one LLM call - and that is intentional. Here, PDI already knows the asset\_id and retrieves the history as structured rows before calling the agent. The agent receives everything it needs in one payload. One call is cleaner, faster, and easier to reason about. The single call is also the hardest part - reading multiple texts and synthesising a cross-entry assessment is more genuinely difficult than two sequential single-task calls. {% endhint %} ``` PDI Transformation [Table Input: maintenance_log] | [Regex Evaluation] ← PDI extracts asset_id from log_text | [Database Join: asset_history] ← PDI retrieves history rows | [MJV: Build agent payload] ← PDI assembles request | [REST Client POST /assess] ───────────────────────────────┐ | │ [MJV: Parse response] Agent (FastAPI / venv) │ | POST /assess │ [Switch/Case: priority] ├─ Read log_text + history │ | ├─ [LLM] Reason across texts │ CRITICAL → alerts_table └─ Return assessment JSON ◄──┘ HIGH → priority_queue MEDIUM → standard_queue LOW → log_archive ```

	PDI Transformation	Maintenance Assessment Agent
Owns	All data flow, retrieval, routing, and persistence	Language understanding and cross-text reasoning
Reads	Structured rows from maintenance_log and asset_history tables	Unstructured text: current log entry + N history entries as text
Produces	Structured rows with priority, fault_type, assessment fields	A single structured JSON assessment per request
Uses	Regex, Database Join, MJV, REST Client, Switch/Case	One LLM call with history context; no tool calls needed
Can be replaced by rules?	Yes - all PDI logic is deterministic	No - language reasoning is not encodable as rules

*** Let's run some tests: **Windows (Powershell)** ```powershell @" { "log_id": "TEST-001", "asset_id": "PUMP-017", "log_text": "Rougher than usual on startup, louder than before January maintenance", "history": [ {"logged_at": "2025-09-12", "log_text": "slight rumble on startup"}, {"logged_at": "2025-11-03", "log_text": "intermittent vibration under load"}, {"logged_at": "2026-01-18", "log_text": "bearing replaced, WO-4412"} ] } "@ | Out-File -Encoding utf8 test.json curl.exe -s -X POST http://localhost:8000/assess ` -H "Content-Type: application/json" ` -d "@test.json" ``` **Linux / macOS** ```json curl -s -X POST http://localhost:8000/assess \ -H "Content-Type: application/json" \ -d '{ "log_id": "TEST-001", "asset_id": "PUMP-017", "log_text": "Rougher than usual on startup, louder than before January maintenance", "history": [ {"logged_at": "2025-09-12", "log_text": "slight rumble on startup"}, {"logged_at": "2025-11-03", "log_text": "intermittent vibration under load"}, {"logged_at": "2026-01-18", "log_text": "bearing replaced, WO-4412"} ] }' ``` Response: ```json { "log_id": "TEST-001", "asset_id": "PUMP-017", "priority": "HIGH", "fault_type": "intermittent_vibration_under_load", "pattern": "RECURRENCE", "assessment": "The pump's performance has worsened since the last maintenance, indicating a potential issue with the replacement bearing.", "confidence": 80 } ``` {% hint style="info" %} The agent is working correctly: * **priority: HIGH** - the model recognises this needs same-day attention, which is correct given the history * **fault\_type: intermittent\_vibration\_under\_load** - picked up the dominant symptom from the history * **pattern: RECURRENCE** - correctly identified that this has happened before (pre-January bearing issues) * **assessment** - the model has read the history and connected the dots: bearing was replaced in January and symptoms are returning * **confidence: 80** - reasonably high, consistent with a clear pattern in the history {% endhint %} {% endtab %} {% tab title="Transformation" %} {% hint style="info" %} #### Maintenance Agent The maintenance\_assessment.ktr has all the steps in a single pipeline to illustrate the workflow. The MJV approach works for this workshop because the dataset is small and runs single-copy. In production with thousands of rows and STEP\_COPIES > 1, the accumulator pattern would break and you would need the sub-transformation approach instead. So, a cleaner alternative that avoids the MJV accumulator entirely is to use a **sub-transformation via Transformation Executor**. For each log entry, a child transformation queries the history database and builds the JSON array, returning one row with the complete payload. This is stateless, parallelism-safe, and easier to reason about. {% endhint %} **Sample Maintenance Logs** ``` log_id | asset_id | logged_at | engineer | log_text -------|-----------|---------------------|----------|------------------------------------------ L-1001 | PUMP-017 | 2026-04-07 08:14:00 | J.Walsh | Rougher than usual on startup, louder | | | | than before the January maintenance. | | | | Vibration settling after ~3 minutes. L-1002 | COMP-004 | 2026-04-07 09:02:00 | S.Okafor | High temperature alarm triggered at | | | | 09:00. Reading: 94C. Limit is 85C. | | | | No prior warnings this shift. L-1003 | FAN-011 | 2026-04-07 10:31:00 | T.Marsh | Fan running normally. Slight hum noted | | | | but within normal range. No action. L-1004 | PUMP-017 | 2026-04-07 11:45:00 | J.Walsh | Vibration increased since this morning. | | | | Getting worse through the shift. L-1005 | VALVE-022 | 2026-04-07 13:10:00 | R.Nkosi | Valve sticking on close. Takes 3-4 | | | | attempts. Never seen this before. ``` **Asset History** ``` PUMP-017 history: 2025-09-12: slight rumble on startup, clears after 2 minutes, logged for monitoring 2025-11-03: intermittent vibration under load, no temperature anomaly, bearing checked ok 2026-01-18: bearing replaced (scheduled maintenance, work order WO-4412) COMP-004 history: 2025-10-05: temperature running slightly high (82C) on hot days, within tolerance 2025-12-14: cooling fan filter cleaned, temperature returned to normal range 2026-02-28: temperature normal, no issues logged FAN-011 history: (no entries in last 12 months) VALVE-022 history: 2025-08-20: valve serviced, seals replaced 2026-01-09: operating normally, no issues ``` *** **Transformation design**

Step Name	Type	What PDI does here
Read Log Entries	Table Input	SELECT unprocessed rows from maintenance_log
Retrieve Asset History	Database Join	LEFT JOIN asset_history ON asset_id, concat history rows into JSON array field
Sort rows	Sort rows	Keeps the rows in chronological order
Remove duplicates	Unique rows	Just in case ..
Aggregate History	Modified JavaScript Value	PDI stream is row based, need to aggregate all the records for asset.
	Group by
Build Agent Payload	Modified JavaScript Value	JSON.stringify the full request including history array
Call Assessment Agent	REST Client	POST ${AGENT_URL}/assess — socket timeout 180000ms
Check HTTP Status	Filter Rows	Route non-200 to error log
Parse Agent Response	Modified JavaScript Value	Extract priority, fault_type, pattern, assessment, confidence
Route by Priority	Switch / Case	Fan out on priority field
Write CRITICAL	Table Output	assessed_log with alert flag set
Write HIGH	Table Output	assessed_log
Write MEDIUM	Table Output	assessed_log
Write LOW	Table Output	assessed_log
Write Errors	Text File Output	HTTP failures for investigation

This section walks through building maintenance\_assessment.ktr step by step in Spoon. Every dialog tab, field name, and script is given in full. Build the steps in order — each one depends on output fields from the previous. **Step 1: Set transformation properties** Double-click anywhere on the empty canvas to open Transformation Properties. Click the Parameters tab.

Parameter	Default Value	Description
AGENT_URL	http://localhost:8000	Base URL of the assessment agent
DB_CONNECTION	path to / maintenance_log	Set path to maintenance_log.db
STEP_COPIES	2	Parallel REST Client copies
HISTORY_LIMIT	10	Max history rows per asset

{% hint style="info" %} Add each parameter using the + button. Enter the parameter name in the first column and the default value in the third column (Default value). Click OK to save. Parameters are accessible as ${PARAM\_NAME} in step configuration fields and via getVariable("PARAM\_NAME","default") in MJV scripts. {% endhint %} **Step 2: Create a database connection** The Table Input and Database Join steps both need a named connection to the SQLite database. Create it once and both steps will share it. 1\. View > Connections > New 2\. Connection name: maintenance\_log\_db 3\. Connection type: SQLite 4\. Database name (file path): ../data/maintenance\_log.db 5\. Click Test — should return "Connection to database \[MAINTENANCE\_DB] is OK" 6\. Click OK {% hint style="info" %} SQLite stores the entire database in a single file. The Database Join step will query asset\_history from the same file if you ATTACH it, or you can create a second connection named HISTORY\_DB pointing to data/asset\_history.db. For this workshop, use separate connection objects for clarity. {% endhint %} **Step 3: Table Input: Read Log Entries** **Add the step** 1. In the Design pane, expand the Input category 2. Drag Table Input onto the canvas 3. Double-click the step to open configuration **Configuration**

Field	Value
Step name	Read Log Entries
Connection	maintenanace_log_db
SQL	See query below
Enable lazy conversion	No (unchecked)
Replace variables	Yes (checked) - required for ${HISTORY_LIMIT}

**SQL query** ```sql SELECT log_id, asset_id, logged_at, engineer, log_text FROM maintenance_log WHERE processed = 0 ORDER BY logged_at ASC ``` {% hint style="info" %} The processed = 0 filter ensures each log entry is only assessed once. After a successful run, update processed = 1 in a downstream Table Output step or via a post-processing Execute SQL Script step in a wrapping Job. {% endhint %} **Step 4: Database Join: Retrieve Asset History** The Database Join step executes a parameterised SQL query once per input row, using the asset\_id field from the stream as the ? parameter. It appends all matching history rows to the stream - one output row per history entry. The subsequent MJV step then aggregates those rows back into a single JSON array field. **Configuration** {% hint style="warning" %} You will need to define a connection to the assets\_history.db located in the ../data folder. {% endhint %}

Field	Value
Step name	Get Asset History
Connection	asset_history_db
SQL	See query below
Outer join	Yes (checked) - ensures assets with no history still produce a row
Number of rows to return	${HISTORY_LIMIT} (resolves to 10)
Use variable substitution	Yes (checked)

**SQL query** ```sql SELECT logged_at AS hist_logged_at, log_text AS hist_log_text FROM asset_history WHERE asset_id = ? ORDER BY logged_at ASC ``` **Parameters tab** Click the Parameters tab. Add one row to bind the ? placeholder to the stream field:

Field (from stream)	Type
asset_id	String

{% hint style="warning" %} Outer join = Yes is critical. Without it, assets with no history rows (like FAN-011) would be silently dropped from the stream. With Outer join enabled, the step outputs one row with NULL values for hist\_logged\_at and hist\_log\_text when no history exists. The next MJV step handles the NULL case by defaulting history\_json to "\[]". {% endhint %} {% hint style="info" %} The Database Join step produces one output row per history entry matched. For PUMP-017 (3 history rows), it produces 3 output rows — all carrying the original log entry fields (log\_id, asset\_id, log\_text, etc.). Step (Aggregate History) collapses those back to one row per log entry. {% endhint %} **Step 5: Aggregate History into JSON Array** The Database Join expanded each log entry into N rows (one per history entry). This MJV step runs after a Group By step to collapse them back into one row per log entry, with the history encoded as a JSON array string in a new field history\_json. **Sort rows** Sort on two fields:

Field name	Ascending	Case sensitive
log_id	Yes	No
hist_logged_at	Yes	No

**Modified JavaScript Value** Step name: Aggregate History Copy & Paste the following script into the script editor: ```javascript // ========================================================================== // AGGREGATE HISTORY ROWS INTO A JSON ARRAY // ========================================================================== // The Database Join produced N rows per log entry (one per history record). // This script accumulates those rows and — on the last row for each log_id — // emits a single row with the full history encoded as a JSON array string. // // PDI runs all step instances in parallel threads. Because Sort Rows has // sorted by log_id ASC, all rows for the same log_id are contiguous. // We use a static accumulator array to collect them. // // Static variables (declared with var in the first script block) persist // across rows within a single step execution. // ========================================================================== // Initialise accumulators on first row if (typeof history_buffer === "undefined") { var history_buffer = []; var current_log_id = null; } // If this is a new log_id, flush the previous buffer first if (current_log_id !== null && current_log_id !== log_id + "") { // This should not happen after Sort Rows — included as a safety guard history_buffer = []; } current_log_id = log_id + ""; // Accumulate this history row (skip NULL entries from Outer Join) if (hist_logged_at !== null && hist_log_text !== null) { history_buffer.push({ "logged_at": hist_logged_at + "", "log_text": hist_log_text + "" }); } // Use a Group By step downstream to get the final row per log_id. // Set history_json as the output field from this script. var history_json = JSON.stringify(history_buffer); ``` **Add output fields in the Fields tab** In the Fields tab at the bottom of the MJV dialog, click Get variables and then add:

Fieldname	Type	Notes
history_json	String	JSON array of history objects, or "[]" if no history

**Group By step to keep only the last row per log\_id** Group field:

Group by field	Notes
log_id	Primary key — one output row per log entry
asset_id	Carry-through field
logged_at	Carry-through field
engineer	Carry-through field
log_text	Carry-through field

Aggregation (to retain the last history\_json value per group):

Name	Subject	Type
history_json	history_json	Last non-null value

{% hint style="info" %} The LAST aggregation type returns the last non-NULL value for history\_json within the group. Because the MJV accumulates entries into the buffer as rows arrive, and Sort Rows guarantees chronological order within each log\_id, the LAST row for each log\_id carries the complete history array. {% endhint %} **Step 6: MJV: Build Agent Payload** Step name: Build Agent Payload ```javascript // ========================================================================== // BUILD THE AGENT REQUEST PAYLOAD // ========================================================================== // This step takes the accumulated history_json array string produced by // the Aggregate History step and constructs the complete JSON payload // for a POST request to the agent's /assess endpoint. // // Input fields expected from the stream: // log_id — unique identifier for this log entry // asset_id — asset identifier extracted by Regex Evaluation // log_text — the raw engineer observation text // history_json — JSON array string from Aggregate History + Group By // // Output field: // agent_payload — serialised JSON string ready for the REST Client body // ========================================================================== // 1. Sanitise log_text to prevent JSON serialisation errors. // Engineers enter free text — quotes and newlines are common. var safe_text = (log_text + "") .replace(/\\/g, "\\\\") // escape backslashes first .replace(/"/g, '\\"') // escape double quotes .replace(/\n/g, " ") // replace newlines with space .replace(/\r/g, ""); // remove carriage returns // 2. Parse the history JSON array. // history_json arrives as a string: '[{"logged_at":"...","log_text":"..."}]' // It may be "[]" for assets with no history (FAN-011). // It may be null if the Group By produced no value — handle defensively. var history_arr = []; try { var raw = history_json + ""; if (raw && raw !== "null" && raw.length > 2) { history_arr = JSON.parse(raw); } } catch(e) { history_arr = []; // default to empty — agent handles gracefully } // 3. Build the complete request object and serialise it. var agent_payload = JSON.stringify({ "log_id": log_id + "", "asset_id": asset_id + "", "log_text": safe_text, "history": history_arr }); ``` **Fields tab - output fields**

Fieldname	Type	Notes
agent_payload	String	Complete JSON payload for POST /assess

**Verify the payload** Before connecting the REST Client, preview the output of this step and verify the agent\_payload field contains valid JSON with the correct structure: ``` // Expected agent_payload for PUMP-017: { "log_id": "L-1001", "asset_id": "PUMP-017", "log_text": "Rougher than usual on startup, louder than before the January maintenance.", "history": [ {"logged_at": "2025-09-12", "log_text": "slight rumble on startup"}, {"logged_at": "2025-11-03", "log_text": "intermittent vibration under load"}, {"logged_at": "2026-01-18", "log_text": "bearing replaced, WO-4412"} ] } // Expected agent_payload for FAN-011 (no history): { "log_id": "L-1003", "asset_id": "FAN-011", "log_text": "Fan running normally. Slight hum noted but within normal range.", "history": [] } ``` **Step 7: REST Client: Call the Assessment Agent** Step name: Call Assessment Agent **General tab**

Field	Value
HTTP method	POST
URL	${AGENT_URL}/assess
Body field	agent_payload
Application type	TEXT PLAIN
Result field name	agent_response
HTTP status code field	response_code
Response time field	response_time_ms

**Headers:** (leave empty - REST Client auto-adds Content-Type) {% hint style="info" %} The actual timeout behaviour is controlled at the JVM level via `kettle.properties`, not per-step. {% endhint %} **Step 8: Filter Rows: Check HTTP Status** Step name: Check HTTP Status **Condition** Click inside the condition area and build the following filter: `response_code = 200` {% hint style="info" %} The Write HTTP Errors step should capture: log\_id, asset\_id, response\_code, agent\_payload (so you can see what was sent), and response\_time\_ms. Common non-200 codes: 422 = malformed payload 502 = Ollama not running 500 = agent parse failure (LLM returned malformed JSON after retries). {% endhint %} **Output hops** Right-click the Filter Rows step. You need two output hops: TRUE path → draw a hop to the next step (Parse Agent Response) FALSE path → draw a hop to a Text File Output step named Write HTTP Errors **Step 8: MJV: Parse Agent Response** Step name: Parse Agent Response ```javascript // ========================================================================== // PARSE THE AGENT RESPONSE // ========================================================================== // The agent returns a clean JSON object — no Ollama wrapper envelope. // The agent_response field contains the raw HTTP response body as a string. // // Expected response schema: // { // "log_id": "L-1001", // "asset_id": "PUMP-017", // "priority": "HIGH", // "fault_type": "bearing_degradation", // "pattern": "RECURRENCE", // "assessment": "Bearing degradation ...", // "confidence": 85 // } // // Error handling: if parsing fails, priority defaults to MEDIUM so the // row is routed to the standard_queue rather than being lost. // parse_error = "Y" flags it for manual review. // ========================================================================== var priority = "MEDIUM"; var fault_type = "unknown"; var pattern = "NEW_FAULT"; var assessment = ""; var confidence = 50; var parse_error = "N"; var parse_error_msg = ""; try { var d = JSON.parse(agent_response + ""); priority = d.priority ? (d.priority + "").toUpperCase() : "MEDIUM"; fault_type = d.fault_type ? (d.fault_type + "") : "unknown"; pattern = d.pattern ? (d.pattern + "").toUpperCase() : "NEW_FAULT"; assessment = d.assessment ? (d.assessment + "") : ""; confidence = d.confidence ? parseInt(d.confidence, 10) : 50; // Clamp confidence to valid range if (confidence < 0) confidence = 0; if (confidence > 100) confidence = 100; } catch(e) { parse_error = "Y"; parse_error_msg = e.message ? e.message + "" : "JSON parse error"; // All output fields retain their safe defaults above. // The row continues downstream with priority = MEDIUM. } ``` **Fields tab - output fields**

Fieldname	Type	Notes
priority	String	CRITICAL \| HIGH \| MEDIUM \| LOW
fault_type	String	Short snake_case label from agent
pattern	String	RECURRENCE \| ESCALATION \| NEW_FAULT \| NORMAL_VARIATION
assessment	String	One-to-two sentence explanation
confidence	Integer	0–100
parse_error	String	Y or N
parse_error_msg	String	Exception message if parse_error = Y

**Step 9: Switch / Case: Route by Priority** Add four Table Output steps (Output category), one for each priority path. They all write to the same assessed\_log table but can be separate tables if your schema requires it. Connect each from the corresponding Switch / Case output hop. x x {% endtab %} {% tab title="RUN" %} {% hint style="info" %} #### RUN Transformation {% endhint %} **Step 1: Start the agent** **Windows (PowerShell)** ```powershell cd $env:USERPROFILE\LLM-PDI-Integration\agent-maintenance .\agent-venv\scripts\Activate.ps1 ``` ```powershell # Keep this terminal open uvicorn agent.agent:app --host 0.0.0.0 --port 8000 ``` **macOS / Linux** ```bash cd /opt/agent-maintenance source agent-venv/bin/activate ``` ```bash # Keep this terminal open uvicorn agent.agent:app --host 0.0.0.0 --port 8000 ``` **Step 2: RUN** 1. Select Run options — leave defaults 2. Click Run

{% hint style="info" %} The Preview panel opens with the Logging tab active. Watch for: * Read Log Entries: should show 5 rows read (one per sample log entry) * Get Asset History: row count will be higher than 5 — this is expected (one row per history match) * Group By: should collapse back to 5 rows * Call Assessment Agent: 5 rows sent; watch response\_time\_ms * Route by Priority: rows distributed across CRITICAL (1), HIGH (1), MEDIUM (2), LOW (1) {% endhint %} CRITICAL

Field	Value
log_id	L-1002
asset_id	COMP-004
logged_at	2026-04-07 09:02:00
engineer	S.Okafor
log_text	High temperature alarm triggered at 09:00. Reading: 94C. Limit is 85C. No prior warnings this shift.
history_json	`[{"logged_at":"2025-10-05","log_text":"temperature running slightly high 82C, within tolerance"},{"logged_at":"2025-12-14","log_text":"cooling fan filter cleaned, temperature normal"},{"logged_at":"2026-02-28","log_text":"temperature normal, no issues"}]`
agent_payload	`{"log_id":"L-1002","asset_id":"COMP-004","log_text":"High temperature alarm triggered...","history":[...3 entries...]}`
agent_response	`{"log_id":"L-1002","asset_id":"COMP-004","priority":"CRITICAL","fault_type":"high_temperature_alarm","pattern":"RECURRENCE","assessment":"High temperature alarm triggered at 09:00 with a reading of 94C, exceeding the limit of 85C.","confidence":100}`
response_code	200
response_time_ms	5094
priority	CRITICAL ✅
fault_type	`high_temperature_alarm` ✅
pattern	`RECURRENCE` ✅
assessment	High temperature alarm triggered at 09:00 with a reading of 94C, exceeding the limit of 85C.
confidence	100
parse_error	N
parse_error_msg	(blank)

log_id	priority	fault_type	pattern
L-1001	HIGH	bearing_degradation / vibration_recurrence	RECURRENCE
L-1002	CRITICAL	high_temperature_alarm	RECURRENCE
L-1003	LOW	normal_variation	NORMAL_VARIATION
L-1004	CRITICAL	increased_vibration / vibration_escalation	ESCALATION
L-1005	MEDIUM	valve_sticking	NEW_FAULT

{% hint style="info" %} KEY POINTS 1\. The boundary is precise and provable. PDI steps handle extraction, lookup, rules, and routing. No PDI step reads multiple unstructured text entries together and reasons about whether they collectively indicate a known pattern. That is the exact and only task the agent owns. 2\. PDI does more work than the agent in this pipeline. Regex extracts the asset ID. Database Join retrieves the history. MJV builds the payload. Switch/Case routes the result. The agent does one thing: language reasoning across text. 3\. One LLM call, grounded in retrieved data, is the clean pattern. The history context is retrieved by PDI as structured rows, formatted by MJV, and passed to the agent in a single payload. The agent does not need tools because PDI already did the lookup. 4\. Confidence scores surface model uncertainty for human review. Entries with low history or ambiguous text should route to a review queue regardless of the priority label. 5\. The test for any agent pattern is: can PDI replicate it? If yes with steps — use steps. Faster, simpler, deterministic. If no — that is where the agent earns its place. {% endhint %} {% endtab %} {% tab title="Performance" %} {% hint style="info" %} #### Performance {% endhint %} History context increases prompt length compared to earlier workshops. A typical prompt for this workshop is 300–600 tokens (system instructions + up to 10 history entries + current entry). At llama3.1:8b on the 3080 12GB, expect 4–8 seconds per assessment.

Configuration	Step Copies	5 entries	Notes
llama3.1:8b, RTX 3080 12GB, no history	1	~15-25 sec	~3-5 sec/call, short prompt
llama3.1:8b, RTX 3080 12GB, 5 history entries	1	~25-40 sec	~5-8 sec/call, longer prompt
llama3.1:8b, RTX 3080 12GB, 10 history entries	1	~30-50 sec	~6-10 sec/call
llama3.1:8b, RTX 3080 12GB, 10 history entries	2	~18-30 sec	Recommended starting point
llama3.1:8b, RTX 3080 12GB, 10 history entries	4	~14-24 sec	Try this; 7GB KV headroom supports it

{% endtab %} {% endtabs %}