Worker

Worker

A Worker converts one image region into one Recognition (text + confidence + metadata). It is one of the two pillars of scriva; see Concepts for the other.

Worker = image region → text + confidence

A Worker is plugged into an Orchestrator at the .recognize(...) step. It can also be called directly on a crop for testing or one-off use:

recognition = await worker(crop)

Building a Worker

Every Worker is built by chaining decorators onto a factory. Each call returns a new immutable Worker.

from scriva import Worker

worker = (
    Worker.openai("gpt-4o")              # engine factory — the base
    .prompt(scriva.prompts.cell)         # prompt template
    .cache(".scriva_cache")              # cache decorator
    .few_shot(samples)                   # few-shot exemplars
    .score(method="rendering")           # confidence scoring
    .retry(times=3)                      # retry on transient error
    .fallback(Worker.tesseract())        # fallback Worker on failure
)

The chain reads bottom-up: a gpt-4o call wrapped in a few-shot prompt, looked up in a cache first, scored after, retried on failure, falling back to tesseract if everything else fails.

Factories

Engine factories

The base of the chain. Each calls one OCR engine.

Factory	Engine	Extra
Worker.openai(model, **kwargs)	OpenAI vision (gpt-4o, …)	openai
Worker.anthropic(model, **kwargs)	Anthropic Claude (claude-…)	anthropic
Worker.bedrock(model, **kwargs)	AWS Bedrock vision models	bedrock
Worker.tesseract(**kwargs)	Tesseract OCR (local)	tesseract
Worker.paddle(**kwargs)	PaddleOCR (local)	paddle
Worker.azure(endpoint, **kwargs)	Azure Document Intelligence	azure
Worker.custom(fn)	Wrap a callable	—

All engine factories accept these common kwargs:

Kwarg	Type	Default	Meaning
prompt	Prompt	cell	Prompt template (VLM engines only)
max_concurrency	int	8	Per-Worker concurrency budget
timeout	float	30.0	Per-call timeout in seconds
language	str | None	None	ISO 639-1 hint

Kind factories

Specialised Workers for one content type. Each is pre-tuned for its domain — different prompt, different scorer, different post-processing. Slot them into .recognize.by_kind({...}):

Factory	Kind	Notes
Worker.text()	"text"	General free text. Default for unclassified regions.
Worker.number()	"number"	Digits, decimals, thousands separators, currency. Scores numerically.
Worker.date()	"date"	Locale-aware parsing; returns ISO 8601.
Worker.time()	"time"	24-hour ISO; locale-aware input.
Worker.checkbox()	"checkbox"	Returns "checked" | "unchecked" | "ambiguous".
Worker.handwriting()	"handwriting"	Cursive / printed handwriting; few-shot-friendly.
Worker.signature()	"signature"	Detects presence + similarity to a reference, not text.
Worker.barcode()	"barcode"	1D/2D barcode + QR.
Worker.email()	"email"	RFC 5322 validation; rejects on parse failure.
Worker.url()	"url"	RFC 3986 validation.
Worker.phone()	"phone"	Locale-aware; returns E.164.
Worker.address()	"address"	Structured address; falls back to free text.
Worker.blank()	"blank"	Returns Recognition(text=None). Free.
Worker.skip()	any	Returns Recognition(text=None). Used to drop regions.

Each kind factory is itself a fully decorated Worker — you can chain further decorators (Worker.number().cache(...).retry()).

By default the kind factories delegate to a configured “house” recogniser (the same one scriva.read picks). Override with .using(...):

Worker.number().using(Worker.tesseract())  # numeric crops via tesseract
Worker.handwriting().using(Worker.openai("gpt-4o"))

Composition factories

These take other Workers and return a new one:

Factory	Behaviour
Worker.cascade(a, b, c, ...)	Try in order; first non-error result wins.
Worker.consensus(a, b, ..., resolve=...)	Run all in parallel; resolve disagreements.
Worker.escalate(primary, oracle, when=...)	Run primary; re-run oracle only when when(recognition) is true.
Worker.router({kind: worker, ...})	Dispatch by region.kind inside a single Worker.
Worker.parallel(a, b, ...)	Run all, return list of recognitions. Power-user use.

