Implementation:Ucbepic Docetl Utils

Knowledge Sources	Ucbepic_Docetl
Domains	Data_Processing, Utility
Last Updated	2026-02-08 00:00 GMT

Overview

Concrete tool for shared utility functions used across the DocETL codebase provided by DocETL.

Description

The utils module provides a collection of general-purpose helper functions used throughout DocETL. It includes the Decryptor class for LZString-based decryption, the StageType enum and CapturedOutput class for optimizer stage tracking, Jinja2 template utilities (has_jinja_syntax, extract_jinja_variables), token counting and data truncation functions (count_tokens, truncate_sample_data), YAML config loading (load_config), stratified sampling (smart_sample), a classproperty descriptor, and output extraction from pipeline results (extract_output_from_json).

Usage

Use these utilities as needed across DocETL modules. They are imported by the runner, optimizers, operations, and experiment scripts for common tasks like template analysis, token management, and data sampling.

Code Reference

Source Location

• Repository: Ucbepic_Docetl
• File: docetl/utils.py
• Lines: 1-453

Signature

class Decryptor:
    def __init__(self, secret_key: str): ...
    def decrypt(self, encrypted_data: str) -> str: ...

def decrypt(encrypted_data: str, secret_key: str) -> str: ...

class StageType(Enum):
    SAMPLE_RUN = "sample_run"
    SHOULD_OPTIMIZE = "should_optimize"
    CANDIDATE_PLANS = "candidate_plans"
    EVALUATION_RESULTS = "evaluation_results"
    END = "end"

def get_stage_description(stage_type: StageType) -> str: ...

class CapturedOutput:
    def __init__(self) -> None: ...
    def set_step(self, step: str) -> None: ...
    def save_optimizer_output(self, stage_type: StageType, output: Any) -> None: ...

def has_jinja_syntax(template_string: str) -> bool: ...
def prompt_user_for_non_jinja_confirmation(prompt_text: str, operation_name: str, prompt_field: str = "prompt") -> bool: ...
def extract_jinja_variables(template_string: str) -> list[str]: ...
def completion_cost(response: ModelResponse) -> float: ...
def load_config(config_path: str) -> dict[str, Any]: ...
def count_tokens(text: str, model: str) -> int: ...
def truncate_sample_data(data: dict[str, Any], available_tokens: int, key_lists: list[list[str]], model: str) -> dict[str, Any]: ...
def smart_sample(input_data: list[dict], sample_size_needed: int, max_unique_values: int = 5) -> list[dict]: ...
def extract_output_from_json(yaml_file_path, json_output_path=None) -> list[dict]: ...

class classproperty:
    def __init__(self, f: Callable[[Any], Any]) -> None: ...
    def __get__(self, obj: Any | None, owner: type) -> Any: ...

Import

from docetl.utils import (
    smart_sample,
    load_config,
    count_tokens,
    truncate_sample_data,
    has_jinja_syntax,
    extract_jinja_variables,
    extract_output_from_json,
    CapturedOutput,
    StageType,
)

I/O Contract

Inputs

Name	Type	Required	Description
template_string	str	Yes	Jinja2 template string for syntax detection or variable extraction
text	str	Yes	Text string for token counting
model	str	Yes	Model name for tokenizer selection (e.g., "gpt-4o")
input_data	list[dict]	Yes	Documents to sample (for smart_sample)
sample_size_needed	int	Yes	Target number of samples
config_path	str	Yes	Path to YAML configuration file (for load_config)
yaml_file_path	str	Yes	Path to YAML file (for extract_output_from_json)

Outputs

Name	Type	Description
has_jinja	bool	Whether the string contains Jinja2 syntax
variables	list[str]	List of variable names found in the template
token_count	int	Number of tokens in the text
sampled_data	list[dict]	Stratified sample of documents biased toward larger items
config	dict[str, Any]	Parsed YAML configuration dictionary
extracted_data	list[dict]	Output data filtered to schema-specified fields

Usage Examples

from docetl.utils import smart_sample, count_tokens, extract_jinja_variables, load_config

# Stratified sampling of documents
documents = [{"category": "A", "text": "long text..."}, ...]
sample = smart_sample(documents, sample_size_needed=50)

# Count tokens for a prompt
tokens = count_tokens("Analyze this document...", model="gpt-4o")

# Extract variables from a Jinja2 template
vars = extract_jinja_variables("{{ input.title }} by {{ input.author }}")
# Returns: ["title", "author"] (without "input" prefix)

# Load a YAML pipeline config
config = load_config("pipeline.yaml")

Related Pages

• Environment:Ucbepic_Docetl_Python_Runtime