Implementation:Ucbepic Docetl Directive DocChunkingTopK
| Knowledge Sources | |
|---|---|
| Domains | Pipeline_Optimization, LLM_Operations |
| Last Updated | 2026-02-08 00:00 GMT |
Overview
Concrete tool for transforming Map or Filter operations into retrieval-augmented chunking pipelines using TopK chunk selection provided by the DocETL reasoning optimizer.
Description
The DocumentChunkingTopKDirective class is a cost optimization directive for documents where only certain portions are relevant to the task. It works with both Map and Filter operations, transforming them into a retrieval-augmented pipeline: splits documents into chunks, uses TopK to retrieve the most relevant chunks (via embedding or full-text search), and processes them in a reduce operation. For Filter operations, it adds a final code_filter step to return boolean results.
Usage
The MOAR agent applies this directive when only certain portions of documents are relevant to the task and at least half of the document content is irrelevant. Ideal for complex filters or targeted extraction from documents with localized relevant sections. Works with both Map (extraction) and Filter (boolean decision) operations.
Code Reference
Source Location
- Repository: Ucbepic_Docetl
- File: docetl/reasoning_optimizer/directives/doc_chunking_topk.py
- Lines: 1-619
Signature
class DocumentChunkingTopKDirective(Directive):
name = "doc_chunking_topk"
description = "Cost optimization directive: Map/Filter => Split -> TopK -> Reduce (-> Code Filter if original was Filter)."
def check_applicability(self, ...) -> Tuple[bool, str]: ...
def apply(self, ...) -> Tuple[List[Dict], List[Dict], str, dict]: ...
Import
from docetl.reasoning_optimizer.directives.doc_chunking_topk import DocumentChunkingTopKDirective
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| op_config | Dict | Yes | Operation configuration to transform |
| pipeline_ops | List[Dict] | Yes | Full pipeline operations list |
| op_idx | int | Yes | Index of target operation |
| dataset_descriptions | Dict | Yes | Dataset schema descriptions |
Outputs
| Name | Type | Description |
|---|---|---|
| new_ops | List[Dict] | Transformed operation configs |
| new_steps | List[Dict] | Updated pipeline steps |
| explanation | str | Human-readable description of changes |
| metadata | dict | Additional metadata about the transformation |
Usage Examples
# Directives are typically invoked by the MOAR agent automatically
# Example of manual invocation:
from docetl.reasoning_optimizer.directives.doc_chunking_topk import DocumentChunkingTopKDirective
directive = DocumentChunkingTopKDirective()
applicable, reason = directive.check_applicability(op_config, pipeline_ops, op_idx, dataset_descriptions)
if applicable:
new_ops, new_steps, explanation, metadata = directive.apply(op_config, pipeline_ops, op_idx, dataset_descriptions)