### Install DocETL Claude Code Skill via Pip Source: https://ucbepic.github.io/docetl/quickstart-claude-code Installs the DocETL Claude Code skill separately using pip. This method is suitable if DocETL is already installed. After installation, the skill is available in `~/.claude/skills/docetl/` and can be used by running `claude` in any directory. To uninstall, use the `--uninstall` flag. ```bash pip install docetl docetl install-skill ``` ```bash docetl install-skill --uninstall ``` -------------------------------- ### Extract and Summarize Product Review Themes with DocETL Python API Source: https://ucbepic.github.io/docetl/python/examples This example demonstrates extracting themes and quotes from product reviews using MapOp, then summarizing these themes across all documents using ReduceOp with the special '_all' reduce key. It requires the 'docetl' library and a CSV file named 'product_reviews.csv'. The output is a JSON file containing the summary. ```python from docetl.api import Pipeline, Dataset, MapOp, ReduceOp, PipelineStep, PipelineOutput # Define dataset - A CSV of product reviews dataset = Dataset( type="file", path="product_reviews.csv" # Contains columns: review_id, product_id, rating, review_text ) # Define operations operations = [ # Extract themes and quotes from each review MapOp( name="extract_themes", type="map", prompt=""" Analyze this product review and extract the key themes and representative quotes: Review: {{ input.review_text }} Rating: {{ input.rating }} Identify 2-3 major themes (e.g., usability, quality, value) and extract direct quotes that best represent each theme. """, output={ "schema": { "themes": "list[string]", "quotes": "list[string]", "sentiment": "string" } } ), # Summarize all themes across reviews using _all key ReduceOp( name="summarize_themes", type="reduce", reduce_key="_all", # Special key to reduce across all items prompt=""" Analyze and synthesize the themes and quotes from these product reviews: {% for item in inputs %} Review ID: {{ item.review_id }} Product: {{ item.product_id }} Rating: {{ item.rating }} Themes: {{ item.themes | join(", ") }} Quotes: {% for quote in item.quotes %} - "{{ quote }}" {% endfor %} Sentiment: {{ item.sentiment }} {% endfor %} Summarize the most frequent themes, the most representative quotes for each theme, and the overall sentiment. """, output={ "schema": { "summary": "string" } } ) ] # Define pipeline step (can consist of multiple operations) step = PipelineStep( name="review_analysis", input="product_reviews", operations=["extract_themes", "summarize_themes"] ) # Define output output = PipelineOutput(type="file", path="review_analysis_summary.json") # Create and run pipeline pipeline = Pipeline( name="review_analysis_pipeline", datasets={"product_reviews": dataset}, operations=operations, steps=[step], output=output, default_model="gpt-4o-mini" ) # Run the pipeline cost = pipeline.run() print(f"Pipeline execution completed. Total cost: ${cost:.2f}") ``` -------------------------------- ### Two-Step Ranking Pipeline Example Source: https://ucbepic.github.io/docetl/operators/rank This example demonstrates a two-step ranking process. The first step, 'extract_hostile_exchanges', uses a map operation to extract structured data about meanness and hostility from debate transcripts. The second step, 'rank_by_meanness', uses a rank operation to order these transcripts based on the extracted hostility information. ```yaml operations: - name: extract_hostile_exchanges type: map output: schema: meanness_summary: "str" hostility_level: "int" key_examples: "list[str]" prompt: | Analyze the following debate transcript for {{ input.title }} on {{ input.date }}: {{ input.content }} Extract and summarize exchanges where candidates are mean or hostile to each other. [... prompt details ...] - name: rank_by_meanness type: rank prompt: | Order these debate transcripts based on how mean or hostile the candidates are to each other. Focus on the meanness summaries and examples that have been extracted. Consider: - The overall hostility level rating - Severity of personal attacks in the key examples [... prompt details ...] input_keys: ["meanness_summary", "hostility_level", "key_examples", "title", "date"] direction: desc rerank_call_budget: 10 pipeline: steps: - name: meanness_analysis input: debates operations: - extract_hostile_exchanges - rank_by_meanness ``` -------------------------------- ### DocETL Aggregation Examples Source: https://ucbepic.github.io/docetl/pandas/operations These examples demonstrate the usage of the `df.semantic.agg` function for aggregation tasks. The first example shows a simple aggregation for summarization, while the second illustrates fuzzy matching with custom resolution configurations. ```python # Simple aggregation df.semantic.agg( reduce_prompt="Summarize these items: {{input.text}}", output={\"schema\": {\"summary\": \"str\"}} ) # Fuzzy matching with custom resolution df.semantic.agg( reduce_prompt="Combine these items: {{input.text}}", output={\"schema\": {\"combined\": \"str\"}}, fuzzy=True, comparison_prompt="Are these items similar: {{input1.text}} vs {{input2.text}}", resolution_prompt="Resolve conflicts between: {{items}}", resolution_output={\"schema\": {\"resolved\": \"str\"}} ) ``` -------------------------------- ### Multi-step Analysis Pipeline with DocETL Pandas Source: https://ucbepic.github.io/docetl/pandas/examples Demonstrates a multi-step analysis pipeline using DocETL with pandas. This example involves extracting structured information from social media posts, filtering relevant posts, and then grouping by product to summarize feedback. ```python # Social media posts posts = pd.DataFrame({ "text": ["Just tried the new iPhone 15!", "Having issues with iOS 17", "Android is better"], "timestamp": ["2024-01-01", "2024-01-02", "2024-01-03"] }) # 1. Extract structured information analyzed = posts.semantic.map( prompt="""Analyze this social media post and extract: 1. Product mentioned 2. Sentiment 3. Issues/Praise points Post: {{input.text}}""", output={ "schema": { "product": "str", "sentiment": "str", "points": "list[str]" } } ) # 2. Filter relevant posts relevant = analyzed.semantic.filter( prompt="Is this post about Apple products? {{input}}" ) # 3. Group by issue and summarize summaries = relevant.semantic.agg( fuzzy=True, reduce_keys=["product"], comparison_prompt="Do these posts discuss the same product?", reduce_prompt="Summarize the feedback about this product", output={ "schema": { "summary": "str", "frequency": "int", "severity": "str" } } ) # Summaries will be a df with the following columns: ``` -------------------------------- ### MOAR Optimizer Configuration Example Source: https://ucbepic.github.io/docetl/optimization/moar/configuration A complete example demonstrating the structure and common fields for MOAR's optimizer configuration. This includes required fields like 'type', 'save_dir', 'available_models', 'evaluation_file', 'metric_key', 'max_iterations', and 'rewrite_agent_model', along with optional fields like 'dataset_path' and 'exploration_weight'. ```yaml optimizer_config: type: moar save_dir: results/moar_optimization available_models: - gpt-4o-mini - gpt-4o - gpt-5.1-mini - gpt-5.1 evaluation_file: evaluate_medications.py metric_key: medication_extraction_score max_iterations: 40 rewrite_agent_model: gpt-5.1 dataset_path: data/sample.json # Optional exploration_weight: 1.414 # Optional ``` -------------------------------- ### Iterate and Format Calibration Prompt (Python) Source: https://ucbepic.github.io/docetl/api-reference/operations This snippet iterates through calibration samples and results to format a prompt for LLM calibration. It constructs a string that includes input-output pairs as examples for the LLM. ```python for i, (input_doc, output_doc) in enumerate( zip(calibration_sample, calibration_results) ): calibration_prompt += f"\n--- Example {i+1} ---" calibration_prompt += f"\nInput: {input_doc}" calibration_prompt += f"\nOutput: {output_doc}" calibration_prompt += """ Based on these examples, provide reference anchors that will be appended to the prompt to help maintain consistency when processing all documents. DO NOT provide generic advice. Instead, use specific examples from above as calibration points. Note that the outputs might be incorrect, because the user's prompt was not calibrated or rich in the first place. You can ignore the outputs if they are incorrect, and focus on the diversity of the inputs. Format as concrete reference points: - \"For reference, consider \'[specific input text]\' \u2192 [output] as a baseline for [category/level]" - \"Documents similar to \'[specific input text]\' should be classified as [output]" Reference anchors:""" ``` -------------------------------- ### Initialize Optimizer Configuration and LLM Client Source: https://ucbepic.github.io/docetl/api-reference/docetl Initializes the Optimizer with runner configuration, sets up the LLM client with specified models and rate limits, and prepares cache directories and paths based on configuration hashing. It also updates sample size mappings from the configuration. ```python def __init__( self, runner, rewrite_agent_model, judge_agent_model, timeout, resume, litellm_kwargs=None ): self.config = runner.config self.console = runner.console self.max_threads = runner.max_threads self.base_name = runner.base_name self.yaml_file_suffix = runner.yaml_file_suffix self.runner = runner self.status = runner.status self.optimized_config = copy.deepcopy(self.config) rate_limits = self.config.get("optimizer_config", {}).get("rate_limits", {}) self.llm_client = LLMClient( runner, rewrite_agent_model, judge_agent_model, rate_limits, **litellm_kwargs, ) self.timeout = timeout self.resume = resume self.captured_output = CapturedOutput() self.sample_cache = {} # Maps operation names to (output_data, sample_size) home_dir = os.environ.get("DOCETL_HOME_DIR", os.path.expanduser("~")) cache_dir = os.path.join(home_dir, f".docetl/cache/{runner.yaml_file_suffix}") os.makedirs(cache_dir, exist_ok=True) config_hash = hashlib.sha256(str(self.config).encode()).hexdigest() self.optimized_ops_path = f"{cache_dir}/{config_hash}.yaml" self.sample_size_map = SAMPLE_SIZE_MAP if self.config.get("optimizer_config", {}).get("sample_sizes", {}): self.sample_size_map.update(self.config["optimizer_config"]["sample_sizes"]) if not self.runner._from_df_accessors: self.print_optimizer_config() ``` -------------------------------- ### Install DocETL with Pandas Integration Source: https://ucbepic.github.io/docetl/pandas Installs the main DocETL package, which includes the pandas integration, using pip. This is the primary method to get started with DocETL's DataFrame capabilities. ```bash pip install docetl ``` -------------------------------- ### Write Clear and Concise LLM Prompts with Instructions Source: https://ucbepic.github.io/docetl/best-practices Demonstrates crafting specific prompts that provide necessary context and instruct the LLM on the desired output format and quantity. This example shows a prompt for summarizing medication side effects and uses. ```yaml prompt: | Here are some transcripts of conversations between a doctor and a patient: {% for value in inputs %} Transcript {{ loop.index }}: {{ value.src }} {% endfor %} For the medication {{ reduce_key }}, please provide the following information based on all the transcripts above: 1. Side Effects: Summarize all mentioned side effects of {{ reduce_key }}. List 2-3 main side effects. 2. Therapeutic Uses: Explain the medical conditions or symptoms for which {{ reduce_key }} was prescribed or recommended. Provide 1-2 primary uses. Ensure your summary: - Is based solely on information from the provided transcripts - Focuses only on {{ reduce_key }}, not other medications - Includes relevant details from all transcripts - Is clear and concise - Includes quotes from the transcripts ``` -------------------------------- ### Python Evaluation Function for DocETL MOAR Source: https://ucbepic.github.io/docetl/optimization/moar/getting-started An example Python function decorated with `@register_eval` that evaluates pipeline output against an original dataset. It calculates metrics like correct medication extraction counts and precision, returning them as a dictionary. ```python # evaluate_medications.py import json from typing import Any, Dict from docetl.utils_evaluation import register_eval @register_eval def evaluate_results(dataset_file_path: str, results_file_path: str) -> Dict[str, Any]: """ Evaluate pipeline output against the original dataset. """ # Load pipeline output with open(results_file_path, 'r') as f: output = json.load(f) # Load original dataset for comparison with open(dataset_file_path, 'r') as f: dataset = json.load(f) # Compute your evaluation metrics correct_count = 0 total_count = len(output) for idx, result in enumerate(output): # Compare result with original data # For example, if your dataset has a 'src' attribute, it's available in the output original_text = result.get("src", "").lower() extracted_items = result.get("medication", []) # Check if extracted items appear in original text for item in extracted_items: if item.lower() in original_text: correct_count += 1 # Return dictionary of metrics return { "medication_extraction_score": correct_count, # This key will be used if metric_key matches "total_extracted": total_count, "precision": correct_count / total_count if total_count > 0 else 0.0, } ``` -------------------------------- ### Initialize DocETL Optimizer Source: https://ucbepic.github.io/docetl/api-reference/docetl Initializes the optimizer with a runner instance and configuration. It sets up optimization parameters, caching, and cost tracking. Dependencies include DSLRunner, LLMClient, CapturedOutput, and copy. The method takes parameters for language models, resume status, and timeout, and initializes various attributes for configuration, console output, threading, file paths, LLM interaction, and caching. ```python def __init__( self, runner: "DSLRunner", rewrite_agent_model: str = "gpt-5.1", judge_agent_model: str = "gpt-4o-mini", litellm_kwargs: dict[str, Any] = {}, resume: bool = False, timeout: int = 60, ): """ Initialize the optimizer with a runner instance and configuration. Sets up optimization parameters, caching, and cost tracking. Args: yaml_file (str): Path to the YAML configuration file. model (str): The name of the language model to use. Defaults to "gpt-5.1". resume (bool): Whether to resume optimization from a previous run. Defaults to False. timeout (int): Timeout in seconds for operations. Defaults to 60. Attributes: config (Dict): Stores the loaded configuration from the YAML file. console (Console): Rich console for formatted output. max_threads (int): Maximum number of threads for parallel processing. base_name (str): Base name used for file paths. yaml_file_suffix (str): Suffix for YAML configuration files. runner (DSLRunner): The DSL runner instance. status: Status tracking for the runner. optimized_config (Dict): A copy of the original config to be optimized. llm_client (LLMClient): Client for interacting with the language model. timeout (int): Timeout for operations in seconds. resume (bool): Whether to resume from previous optimization. captured_output (CapturedOutput): Captures output during optimization. sample_cache (Dict): Maps operation names to tuples of (output_data, sample_size). optimized_ops_path (str): Path to store optimized operations. sample_size_map (Dict): Maps operation types to sample sizes. The method also calls print_optimizer_config() to display the initial configuration. """ self.config = runner.config self.console = runner.console self.max_threads = runner.max_threads self.base_name = runner.base_name self.yaml_file_suffix = runner.yaml_file_suffix self.runner = runner self.status = runner.status self.optimized_config = copy.deepcopy(self.config) # Get the rate limits from the optimizer config rate_limits = self.config.get("optimizer_config", {}).get("rate_limits", {}) self.llm_client = LLMClient( runner, rewrite_agent_model, judge_agent_model, rate_limits, **litellm_kwargs, ) self.timeout = timeout self.resume = resume self.captured_output = CapturedOutput() # Add sample cache for build operations self.sample_cache = {} # Maps operation names to (output_data, sample_size) ``` -------------------------------- ### Run DocETL MOAR Optimization via CLI Source: https://ucbepic.github.io/docetl/optimization/moar/getting-started Command to initiate the DocETL build process with MOAR optimization enabled. This command points to the pipeline configuration file and specifies 'moar' as the optimizer. ```bash docetl build pipeline.yaml --optimizer moar ``` -------------------------------- ### Initialize Optimizer Components Source: https://ucbepic.github.io/docetl/api-reference/optimizers Initializes various components required for the optimization process, including a plan generator, evaluator, and prompt generator. It sets up configurations for LLM clients, console logging, and operational parameters like maximum threads and timeouts. ```python def __init__( self, runner, run_operation, depth, is_filter (bool, optional): If True, the operation is a filter operation. Defaults to False. ): """ self.runner = runner self.config = runner.config self.console = runner.console self.llm_client = runner.optimizer.llm_client self._run_operation = run_operation self.max_threads = runner.max_threads self.timeout = runner.optimizer.timeout self._num_plans_to_evaluate_in_parallel = 5 self.is_filter = is_filter self.k_to_pairwise_compare = 6 self.plan_generator = PlanGenerator( runner, self.llm_client, self.console, self.config, run_operation, self.max_threads, is_filter, depth, ) self.evaluator = Evaluator( self.llm_client, self.console, self._run_operation, self.timeout, self._num_plans_to_evaluate_in_parallel, self.is_filter, ) self.prompt_generator = PromptGenerator( self.runner, self.llm_client, self.console, self.config, self.max_threads, self.is_filter, ) ``` -------------------------------- ### Create DocETL Pipeline YAML for MOAR Source: https://ucbepic.github.io/docetl/optimization/moar/getting-started Defines a standard DocETL pipeline structure with dataset configuration, a default model, and an operation to extract medications. This YAML serves as the base for MOAR optimization. ```yaml datasets: transcripts: path: data/transcripts.json type: file default_model: gpt-4o-mini operations: - name: extract_medications type: map output: schema: medication: list[str] prompt: | Extract all medications mentioned in: {{ input.src }} pipeline: steps: - name: medication_extraction input: transcripts operations: - extract_medications output: type: file path: results.json ``` -------------------------------- ### Build Optimized Pipeline with MOAR Optimizer (CLI) Source: https://ucbepic.github.io/docetl/concepts/optimization This command-line instruction demonstrates how to use the DocETL CLI to build an optimized pipeline using the MOAR optimizer. The MOAR optimizer is recommended for finding Pareto-optimal solutions balancing accuracy and cost. ```bash docetl build your_pipeline.yaml --optimizer moar ``` -------------------------------- ### Configure DocETL MOAR Optimizer Source: https://ucbepic.github.io/docetl/optimization/moar/getting-started Specifies the MOAR optimizer configuration within a YAML file. It includes the optimizer type, save directory, available models, the evaluation file path, the metric key for optimization, and maximum iterations. ```yaml optimizer_config: type: moar save_dir: results/moar_optimization available_models: # LiteLLM model names - ensure API keys are set in your environment - gpt-4o-mini - gpt-4o - gpt-5.1-mini - gpt-5.1 evaluation_file: evaluate_medications.py metric_key: medication_extraction_score # This must match a key in your evaluation function's return dictionary max_iterations: 40 rewrite_agent_model: gpt-5.1 dataset_path: data/transcripts_sample.json # Optional: use sample/hold-out dataset ``` -------------------------------- ### Print Pipeline Steps and Query Plan (Python) Source: https://ucbepic.github.io/docetl/api-reference/docetl This function visualizes pipeline steps with assigned colors and prints the full query plan. It uses a predefined color list and iterates through step boundaries to assign colors. Dependencies include 'rich' for console logging and a helper function '_print_op'. ```python colors = ["cyan", "magenta", "green", "yellow", "blue", "red"] step_names = [b.name.split("/")[0] for b in step_boundaries] step_colors = { name: colors[i % len(colors)] for i, name in enumerate(step_names) } # Print the legend self.console.log("\n[bold]Pipeline Steps:[/bold]") for step_name, color in step_colors.items(): self.console.log(f"[{color}]■[/{color}] {step_name}") # Print the full query plan starting from the last step boundary query_plan = _print_op(self.last_op_container, step_colors=step_colors) self.console.log(Panel(query_plan, title="Query Plan", width=100)) self.console.log() ``` -------------------------------- ### Configure MOAR for Medication Extraction (YAML) Source: https://ucbepic.github.io/docetl/optimization/moar/examples This YAML configuration defines a pipeline for extracting medications from medical transcripts using MOAR. It specifies datasets, models, optimization parameters, system prompts, and operations. The `metric_key` is set to `medication_extraction_score` to guide the optimization process. ```yaml datasets: transcripts: path: workloads/medical/raw.json type: file default_model: gpt-4o-mini bypass_cache: true optimizer_config: type: moar dataset_path: workloads/medical/raw_sample.json # Use sample for faster optimization save_dir: workloads/medical/moar_results available_models: # LiteLLM model names - ensure API keys are set in your environment - gpt-5.1-nano - gpt-5.1-mini - gpt-5.1 - gpt-4o - gpt-4o-mini evaluation_file: workloads/medical/evaluate_medications.py metric_key: medication_extraction_score max_iterations: 40 rewrite_agent_model: gpt-5.1 system_prompt: dataset_description: a collection of transcripts of doctor visits persona: a medical practitioner analyzing patient symptoms and reactions to medications operations: - name: extract_medications type: map output: schema: medication: list[str] prompt: | Analyze the following transcript of a conversation between a doctor and a patient: {{ input.src }} Extract and list all medications mentioned in the transcript. If no medications are mentioned, return an empty list. pipeline: steps: - name: medication_extraction input: transcripts operations: - extract_medications output: type: file path: workloads/medical/extracted_medications_results.json ``` -------------------------------- ### Equijoin: Match Candidates to Jobs using LLM Prompt Source: https://ucbepic.github.io/docetl/operators/equijoin This example demonstrates an Equijoin operation configured to match job candidates to job postings. It uses a 'comparison_prompt' that leverages Jinja2 syntax to pass candidate and job details to an LLM for a similarity-based match. The prompt guides the LLM to consider skills and experience for a 'True'/'False' response. ```yaml - name: match_candidates_to_jobs type: equijoin comparison_prompt: | Compare the following job candidate and job posting: Candidate Skills: {{ left.skills }} Candidate Experience: {{ left.years_experience }} Job Required Skills: {{ right.required_skills }} Job Desired Experience: {{ right.desired_experience }} Is this candidate a good match for the job? Consider both the overlap in skills and the candidate's experience level. Respond with "True" if it's a good match, or "False" if it's not a suitable match. ``` -------------------------------- ### Load Datasets from Configuration (Python) Source: https://ucbepic.github.io/docetl/api-reference/docetl This method loads all datasets specified in the project's configuration. It supports 'file' and 'memory' types, extracting necessary information like path and parsing rules. It creates 'Dataset' objects and logs the loading status. Dependencies include a 'Dataset' class and 'rich' for console logging. ```python def load(self) -> None: """ Load all datasets defined in the configuration. """ datasets = {} self.console.rule("[bold]Loading Datasets[/bold]") for name, dataset_config in self.config["datasets"].items(): if dataset_config["type"] == "file": datasets[name] = Dataset( self, "file", dataset_config["path"], source="local", parsing=dataset_config.get("parsing", []), user_defined_parsing_tool_map=self.parsing_tool_map, ) self.console.log( f"[green]✓[/green] Loaded dataset '{name}' from {dataset_config['path']}" ) elif dataset_config["type"] == "memory": datasets[name] = Dataset( self, "memory", dataset_config["path"], source="local", parsing=dataset_config.get("parsing", []), user_defined_parsing_tool_map=self.parsing_tool_map, ) self.console.log( f"[green]✓[/green] Loaded dataset '{name}' from in-memory data" ) else: raise ValueError(f"Unsupported dataset type: {dataset_config['type']}") self.datasets = { name: ( dataset if isinstance(dataset, Dataset) else Dataset(self, "memory", dataset) ) for name, dataset in datasets.items() } self.console.log() ``` -------------------------------- ### Configure Value Sampling Method and Explanation (Python) Source: https://ucbepic.github.io/docetl/api-reference/optimizers This snippet demonstrates how to configure the value sampling method and generate an explanation using an LLM client. It parses the LLM response to extract the chosen method and initializes a value sampling configuration object. ```python parameters = { "type": "object", "properties": { "method": {"type": "string", "enum": ["random", "cluster", "sem_sim"]}, "explanation": {"type": "string"}, }, "required": ["method", "explanation"], } response = self.llm_client.generate_rewrite( [{"role": "user", "content": prompt}], system_prompt, parameters, ) result = json.loads(response.choices[0].message.content) method = result["method"] value_sampling_config = { "enabled": True, "method": method, "sample_size": 100, # Default sample size "embedding_model": "text-embedding-3-small", } ``` -------------------------------- ### JoinOptimizer Class Initialization (Python) Source: https://ucbepic.github.io/docetl/api-reference/optimizers Initializes the JoinOptimizer with a runner, operation configuration, and parameters for selectivity estimation and sampling. It sets up logging and internal state for optimization. ```python class JoinOptimizer: def __init__( self, runner, op_config: dict[str, Any], target_recall: float = 0.95, sample_size: int = 500, sampling_weight: float = 20, agent_max_retries: int = 5, estimated_selectivity: float | None = None, ): self.runner = runner self.config = runner.config self.op_config = op_config self.llm_client = runner.optimizer.llm_client self.max_threads = runner.max_threads self.console = runner.console self.target_recall = target_recall self.sample_size = sample_size self.sampling_weight = sampling_weight self.agent_max_retries = agent_max_retries self.estimated_selectivity = estimated_selectivity self.console.log(f"Target Recall: {self.target_recall}") self.status = self.runner.status self.max_comparison_sampling_attempts = 5 self.synthesized_keys = [] # if self.estimated_selectivity is not None: ``` -------------------------------- ### Map-Unnest-Resolve-Reduce on Theme Keys with DocETL Python API Source: https://ucbepic.github.io/docetl/python/examples This Python example demonstrates a DocETL pipeline that extracts theme-quote pairs, unnests them, resolves similar themes using ResolveOp, and then reduces the results based on the theme key. It utilizes MapOp, UnnestOp, ResolveOp, and ReduceOp, requiring the 'docetl' library and a CSV file for input. This example also focuses on pipeline optimization. ```python from docetl.api import Pipeline, Dataset, MapOp, UnnestOp, ResolveOp, ReduceOp, PipelineStep, PipelineOutput # Define dataset - A JSON file with product reviews dataset = Dataset( type="file", path="product_reviews.csv" # Same csv as previously ) ``` -------------------------------- ### Using Sample/Hold-Out Dataset for Optimization Source: https://ucbepic.github.io/docetl/optimization/moar/configuration Demonstrates the best practice of specifying a separate 'dataset_path' for optimization (e.g., a sample or hold-out set) to prevent overfitting to the test set, while the full dataset is used in the final pipeline. ```yaml optimizer_config: dataset_path: data/sample_dataset.json # Use sample/hold-out for optimization # ... other config ... datasets: transcripts: path: data/full_dataset.json # Full dataset for final pipeline ``` -------------------------------- ### Containment-Based Blocking Rule Example (Python) Source: https://ucbepic.github.io/docetl/api-reference/optimizers An example of a containment-based blocking rule using Python's `eval()` function. This rule checks if all words in the right field are present in the left field, ignoring case. It's designed for efficient pre-filtering of potential matches. ```python "all(word in left['{{left_key}}'].lower() for word in right['{{right_key}}'].lower().split())" ``` -------------------------------- ### EquijoinOperation Initialization with User Confirmation Source: https://ucbepic.github.io/docetl/api-reference/operations Handles the initialization of the EquijoinOperation, specifically checking if the 'comparison_prompt' lacks Jinja2 syntax. If it does, it prompts the user for confirmation before proceeding, raising a ValueError if the user cancels. ```python def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Check for non-Jinja prompts and prompt user for confirmation if "comparison_prompt" in self.config and not has_jinja_syntax( self.config["comparison_prompt"] ): if not prompt_user_for_non_jinja_confirmation( self.config["comparison_prompt"], self.config["name"], "comparison_prompt", ): raise ValueError( f"Operation '{self.config['name']}' cancelled by user. Please add Jinja2 template syntax to your comparison_prompt." ) ``` -------------------------------- ### Synthetic Data Generation with DocETL Source: https://ucbepic.github.io/docetl/pandas/examples Shows how to generate multiple variations of output for each input row using the `n` parameter in DocETL's map operation. This is useful for synthetic data generation, A/B testing, or creating alternative content. The example generates marketing headlines and then filters them for quality. ```python # Starter concepts for content generation df = pd.DataFrame({ "product": ["Smart Watch", "Wireless Earbuds", "Home Security Camera"], "target_audience": ["Fitness Enthusiasts", "Music Lovers", "Homeowners"] }) # Generate 5 marketing headlines for each product-audience combination variations = df.semantic.map( prompt="""Generate a compelling marketing headline for the following product and target audience: Product: {{input.product}} Target Audience: {{input.target_audience}} The headline should: - Be attention-grabbing and memorable - Speak directly to the target audience's needs or desires - Highlight the key benefit of the product - Be between 5-10 words """, output={"schema": {"headline": "str"}, "n": 5} # Generate 5 variations for each input row ) print(f"Original dataframe rows: {len(df)}") print(f"Generated variations: {len(variations)}") # Should be 5x the original count # Variations dataframe will contain: # - All original columns (product, target_audience) # - The new headline column # - 5 rows for each original row (15 total for this example) # You can also combine this with other operations # Filter to only keep the best headlines best_headlines = variations.semantic.filter( prompt="""Is this headline exceptional and likely to drive high engagement? Product: {{input.product}} Target Audience: {{input.target_audience}} Headline: {{input.headline}} Consider catchiness, emotional appeal, and clarity. """ ) ``` -------------------------------- ### Initialize DocETL Optimizer Source: https://ucbepic.github.io/docetl/api-reference/docetl Initializes the Optimizer class with a DSLRunner instance and configuration parameters. It sets up the language model, resume behavior, and operation timeouts. Dependencies include the DSLRunner class and potentially LiteLLM for LLM interactions. ```python class Optimizer: """ Orchestrates the optimization of a DocETL pipeline by analyzing and potentially rewriting operations marked for optimization. Works with the runner's pull-based execution model to maintain lazy evaluation while improving pipeline efficiency. """ def __init__( self, runner: "DSLRunner", rewrite_agent_model: str = "gpt-5.1", judge_agent_model: str = "gpt-4o-mini", litellm_kwargs: dict[str, Any] = {}, resume: bool = False, timeout: int = 60, ): """ Initialize the optimizer with a runner instance and configuration. Sets up optimization parameters, caching, and cost tracking. Args: yaml_file (str): Path to the YAML configuration file. model (str): The name of the language model to use. Defaults to "gpt-5.1". resume (bool): Whether to resume optimization from a previous run. Defaults to False. timeout (int): Timeout in seconds for operations. Defaults to 60. Attributes: config (Dict): Stores the loaded configuration from the YAML file. console (Console): Rich console for formatted output. max_threads (int): Maximum number of threads for parallel processing. base_name (str): Base name used for file paths. yaml_file_suffix (str): Suffix for YAML configuration files. runner (DSLRunner): The DSL runner instance. status: Status tracking for the runner. optimized_config (Dict): A copy of the original config to be optimized. llm_client (LLMClient): Client for interacting with the language model. timeout (int): Timeout for operations in seconds. resume (bool): Whether to resume from previous optimization. ``` -------------------------------- ### Parallel Map Operation Configuration and Example Source: https://ucbepic.github.io/docetl/operators/parallel-map This snippet demonstrates the configuration of a Parallel Map operation in DocETL. It includes required and optional parameters, and an example of processing job applications by extracting skills, calculating experience, and evaluating cultural fit concurrently. The 'output' schema defines the structure of the combined results. ```yaml - name: process_job_application type: parallel_map prompts: - name: extract_skills prompt: "Given the following resume: '{{ input.resume }}', list the top 5 relevant skills for a software engineering position." output_keys: - skills gleaning: num_rounds: 1 validation_prompt: | Confirm the skills list contains **exactly** 5 distinct skills and each skill is one or two words long. model: gpt-4o-mini - name: calculate_experience prompt: "Based on the work history in this resume: '{{ input.resume }}', calculate the total years of relevant experience for a software engineering role." output_keys: - years_experience model: gpt-4o-mini - name: evaluate_cultural_fit prompt: "Analyze the following cover letter: '{{ input.cover_letter }}'. Rate the candidate's potential cultural fit on a scale of 1-10, where 10 is the highest." output_keys: - cultural_fit_score model: gpt-4o-mini output: schema: skills: list[string] years_experience: float cultural_fit_score: integer ``` -------------------------------- ### Setting up Retrievers in Python Source: https://ucbepic.github.io/docetl/api-reference/docetl Instantiates and configures retrievers, specifically supporting LanceDB retrievers. It validates retriever configurations, sets default parameters, and handles lazy index creation. ```python def _setup_retrievers(self) -> None: """Instantiate retrievers from configuration (lazy index creation).""" from docetl.retrievers.lancedb import LanceDBRetriever self.retrievers: dict[str, Any] = {} retrievers_cfg = self.config.get("retrievers", {}) or {} for name, rconf in retrievers_cfg.items(): if not isinstance(rconf, dict): raise ValueError(f"Invalid retriever '{name}' configuration") if rconf.get("type") != "lancedb": raise ValueError( f"Unsupported retriever type '{rconf.get('type')}' for '{name}'. Only 'lancedb' is supported." ) required = ["dataset", "index_dir", "index_types"] for key in required: if key not in rconf: raise ValueError( f"Retriever '{name}' missing required key '{key}'." ) # Defaults rconf.setdefault("query", {"top_k": 5}) rconf.setdefault("build_index", "if_missing") self.retrievers[name] = LanceDBRetriever(self, name, rconf) ``` -------------------------------- ### Batch Processing Example for Document Classification Source: https://ucbepic.github.io/docetl/operators/map Demonstrates how to use the `batch_prompt` parameter to process multiple documents in a single LLM call. This is efficient for simpler tasks and shorter documents, but requires careful adjustment of `max_batch_size` to avoid incorrect results. The `batch_prompt` receives an `inputs` list, and a fallback `prompt` is required. ```yaml - name: classify_documents type: map max_batch_size: 5 # Process up to 5 documents in a single LLM call batch_prompt: | Classify each of the following documents into categories (technology, business, or science): {% for doc in inputs %} Document {{loop.index}}: {{doc.text}} {% endfor %} Provide a classification for each document. prompt: | Classify the following document: {{input.text}} output: schema: category: string ``` -------------------------------- ### Define Enum Schema in DocETL Source: https://ucbepic.github.io/docetl/concepts/schemas This example shows how to define an enum type for a sentiment field in DocETL, restricting the output to a predefined set of values. ```yaml output: schema: sentiment: "enum[positive, negative, neutral]" ``` -------------------------------- ### Generate Reduce Plans with Varying Batch Sizes and Fold Prompts Source: https://ucbepic.github.io/docetl/api-reference/optimizers This method generates a list of reduce plans by systematically varying batch sizes and fold prompts. It dynamically calculates optimal batch sizes by considering the LLM's context window length, prompt size, and estimated input/output token counts. The function also attempts to synthesize multiple fold prompts and combines them with the calculated batch sizes to create diverse plans. Dependencies include `model_cost`, `count_tokens`, `extract_jinja_variables`, `mean`, and `_run_operation`, `_synthesize_fold_prompts`. ```python def _generate_reduce_plans(self, op_config: dict[str, Any], input_data: list[dict[str, Any]], is_associative: bool, ) -> list[dict[str, Any]]: """ Generates various reduce plans by varying batch sizes and fold prompts. It takes into account the LLM's context window size to determine appropriate batch sizes. Args: op_config (dict[str, Any]): Configuration for the reduce operation. input_data (list[dict[str, Any]]): Input data for the reduce operation. is_associative (bool): Flag indicating whether the reduce operation is associative. Returns: list[dict[str, Any]]: A list of reduce plans, each with different batch sizes and fold prompts. """ model = op_config.get("model", "gpt-4o-mini") model_input_context_length = model_cost.get(model, {}).get( "max_input_tokens", 8192 ) # Estimate tokens for prompt, input, and output prompt_tokens = count_tokens(op_config["prompt"], model) sample_input = input_data[:100] sample_output = self._run_operation(op_config, input_data[:100]) prompt_vars = extract_jinja_variables(op_config["prompt"]) prompt_vars = [var.split(".")[-1] for var in prompt_vars] avg_input_tokens = mean( [ count_tokens( json.dumps({k: item[k] for k in prompt_vars if k in item}), model ) for item in sample_input ] ) avg_output_tokens = mean( [ count_tokens( json.dumps({k: item[k] for k in prompt_vars if k in item}), model ) for item in sample_output ] ) # Calculate max batch size that fits in context window max_batch_size = ( model_input_context_length - prompt_tokens - avg_output_tokens ) // avg_input_tokens # Generate 6 candidate batch sizes batch_sizes = [ max(1, int(max_batch_size * ratio)) for ratio in [0.1, 0.2, 0.4, 0.6, 0.75, 0.9] ] # Log the generated batch sizes self.console.log("[cyan]Generating plans for batch sizes:[/cyan]") for size in batch_sizes: self.console.log(f" - {size}") batch_sizes = sorted(set(batch_sizes)) # Remove duplicates and sort plans = [] # Generate multiple fold prompts max_retries = 5 retry_count = 0 fold_prompts = [] while retry_count < max_retries and not fold_prompts: try: fold_prompts = self._synthesize_fold_prompts( op_config, sample_input, sample_output, num_prompts=self.num_fold_prompts, ) fold_prompts = list(set(fold_prompts)) if not fold_prompts: raise ValueError("No fold prompts generated") except Exception as e: retry_count += 1 if retry_count == max_retries: raise RuntimeError( f"Failed to generate fold prompts after {max_retries} attempts: {str(e)}" ) self.console.log( f"Retry {retry_count}/{max_retries}: Failed to generate fold prompts. Retrying..." ) for batch_size in batch_sizes: for fold_idx, fold_prompt in enumerate(fold_prompts): plan = op_config.copy() plan["fold_prompt"] = fold_prompt plan["fold_batch_size"] = batch_size plan["associative"] = is_associative plan["name"] = f"{op_config['name']}_bs_{batch_size}_fp_{fold_idx}" plans.append(plan) return plans ``` -------------------------------- ### Initialize ReduceOptimizer in Python Source: https://ucbepic.github.io/docetl/api-reference/optimizers Initializes the ReduceOptimizer with necessary components like a runner, configuration, LLM client, and a function to execute operations. It sets up parameters for prompt generation and validation, and defines the maximum threads for parallel processing. ```python class ReduceOptimizer: """ A class that optimizes reduce operations in data processing pipelines. This optimizer analyzes the input and output of a reduce operation, creates and evaluates multiple reduce plans, and selects the best plan for optimizing the operation's performance. Attributes: config (dict[str, Any]): Configuration dictionary for the optimizer. console (Console): Rich console object for pretty printing. llm_client (LLMClient): Client for interacting with a language model. _run_operation (Callable): Function to run an operation. max_threads (int): Maximum number of threads to use for parallel processing. num_fold_prompts (int): Number of fold prompts to generate. num_samples_in_validation (int): Number of samples to use in validation. """ def __init__( self, runner, run_operation: Callable, num_fold_prompts: int = 1, num_samples_in_validation: int = 10, ): """ Initialize the ReduceOptimizer. Args: config (dict[str, Any]): Configuration dictionary for the optimizer. console (Console): Rich console object for pretty printing. llm_client (LLMClient): Client for interacting with a language model. max_threads (int): Maximum number of threads to use for parallel processing. run_operation (Callable): Function to run an operation. num_fold_prompts (int, optional): Number of fold prompts to generate. Defaults to 1. num_samples_in_validation (int, optional): Number of samples to use in validation. Defaults to 10. """ self.runner = runner self.config = self.runner.config self.console = self.runner.console self.llm_client = self.runner.optimizer.llm_client self._run_operation = run_operation self.max_threads = self.runner.max_threads self.num_fold_prompts = num_fold_prompts self.num_samples_in_validation = num_samples_in_validation self.status = self.runner.status ``` -------------------------------- ### Define Complex Schema in DocETL Source: https://ucbepic.github.io/docetl/concepts/schemas This example demonstrates a complex schema definition in DocETL, featuring a list of objects for insights and a nested object for metadata. ```yaml output: schema: insights: "list[{insight: string, confidence: number}]" metadata: "{timestamp: string, source: string}" ```