Implementation:Ucbepic Docetl ExperimentSimpleAgent

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

Overview

Concrete tool for a simple baseline LLM agent that generates and evaluates DocETL pipelines for reasoning experiments provided by DocETL.

Description

The run_simple_agent module implements a baseline agent for comparison against the MOAR optimizer. The agent uses an iterative loop where it reads operator documentation via DataAnalyzer, examines sample data, generates pipeline operator configurations through LLM tool calls via AgentCommunicator, executes pipelines using PipelineExecutor, evaluates results, and decides whether to refine the pipeline or return it as final via the AgentAction schema. The module includes PathResolver for local/Modal path handling, supports remote execution on Modal, and tracks cumulative LLM costs. It serves as a controlled comparison baseline against the MOAR tree-search approach.

Usage

Use this module to run baseline experiments comparing a simple iterative LLM agent against the MOAR optimizer for DocETL pipeline generation. It is designed for research evaluation across standard benchmark datasets.

Code Reference

Source Location

• Repository: Ucbepic_Docetl
• File: experiments/reasoning/run_simple_agent.py
• Lines: 1-986

Signature

DEFAULT_MODEL = "gpt-5"
DEFAULT_OUTPUT_DIR = "outputs/simple_agent"

class AgentAction(BaseModel):
    action: Literal["try_pipeline", "return_pipeline"]
    reasoning: str

class PathResolver:
    @staticmethod
    def resolve_in_volume(path: str | None) -> str | None: ...
    @staticmethod
    def get_data_path(dataset: str) -> Path: ...

class PipelineExecutor:
    def __init__(self, experiment_dir: Path): ...
    def create_yaml(self, operators: List[Dict], dataset: str, prefix: str = "pipeline") -> str: ...
    def rewrite_for_modal(self, yaml_path: str) -> str: ...
    def execute(self, operators: List[Dict], dataset: str, prefix: str = "test_pipeline") -> Dict[str, Any]: ...

class DataAnalyzer:
    @staticmethod
    def load_documentation(doc_path: str = "...") -> str: ...
    @staticmethod
    def load_sample_data(dataset: str, limit: int = 5) -> List[Dict]: ...
    @staticmethod
    def analyze_sample_data(sample_data: List[Dict]) -> str: ...

class AgentCommunicator:
    def __init__(self, model: str, agent=None): ...
    def safe_json_parse(self, response_content: str, fallback: Dict = None) -> Dict: ...
    def get_action_decision(self, messages: List[Dict]) -> Optional[AgentAction]: ...
    def get_operators(self, messages: List[Dict], request_msg: str) -> List[Dict]: ...

Import

from experiments.reasoning.run_simple_agent import (
    PipelineExecutor,
    DataAnalyzer,
    AgentCommunicator,
    AgentAction,
)

I/O Contract

Inputs

Name	Type	Required	Description
dataset	str	Yes	Dataset name (e.g., "cuad", "blackvault", "game_reviews")
model	str	No	LLM model for the agent (default: "gpt-5")
experiment_dir	Path	Yes	Directory for saving pipeline YAMLs and outputs
operators	List[Dict]	Yes	List of operator configuration dicts to build a pipeline
doc_path	str	No	Path to operator documentation file
limit	int	No	Maximum number of sample data items to load (default: 5)

Outputs

Name	Type	Description
execution_result	Dict[str, Any]	Pipeline execution result with success, cost, sample_outputs, error, output_path
action	AgentAction	Agent's decision (try_pipeline or return_pipeline) with reasoning
operators	List[Dict]	Generated operator configurations from LLM
analysis	str	Formatted analysis of sample data structure

Usage Examples

from experiments.reasoning.run_simple_agent import PipelineExecutor, DataAnalyzer
from pathlib import Path

# Set up the experiment
experiment_dir = Path("outputs/simple_agent/cuad_run1")
experiment_dir.mkdir(parents=True, exist_ok=True)

# Load and analyze data
sample = DataAnalyzer.load_sample_data("cuad", limit=5)
analysis = DataAnalyzer.analyze_sample_data(sample)
print(analysis)

# Execute a pipeline with operators
executor = PipelineExecutor(experiment_dir)
result = executor.execute(
    operators=[{
        "name": "extract_clauses",
        "type": "map",
        "model": "gpt-4o-mini",
        "prompt": "Extract legal clauses from {{ input.text }}",
        "output": {"schema": {"clauses": "list[string]"}},
    }],
    dataset="cuad",
)

if result["success"]:
    print(f"Pipeline cost: ${result['cost']:.2f}")

Related Pages

• Environment:Ucbepic_Docetl_Python_Runtime