Documentation / Agents / Patterns

Build data and evaluation Agents

Build governed analysts, batch enrichment pipelines, document transforms, and evaluators with Resources, 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. For an evaluator, give turn a gate with a stable rubric. Commit the score and evidence, not just prose.
  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. If classification requires model judgment, 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 Scale with map and reduce, Durable execution, Analytics, and Research and knowledge Agents.