Documentation / Agents / Patterns

Build data and evaluation Agents

Build governed analysts, batch pipelines, and evaluators with Resources, typed Model evaluation, map, reduce, and turn gates.

Before you begin

Separate bounded analytical judgment from data volume. A model may decide which approved query to run or interpret a bounded result, while large extraction and transformation belong in registered map functions. Deterministic aggregation belongs in reduce. Define schemas, row limits, privacy rules, cost budgets, and the exact database or table Resources before authoring prompts.

Steps

  1. For an analyst, use a turn to produce validated query intent, invoke a read-only Resource, then use a second turn to explain the bounded result.
  2. For batch work, register versioned partition and fold functions. Keep partition effects pure, read-only, or idempotent.
  3. To judge one state against named Choice, Score, or Noul questions, call ctx.evaluate() with a bound evaluation Model. For a generated turn whose output must pass a Scorer, use a turn gate. Commit the resulting evidence when the workflow needs it in the ledger.
  4. Use a content-addressed manifest for large input. Do not place an unbounded dataset inside a prompt or spawn one child Run per row.
ts
import { agent, validateFoldFn, validatePartitionFn } from "@constal/sdk";

const classify = validatePartitionFn<{ id: string; text: string }, { label: string; count: number }, never>({
  id: "classify-record", version: "1", effects: "pure", batch: { rows: 100 },
  async run(batch, out) {
    for (const row of batch) out.emit({ label: row.text.length > 80 ? "long" : "short", count: 1 });
  },
});

const countByLabel = validateFoldFn<{ label: string; count: number }, { label: string; count: number }>({
  id: "count-by-label", version: "1", deterministic: true, associative: true,
  async run(group, out) {
    let count = 0;
    for await (const row of group.rows) count += row.count;
    out.emit({ label: group.key!, count });
  },
});

export default agent({
  id: "data-profiler", version: "1.0.0", model: "model",
  partitionFns: [classify], foldFns: [countByLabel],
  async onMessage(rows: { id: string; text: string }[], ctx) {
    const mapped = await ctx.map(classify, rows, { partition: { rows: 100 }, concurrency: 8 });
    return ctx.reduce(countByLabel, mapped, { by: "label", scope: "key" });
  },
});

The same structure covers document extraction, catalog enrichment, classification, deduplication, and offline evaluation. For typed Model judgment, follow Evaluate state with a Model. If classification requires model judgment at volume, invoke a governed Model capability from an idempotent partition implementation and enforce a stage budget. Keep aggregation deterministic so retries and tree reduction preserve meaning.

Verify

Run a small known fixture before production volume. Confirm schemas, partition counts, row totals, residue, retry behavior, and aggregate results. Re-run failed partitions and verify committed output is identical. For evaluators, test adversarial examples and gate exhaustion. Analytics should distinguish model usage, Resource failures, rejected rows, and stage cost.

Next steps

Read Evaluate state with a Model, Scale with map and reduce, Durable execution, and Analytics.