Implementation:Ucbepic Docetl MapOptimizer PlanGenerators

Knowledge Sources	Ucbepic_Docetl DocETL Docs
Domains	Data_Processing, Optimization
Last Updated	2026-02-08 00:00 GMT

Overview

Concrete tool for generating chunk-based, gleaning, parallel, and chain decomposition optimization plans for map operations provided by DocETL.

Description

The PlanGenerator class is a component of the MapOptimizer subsystem responsible for generating candidate optimization plans for map operations. It creates four types of plans: (1) chunk size plans that split long documents into chunks with configurable peripheral context and metadata extraction, (2) gleaning plans that iteratively refine outputs using validation feedback, (3) parallel plans that decompose a task into independent subtasks producing different output schema keys, and (4) chain plans that decompose a task into sequential dependent subtasks. The class supports recursive optimization of sub-operations (map and reduce sub-plans) up to a configurable maximum depth.

Usage

Use PlanGenerator when you need to generate candidate optimization plans for a map operation as part of the MapOptimizer pipeline. It is used internally by MapOptimizer to create the set of candidate plans that are then evaluated by the Evaluator. It is relevant when operations need chunking for long documents, gleaning for quality refinement, or decomposition into simpler parallel or sequential sub-tasks.

Code Reference

Source Location

Repository: Ucbepic_Docetl
File: docetl/optimizers/map_optimizer/plan_generators.py
Lines: 1-1047

Signature

class PlanGenerator:
    def __init__(
        self,
        runner,
        llm_client: LLMClient,
        console: Console,
        config: dict[str, Any],
        run_operation: Callable[[dict[str, Any], list[dict[str, Any]]], list[dict[str, Any]]],
        max_threads: int,
        is_filter: bool = False,
        depth: int = 1,
    ) -> None: ...

    def _generate_chunk_size_plans(
        self,
        op_config: dict[str, Any],
        input_data: list[dict[str, Any]],
        validator_prompt: str,
        token_limit: int,
    ) -> dict[str, list[dict[str, Any]]]: ...

    def generate_info_extraction_prompt(
        self,
        subprompt: str,
        split_key: str,
        sample_chunk_1: str,
        sample_chunk_2: str,
    ) -> str: ...

    def _generate_gleaning_plans(
        self,
        op_config: dict[str, Any],
        validation_prompt: str,
    ) -> dict[str, list[dict[str, Any]]]: ...

    def _generate_parallel_plans(
        self,
        op_config: dict[str, Any],
        input_data: list[dict[str, Any]],
    ) -> dict[str, list[dict[str, Any]]]: ...

    def _generate_chain_plans(
        self,
        op_config: dict[str, Any],
        input_data: list[dict[str, Any]],
    ) -> dict[str, list[dict[str, Any]]]: ...

    def _recursively_optimize_subtask(
        self,
        subtask_config: dict[str, Any],
        input_data: list[dict[str, Any]],
        subtask_name: str,
        plan_types: list[str],
    ) -> tuple[list[dict[str, Any]], float]: ...

Import

from docetl.optimizers.map_optimizer.plan_generators import PlanGenerator

I/O Contract

Inputs

Name	Type	Required	Description
runner	Runner	Yes	The pipeline runner object
llm_client	LLMClient	Yes	Client for interacting with the language model
console	Console	Yes	Rich console object for output logging
config	dict[str, Any]	Yes	The pipeline configuration dictionary
run_operation	Callable	Yes	Function to execute operations on input data
max_threads	int	Yes	Maximum number of threads for parallel execution
is_filter	bool	No	Whether the operation is a filter operation (default: False)
depth	int	No	Current recursion depth for nested optimization (default: 1)
op_config	dict[str, Any]	Yes	The map operation configuration
input_data	list[dict[str, Any]]	Yes	Input data for the operation
validator_prompt	str	Yes	Prompt for validating output quality (for chunk and gleaning plans)
token_limit	int	Yes	Maximum token limit for the model context window (for chunk plans)

Outputs

Name	Type	Description
plans	dict[str, list[dict[str, Any]]]	Dictionary mapping plan names to lists of operation configurations
subplan_optimizer_cost	float	Accumulated cost from recursive sub-plan optimization

Usage Examples

from docetl.optimizers.map_optimizer.plan_generators import PlanGenerator

plan_gen = PlanGenerator(
    runner=pipeline_runner,
    llm_client=llm_client,
    console=console,
    config=pipeline_config,
    run_operation=runner.run_operation,
    max_threads=4,
    is_filter=False,
    depth=1,
)

# Generate gleaning plans
gleaning_plans = plan_gen._generate_gleaning_plans(
    op_config=map_op_config,
    validation_prompt=validator_prompt,
)
print(f"Generated {len(gleaning_plans)} gleaning plans")

# Generate parallel decomposition plans
parallel_plans = plan_gen._generate_parallel_plans(
    op_config=map_op_config,
    input_data=sample_data,
)
for name, plan in parallel_plans.items():
    print(f"Plan '{name}': {len(plan)} operations")

# Generate chunk size plans
chunk_plans = plan_gen._generate_chunk_size_plans(
    op_config=map_op_config,
    input_data=sample_data,
    validator_prompt=validator_prompt,
    token_limit=8192,
)
print(f"Generated {len(chunk_plans)} chunk plans")

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment