Skip to content

Tasks

See the tasks guide for the built-in tasks.

LegalTask

Bases: ABC

Abstract base for all Contractex legal NLP tasks.

Subclasses must set class-level attributes and implement run().

Class attributes

task_id: str Unique snake_case identifier (e.g. "contract_extraction"). Used as the key in TaskRegistry. doc_types: list[DocType] Document types this task can process. Empty list means all types. requires_llm: bool Whether the task calls an external or local LLM. TaskPipeline refuses to run such tasks on documents whose privacy profile is blocked. router: PrivacyAwareLLMRouter | None Router used by llm_for(). None uses the shared default.

run(doc: LegalDoc, **kwargs: Any) -> LegalDoc abstractmethod

Execute the task on doc.

Parameters

doc: Input document. May be mutated in-place or a new LegalDoc may be returned — callers should always use the returned value. **kwargs: Task-specific options (e.g. confidence_threshold).

Returns

LegalDoc Document with task results merged into doc.extracted.

llm_for(*docs: LegalDoc) -> LLMProvider

Return this task's provider wrapped so that every call on it enforces the strictest privacy profile among docs (block, local-only, redact). Pass every document whose text goes into the prompt.

run_async(doc: LegalDoc, **kwargs: Any) -> LegalDoc async

Async version of run().

Default implementation wraps the synchronous run() in a thread executor so it does not block the event loop. Override for native async providers.

estimate_cost(doc: LegalDoc) -> float

Estimate the API cost (USD) for processing doc with this task.

Default returns 0.0. Override for tasks that call external LLMs.

supports(doc: LegalDoc) -> bool

Return True if this task can process doc.

Returns True for all doc types if self.doc_types is empty.

TaskPipeline

An ordered sequence of LegalTask objects applied to a LegalDoc.

Tasks are applied in order. If a task does not support the document's doc_type, it is skipped with a warning rather than raising.

Usage

::

pipeline = TaskPipeline([pii_task, extraction_task, risk_task])
doc = pipeline.run(doc)
# doc.extracted now contains results from all tasks

Parameters

tasks: Ordered list of LegalTask instances. skip_unsupported: If True (default), tasks that don't support the document type are silently skipped. If False, an error is raised. record_timings: If True (default), task run times are written to doc.extracted["_task_timings"].

run(doc: LegalDoc, **kwargs: Any) -> LegalDoc

Run all tasks in order and return the final LegalDoc.

Parameters

doc: Input document. **kwargs: Passed to every task's run() method.

Returns

LegalDoc Document with all task results attached to doc.extracted.

run_async(doc: LegalDoc, **kwargs: Any) -> LegalDoc async

Async version of run(). Awaits each task in sequence.

estimate_total_cost(doc: LegalDoc) -> float

Sum cost estimates across all tasks that support doc.

TaskRegistry

Registry of available LegalTask implementations.

Attributes

task_ids: set[str] IDs of all currently registered tasks.

task_ids: set[str] property

Set of all registered task IDs (does not trigger lazy loading).

default() -> TaskRegistry classmethod

Return the shared global registry.

Built-in tasks are loaded lazily on first access.

register(task: type[LegalTask] | LegalTask) -> None

Register a task class or instance.

Parameters

task: A LegalTask subclass or instance. The task_id class attribute is used as the registry key.

Raises

ValueError If task_id is empty or already registered.

unregister(task_id: str) -> None

Remove a task from the registry.

get(task_id: str) -> LegalTask

Return a task instance by task_id.

Built-in tasks are loaded lazily the first time they are requested.

Raises

KeyError If task_id is not registered.

get_class(task_id: str) -> type[LegalTask]

Return the task class for task_id.

build_pipeline(task_ids: list[str], task_kwargs: dict[str, dict[str, Any]] | None = None, **pipeline_kwargs: Any) -> TaskPipeline

Build a TaskPipeline from an ordered list of task IDs.

Parameters

task_ids: Ordered list of task IDs to include. task_kwargs: Optional per-task constructor kwargs, e.g. {"contract_extraction": {"confidence_threshold": 0.8}}. **pipeline_kwargs: Passed to TaskPipeline.__init__.

Returns

TaskPipeline

Raises

KeyError If any task_id is not registered.