Agents in ETL steps (`WithAgent`)¶

A step that needs a model to answer something about each row (classify a message, extract a field, label a reply) declares the agent with WithAgent, the way it declares a config value with FromConfig. The agent itself is an agent artifact in the repository; the model, its region and its price are deployment config under ai:. The step stays synchronous and Polars: it receives an AgentMapper and calls map on a frame.

pip install "loom-kernel[etl-polars,ai-bedrock]"

A complete example¶

The output type, next to the step that uses it. It must reject unknown fields, as every type_ref output does:

# src/pipelines/respondio/agents/seller_reply.py
from typing import Annotated, Literal

import msgspec


class SellerReply(msgspec.Struct, frozen=True, forbid_unknown_fields=True):
    respuesta: Literal["acepta", "contraoferta", "evalua", "rechaza", "otro"]
    motivo: Literal["ya_vendida", "precio", "competencia", "sin_motivo"] | None
    confianza: Annotated[float, msgspec.Meta(ge=0, le=1)]

The artifact, reviewed in pull requests like any other code:

# src/pipelines/respondio/agents/seller_reply.agent.yaml
spec_version: 1
name: seller_reply
description: Classifies the seller's answer to our appraisal.
model_role: classifier
instructions:
  - name: guide
    text: |
      Classify the seller's message as one of acepta, contraoferta, evalua,
      rechaza or otro. Give a reason only when the answer is rechaza.
output: {kind: type_ref, ref: "pipelines.respondio.agents.seller_reply:SellerReply"}
policies: {retries: 2, run_timeout_ms: 60000, max_requests: 3, max_usd: 0.01, on_unpriced_spend: refuse}

The pipeline config, next to storage:. specs resolve against the directory holding the YAML and may not leave it (see below for a config kept in a subfolder):

# etl.yaml, at the repository root
storage:
  engine: polars
  # ...

ai:
  engine: pydantic-ai
  specs: ["src/pipelines/respondio/agents/*.agent.yaml"]
  max_concurrent_runs: 8
  models:
    classifier:
      provider: bedrock
      model: eu.anthropic.claude-haiku-5-5
      region: eu-west-1
      output_mode: tool
      options: {max_tokens: 3000}
  prices:                           # USD per million tokens
    eu.anthropic.claude-haiku-5-5:
      input: 0.11
      output: 0.55
      cache_read: 0.011
      cache_write: 0.1375
      source: AWS Pricing API eu-west-1
      as_of: 2026-10-09

A config in a subfolder¶

specs resolve against ai.root, a path relative to the directory holding the YAML; without ai.root, against that directory itself. A glob may not leave the directory it resolves against, so a config kept under config/pipelines/ names the repository root with ai.root and its globs from there:

# config/pipelines/respondio.yaml
ai:
  engine: pydantic-ai
  root: ../..                       # the repository root
  specs: ["src/pipelines/respondio/agents/*.agent.yaml"]
  # models, prices, ...

ai.root is trusted config and may point anywhere; .. in a specs glob is still refused (AGENT_COMPILATION_FAILED carrying AGENT_SPECS_ESCAPE_ROOT).

The step reads the messages and a bounded window of what is already labelled, keeps the pending ones with an anti-join, and maps them:

from decimal import Decimal

import polars as pl

from loom.etl import AgentMapper, ETLStep, FromTable, IntoTable, WithAgent, col, params
from pipelines.respondio.agents.seller_reply import SellerReply

KEYS = ("workspace", "message_id")
TARGET = "fact.seller_reply"


class SellerReplyStep(ETLStep[DailyParams]):
    messages = FromTable("prep.seller_reply").where(col("message_date") == params.run_date)
    stored = FromTable(TARGET).where(col("message_date") == params.run_date)
    agent = WithAgent("seller_reply", output=SellerReply, max_usd=Decimal("2.00"))
    target = IntoTable(TARGET).upsert(keys=(*KEYS, "agent_version"))

    def execute(
        self,
        params: DailyParams,
        *,
        messages: pl.LazyFrame,
        stored: pl.LazyFrame,
        agent: AgentMapper,
    ) -> pl.LazyFrame:
        done = stored.filter(pl.col("agent_version") == agent.version).select(KEYS)
        pending = messages.join(done, on=list(KEYS), how="anti")
        prompt = pl.format(
            "<our_message>\n{}\n</our_message>\n<seller_message>\n{}\n</seller_message>",
            pl.col("our_text").fill_null("(none)"),
            pl.col("seller_text"),
        )
        return agent.map(pending, keys=KEYS, prompt=prompt).lazy()

