LLM-Assisted Geoprocessing Pipelines

“Which residential parcels are within 200 metres of a watercourse and outside the flood zone” is one sentence and four geoprocessing operations, each of which can succeed while producing something the next step cannot use. Planning that chain, and validating between its steps, is what separates an analysis you can defend from a sequence of plausible operations ending in a number.

This topic belongs to geospatial prompt engineering and tool routing and sits above GeoPandas and PostGIS tool routing, which decides where each planned step runs. Chains long enough to need checkpointing belong to multi-step spatial agent orchestration.

Foundational Principles

Decompose before executing. The whole chain is planned, reviewed and estimated before the first operation runs. A pipeline assembled step by step as the model reacts to each result cannot be costed, cannot be checkpointed, and cannot be explained afterwards.

Every intermediate result is validated. Non-empty, valid, in the expected frame, and of a plausible size. Those four checks catch nearly every silent failure, and they are cheap relative to the operations they guard.

An empty intermediate is a stop, not a value. A buffer that produced nothing, an intersection that found nothing — these mean the analysis has already failed, and continuing produces an answer about nothing that reads like an answer about something.

Step-by-Step Implementation Pipeline

1. Decompose the request into a typed plan

The plan is a list of typed steps with named inputs and outputs, produced once and reviewable as data.

import logging
from dataclasses import dataclass
from typing import Literal, Optional, Sequence

log = logging.getLogger("geoprocessing_plan")

Op = Literal["select", "buffer", "intersect", "difference", "dissolve", "aggregate"]


@dataclass(frozen=True)
class Step:
    name: str
    op: Op
    inputs: tuple[str, ...]              # names of earlier steps or source layers
    params: dict


@dataclass(frozen=True)
class Plan:
    steps: tuple[Step, ...]
    result: str                          # the step whose output is the answer


class PlanRejected(ValueError):
    """The plan cannot be executed as written."""

2. Check the plan before running any of it

Most bad plans are detectable statically: a step referencing an output that does not exist, an operation applied to the wrong geometry type, a chain whose result is not produced by any step.

GEOMETRY_REQUIREMENTS = {
    "buffer":     {"Point", "LineString", "Polygon", "MultiPolygon"},
    "intersect":  {"Polygon", "MultiPolygon"},
    "difference": {"Polygon", "MultiPolygon"},
    "dissolve":   {"Polygon", "MultiPolygon"},
}


def check_plan(plan: Plan, sources: dict[str, str]) -> Plan:
    """Static checks: references resolve, types match, the result exists."""
    produced: dict[str, str] = dict(sources)          # name -> geometry type
    for step in plan.steps:
        for ref in step.inputs:
            if ref not in produced:
                raise PlanRejected(f"step {step.name!r} references unknown input {ref!r}")
        allowed = GEOMETRY_REQUIREMENTS.get(step.op)
        if allowed:
            for ref in step.inputs:
                if produced[ref] not in allowed:
                    raise PlanRejected(
                        f"{step.op} cannot take {produced[ref]} from {ref!r}")
        produced[step.name] = _output_type(step.op, [produced[r] for r in step.inputs])
    if plan.result not in produced:
        raise PlanRejected(f"the plan's result {plan.result!r} is not produced by any step")
    return plan

Type-checking the chain statically catches the most common planning error — buffering a polygon and then intersecting with a line, or dissolving something that is not an area — before any expensive operation runs. The decomposition itself is developed in decomposing natural language into geoprocessing steps.

3. Estimate the whole chain, not each step

A chain’s cost is dominated by the step with the largest intermediate, which is frequently not the step that looks expensive. Estimating the whole plan lets the agent trade a cheaper decomposition before anything runs.

