Let the big model think, let the small model work: Splitting LLM costs in Elastic Workflows
Build an Elastic workflow that sends a data sample to a large model to propose classification labels. A human signs off, then a smaller model applies them across the full corpus.
Split the expensive part of large language model (LLM) classification from the cheap part. This article builds an Elastic workflow where Claude Sonnet reads a stratified sample of NASA pilot incident reports and proposes classification labels based on what it finds. A human reviews the schema and signs off, and then Mistral Small 3.1 applies the labels across the full corpus. The routing is YAML, the results land in Elasticsearch as structured data, and the pattern works wherever you have free text that needs labeling.
NASA Aviation Safety Reporting System (ASRS) reports describe unusual events during flights, such as missed altitudes, confusing clearances, runway issues, or mechanical problems. Each report already has an official category, like altitude deviation, course deviation, or ground encounter. In this article, we ask a different question: What does this report reveal about the pilot who wrote it? The idea is to ask a model to infer a schema grounded on the data to classify the report based on criteria that help us figure out information about the report writers. Then ask a second model to apply the labels.
You can find the full workflow definitions and helper scripts here.
What you need to run this LLM pipeline
Elastic Stack 9.4+ or Elastic Cloud Serverless. Elastic Workflows has been generally available (GA) since 9.4.
Elastic Agent Builder enabled in your deployment.
A Kibana generative AI (GenAI) connector pointing at Claude Sonnet (or an equivalent reasoning model). This is the planner.
A Mistral API key. We’ll use it to register an Elasticsearch inference endpoint.
Python 3.10+ with
elasticsearch>=9.0andpandas. Used by the dataset loader.
How two-tier LLM orchestration works
The workflow has two jobs: Decide what labels should exist, and then apply those labels to every report.
The first job is open-ended. A large model reads a varied sample of reports and proposes a small schema of categorical fields. A field is one way to describe the writer, such as attribution_style or procedure_orientation. Each field has a few allowed values, such as self_critical, system_attributing, or balanced.
The second job is repeatable. After a human approves the schema, a smaller model reads each report and chooses one value for each field.
We use Elastic Workflows because the steps are known ahead of time: sample reports, propose labels, wait for approval, classify every document, and store the results. Writing those steps in YAML makes the process reproducible, observable, and cheaper to rerun.
Why split LLM work across two model tiers?
A small model could handle classification, but schema discovery is a different shape of problem. It requires reading a diverse sample, spotting latent patterns, and proposing complex structures. In practice, smaller models over-anchor on surface keywords and produce redundant or nonexclusive fields.
Classification is simpler, the schema exists, the values are enumerated, and the task is to pick one per field. A smaller model handles this reliably and at a fraction of the cost, since it runs once per document across the entire corpus.
Large and small here mean reasoning capability. In this article, Claude Sonnet plays the planner and Mistral Small 3.1 plays the executor.
Classifying NASA pilot reports with a two-tier LLM pipeline
We’ll use the NASA ASRS database, which collects voluntary, anonymous incident reports from pilots, controllers, and mechanics. The dataset is public, and the reports are written as free-text narratives.
What we want to ask is:
What does this report reveal about the pilot who wrote it?
The planner reads a varied sample of reports and decides which distinctions are meaningful based on how the reports are actually written.
Step | Role | Model tier |
|---|---|---|
| Pull a diverse subset of reports from the corpus. | (no LLM) |
| Read the question and the sample, propose a schema of fields with enum values. | Large |
| Human reviews the proposed schema and approves or edits it. | (Human via |
| Iterate over the corpus, assign one value per field to each report. | Small |
| Write the schema and the per-document field values to Elasticsearch. | (No LLM) |
Registering Mistral and Claude as Elasticsearch inference endpoints
The small model will be registered as an Elasticsearch inference endpoint using the native mistral service integration.
INFERENCE_ID = "mistral-small-extractor"
es.inference.put(
task_type="chat_completion",
inference_id=INFERENCE_ID,
inference_config={
"service": "mistral",
"service_settings": {
"api_key": MISTRAL_API_KEY,
"model": "mistral-small-latest",
# 6 RPM is conservative for the Mistral free tier to avoid 429s.
"rate_limit": {"requests_per_minute": 6},
},
},
)The alias mistral-small-latest resolves to Mistral Small 3.1. It has a 128k context window and supports JSON-mode output.
The large model will be an AI connector pointing at Claude Sonnet. The Agent Builder UI walks you through creating the connector. Take a note of the connector ID since we’ll reference it from the workflow.
Indexing NASA ASRS incident reports into Elasticsearch
The ASRS dataset is indexed with keyword mappings for aggregation fields and text mappings for the narratives the models will read.
Download the ASRS CSV (the database publishes quarterly extracts at the ASRS Database Online page), and index it. The mappings are:
{
"properties": {
"acn": { "type": "keyword" },
"flight_phase": { "type": "keyword" },
"anomaly": { "type": "keyword" },
"synopsis": { "type": "text" },
"narrative": { "type": "text" }
}
}The mapping types follow how each field is used. flight_phase and anomaly are mapped as keyword because we’ll run terms aggregations on them to build the sample, and aggregations need exact, non-analyzed values. narrative and synopsis are mapped as text because they hold free-form prose that the models will read. The companion notebook has the full loader script that reads the CSV and bulk-indexes the documents.
Building a stratified sample for the planning LLM
The YAML snippets in this and the following sections are steps of the workflow definition that the notebook registers via the Workflows API. The first two steps generate a representative sample: They aggregate by flight phase and by anomaly and pull a few documents per bucket with top_hits.
- name: by_phase
type: elasticsearch.request
with:
method: POST
path: "/incident_reports/_search"
body:
size: 0
aggs:
per_phase:
terms:
field: flight_phase
size: 8
aggs:
sampled_docs:
top_hits:
size: 5
_source: ["acn", "synopsis", "narrative"]
- name: by_anomaly
type: elasticsearch.request
with:
method: POST
path: "/incident_reports/_search"
body:
size: 0
aggs:
per_anomaly:
terms:
field: anomaly
size: 8
aggs:
sampled_docs:
top_hits:
size: 3
_source: ["acn", "synopsis", "narrative"]How the large LLM discovers a classification schema from the data
The prompt needs both the question and the sample. A question alone may produce generic labels disconnected from the corpus, and a sample alone produces descriptive clusters that ignore the angle of the question.
When both are present and the output is structured, the model produces labels that are grounded in the data and oriented to the task: a schema of categorical fields, each with two to four mutually exclusive value options backed by evidence from the sample.
Here’s the planner step from the workflow:
- name: discover
type: ai.prompt
connector-id: "claude-sonnet"
with:
systemPrompt: |
You design categorical schemas for use by downstream classifiers. A
schema is a small set of fields, each with a few mutually exclusive
values. Every value you propose must be grounded in evidence from the
provided sample and must serve the stated question. You do not invent
values that are not supported by at least two documents in the sample.
You do not propose fields that a reasonable analyst could have written
without reading the documents.
prompt: |
Question:
${{ inputs.goal }}
Sample documents stratified by flight phase:
${{ steps.by_phase.output.aggregations.per_phase.buckets | json }}
Sample documents stratified by anomaly type:
${{ steps.by_anomaly.output.aggregations.per_anomaly.buckets | json }}
Propose between 2 and 4 categorical fields that:
- serve the question (you can explain how)
- depend on patterns visible in the sample (you can cite document IDs)
- would not be obvious to someone who has not read the sample
For each field, return: name (snake_case), definition, why_useful,
and values (2 to 4 mutually exclusive options).
For each value, return: value (snake_case) and definition.
schema:
type: object
properties:
fields:
type: array
minItems: 2
maxItems: 4
items:
type: object
required: [name, definition, why_useful, values]
properties:
name: { type: string }
definition: { type: string }
why_useful: { type: string }
values:
type: array
minItems: 2
maxItems: 4
items:
type: object
required: [value, definition]
properties:
value: { type: string }
definition: { type: string }
temperature: 0.3The structured output schema enforces the shape of the response:
name: Identifier for the categorical field.definition: What this field measures, in one sentence.why_useful: How this field serves the question; this also helps the downstream classifier understand the intent.values: Two to four mutually exclusive options. Each has avalueand adefinition.
Here’s an example of the produced schema. We can see how the writer is being classified and the reasons why the model decided to create the category. definitionand why_useful fields are used by the second model to classify the documents.
{
"fields": [
{
"name": "attribution_style",
"definition": "How the reporter frames responsibility for what happened.",
"why_useful": "Surfaces reporting culture independent of the technical event. Useful for training and safety-management programmes that want to distinguish reporter style from incident type.",
"values": [
{
"value": "self_critical",
"definition": "Assigns the cause primarily to their own action, even when external factors clearly contributed."
},
{
"value": "system_attributing",
"definition": "Frames the cause as external: ATC, equipment, weather, or organisational factors."
},
{
"value": "balanced",
"definition": "Distributes responsibility across self and system without emphasising either."
}
]
},
{
"name": "procedure_orientation",
"definition": "How the reporter relates to written procedure.",
"why_useful": "Distinguishes pilots who frame events through SOPs from those who frame them through personal judgment.",
"values": [
// procedure_first, experience_first (same structure as above)
]
}
]
}Human-in-the-loop schema approval with waitForInput
The proposed schema is now passed to a person for approval. Elastic Workflows has a waitForInput step that pauses the workflow with a schema, exposes a form, and resumes when the input is submitted.
waitForInput has no timeout of its own, so if nobody responds, the execution waits indefinitely. To put a limit on that, set a workflow-level settings.timeout; if it elapses before the reviewer submits the form, the execution is canceled.
- name: human_gate
type: waitForInput
with:
message: "Review and edit the proposed schema. The approved fields will be applied across the full corpus."
schema:
type: object
required: [approved_fields]
properties:
approved_fields:
type: array
items:
type: object
properties:
name: { type: string }
definition: { type: string }
values:
type: array
items:
type: object
properties:
value: { type: string }
notes:
type: stringWhen the workflow reaches this step, the execution pauses and the Kibana UI shows an "Action is required" badge. Clicking Provide action opens a form where the reviewer can paste or edit the schema JSON. Since waitForInput cannot be prepopulated from a previous step, the code polls the discover step output and prints a paste-ready JSON block that can be copied directly into this form.
discover = step_output(execution_id, "discover") # polls until the step completes
# Strip
why_useful
(not part of the human_gate form) and wrap in the shape
# expected by the waitForInput form so this is paste-ready.
approved_fields = [
{
"name": field["name"],
"definition": field["definition"],
"values": [
{"value": v["value"], "definition": v["definition"]}
for v in field["values"]
],
}
for field in discover["content"]["fields"]
]
print(json.dumps({"approved_fields": approved_fields, "notes": ""}, indent=2))The step_output helper (in the notebook) polls the execution via GET /api/workflows/executions/{id} until the discover step completes and then returns its output.
Code JSON output pasted on Kibana:
The reviewer can keep useful fields, rewrite unclear ones, merge overlapping values, and add notes. After approval, the workflow resumes and sends the final schema to the executor step.
For a new corpus, keep this human gate in place. Once the schema is stable, you can auto-approve and only fall back to review when it’s worth it: Route just the low-confidence extractions to a person, or compare a new discovery run against the schema stored in the schemas index and trigger review only when fields or values change beyond a threshold.
Classifying the full corpus with a smaller LLM
By the time the workflow reaches this step, the open-ended part of the job is over. From here, the small model takes over and classifies each report against the approved schema.
- name: fetch_corpus
type: elasticsearch.request
with:
method: POST
path: "/incident_reports/_search"
body:
size: 100
_source: ["acn", "narrative"]
query:
match_all: {}
- name: classify_all
type: foreach
foreach: "${{ steps.fetch_corpus.output.hits.hits }}"
iteration-on-failure:
retry:
max-attempts: 5
delay: "3s"
fallback:
- name: notify_failure
type: slack_api.postMessage
connector-id: "team-alerts"
with:
channelNames:
- "#pipeline-alerts"
text: "Classification failed for ACN ${{ foreach.item._source.acn }} after all retries."
continue: true
steps:
- name: classify
type: ai.agent
inference-id: "mistral-small-extractor"
timeout: "120s"
with:
message: |
You will classify the following report against a fixed schema.
For each field in the schema, assign exactly one of its value
options, or null if none of the values clearly applies. Include
the short quote that supports the assignment and a confidence
score between 0 and 1. Set review_required to true if any field
returned null or any confidence is below 0.5.
Schema:
${{ steps.human_gate.output.approved_fields | json }}
Report:
${{ foreach.item._source.narrative }}
schema:
type: object
properties:
field_values:
type: object
additionalProperties: true
review_required: { type: boolean }
- name: write_extraction
type: elasticsearch.index
with:
index: extractions
document:
acn: "${{ foreach.item._source.acn }}"
field_values: "${{ steps.classify.output.structured_output.field_values }}"
review_required: "${{ steps.classify.output.structured_output.review_required }}"Note: The classification step uses ai.agent instead of ai.prompt because ai.agent accepts an inference-id, which lets it call the Elasticsearch _inference endpoint directly, while ai.prompt only accepts a connector-id.
The fetch_corpus step is the third elasticsearch.request in the workflow, so it’s worth saying why we read from the index again. The first two (by_phase and by_anomaly) only pulled a small stratified sample for the planner to reason over, not the data to label. Now that the schema is approved, fetch_corpus pulls the documents we actually want to classify. We cap it at 100 with match_all to keep the demo fast; this is where you would page through the full corpus.
For every field, it returns a value (or null), a confidence, and a short quote. Setting additionalProperties: true in the JSON schema lets the step return one entry per field without the workflow having to know the field names ahead of time. A stored extraction looks like this:
{
"acn": "2238341",
"field_values": {
"attribution_style": {
"value": "self_critical",
"confidence": 0.82,
"quote": "I should have caught the altitude bust earlier"
},
"procedure_orientation": {
"value": "procedure_first",
"confidence": 0.44,
"quote": "we ran the QRH before doing anything else"
}
},
"review_required": true
}Here, review_required is true because procedure_orientation came back at 0.44 confidence, below our 0.5 threshold, which is the signal a confidence-based quality gate would act on.
The fetch_corpus step pulls the documents to classify. The foreachstep iterates over them sequentially, and iteration-on-failure handles the errors: retry covers transient API errors from the inference endpoint, and, if all attempts fail, the fallback step posts to Slack so the failure doesn’t pass silently. (An email connector works the same way.) continue: true then lets the loop move on to the next document instead of failing the whole run.
For production-scale corpora, consider using executeAsync, which is the fan-out version of execute.
Writing schemas and extractions back to Elasticsearch
The workflow produces two things: the approved schema and the per-document field values. The store_schema step runs right after the human gate, before the classification step fans out:
- name: store_schema
type: elasticsearch.index
with:
index: schemas
document:
question: "${{ inputs.goal }}"
approved_fields: "${{ steps.human_gate.output.approved_fields }}"
reviewer_notes: "${{ steps.human_gate.output.notes }}"Each extraction is written inside the foreach loop, so results are persisted as they’re produced rather than batched at the end.
The schemas index holds one document per discovery run (question, approved fields, reviewer notes). The extractions index holds one document per report per schema version.
What this two-tier LLM orchestration pattern gives you
We built one Elastic workflow that pulls a stratified sample from an incident report index, sends it with a question to a large reasoning model to generate a classification schema based on the data and a user-defined angle, pauses for human approval, and then iterates over the full corpus with a small Mistral model that assigns one value per field.
The approved schema and per-document field values are written back to Elasticsearch as structured data.
The point of the exercise is that two different shapes of work, schema discovery, and schema application can use two different model tiers and that a workflow lets you write the routing decision down.
Next steps for your own LLM pipeline
Try it on a corpus of your own. The pattern doesn’t care whether the input is incident reports, customer feedback, weekly status updates, or property listings.
Promote the
foreachstep toworkflow.executeAsynconce you’re comfortable for parallel fan-out at scale.Schedule the rediscovery workflow on a cron trigger so you can discover different schema variations based on the data that comes in.
Read the Elastic Workflows documentation for the full step catalog.
Related reading
Build AI agents with Elastic Inference Service (EIS) covers the broader multi-model wiring pattern via EIS, complementary to the Workflows-orchestrated split shown here.
How to build AI agentic workflows with Elasticsearch is a higher-level survey of how Agent Builder and Workflows fit together.
LangExtract and Elasticsearch tutorial explores a different extraction pattern using a hand-authored schema; useful for contrast with the discover-then-apply approach above.