runner = ETLRunner.from_yaml("etl.yaml")
runner.run(RespondioPipeline, DailyParams(run_date=date(2026, 10, 9)))

What `map` returns¶

One row per input row, in the input order, failures included:

Column	Content
the `keys`	copied from the input row
one per field of `output`, named as the field is on the wire (its `rename` or alias)	the answer, null on an error row
`agent_version`	the agent’s fingerprint (see below)
`agent_status`	`ok` or `error`
`agent_error`	null, a run error code (`PROVIDER_UNAVAILABLE`, `OUTPUT_SCHEMA_VIOLATION`, …), `BUDGET_EXHAUSTED`, `PROMPT_MISSING`, `OUTPUT_UNREPRESENTABLE` or `UNEXPECTED_ERROR`
`agent_input_tokens`, `agent_output_tokens`, `agent_cache_read_tokens`, `agent_cache_write_tokens`	what the run consumed, `0` when nothing was sent
`agent_cost_usd`	what the run cost in USD, as `Decimal(38, 12)` rounded half to even to 12 decimal places, null when it could not be priced

prompt is a Polars expression or a column name; a null prompt is not sent. A key named like a column map writes is refused with a ValueError before anything runs; an output field named like an agent_* column fails compilation (AGENT_COLUMN_COLLISION).

The answer columns take their Polars type from the field, never from the rows, so a batch without rows or with errors only has the same schema as any other and upserts into the same table:

Field type	Polars type
`str`, a `Literal` or `Enum` of strings	`String`
`int`, a `Literal` or `Enum` of integers	`Int64`
`float`	`Float64`
`bool`	`Boolean`
`date`	`Date`
`datetime`	`Datetime("us", "UTC")`; a naive value is read as UTC. `Datetime("us")` when the field requires naive values (`Meta(tz=False)`)
`Decimal`	`Decimal(38, 9)`, keeping 9 decimal places
`UUID`, `time`, `timedelta`, `bytes`	`String` holding the text the output type serialises the value to (for a msgspec struct: canonical UUID, ISO time, ISO 8601 duration such as `P1D`, base64)
`list[T]`, `set[T]`, `tuple[T, ...]`	`List` of the type of `T`
a nested struct or pydantic model	`Struct` of its fields’ types
a struct or pydantic model nested in itself, directly or through another type	`String` holding the value’s JSON from the first repetition down
`T \| None`	the type of `T`
anything else (`dict`, `Any`, a union of several types, a fixed-length tuple, a `Literal` mixing value types)	`String` holding the value’s JSON

An answer holding a value its column cannot (an int past Int64, a Decimal past Decimal(38, 9), …) becomes an error row with agent_error = OUTPUT_UNREPRESENTABLE, its answer columns null and its tokens and cost kept; the other rows of the batch are unaffected.

A failed row never aborts the batch: an Exception that is not a run error becomes UNEXPECTED_ERROR, keeps its reservation as spent and is logged by its type. What is not an Exception aborts map: KeyboardInterrupt and SystemExit reach the caller as themselves, any other inside a BaseExceptionGroup, and a run cancelled while the batch was not raises a RuntimeError naming the rows left without an outcome. Whether to write error rows, drop them or count them against a threshold is the step’s decision. Writing them with agent_version in the upsert keys and filtering agent_status == "ok" in the anti-join retries them on the next execution; leaving them in the done side does not.

How it runs¶

One runtime per call. Each map builds, enters and closes a runtime of its own on an event loop of its own (a dedicated thread when the calling thread already runs one, as under an async orchestrator; that thread sees the caller’s context variables). Two steps mapping at once share nothing. A frame without rows compiles nothing and opens no runtime. version and the compile-time checks compile each agent once per runner; each map recompiles it and version then answers that compilation, so it always equals the agent_version the last map wrote.
Concurrency waits. At most ai.max_concurrent_runs runs are in flight; the others wait for a slot instead of being refused with TOO_MANY_RUNS.
Budget per execution. WithAgent(..., max_usd=...) bounds what one map call may spend, whatever max_concurrent_runs. The budget is a Decimal or an int (converted exactly) above zero. A float is deprecated until loom 3.0: it emits a DeprecationWarning and is converted through its str(), so 0.1 becomes Decimal("0.1"). Any other type, bool included, raises TypeError; zero, a negative amount, NaN or Infinity raise ValueError when the WithAgent is constructed. Before each run its policies.max_usd is set aside, and replaced by the real cost when the run ends (the reservation itself when the cost is unknown or the run failed unexpectedly). The first run that does not fit stops the batch: it and every row after it are returned with agent_error = BUDGET_EXHAUSTED and nothing spent, and the anti-join picks them up on the next execution. A step budget therefore needs policies.max_usd on the artifact, and one run’s policies.max_usd must fit in it; otherwise the step does not compile (AGENT_BUDGET_UNENFORCEABLE), and map called directly with a budget raises ValueError.

