Agent-as-a-Service
Pattern reasoning across unstructured data ..
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.
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:
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
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)
Start the agent
Windows (PowerShell)
macOS / Linux
Leave this terminal open while running the PDI pipeline. The agent process must remain running to serve assessment requests.
Verify the setup
Run the checks that match your platform.
Windows (PowerShell)
macOS / Linux
Success means:
data/asset_history.dbexistsOllama responds on
localhost:11434The agent starts without import errors
FastAPI responds on port
8000
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.
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
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.
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.
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.
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.
Given these four history entries for asset PUMP-017:
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.
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.
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.
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)
Linux / macOS
Response:
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
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.
Sample Maintenance Logs
Asset History
Transformation design
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.
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
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.
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
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.
Step 3: Table Input: Read Log Entries
Add the step
In the Design pane, expand the Input category
Drag Table Input onto the canvas
Double-click the step to open configuration
Configuration
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
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.
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
You will need to define a connection to the assets_history.db located in the ../data folder.
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
Parameters tab
Click the Parameters tab. Add one row to bind the ? placeholder to the stream field:
asset_id
String
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 "[]".
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.
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:
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:
Add output fields in the Fields tab
In the Fields tab at the bottom of the MJV dialog, click Get variables and then add:
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:
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):
history_json
history_json
Last non-null value
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.
Step 6: MJV: Build Agent Payload
Step name: Build Agent Payload
Fields tab - output fields
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:
Step 7: REST Client: Call the Assessment Agent
Step name: Call Assessment Agent
General tab
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)
The actual timeout behaviour is controlled at the JVM level via kettle.properties, not per-step.
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
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).
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
Fields tab - output fields
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
Step 1: Start the agent
Windows (PowerShell)
macOS / Linux
Step 2: RUN
Select Run options — leave defaults
Click Run
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)
CRITICAL
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)
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
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.
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.
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
Last updated
Was this helpful?