Worker.router and the Orchestrator’s .recognize.by_kind({...}) overlap intentionally — use the router when you want the dispatch logic to live in the Worker (and be reusable across recipes); use .by_kind when you want it visible in the recipe.

Decorators

Every decorator is a method that returns a new Worker.

.prompt(prompt)

Sets the prompt template (VLM engines only).

from scriva.prompts import Prompt

worker = Worker.openai("gpt-4o").prompt(Prompt.ocr_strict)
worker = Worker.openai("gpt-4o").prompt("Read the cell. Return only the text.")

A Prompt is a small templating helper that injects {hint}, {language}, {kind}, and {exemplars} if they are provided. Plain strings are treated as a static prompt.

.cache(cache)

Wraps the Worker in a cache. cache may be a path (becomes FileSystemCache), a Cache instance, or a layered Cache.layered(path).

worker = Worker.openai("gpt-4o").cache(".scriva_cache")
worker = Worker.openai("gpt-4o").cache(Cache.layered(".scriva_cache"))

Hits surface on recognition.cache and in the event stream.

.few_shot(samples)

Adds few-shot exemplars to the prompt. samples may be:

• a SampleStore — exemplars are retrieved per-region by similarity
• a list of (crop, label) pairs — static exemplars on every call
• a callable (region) -> list[(crop, label)] — fully custom

worker = Worker.handwriting().few_shot(samples=SampleStore.fs(".scriva_samples"))
worker = Worker.number().few_shot([(crop1, "1,234"), (crop2, "56.78")])

See samples.md for the full SampleStore reference.

.score(method=...)

Adds confidence scoring after recognition.

method	What it does
"rendering"	Render the text back to pixels, embed both, cosine similarity.
"logprob"	Use the engine’s reported logprobs (engine-dependent).
"self_check"	Ask the model to re-read its own output; agreement = confidence.
"voting"	Run the Worker N times, score = agreement fraction.

worker = Worker.openai("gpt-4o").score(method="rendering")
worker = Worker.openai("gpt-4o").score(method="voting", n=3)

The score lands on recognition.confidence in [0.0, 1.0].

.retry(times=, on=, backoff=)

Retry on transient errors.

worker = (
    Worker.openai("gpt-4o")
    .retry(times=3, on=(EngineError, TimeoutError), backoff="expo")
)

backoff is "expo", "linear", or a callable (attempt) -> seconds.

.fallback(other)

Use other if this Worker raises after retries are exhausted.

worker = Worker.openai("gpt-4o").fallback(Worker.tesseract())

fallback chains: .fallback(a).fallback(b) tries self → a → b.

.escalate_to(stronger, when=...)

Run the Worker, then re-run stronger when when(recognition) is true. Use to cheap-first / oracle-second cascade:

worker = (
    Worker.openai("gpt-4o-mini")
    .score(method="rendering")
    .escalate_to(
        Worker.openai("gpt-4o"),
        when=lambda r: (r.confidence or 0) < 0.7,
    )
)

.parallel(max=...)

Set the per-Worker concurrency budget.

worker = Worker.openai("gpt-4o").parallel(max=16)

Inside an Orchestrator this is the semaphore the Orchestrator enforces. Default is 8.

.with_language(lang)

Set an ISO 639-1 language hint.

worker = Worker.openai("gpt-4o").with_language("ja")

.hint(hint)

Attach a static hint that the prompt template can splice. For per-region hints, use RecognitionHint (below).

worker = Worker.openai("gpt-4o").hint("This is an invoice number; digits only.")

Composition patterns

Cascade — try cheap first

worker = Worker.cascade(
    Worker.tesseract(),                       # free; try it first
    Worker.openai("gpt-4o-mini"),             # cheap VLM
    Worker.openai("gpt-4o"),                  # final fallback
)

cascade stops at the first Worker that returns a non-error recognition. To stop at the first high-confidence result, add a score and a predicate:

worker = Worker.cascade(
    Worker.tesseract().score(method="logprob"),
    Worker.openai("gpt-4o-mini").score(method="rendering"),
    Worker.openai("gpt-4o").score(method="rendering"),
    accept=lambda r: (r.confidence or 0) >= 0.8,
)

Consensus — run multiple, resolve