What the step spends never exceeds its budget except by the elastic excess of the last response of each run in flight, so by at most max_concurrent_runs responses. policies.max_usd is elastic: the engine checks it against the run’s spend, summed over all its attempts, after each response, so a run stops with USAGE_LIMIT_EXCEEDED once a response takes it past the cap, and that response is billed and counted. Retries do not multiply the reservation: the attempts of one run share the same cap.
Writes happen once, after execute returns. A process killed mid-batch loses what that batch paid for, so a large backfill goes in bounded chunks (one day per execution, for instance).
Polars only. map takes a Polars DataFrame or LazyFrame and returns a DataFrame. On a Spark runner every WithAgent fails compilation (AGENT_UNSUPPORTED_ENGINE) and loom.ai.etl is never loaded.

Versions¶

agent.version, and the agent_version column, is the agent’s fingerprint: it changes when the instructions, the output schema, the output check, the declared tools (the kind and declared facts of each capability, including the include/exclude filters of MCP and A2A) or the model binding change, and stays when any policy (retries, the timeouts, max_iterations, max_tool_calls, max_requests, the token caps, max_usd, on_unpriced_spend, max_history_bytes), the step budget (WithAgent(max_usd=...)), the region, the credentials, the price or where a tool is served from change. Editing a skill’s text or a tool’s code does not change it either. Keeping agent_version in the upsert keys and in the anti-join re-runs, and pays for, every row of the anti-join window the day the agent changes, and leaves the old answers in place for comparison.

What is checked before anything is spent¶

ETLRunner.run compiles the pipeline and validates every WithAgent before any step runs:

Code	Cause
`AGENT_RUNNER_NOT_CONFIGURED`	the config has no `ai:` section, so no agent runner serves the step
`AGENT_NOT_FOUND`	no artifact under `ai.specs` declares the name
`AGENT_COMPILATION_FAILED`	the artifact does not compile; the message carries every compiler issue with its code
`AGENT_OUTPUT_MISMATCH`	`output=` is not the artifact’s `type_ref` output (a `json_schema` output never matches)
`AGENT_UNPRICED_BUDGET`	a budget is declared (`WithAgent(max_usd=...)` or the artifact’s `policies.max_usd`) for a model neither `ai.prices` nor the engine can price
`AGENT_BUDGET_UNENFORCEABLE`	`WithAgent(max_usd=...)` is declared but the artifact has no `policies.max_usd`, or that `policies.max_usd` exceeds it
`AGENT_COLUMN_COLLISION`	a field of `output` is named like an `agent_*` column `map` writes
`AGENT_UNSUPPORTED_ENGINE`	the runner executes on Spark; agents map Polars frames only

ETLCompiler() alone checks the shape, with the codes FromConfig uses: a keyword-only parameter named like the attribute (MISSING_CONFIG_PARAMS), no name shared with a source or client (CONFIG_ALIAS_CONFLICT), and no WithAgent on a StepSQL (UNSUPPORTED_CONFIG_VALUE).

Without `from_yaml`¶

ETLRunner, ETLRunner.from_config, ETLCompiler and ETLExecutor take agents=, any implementation of loom.etl.runtime.AgentBatchRunner. loom.ai.etl.PolarsAgentRunner(ai_config, root=...) is the one loom ships; its engine_provider= lets a test serve the agent from a pydantic-ai TestModel or FunctionModel:

from pydantic_ai.models.test import TestModel

from loom.ai.engines.pydantic_ai import PydanticAIEngineProvider
from loom.ai.etl import PolarsAgentRunner

agents = PolarsAgentRunner(
    ai_config,
    root=repo_root,
    engine_provider=lambda: PydanticAIEngineProvider(model_resolver=lambda _: TestModel()),
)
runner = ETLRunner.from_config(storage_config, agents=agents)

loom.etl never imports loom.ai: from_yaml loads loom.ai.etl by module path only when the config has an ai: section.

Agents in ETL steps (WithAgent)¶