A pile of messages
A text to records pipeline turns messy messages into checked records a program can store. It matters because each step you have met so far is small, and a real job needs all of them together.
A school office receives a pile of free-text supply requests. Somebody wants them as rows: the item, how many and whether it is urgent. No one wants to type three hundred of them.
By the end of this lesson
Note
I can build a small pipeline from messy text to checked records.
Key words
| Word | What it means |
|---|---|
| Pipeline | steps that each hand their result to the next |
| Batch | a group of texts processed together |
| Reject | to keep a text out, with a reason |
| Reject rate | the share of texts that were rejected |
| Save | to write the checked records somewhere |
| Personal data | facts that identify a person |
The cells students fill in are left blank in this preview.
Four steps, one after another
Each message goes through the same steps. Extract: ask the model for the fields. Validate: check the record against the schema. Repair: if it fails, ask once more with the fault. Save: write the good record.
A message that still fails is rejected. It is not saved, and it is not lost: you keep the text and the reason.
Count the rejects and say why
A reject is information. Keep the text and the reason, and work out the reject rate. If it is high, the prompt or the schema needs work.
Be careful with personal data. Send the model only the text it needs, and write the reason to the log, not the text of a message about a person.
Write the model helper
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'const base = process.env.AI_URL || \"http://ai:4000\";' 'async function ask(messages, sample, max) {' ' try {' ' const res = await fetch(base + \"/v1/chat/completions\", {' ' method: \"POST\",' ' headers: { \"content-type\": \"application/json\", authorization: \"Bearer lab\" },' ' body: JSON.stringify({ model: \"lab\", messages: messages, max_tokens: max || 200 }),' ' signal: AbortSignal.timeout(30000),' ' });' ' if (res.ok === false) throw new Error(\"status \" + res.status);' ' return (await res.json()).choices[0].message.content;' ' } catch (error) {' ' console.log(\"Recorded sample: no model answered (\" + error.message + \").\");' ' const list = [].concat(sample);' ' const turn = messages.filter(function (m) { return m.role === \"assistant\"; }).length;' ' return list[Math.min(turn, list.length - 1)];' ' }' '}' 'module.exports = { ask };' > model.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 1
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'const fs = require(\"fs\");' 'const { ask } = require(\"./model\");' '' 'const texts = [' ' \"Please send 12 pencils to the art room, it is not urgent.\",' ' \"We need 3 boxes of chalk today, as soon as possible.\",' ' \"Can someone tell me where the staff room is?\",' '];' > pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 2
docker exec -w /home/node/project agent sh -c "printf '%s\n' '// Recorded samples, used only when no model answers: a clean reply, a reply that needs one repair,' '// and a message that holds no order, which is rejected.' 'const samples = [' ' JSON.stringify({ item: \"pencils\", quantity: 12, urgent: false }),' ' [\"Here you go: \" + JSON.stringify({ item: \"chalk\", quantity: \"three\", urgent: true }), JSON.stringify({ item: \"chalk\", quantity: 3, urgent: true })],' ' [JSON.stringify({ item: null, quantity: null, urgent: false }), \"I cannot find an order in this message.\"],' '];' >> pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 3
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'function pullJson(text) {' ' const start = text.indexOf(\"{\");' ' const end = text.lastIndexOf(\"}\");' ' if (start === -1 || end < start) return null;' ' try { return JSON.parse(text.slice(start, end + 1)); } catch (error) { return null; }' '}' >> pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 4
docker exec -w /home/node/project agent sh -c "printf '%s\n' '// The first problem, or an empty text when the record is good.' 'function problemOf(record) {' ' if (record === null) return \"not JSON\";' ' if ((typeof record.item === \"string\") === false || record.item === \"\") return \"bad item\";' ' if (Number.isInteger(record.quantity) === false || record.quantity < 1 || record.quantity > 20) return \"bad quantity\";' ' if ((typeof record.urgent === \"boolean\") === false) return \"bad urgent\";' ' return \"\";' '}' >> pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 5
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'async function handle(text, sample) {' ' const prompt = \"Extract item, quantity (1 to 20) and urgent (true or false) from this message. Reply with only the JSON.\n\nMessage: \" + text;' ' const messages = [{ role: \"user\", content: prompt }];' ' let reply = await ask(messages, sample, 80);' ' let problem = problemOf(pullJson(reply));' ' if (problem.length > 0) {' ' // One repair: send the bad reply and the problem back, then ask once more.' ' messages.push({ role: \"assistant\", content: reply });' ' messages.push({ role: \"user\", content: \"Fix this problem and reply with only the JSON: \" + problem });' ' reply = await ask(messages, sample, 80);' ' problem = problemOf(pullJson(reply));' ' }' ' return { record: pullJson(reply), problem: problem };' '}' >> pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Write the pipeline, part 6
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'async function main() {' ' const saved = [];' ' const rejected = [];' ' for (let at = 0; at < texts.length; at = at + 1) {' ' const result = await handle(texts[at], samples[at]);' ' if (result.problem.length > 0) rejected.push({ text: texts[at], why: result.problem });' ' else saved.push({ item: result.record.item, quantity: result.record.quantity, urgent: result.record.urgent });' ' }' ' fs.writeFileSync(\"records.json\", JSON.stringify(saved, null, 2));' ' console.log(\"Saved \" + saved.length + \" records to records.json\");' ' for (const one of rejected) console.log(\"Rejected: \" + one.text + \" (\" + one.why + \")\");' ' console.log(\"Reject rate: \" + Math.floor(rejected.length * 100 / texts.length) + \" percent\");' '}' 'main();' >> pipeline6.js"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Run it on three messages
docker exec -w /home/node/project agent node pipeline6.jsShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
One possible output
| Line | What it may show |
|---|---|
| Saved | Saved 2 records to records.json, for example |
| Rejected | The message about the staff room, with a reason such as bad item |
| Reject rate | Reject rate: 33 percent if one of the three is rejected |
The cells students fill in are left blank in this preview.
A common mistake
Warning
Many people think a rejected message means the pipeline failed. Rejecting is right when a message holds no order. What fails is rejecting without counting or saying why.