def estimate_chain(plan: Plan, sizes: dict[str, int]) -> tuple[int, str]:
    """Track the largest intermediate; that is what the chain will cost."""
    current = dict(sizes)
    peak, peak_step = 0, ""
    for step in plan.steps:
        inputs = [current.get(r, 0) for r in step.inputs]
        if step.op == "buffer":
            out = inputs[0]
        elif step.op in {"intersect", "difference"}:
            out = int(max(inputs) * 1.2)              # overlays can add vertices and parts
        elif step.op == "dissolve":
            out = max(1, inputs[0] // 10)
        else:
            out = inputs[0] if inputs else 0
        current[step.name] = out
        if out > peak:
            peak, peak_step = out, step.name
    return peak, peak_step

4. Validate between every pair of steps

The four checks are the same each time, and running them uniformly is what makes the chain’s failure modes predictable.

@dataclass(frozen=True)
class Check:
    ok: bool
    reason: str


def validate_intermediate(result, expected_epsg: int, input_count: int,
                          step_name: str) -> Check:
    """Four checks that catch nearly every silent intermediate failure."""
    if result is None or len(result) == 0:
        return Check(False, f"{step_name} produced no features")
    invalid = sum(1 for g in result.geometry if g is not None and not g.is_valid)
    if invalid:
        return Check(False, f"{step_name} produced {invalid} invalid geometries")
    epsg = getattr(result.crs, "to_epsg", lambda: None)()
    if epsg != expected_epsg:
        return Check(False, f"{step_name} output is in EPSG:{epsg}, expected {expected_epsg}")
    if input_count and len(result) > input_count * 50:
        return Check(False,
                     f"{step_name} produced {len(result)} features from {input_count} "
                     "inputs — check for an unintended cross join")
    return Check(True, "")

The size sanity check is the one that catches the most expensive mistake. An overlay against the wrong layer produces a cross product, and the result is a valid, non-empty, correctly framed dataset of two million slivers that the next step will happily consume. The full set of checks is developed in validating intermediate geoprocessing outputs.

5. Stop on a failed check, with the step named

A failed intermediate is a stop, and the message needs the step name and the reason — that is what turns a dead chain into a correctable plan.

def run_chain(plan: Plan, execute, expected_epsg: int) -> dict:
    """Execute with a gate after every step. Stops at the first failure."""
    results: dict[str, object] = {}
    for step in plan.steps:
        inputs = [results.get(r) for r in step.inputs]
        try:
            out = execute(step, inputs)
        except Exception as exc:
            return {"ok": False, "failed_at": step.name, "reason": f"execution failed: {exc}"}
        check = validate_intermediate(out, expected_epsg,
                                      sum(len(i) for i in inputs if i is not None),
                                      step.name)
        if not check.ok:
            log.info("chain stopped at %s: %s", step.name, check.reason)
            return {"ok": False, "failed_at": step.name, "reason": check.reason}
        results[step.name] = out
    return {"ok": True, "result": results[plan.result], "steps": list(results)}

6. Keep the plan and the result together

An analysis result without its plan cannot be defended or repeated. Returning both, with the parameters that produced each step, is what makes a number a finding rather than an assertion.

def to_finding(plan: Plan, result, notes: Sequence[str]) -> dict:
    return {
        "value": summarise(result),
        "method": [{"step": s.name, "op": s.op, "params": s.params} for s in plan.steps],
        "notes": list(notes),
    }

7. Prefer a plan the database can run in one pass

A chain of four operations executed as four round trips through a process is slower and more fragile than the same chain expressed as one query where the database can do it. Detecting that case is worth the effort for common shapes.

FUSIBLE = ({"select", "buffer", "intersect"}, {"select", "intersect", "difference"})


def can_fuse(plan: Plan) -> bool:
    """Can this chain be expressed as a single query?"""
    ops = {s.op for s in plan.steps}
    return any(ops <= fusible for fusible in FUSIBLE) and len(plan.steps) <= 4

Fusing is an optimisation and not a requirement, and it should never change the result. Where a fused query and a stepwise chain disagree, the stepwise chain is the reference — it is the one whose intermediates were checked. The specific pattern of buffer, overlay and dissolve chains is covered in planning buffer, overlay and dissolve chains from a prompt.

8. Record what each step actually produced

Row counts per step are the cheapest diagnostic in the whole topic. A chain that ended in zero is explained instantly by the step where the count first became zero, and that number costs nothing to record.

def step_counts(results: dict[str, object]) -> dict[str, int]:
    return {name: (len(value) if value is not None else 0)
            for name, value in results.items()}

9. Resolve source layers against the catalog at plan time

A plan referencing a layer that does not exist should fail while it is still a plan, not three steps into execution. Resolving names against the live catalog during the static check is what makes that happen, and it also catches the more dangerous case of a name that resolves to something other than what was meant.

def resolve_sources(plan: Plan, catalog) -> dict[str, str]:
    """Map every source name to a real layer and its geometry type, or reject."""
    sources: dict[str, str] = {}
    referenced = {ref for step in plan.steps for ref in step.inputs}
    step_names = {step.name for step in plan.steps}
    for ref in sorted(referenced - step_names):
        record = catalog.get(ref)
        if record is None:
            raise PlanRejected(
                f"source layer {ref!r} does not exist; available layers are "
                f"{', '.join(sorted(catalog)[:8])}")
        if record.get("deprecated"):
            log.info("plan uses deprecated layer %r (successor: %s)",
                     ref, record.get("successor"))
        sources[ref] = record["geometry_type"]
    return sources

Listing the available layers in the rejection is what makes it recoverable in one turn. A model told that a layer does not exist will invent another plausible name; told which names exist, it picks one.

10. Keep the plan reproducible across data versions

A finding produced against last month’s data cannot be compared with one produced today unless both record which version they ran against. Stamping the source versions onto the plan makes a re-run either identical or explicitly different.

def stamp_versions(plan: Plan, catalog) -> dict:
    """Record the version of every source the plan touches."""
    referenced = {ref for step in plan.steps for ref in step.inputs}
    step_names = {step.name for step in plan.steps}
    return {ref: catalog[ref].get("version", "unversioned")
            for ref in sorted(referenced - step_names) if ref in catalog}

The unversioned default is deliberately visible. A source with no version is one whose findings cannot be reproduced, and seeing that string in a stored result is more useful than an absent field nobody notices — it is a prompt to fix the source rather than the analysis.

Operating This Stage Over Time

Plans acquire steps. A chain that started as three operations grows a filter, then a reprojection, then a cleanup, and each addition is individually justified. The cost is not just latency: every additional step is another place an intermediate can be empty, and a seven-step chain fails more often than a three-step one for reasons that have nothing to do with the data. Reviewing chain length distribution occasionally, and asking whether the longest ones could be expressed differently, is worth more than optimising any single step.

Source layers change beneath a plan. A layer renamed or re-partitioned upstream turns a working plan into a static check failure, which is the good case, or into a plan that runs against a different layer with the same name, which is not. Validating source names against the live catalog at plan time — not at execution time — turns the second case into the first.

The size sanity threshold needs attention as data grows. A multiplier that flagged cross joins against a corpus of thousands will flag legitimate results against a corpus of millions, and once it produces false positives it gets raised until it no longer fires. Derive it from the operation rather than from a constant: an intersection legitimately produces more features than its inputs, a dissolve produces fewer, and a buffer produces the same number.

Finally, keep the reference implementation stepwise even after fusing is introduced. The fused path is faster and unvalidated between steps, and the day the two disagree you will want the one whose intermediates were checked to be the one you trust.

Failure Modes & Root Causes

The confident zero. Every step succeeds, the answer is zero, and no step produced anything after the second. Root cause: no gate on empty intermediates. Mitigation: stop at the first empty result, naming the step.

The cross join. An overlay against the wrong layer produces millions of slivers, and the chain continues. Root cause: no size sanity check. Mitigation: bound the output size relative to the inputs, per operation.

The frame slip. One step returns geometry in a different frame and the next overlays it against the original, producing an empty or nonsensical result. Root cause: frames checked at ingestion but not between steps. Mitigation: check the frame on every intermediate.

The unrepeatable finding. A number is produced and nobody can reconstruct how. Root cause: the plan discarded after execution. Mitigation: return the plan with the result, always.

Production Validation Protocols

1. Static plan check. Assert every plan passes reference resolution and geometry-type checking before execution; a plan that fails should never reach a database.
2. Empty-intermediate test. Assert a chain whose second step produces nothing stops there, with that step named.
3. Cross-join guard test. Assert an overlay producing far more features than its inputs is stopped, using a fixture that triggers it.
4. Frame-consistency test. Assert an intermediate in an unexpected frame stops the chain rather than being reprojected silently.
5. Fusion equivalence test. Where fusing is enabled, assert the fused and stepwise paths produce identical results on a fixture set.
6. Step-count publication. Publish per-step row counts with every finding; they are the diagnostic that explains a surprising answer.

Frequently Asked Questions

Related

• Up to the section overview: Geospatial Prompt Engineering and Tool Routing
• Technique: Decomposing Natural Language into Geoprocessing Steps
• Technique: Validating Intermediate Geoprocessing Outputs
• Technique: Planning Buffer, Overlay and Dissolve Chains from a Prompt
• Peer topic: Multi-Step Spatial Agent Orchestration
• Peer topic: Topology Rule Enforcement via LLMs