worker = Worker.consensus(
    Worker.openai("gpt-4o"),
    Worker.anthropic("claude-opus-4-7"),
    Worker.bedrock("qwen.qwen3-vl-235b-a22b"),
    resolve="confidence",   # or "majority" | "tiebreaker" | callable
)

resolve	Behaviour
"confidence"	The highest confidence wins. Default.
"majority"	Most-common text wins; ties → "confidence".
"tiebreaker"	A fourth Worker passed as tiebreaker= decides ties.
Callable	(list[Recognition]) -> Recognition.

Every member’s answer is recorded on recognition.alternatives so you can audit disagreements after the fact.

Escalate — uncertainty-first

worker = Worker.escalate(
    primary=Worker.openai("gpt-4o-mini").score(method="rendering"),
    oracle=Worker.openai("gpt-4o"),
    when=lambda r: (r.confidence or 0) < 0.7,
)

The oracle is only called for crops the primary was unsure about. This is the cheap shape that scales to large batches.

Router — kind-aware in one Worker

worker = Worker.router({
    "text":         Worker.openai("gpt-4o"),
    "number":       Worker.number().using(Worker.tesseract()),
    "date":         Worker.date(),
    "checkbox":     Worker.checkbox(),
    "handwriting":  Worker.handwriting().few_shot(samples),
    "blank":        Worker.skip(),
})

A router uses region.kind to pick the child Worker. Use this when you want the kind→Worker mapping to be reusable across recipes.

Calling a Worker directly

A Worker is callable on a crop:

import scriva
crop = scriva.crop_from_path("cell.png")            # PIL or numpy
recognition = await worker(crop)

Or synchronously:

recognition = worker.sync(crop)

Pass extra context if needed:

recognition = await worker(crop, hint="invoice number", language="en")

For testing, Worker.constant(text, confidence=1.0) returns a fixed result without calling any engine.

RecognitionHint

RecognitionHint is the typed envelope for per-region hints — text the Worker can use to bias its read.

from scriva import RecognitionHint

hint = RecognitionHint(text="Previous read: 1,234.56", confidence=0.4)
recognition = await worker(crop, hint=hint)

Common factories:

Factory	Hint source
RecognitionHint.from_result(result)	Per-region text from a prior DocumentResult.
RecognitionHint.from_store(store, near=...)	Few-shot exemplars from a SampleStore.
RecognitionHint.from_dict({rid: text})	Static per-region hints.

The Orchestrator threads hint= to the Worker for each region; custom Workers can read it directly off the call.

Writing a custom Worker

Two paths.

Subclass — for stateful Workers

from scriva import Worker, Recognition

class MyWorker(Worker):
    name = "my-worker"
    capabilities = frozenset({Capability.CONFIDENCE})

    async def recognize(self, crop, *, hint=None, region=None) -> Recognition:
        ...
        return Recognition(text=..., confidence=...)

Decorator — for stateless ones

from scriva import worker

@worker(name="strip-quotes", capabilities={Capability.TEXT})
async def strip_quotes(crop, *, hint=None, region=None):
    text = await my_engine.read(crop)
    return Recognition(text=text.strip('"'))

Both forms produce a Worker you can plug into any decorator chain (my_worker.cache(...).retry(...)) and any Orchestrator.

Capability matrix

worker.capabilities is a frozenset[Capability] that lets the Orchestrator validate wiring at build time. Common capabilities:

Capability	Meaning
TEXT	Returns text. Almost every Worker has this.
CONFIDENCE	Returns a meaningful confidence.
LANGUAGE_DETECTION	Sets recognition.language.
ALTERNATIVES	Returns an n-best list.
TOKEN_BOXES	Returns per-token bounding boxes (in recognition.attrs["tokens"]).
STREAMING	Emits progress events with partial_text.
STRUCTURED	Returns a structured value (date, number, etc.) on recognition.attrs.

Wiring a step that requires a capability the Worker lacks raises ConfigurationError at recipe-build time.

What to read next

• Orchestrator — where Workers plug in.
• Caching — Worker.cache(...) in depth.
• Sample stores — Worker.few_shot(...) and active learning.
• Reliability — retries, rate limits, timeouts, cost caps.