corpus_level

corpus_level

Corpus-level checklist generators (one checklist for entire dataset).

InductiveGenerator

Bases: CorpusChecklistGenerator

Generator that induces checklists from observations.

Takes a collection of evaluative observations (e.g., reviewer feedback, user complaints, quality notes, strengths/weaknesses) and generates a comprehensive yes/no checklist that addresses them.

The pipeline applies multiple refinement steps: - Deduplication (merge similar questions) - Tagging (filter for applicability) - Selection (beam search for diverse subset)

Source code in autochecklist/generators/corpus_level/inductive.py

class InductiveGenerator(CorpusChecklistGenerator):
    """Generator that induces checklists from observations.

    Takes a collection of evaluative observations (e.g., reviewer feedback,
    user complaints, quality notes, strengths/weaknesses) and generates a
    comprehensive yes/no checklist that addresses them.

    The pipeline applies multiple refinement steps:
    - Deduplication (merge similar questions)
    - Tagging (filter for applicability)
    - Selection (beam search for diverse subset)
    """

    @property
    def method_name(self) -> str:
        return "feedback"

    def __init__(
        self,
        model: Optional[str] = None,
        temperature: float = 0.7,
        api_key: Optional[str] = None,
        # Refinement parameters
        dedup_threshold: float = 0.85,
        max_questions: int = 20,
        beam_width: int = 5,
        # Batching
        batch_size: int = 100,
        # Unit testing
        max_unit_test_references: int = 20,
        # API keys
        embedding_api_key: Optional[str] = None,
        # Custom prompt
        custom_prompt: Optional[Union[str, Path]] = None,
        **kwargs,
    ):
        super().__init__(model=model, api_key=api_key, **kwargs)
        self.temperature = temperature
        self.dedup_threshold = dedup_threshold
        self.max_questions = max_questions
        self.beam_width = beam_width
        self.batch_size = batch_size
        self.max_unit_test_references = max_unit_test_references
        self.embedding_api_key = embedding_api_key

        # Load generation prompt template (custom or default)
        if custom_prompt is not None:
            if isinstance(custom_prompt, Path):
                template_str = custom_prompt.read_text(encoding="utf-8")
            else:
                template_str = custom_prompt
        else:
            template_str = load_template("generators/feedback", "generate")
        self._generate_template = PromptTemplate(template_str)

    def generate(
        self,
        observations: List[str],
        domain: str = "general responses",
        skip_dedup: bool = False,
        skip_tagging: bool = False,
        skip_selection: bool = False,
        skip_unit_testing: bool = True,
        references: Optional[List[str]] = None,
        verbose: bool = False,
        **kwargs: Any,
    ) -> Checklist:
        """Generate a checklist from observations.

        Args:
            observations: List of evaluative observation strings (feedback,
                review comments, quality notes, etc.)
            domain: Domain description for the prompt
            skip_dedup: Skip deduplication step
            skip_tagging: Skip tagging/filtering step
            skip_selection: Skip subset selection step
            skip_unit_testing: Skip unit testing step (default True)
            references: Optional reference targets for unit testing
            verbose: Print progress at each pipeline stage
            **kwargs: Additional arguments

        Returns:
            Generated and refined Checklist
        """
        if not observations:
            return Checklist(
                items=[],
                source_method="feedback",
                generation_level="corpus",
                metadata={"observation_count": 0},
            )

        # Step 1: Generate initial questions from observations
        raw_questions = self._generate_questions(observations, domain=domain)

        # Convert to checklist items
        items = []
        for i, q in enumerate(raw_questions):
            items.append(
                ChecklistItem(
                    id=f"fb-{i}",
                    question=q["question"],
                    metadata={
                        "source_feedback_indices": q.get("source_feedback_indices", []),
                    },
                )
            )

        checklist = Checklist(
            items=items,
            source_method="feedback",
            generation_level="corpus",
            metadata={
                "observation_count": len(observations),
                "raw_question_count": len(items),
            },
        )

        if verbose:
            print(f"[InductiveGenerator] Generated {len(items)} raw questions from {len(observations)} observations")

        # Step 2: Deduplicate
        if not skip_dedup and len(checklist.items) > 1:
            before_count = len(checklist.items)
            dedup = Deduplicator(
                similarity_threshold=self.dedup_threshold,
                model=self.model,
                api_key=self.api_key,
                embedding_api_key=self.embedding_api_key,
            )
            checklist = dedup.refine(checklist)
            if verbose:
                after_count = len(checklist.items)
                clusters_merged = checklist.metadata.get("clusters_merged", 0)
                print(f"[InductiveGenerator] Deduplication: {before_count} → {after_count} questions ({clusters_merged} clusters merged)")

        # Step 3: Tag and filter
        if not skip_tagging and len(checklist.items) > 0:
            before_count = len(checklist.items)
            tagger = Tagger(
                model=self.model,
                api_key=self.api_key,
            )
            checklist = tagger.refine(checklist)
            if verbose:
                after_count = len(checklist.items)
                filtered_count = checklist.metadata.get("filtered_count", 0)
                print(f"[InductiveGenerator] Tagging: {before_count} → {after_count} questions ({filtered_count} filtered out)")

        # Step 4: Unit test (optional, requires references)
        if not skip_unit_testing and references is not None and len(checklist.items) > 0:
            import random as _random
            before_count = len(checklist.items)
            refs = references
            if len(refs) > self.max_unit_test_references:
                _random.seed(0)
                refs = _random.sample(refs, self.max_unit_test_references)
            raw_samples = [{"id": f"ref-{i}", "text": r} for i, r in enumerate(refs)]
            unit_tester = UnitTester(
                model=self.model,
                api_key=self.api_key,
            )
            checklist = unit_tester.refine(checklist, raw_samples=raw_samples)
            if verbose:
                after_count = len(checklist.items)
                print(f"[InductiveGenerator] Unit testing: {before_count} → {after_count} questions")

        # Step 5: Select diverse subset
        if not skip_selection and len(checklist.items) > self.max_questions:
            before_count = len(checklist.items)
            selector = Selector(
                max_questions=self.max_questions,
                beam_width=self.beam_width,
                embedding_api_key=self.embedding_api_key,
                observations=observations,
            )
            checklist = selector.refine(checklist)
            if verbose:
                after_count = len(checklist.items)
                diversity_score = checklist.metadata.get("diversity_score", 0)
                print(f"[InductiveGenerator] Selection: {before_count} → {after_count} questions (diversity={diversity_score:.3f})")

        if verbose:
            print(f"[InductiveGenerator] Final checklist: {len(checklist.items)} questions")

        return checklist

    def _generate_questions(
        self,
        observations: List[str],
        domain: str = "general responses",
        verbose: bool = False,
    ) -> List[Dict[str, Any]]:
        """Generate questions from observations using LLM, batched for scalability.

        Splits observations into batches of ``self.batch_size`` items, makes one
        LLM call per batch, and pools results. Observation indices in each batch
        prompt use **global** numbering so ``source_feedback_indices`` are
        correct across batches.

        Args:
            observations: List of observation strings
            domain: Domain description
            verbose: Print per-batch progress

        Returns:
            List of question dicts with 'question' and 'source_feedback_indices'
        """
        response_format = {
            "type": "json_schema",
            "json_schema": {
                "name": "generated_questions",
                "strict": True,
                "schema": {
                    "type": "object",
                    "properties": {
                        "questions": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "question": {"type": "string"},
                                    "source_feedback_indices": {
                                        "type": "array",
                                        "items": {"type": "integer"},
                                    },
                                },
                                "required": ["question", "source_feedback_indices"],
                                "additionalProperties": False,
                            },
                        },
                    },
                    "required": ["questions"],
                    "additionalProperties": False,
                },
            },
        }

        client = self._get_or_create_client()
        n_batches = math.ceil(len(observations) / self.batch_size)
        all_questions: List[Dict[str, Any]] = []

        for batch_idx in range(n_batches):
            start = batch_idx * self.batch_size
            end = min(start + self.batch_size, len(observations))
            batch = observations[start:end]

            # Format with global indices
            feedback_text = "\n".join(
                f"[{start + i}] {f}" for i, f in enumerate(batch)
            )

            prompt = self._generate_template.format(
                domain=domain,
                feedback=feedback_text,
            )

            response = client.chat_completion(
                model=self.model or "openai/gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                temperature=self.temperature,
                max_tokens=self.max_tokens,
                response_format=response_format,
            )

            content = response["choices"][0]["message"]["content"]

            try:
                result = json.loads(content)
                batch_questions = result.get("questions", [])
            except json.JSONDecodeError:
                batch_questions = []

            all_questions.extend(batch_questions)

            if verbose:
                print(
                    f"[InductiveGenerator] Batch {batch_idx + 1}/{n_batches}: "
                    f"items [{start}..{end - 1}] → {len(batch_questions)} questions"
                )

        return all_questions

generate(observations, domain='general responses', skip_dedup=False, skip_tagging=False, skip_selection=False, skip_unit_testing=True, references=None, verbose=False, **kwargs)

Generate a checklist from observations.

Parameters:

Name	Type	Description	Default
observations	List[str]	List of evaluative observation strings (feedback, review comments, quality notes, etc.)	required
domain	str	Domain description for the prompt	'general responses'
skip_dedup	bool	Skip deduplication step	False
skip_tagging	bool	Skip tagging/filtering step	False
skip_selection	bool	Skip subset selection step	False
skip_unit_testing	bool	Skip unit testing step (default True)	True
references	Optional[List[str]]	Optional reference targets for unit testing	None
verbose	bool	Print progress at each pipeline stage	False
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
Checklist	Generated and refined Checklist

Source code in autochecklist/generators/corpus_level/inductive.py

def generate(
    self,
    observations: List[str],
    domain: str = "general responses",
    skip_dedup: bool = False,
    skip_tagging: bool = False,
    skip_selection: bool = False,
    skip_unit_testing: bool = True,
    references: Optional[List[str]] = None,
    verbose: bool = False,
    **kwargs: Any,
) -> Checklist:
    """Generate a checklist from observations.

    Args:
        observations: List of evaluative observation strings (feedback,
            review comments, quality notes, etc.)
        domain: Domain description for the prompt
        skip_dedup: Skip deduplication step
        skip_tagging: Skip tagging/filtering step
        skip_selection: Skip subset selection step
        skip_unit_testing: Skip unit testing step (default True)
        references: Optional reference targets for unit testing
        verbose: Print progress at each pipeline stage
        **kwargs: Additional arguments

    Returns:
        Generated and refined Checklist
    """
    if not observations:
        return Checklist(
            items=[],
            source_method="feedback",
            generation_level="corpus",
            metadata={"observation_count": 0},
        )

    # Step 1: Generate initial questions from observations
    raw_questions = self._generate_questions(observations, domain=domain)

    # Convert to checklist items
    items = []
    for i, q in enumerate(raw_questions):
        items.append(
            ChecklistItem(
                id=f"fb-{i}",
                question=q["question"],
                metadata={
                    "source_feedback_indices": q.get("source_feedback_indices", []),
                },
            )
        )

    checklist = Checklist(
        items=items,
        source_method="feedback",
        generation_level="corpus",
        metadata={
            "observation_count": len(observations),
            "raw_question_count": len(items),
        },
    )

    if verbose:
        print(f"[InductiveGenerator] Generated {len(items)} raw questions from {len(observations)} observations")

    # Step 2: Deduplicate
    if not skip_dedup and len(checklist.items) > 1:
        before_count = len(checklist.items)
        dedup = Deduplicator(
            similarity_threshold=self.dedup_threshold,
            model=self.model,
            api_key=self.api_key,
            embedding_api_key=self.embedding_api_key,
        )
        checklist = dedup.refine(checklist)
        if verbose:
            after_count = len(checklist.items)
            clusters_merged = checklist.metadata.get("clusters_merged", 0)
            print(f"[InductiveGenerator] Deduplication: {before_count} → {after_count} questions ({clusters_merged} clusters merged)")

    # Step 3: Tag and filter
    if not skip_tagging and len(checklist.items) > 0:
        before_count = len(checklist.items)
        tagger = Tagger(
            model=self.model,
            api_key=self.api_key,
        )
        checklist = tagger.refine(checklist)
        if verbose:
            after_count = len(checklist.items)
            filtered_count = checklist.metadata.get("filtered_count", 0)
            print(f"[InductiveGenerator] Tagging: {before_count} → {after_count} questions ({filtered_count} filtered out)")

    # Step 4: Unit test (optional, requires references)
    if not skip_unit_testing and references is not None and len(checklist.items) > 0:
        import random as _random
        before_count = len(checklist.items)
        refs = references
        if len(refs) > self.max_unit_test_references:
            _random.seed(0)
            refs = _random.sample(refs, self.max_unit_test_references)
        raw_samples = [{"id": f"ref-{i}", "text": r} for i, r in enumerate(refs)]
        unit_tester = UnitTester(
            model=self.model,
            api_key=self.api_key,
        )
        checklist = unit_tester.refine(checklist, raw_samples=raw_samples)
        if verbose:
            after_count = len(checklist.items)
            print(f"[InductiveGenerator] Unit testing: {before_count} → {after_count} questions")

    # Step 5: Select diverse subset
    if not skip_selection and len(checklist.items) > self.max_questions:
        before_count = len(checklist.items)
        selector = Selector(
            max_questions=self.max_questions,
            beam_width=self.beam_width,
            embedding_api_key=self.embedding_api_key,
            observations=observations,
        )
        checklist = selector.refine(checklist)
        if verbose:
            after_count = len(checklist.items)
            diversity_score = checklist.metadata.get("diversity_score", 0)
            print(f"[InductiveGenerator] Selection: {before_count} → {after_count} questions (diversity={diversity_score:.3f})")

    if verbose:
        print(f"[InductiveGenerator] Final checklist: {len(checklist.items)} questions")

    return checklist

DeductiveGenerator

Bases: CorpusChecklistGenerator

Generate checklists from evaluation dimension definitions.

CheckEval creates binary yes/no evaluation questions organized by dimension and sub-dimension. Questions can be provided as seeds or generated from dimension definitions.

Parameters:

Name	Type	Description	Default
model	Optional[str]	OpenRouter model ID for generation	None
augmentation_mode	str | AugmentationMode	One of seed, elaboration, or diversification	SEED
task_type	str	Type of task being evaluated (e.g., "summarization", "dialog")	'general'

Example

checkeval = DeductiveGenerator(model="openai/gpt-4o-mini") dimensions = [ ... DeductiveInput( ... name="coherence", ... definition="The response should maintain logical flow.", ... sub_dimensions=["Logical Flow", "Consistency"] ... ) ... ] checklist = checkeval.generate(dimensions=dimensions)

Source code in autochecklist/generators/corpus_level/deductive.py

class DeductiveGenerator(CorpusChecklistGenerator):
    """Generate checklists from evaluation dimension definitions.

    CheckEval creates binary yes/no evaluation questions organized by
    dimension and sub-dimension. Questions can be provided as seeds
    or generated from dimension definitions.

    Args:
        model: OpenRouter model ID for generation
        augmentation_mode: One of seed, elaboration, or diversification
        task_type: Type of task being evaluated (e.g., "summarization", "dialog")

    Example:
        >>> checkeval = DeductiveGenerator(model="openai/gpt-4o-mini")
        >>> dimensions = [
        ...     DeductiveInput(
        ...         name="coherence",
        ...         definition="The response should maintain logical flow.",
        ...         sub_dimensions=["Logical Flow", "Consistency"]
        ...     )
        ... ]
        >>> checklist = checkeval.generate(dimensions=dimensions)
    """

    def __init__(
        self,
        model: Optional[str] = None,
        augmentation_mode: str | AugmentationMode = AugmentationMode.SEED,
        task_type: str = "general",
        api_key: Optional[str] = None,
        **kwargs,
    ):
        super().__init__(model=model, api_key=api_key, **kwargs)

        # Handle string augmentation mode
        if isinstance(augmentation_mode, str):
            augmentation_mode = AugmentationMode(augmentation_mode.lower())
        self.augmentation_mode = augmentation_mode

        self.task_type = task_type

        # Load prompt templates
        self._generate_template = PromptTemplate(load_template("generators/checkeval", "generate"))
        self._augment_template = PromptTemplate(load_template("generators/checkeval", "augment"))
        self._filter_template = PromptTemplate(load_template("generators/checkeval", "filter"))

    @property
    def method_name(self) -> str:
        """Return the method name for this generator."""
        return "checkeval"

    def generate(
        self,
        dimensions: List[DeductiveInput],
        seed_questions: Optional[Dict[str, Dict[str, List[str]]]] = None,
        augment: bool = True,
        max_questions: Optional[int] = None,
        apply_filtering: bool = False,
        verbose: bool = False,
        **kwargs: Any,
    ) -> Checklist:
        """Generate a checklist from dimension definitions.

        Args:
            dimensions: List of DeductiveInput with name, definition, sub_dimensions
            seed_questions: Optional pre-defined questions by dimension/sub-aspect
                Format: {dimension: {sub_aspect: [questions]}}
            augment: Whether to augment seed questions (default True)
            max_questions: Maximum number of questions to include
            apply_filtering: Whether to apply Tagger and Deduplicator refiners (default False)
            verbose: Print progress at each stage (default False)
            **kwargs: Additional parameters passed to generation

        Returns:
            Checklist with generated questions
        """
        if not dimensions:
            return Checklist(
                id=str(uuid.uuid4()),
                items=[],
                source_method="checkeval",
                generation_level="corpus",
                metadata={
                    "dimension_count": 0,
                    "dimensions": [],
                    "augmentation_mode": self.augmentation_mode.value,
                },
            )

        all_questions = []

        for dimension in dimensions:
            # Check if seed questions provided for this dimension
            dim_seed_questions = (
                seed_questions.get(dimension.name) if seed_questions else None
            )

            if dim_seed_questions and not augment:
                # Use seed questions directly without augmentation
                questions = self._convert_seed_to_questions(
                    dim_seed_questions, dimension.name
                )
            elif self.augmentation_mode == AugmentationMode.COMBINED:
                # Paper-faithful: generate seeds, then run both augmentation
                # modes independently, then merge all three pools
                if dim_seed_questions:
                    seeds = self._convert_seed_to_questions(
                        dim_seed_questions, dimension.name
                    )
                else:
                    seeds = self._generate_questions(
                        dimension,
                        mode=AugmentationMode.SEED,
                        task_type=self.task_type,
                    )

                # Tag seeds
                for q in seeds:
                    q["augmentation_source"] = "seed"

                # Run elaboration independently from seeds
                elaborated = self._augment_questions(
                    seeds, dimension,
                    mode=AugmentationMode.ELABORATION,
                    task_type=self.task_type,
                )
                for q in elaborated:
                    q.setdefault("augmentation_source", "elaboration")

                # Run diversification independently from seeds
                diversified = self._augment_questions(
                    seeds, dimension,
                    mode=AugmentationMode.DIVERSIFICATION,
                    task_type=self.task_type,
                )
                for q in diversified:
                    q.setdefault("augmentation_source", "diversification")

                # Merge all three pools, deduplicating by question text
                questions = self._merge_and_deduplicate(seeds, elaborated, diversified)
            elif dim_seed_questions and augment:
                # Augment from seed questions
                seed_list = self._convert_seed_to_questions(
                    dim_seed_questions, dimension.name
                )
                questions = self._augment_questions(
                    seed_list,
                    dimension,
                    mode=self.augmentation_mode,
                    task_type=self.task_type,
                )
            else:
                # Generate questions from dimension definition
                questions = self._generate_questions(
                    dimension,
                    mode=self.augmentation_mode,
                    task_type=self.task_type,
                )

            all_questions.extend(questions)

        # Apply max_questions limit if specified
        if max_questions and len(all_questions) > max_questions:
            all_questions = all_questions[:max_questions]

        # Convert to ChecklistItems
        items = []
        for q in all_questions:
            meta = {
                "dimension": q.get("dimension"),
                "sub_aspect": q.get("sub_aspect"),
            }
            if "augmentation_source" in q:
                meta["augmentation_source"] = q["augmentation_source"]
            item = ChecklistItem(
                id=str(uuid.uuid4()),
                question=q["question"],
                category=q.get("dimension"),
                metadata=meta,
            )
            items.append(item)

        checklist = Checklist(
            id=str(uuid.uuid4()),
            items=items,
            source_method="checkeval",
            generation_level="corpus",
            metadata={
                "dimension_count": len(dimensions),
                "dimensions": [d.name for d in dimensions],
                "augmentation_mode": self.augmentation_mode.value,
                "task_type": self.task_type,
            },
        )

        if verbose:
            print(f"[DeductiveGenerator] Generated {len(checklist.items)} questions from {len(dimensions)} dimensions")

        # Apply filtering if enabled (CheckEval paper §4.2)
        if apply_filtering and len(checklist.items) > 0:
            # Build dimension lookup for consistency checks
            dim_lookup = {d.name: d.definition for d in dimensions}

            # Stage 1 & 2: Alignment + Dimension Consistency (CheckEval-specific)
            before_count = len(checklist.items)
            filtered_items, filter_stats = self._filter_questions(
                checklist.items, dim_lookup, verbose
            )

            # Update checklist with filtered items
            checklist = Checklist(
                id=checklist.id,
                items=filtered_items,
                source_method=checklist.source_method,
                generation_level=checklist.generation_level,
                input=checklist.input,
                metadata={
                    **checklist.metadata,
                    "alignment_filtered": filter_stats["alignment_filtered"],
                    "consistency_filtered": filter_stats["consistency_filtered"],
                },
            )

            if verbose:
                after_count = len(checklist.items)
                print(
                    f"[DeductiveGenerator] Filtering: {before_count} → {after_count} questions "
                    f"({filter_stats['alignment_filtered']} alignment, "
                    f"{filter_stats['consistency_filtered']} consistency)"
                )

            # Stage 3: Redundancy Removal (via Deduplicator)
            if len(checklist.items) > 1:
                before_count = len(checklist.items)
                dedup = Deduplicator(
                    similarity_threshold=0.85,
                    model=self.model,
                    api_key=self.api_key,
                )
                checklist = dedup.refine(checklist)
                if verbose:
                    after_count = len(checklist.items)
                    merged = checklist.metadata.get("clusters_merged", 0)
                    print(f"[DeductiveGenerator] Deduplication: {before_count} → {after_count} questions ({merged} clusters merged)")

        if verbose:
            print(f"[DeductiveGenerator] Final checklist: {len(checklist.items)} questions")

        return checklist

    def generate_grouped(
        self,
        dimensions: List[DeductiveInput],
        **kwargs: Any,
    ) -> Dict[str, "Checklist"]:
        """Generate a checklist and split it by dimension category.

        Convenience wrapper around ``generate().by_category()``.

        Args:
            dimensions: List of DeductiveInput with name, definition, sub_dimensions
            **kwargs: Additional arguments passed to generate()

        Returns:
            Dict mapping dimension name to sub-Checklist
        """
        checklist = self.generate(dimensions=dimensions, **kwargs)
        return checklist.by_category()

    def _generate_questions(
        self,
        dimension: DeductiveInput,
        mode: AugmentationMode,
        task_type: str,
    ) -> List[Dict[str, Any]]:
        """Generate questions from a dimension definition.

        Args:
            dimension: Dimension with name, definition, sub_dimensions
            mode: Augmentation mode (affects how many questions)
            task_type: Type of task being evaluated

        Returns:
            List of question dicts with question, sub_aspect, dimension
        """
        # Determine questions per sub-dimension based on mode
        questions_per_sub = {
            AugmentationMode.SEED: 2,
            AugmentationMode.ELABORATION: 5,
            AugmentationMode.DIVERSIFICATION: 4,
            AugmentationMode.COMBINED: 2,  # Seeds only; augmentation happens separately
        }.get(mode, 2)

        # Format sub-dimensions
        sub_dims = dimension.sub_dimensions or ["General"]
        sub_dims_text = "\n".join(f"- {sd}" for sd in sub_dims)

        # Render prompt
        prompt = self._generate_template.format(
            task_type=task_type,
            dimension_name=dimension.name,
            definition=dimension.definition,
            sub_dimensions=sub_dims_text,
            questions_per_sub=questions_per_sub,
        )

        # Call LLM (request JSON output for reliable parsing with reasoning models)
        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        # Parse response
        content = response["choices"][0]["message"]["content"]
        questions = self._parse_questions_response(content)

        # Ensure dimension is set
        for q in questions:
            q["dimension"] = dimension.name
            if "sub_aspect" not in q:
                q["sub_aspect"] = "General"

        return questions

    def _augment_questions(
        self,
        existing_questions: List[Dict[str, Any]],
        dimension: DeductiveInput,
        mode: AugmentationMode,
        task_type: str,
    ) -> List[Dict[str, Any]]:
        """Augment existing questions using the specified mode.

        Args:
            existing_questions: List of existing question dicts
            dimension: Dimension definition
            mode: Augmentation mode
            task_type: Type of task being evaluated

        Returns:
            List of augmented question dicts
        """
        # Format existing questions
        existing_text = "\n".join(
            f"- [{q.get('sub_aspect', 'General')}] {q['question']}"
            for q in existing_questions
        )

        # Get augmentation instructions
        instructions = AUGMENTATION_INSTRUCTIONS.get(mode, "")

        # Render prompt
        prompt = self._augment_template.format(
            task_type=task_type,
            dimension_name=dimension.name,
            definition=dimension.definition,
            existing_questions=existing_text,
            augmentation_mode=mode.value,
            augmentation_instructions=instructions,
        )

        # Call LLM (request JSON output for reliable parsing with reasoning models)
        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        # Parse response
        content = response["choices"][0]["message"]["content"]
        questions = self._parse_questions_response(content)

        # Ensure dimension is set
        for q in questions:
            q["dimension"] = dimension.name
            if "sub_aspect" not in q:
                q["sub_aspect"] = "General"

        return questions

    def _convert_seed_to_questions(
        self,
        seed_dict: Dict[str, List[str]],
        dimension_name: str,
    ) -> List[Dict[str, Any]]:
        """Convert seed question dict to list of question dicts.

        Args:
            seed_dict: {sub_aspect: [questions]}
            dimension_name: Name of the dimension

        Returns:
            List of question dicts
        """
        questions = []
        for sub_aspect, question_list in seed_dict.items():
            for q in question_list:
                questions.append(
                    {
                        "question": q,
                        "sub_aspect": sub_aspect,
                        "dimension": dimension_name,
                    }
                )
        return questions

    @staticmethod
    def _merge_and_deduplicate(
        *pools: List[Dict[str, Any]],
    ) -> List[Dict[str, Any]]:
        """Merge multiple question pools, deduplicating by question text.

        Questions from earlier pools take priority (seeds first, then
        elaborated, then diversified). This preserves the augmentation_source
        tag from the first occurrence.

        Args:
            *pools: Variable number of question lists to merge

        Returns:
            Deduplicated list of question dicts
        """
        seen = set()
        merged = []
        for pool in pools:
            for q in pool:
                text = q["question"].strip()
                if text not in seen:
                    seen.add(text)
                    merged.append(q)
        return merged

    def _parse_questions_response(self, content: str) -> List[Dict[str, Any]]:
        """Parse LLM response to extract questions.

        Args:
            content: Raw LLM response

        Returns:
            List of question dicts
        """
        # Try to find JSON in the response
        try:
            # Look for JSON block
            if "```json" in content:
                start = content.find("```json") + 7
                end = content.find("```", start)
                json_str = content[start:end].strip()
            elif "```" in content:
                start = content.find("```") + 3
                end = content.find("```", start)
                json_str = content[start:end].strip()
            elif "{" in content:
                # Find JSON object
                start = content.find("{")
                end = content.rfind("}") + 1
                json_str = content[start:end]
            else:
                return []

            data = json.loads(json_str)

            if isinstance(data, dict) and "questions" in data:
                return data["questions"]
            elif isinstance(data, list):
                return data
            else:
                return []

        except (json.JSONDecodeError, ValueError):
            return []

    def _filter_questions(
        self,
        items: List[ChecklistItem],
        dim_lookup: Dict[str, str],
        verbose: bool = False,
    ) -> tuple[List[ChecklistItem], Dict[str, int]]:
        """Filter questions using CheckEval-specific criteria.

        Implements the CheckEval paper's filtering stages:
        1. Alignment: "YES" should indicate higher quality
        2. Dimension Consistency: Question matches its dimension definition

        Args:
            items: List of checklist items to filter
            dim_lookup: Dict mapping dimension name to definition
            verbose: Print per-item filtering details

        Returns:
            Tuple of (filtered items, stats dict)
        """
        filtered_items = []
        stats = {"alignment_filtered": 0, "consistency_filtered": 0}

        client = self._get_or_create_client()
        for item in items:
            # Get dimension info from item metadata
            dimension_name = item.metadata.get("dimension", "unknown") if item.metadata else "unknown"
            dimension_def = dim_lookup.get(dimension_name, "General evaluation criterion")

            # Build filter prompt
            prompt = self._filter_template.format(
                dimension_name=dimension_name,
                dimension_definition=dimension_def,
                question=item.question,
            )

            response_format = {
                "type": "json_schema",
                "json_schema": {
                    "name": "filter_result",
                    "strict": True,
                    "schema": {
                        "type": "object",
                        "properties": {
                            "reasoning": {"type": "string"},
                            "alignment_pass": {"type": "boolean"},
                            "dimension_consistent": {"type": "boolean"},
                        },
                        "required": ["reasoning", "alignment_pass", "dimension_consistent"],
                        "additionalProperties": False,
                    },
                },
            }

            try:
                response = client.chat_completion(
                    model=self.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.0,
                    response_format=response_format,
                )
                result = json.loads(response["choices"][0]["message"]["content"])

                alignment_pass = result.get("alignment_pass", False)
                dimension_consistent = result.get("dimension_consistent", False)

                if alignment_pass and dimension_consistent:
                    # Question passes both checks
                    filtered_items.append(item)
                else:
                    # Track why it was filtered
                    if not alignment_pass:
                        stats["alignment_filtered"] += 1
                    if not dimension_consistent:
                        stats["consistency_filtered"] += 1

            except (json.JSONDecodeError, KeyError, TypeError):
                # On parse error, be conservative and keep the question
                filtered_items.append(item)

        return filtered_items, stats

method_name property

Return the method name for this generator.

generate(dimensions, seed_questions=None, augment=True, max_questions=None, apply_filtering=False, verbose=False, **kwargs)

Generate a checklist from dimension definitions.

Parameters:

Name	Type	Description	Default
dimensions	List[DeductiveInput]	List of DeductiveInput with name, definition, sub_dimensions	required
seed_questions	Optional[Dict[str, Dict[str, List[str]]]]	Optional pre-defined questions by dimension/sub-aspect Format: {dimension: {sub_aspect: [questions]}}	None
augment	bool	Whether to augment seed questions (default True)	True
max_questions	Optional[int]	Maximum number of questions to include	None
apply_filtering	bool	Whether to apply Tagger and Deduplicator refiners (default False)	False
verbose	bool	Print progress at each stage (default False)	False
**kwargs	Any	Additional parameters passed to generation	{}

Returns:

Type	Description
Checklist	Checklist with generated questions

Source code in autochecklist/generators/corpus_level/deductive.py

def generate(
    self,
    dimensions: List[DeductiveInput],
    seed_questions: Optional[Dict[str, Dict[str, List[str]]]] = None,
    augment: bool = True,
    max_questions: Optional[int] = None,
    apply_filtering: bool = False,
    verbose: bool = False,
    **kwargs: Any,
) -> Checklist:
    """Generate a checklist from dimension definitions.

    Args:
        dimensions: List of DeductiveInput with name, definition, sub_dimensions
        seed_questions: Optional pre-defined questions by dimension/sub-aspect
            Format: {dimension: {sub_aspect: [questions]}}
        augment: Whether to augment seed questions (default True)
        max_questions: Maximum number of questions to include
        apply_filtering: Whether to apply Tagger and Deduplicator refiners (default False)
        verbose: Print progress at each stage (default False)
        **kwargs: Additional parameters passed to generation

    Returns:
        Checklist with generated questions
    """
    if not dimensions:
        return Checklist(
            id=str(uuid.uuid4()),
            items=[],
            source_method="checkeval",
            generation_level="corpus",
            metadata={
                "dimension_count": 0,
                "dimensions": [],
                "augmentation_mode": self.augmentation_mode.value,
            },
        )

    all_questions = []

    for dimension in dimensions:
        # Check if seed questions provided for this dimension
        dim_seed_questions = (
            seed_questions.get(dimension.name) if seed_questions else None
        )

        if dim_seed_questions and not augment:
            # Use seed questions directly without augmentation
            questions = self._convert_seed_to_questions(
                dim_seed_questions, dimension.name
            )
        elif self.augmentation_mode == AugmentationMode.COMBINED:
            # Paper-faithful: generate seeds, then run both augmentation
            # modes independently, then merge all three pools
            if dim_seed_questions:
                seeds = self._convert_seed_to_questions(
                    dim_seed_questions, dimension.name
                )
            else:
                seeds = self._generate_questions(
                    dimension,
                    mode=AugmentationMode.SEED,
                    task_type=self.task_type,
                )

            # Tag seeds
            for q in seeds:
                q["augmentation_source"] = "seed"

            # Run elaboration independently from seeds
            elaborated = self._augment_questions(
                seeds, dimension,
                mode=AugmentationMode.ELABORATION,
                task_type=self.task_type,
            )
            for q in elaborated:
                q.setdefault("augmentation_source", "elaboration")

            # Run diversification independently from seeds
            diversified = self._augment_questions(
                seeds, dimension,
                mode=AugmentationMode.DIVERSIFICATION,
                task_type=self.task_type,
            )
            for q in diversified:
                q.setdefault("augmentation_source", "diversification")

            # Merge all three pools, deduplicating by question text
            questions = self._merge_and_deduplicate(seeds, elaborated, diversified)
        elif dim_seed_questions and augment:
            # Augment from seed questions
            seed_list = self._convert_seed_to_questions(
                dim_seed_questions, dimension.name
            )
            questions = self._augment_questions(
                seed_list,
                dimension,
                mode=self.augmentation_mode,
                task_type=self.task_type,
            )
        else:
            # Generate questions from dimension definition
            questions = self._generate_questions(
                dimension,
                mode=self.augmentation_mode,
                task_type=self.task_type,
            )

        all_questions.extend(questions)

    # Apply max_questions limit if specified
    if max_questions and len(all_questions) > max_questions:
        all_questions = all_questions[:max_questions]

    # Convert to ChecklistItems
    items = []
    for q in all_questions:
        meta = {
            "dimension": q.get("dimension"),
            "sub_aspect": q.get("sub_aspect"),
        }
        if "augmentation_source" in q:
            meta["augmentation_source"] = q["augmentation_source"]
        item = ChecklistItem(
            id=str(uuid.uuid4()),
            question=q["question"],
            category=q.get("dimension"),
            metadata=meta,
        )
        items.append(item)

    checklist = Checklist(
        id=str(uuid.uuid4()),
        items=items,
        source_method="checkeval",
        generation_level="corpus",
        metadata={
            "dimension_count": len(dimensions),
            "dimensions": [d.name for d in dimensions],
            "augmentation_mode": self.augmentation_mode.value,
            "task_type": self.task_type,
        },
    )

    if verbose:
        print(f"[DeductiveGenerator] Generated {len(checklist.items)} questions from {len(dimensions)} dimensions")

    # Apply filtering if enabled (CheckEval paper §4.2)
    if apply_filtering and len(checklist.items) > 0:
        # Build dimension lookup for consistency checks
        dim_lookup = {d.name: d.definition for d in dimensions}

        # Stage 1 & 2: Alignment + Dimension Consistency (CheckEval-specific)
        before_count = len(checklist.items)
        filtered_items, filter_stats = self._filter_questions(
            checklist.items, dim_lookup, verbose
        )

        # Update checklist with filtered items
        checklist = Checklist(
            id=checklist.id,
            items=filtered_items,
            source_method=checklist.source_method,
            generation_level=checklist.generation_level,
            input=checklist.input,
            metadata={
                **checklist.metadata,
                "alignment_filtered": filter_stats["alignment_filtered"],
                "consistency_filtered": filter_stats["consistency_filtered"],
            },
        )

        if verbose:
            after_count = len(checklist.items)
            print(
                f"[DeductiveGenerator] Filtering: {before_count} → {after_count} questions "
                f"({filter_stats['alignment_filtered']} alignment, "
                f"{filter_stats['consistency_filtered']} consistency)"
            )

        # Stage 3: Redundancy Removal (via Deduplicator)
        if len(checklist.items) > 1:
            before_count = len(checklist.items)
            dedup = Deduplicator(
                similarity_threshold=0.85,
                model=self.model,
                api_key=self.api_key,
            )
            checklist = dedup.refine(checklist)
            if verbose:
                after_count = len(checklist.items)
                merged = checklist.metadata.get("clusters_merged", 0)
                print(f"[DeductiveGenerator] Deduplication: {before_count} → {after_count} questions ({merged} clusters merged)")

    if verbose:
        print(f"[DeductiveGenerator] Final checklist: {len(checklist.items)} questions")

    return checklist

generate_grouped(dimensions, **kwargs)

Generate a checklist and split it by dimension category.

Convenience wrapper around generate().by_category().

Parameters:

Name	Type	Description	Default
dimensions	List[DeductiveInput]	List of DeductiveInput with name, definition, sub_dimensions	required
**kwargs	Any	Additional arguments passed to generate()	{}

Returns:

Type	Description
Dict[str, Checklist]	Dict mapping dimension name to sub-Checklist

Source code in autochecklist/generators/corpus_level/deductive.py

def generate_grouped(
    self,
    dimensions: List[DeductiveInput],
    **kwargs: Any,
) -> Dict[str, "Checklist"]:
    """Generate a checklist and split it by dimension category.

    Convenience wrapper around ``generate().by_category()``.

    Args:
        dimensions: List of DeductiveInput with name, definition, sub_dimensions
        **kwargs: Additional arguments passed to generate()

    Returns:
        Dict mapping dimension name to sub-Checklist
    """
    checklist = self.generate(dimensions=dimensions, **kwargs)
    return checklist.by_category()

AugmentationMode

Bases: str, Enum

Augmentation modes for question generation.

Source code in autochecklist/generators/corpus_level/deductive.py

class AugmentationMode(str, Enum):
    """Augmentation modes for question generation."""

    SEED = "seed"  # Minimal questions (1-3 per sub-dimension)
    ELABORATION = "elaboration"  # Expanded with more detail
    DIVERSIFICATION = "diversification"  # Alternative framings
    COMBINED = "combined"  # Both elaboration + diversification from seeds (paper-faithful)

InteractiveGenerator

Bases: CorpusChecklistGenerator

Generate checklists from interactive think-aloud attributes.

InteractEval takes pre-collected think-aloud attributes (considerations about evaluating a dimension) and transforms them through a 5-stage pipeline into a validated checklist of yes/no questions.

Parameters:

Name	Type	Description	Default
model	Optional[str]	OpenRouter model ID for generation	None
max_components	int	Maximum number of components to extract (default 5)	5

Example

interacteval = InteractiveGenerator(model="openai/gpt-4o-mini") input = InteractiveInput( ... source="human_llm", ... dimension="coherence", ... attributes=["Check for logical flow", "Ensure consistency"], ... ) checklist = interacteval.generate( ... inputs=[input], ... rubric="Coherence measures the logical flow and consistency...", ... )

Source code in autochecklist/generators/corpus_level/interactive.py

class InteractiveGenerator(CorpusChecklistGenerator):
    """Generate checklists from interactive think-aloud attributes.

    InteractEval takes pre-collected think-aloud attributes (considerations
    about evaluating a dimension) and transforms them through a 5-stage
    pipeline into a validated checklist of yes/no questions.

    Args:
        model: OpenRouter model ID for generation
        max_components: Maximum number of components to extract (default 5)

    Example:
        >>> interacteval = InteractiveGenerator(model="openai/gpt-4o-mini")
        >>> input = InteractiveInput(
        ...     source="human_llm",
        ...     dimension="coherence",
        ...     attributes=["Check for logical flow", "Ensure consistency"],
        ... )
        >>> checklist = interacteval.generate(
        ...     inputs=[input],
        ...     rubric="Coherence measures the logical flow and consistency...",
        ... )
    """

    def __init__(
        self,
        model: Optional[str] = None,
        max_components: int = 5,
        api_key: Optional[str] = None,
        **kwargs,
    ):
        super().__init__(model=model, api_key=api_key, **kwargs)
        self.max_components = max_components

        # Load prompt templates for each pipeline stage
        self._component_extraction_template = PromptTemplate(
            load_template("generators/interacteval", "component_extraction")
        )
        self._attributes_clustering_template = PromptTemplate(
            load_template("generators/interacteval", "attributes_clustering")
        )
        self._question_generation_template = PromptTemplate(
            load_template("generators/interacteval", "question_generation")
        )
        self._sub_question_generation_template = PromptTemplate(
            load_template("generators/interacteval", "sub_question_generation")
        )
        self._question_validation_template = PromptTemplate(
            load_template("generators/interacteval", "question_validation")
        )

    @property
    def method_name(self) -> str:
        """Return the method name for this generator."""
        return "interacteval"

    def generate(
        self,
        inputs: List[InteractiveInput],
        rubric: str = "",
        max_questions: Optional[int] = None,
        **kwargs: Any,
    ) -> Checklist:
        """Generate a checklist from think-aloud attributes.

        Args:
            inputs: List of InteractiveInput with attributes from human/LLM sources
            rubric: Definition/rubric for the evaluation dimension
            max_questions: Maximum number of questions to include
            **kwargs: Additional parameters

        Returns:
            Checklist with generated questions
        """
        if not inputs:
            return self._empty_checklist()

        # Combine attributes from all inputs
        all_attributes = []
        dimension = inputs[0].dimension
        source = inputs[0].source

        for inp in inputs:
            all_attributes.extend(inp.attributes)
            # Use the first non-None dimension
            if not dimension and inp.dimension:
                dimension = inp.dimension
            # Track combined sources
            if inp.source != source:
                source = "human_llm"

        if not all_attributes:
            return self._empty_checklist(dimension=dimension)

        # Run the 5-stage pipeline
        validated_questions = self._run_pipeline(
            all_attributes, rubric, dimension or "quality"
        )

        # Apply max_questions limit if specified
        if max_questions and len(validated_questions) > max_questions:
            validated_questions = validated_questions[:max_questions]

        # Convert to ChecklistItems
        items = []
        for q in validated_questions:
            item = ChecklistItem(
                id=str(uuid.uuid4()),
                question=q,
                category=dimension,
                metadata={
                    "dimension": dimension,
                },
            )
            items.append(item)

        return Checklist(
            id=str(uuid.uuid4()),
            items=items,
            source_method="interacteval",
            generation_level="corpus",
            metadata={
                "dimension": dimension,
                "attribute_count": len(all_attributes),
                "source": source,
                "rubric": rubric[:200] if rubric else None,  # Truncate for metadata
            },
        )

    def _run_pipeline(
        self,
        attributes: List[str],
        rubric: str,
        dimension: str,
    ) -> List[str]:
        """Run the 5-stage pipeline to generate validated questions.

        Args:
            attributes: List of think-aloud attributes
            rubric: Dimension definition/rubric
            dimension: Name of the evaluation dimension

        Returns:
            List of validated question strings
        """
        # Stage 1: Extract components
        components = self._extract_components(attributes, rubric, dimension)
        if not components:
            return []

        # Stage 2: Cluster attributes under components
        clustered = self._cluster_attributes(components, attributes)

        # Stage 3: Generate key questions
        key_questions = self._generate_key_questions(clustered, rubric, dimension)

        # Stage 4: Generate sub-questions
        sub_questions = self._generate_sub_questions(key_questions, rubric, dimension)

        # Stage 5: Validate and refine
        validated = self._validate_questions(sub_questions, rubric, dimension)

        return validated

    def _extract_components(
        self,
        attributes: List[str],
        rubric: str,
        dimension: str,
    ) -> List[str]:
        """Stage 1: Extract recurring components/themes from attributes."""
        attributes_text = "\n".join(f"- {attr}" for attr in attributes)

        prompt = self._component_extraction_template.format(
            dimension=dimension,
            rubric=rubric,
            attributes=attributes_text,
            max_components=self.max_components,
        )

        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        content = response["choices"][0]["message"]["content"]
        return self._parse_list_response(content)

    def _cluster_attributes(
        self,
        components: List[str],
        attributes: List[str],
    ) -> Dict[str, List[str]]:
        """Stage 2: Cluster attributes under components."""
        components_text = "\n".join(f"- {c}" for c in components)
        attributes_text = "\n".join(f"- {attr}" for attr in attributes)

        prompt = self._attributes_clustering_template.format(
            components=components_text,
            attributes=attributes_text,
        )

        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        content = response["choices"][0]["message"]["content"]
        return self._parse_dict_response(content)

    def _generate_key_questions(
        self,
        clustered: Dict[str, List[str]],
        rubric: str,
        dimension: str,
    ) -> Dict[str, str]:
        """Stage 3: Generate key questions for each component."""
        components_attributes_text = ""
        for component, attrs in clustered.items():
            components_attributes_text += f"\n**{component}**:\n"
            for attr in attrs:
                components_attributes_text += f"  - {attr}\n"

        prompt = self._question_generation_template.format(
            dimension=dimension,
            rubric=rubric,
            components_attributes=components_attributes_text,
        )

        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        content = response["choices"][0]["message"]["content"]
        return self._parse_dict_response(content)

    def _generate_sub_questions(
        self,
        key_questions: Dict[str, str],
        rubric: str,
        dimension: str,
    ) -> Dict[str, List[str]]:
        """Stage 4: Generate sub-questions for each key question."""
        key_questions_text = ""
        for component, question in key_questions.items():
            key_questions_text += f"\n**{component}**: {question}\n"

        prompt = self._sub_question_generation_template.format(
            dimension=dimension,
            rubric=rubric,
            key_questions=key_questions_text,
        )

        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            response_format={"type": "json_object"},
        )

        content = response["choices"][0]["message"]["content"]
        return self._parse_dict_list_response(content)

    def _validate_questions(
        self,
        sub_questions: Dict[str, List[str]],
        rubric: str,
        dimension: str,
    ) -> List[str]:
        """Stage 5: Validate and refine questions."""
        # Flatten all questions
        all_questions = []
        for component, questions in sub_questions.items():
            all_questions.extend(questions)

        all_questions_text = "\n".join(f"- {q}" for q in all_questions)

        prompt = self._question_validation_template.format(
            dimension=dimension,
            rubric=rubric,
            all_questions=all_questions_text,
        )

        client = self._get_or_create_client()
        response = client.chat_completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5,  # Lower temperature for validation
            response_format={"type": "json_object"},
        )

        content = response["choices"][0]["message"]["content"]
        return self._parse_list_response(content)

    def _empty_checklist(self, dimension: str = None) -> Checklist:
        """Return an empty checklist."""
        return Checklist(
            id=str(uuid.uuid4()),
            items=[],
            source_method="interacteval",
            generation_level="corpus",
            metadata={
                "dimension": dimension,
                "attribute_count": 0,
                "source": None,
            },
        )

    def _parse_list_response(self, content: str) -> List[str]:
        """Parse LLM response to extract a list of strings."""
        try:
            # Try to find JSON in the response
            if "```json" in content:
                start = content.find("```json") + 7
                end = content.find("```", start)
                json_str = content[start:end].strip()
            elif "```" in content:
                start = content.find("```") + 3
                end = content.find("```", start)
                json_str = content[start:end].strip()
            else:
                json_str = content.strip()

            data = json.loads(json_str)
            if isinstance(data, list):
                return [str(item) for item in data]
            # Handle JSON objects containing a list value (e.g., {"components": [...]})
            if isinstance(data, dict):
                for value in data.values():
                    if isinstance(value, list) and value:
                        return [str(item) for item in value]
            return []

        except (json.JSONDecodeError, ValueError):
            # Fallback: try to extract a JSON array via bracket matching
            if "[" in content:
                try:
                    start = content.find("[")
                    end = content.rfind("]") + 1
                    arr = json.loads(content[start:end])
                    if isinstance(arr, list):
                        return [str(item) for item in arr]
                except (json.JSONDecodeError, ValueError):
                    pass
            return []

    def _parse_dict_response(self, content: str) -> Dict[str, Any]:
        """Parse LLM response to extract a dictionary."""
        try:
            if "```json" in content:
                start = content.find("```json") + 7
                end = content.find("```", start)
                json_str = content[start:end].strip()
            elif "```" in content:
                start = content.find("```") + 3
                end = content.find("```", start)
                json_str = content[start:end].strip()
            elif "{" in content:
                start = content.find("{")
                end = content.rfind("}") + 1
                json_str = content[start:end]
            else:
                return {}

            data = json.loads(json_str)
            if isinstance(data, dict):
                return data
            return {}

        except (json.JSONDecodeError, ValueError):
            return {}

    def _parse_dict_list_response(self, content: str) -> Dict[str, List[str]]:
        """Parse LLM response to extract a dict of lists."""
        result = self._parse_dict_response(content)
        # Ensure all values are lists
        for key, value in result.items():
            if not isinstance(value, list):
                result[key] = [value] if value else []
        return result

method_name property

Return the method name for this generator.

generate(inputs, rubric='', max_questions=None, **kwargs)

Generate a checklist from think-aloud attributes.

Parameters:

Name	Type	Description	Default
inputs	List[InteractiveInput]	List of InteractiveInput with attributes from human/LLM sources	required
rubric	str	Definition/rubric for the evaluation dimension	''
max_questions	Optional[int]	Maximum number of questions to include	None
**kwargs	Any	Additional parameters	{}

Returns:

Type	Description
Checklist	Checklist with generated questions

Source code in autochecklist/generators/corpus_level/interactive.py

def generate(
    self,
    inputs: List[InteractiveInput],
    rubric: str = "",
    max_questions: Optional[int] = None,
    **kwargs: Any,
) -> Checklist:
    """Generate a checklist from think-aloud attributes.

    Args:
        inputs: List of InteractiveInput with attributes from human/LLM sources
        rubric: Definition/rubric for the evaluation dimension
        max_questions: Maximum number of questions to include
        **kwargs: Additional parameters

    Returns:
        Checklist with generated questions
    """
    if not inputs:
        return self._empty_checklist()

    # Combine attributes from all inputs
    all_attributes = []
    dimension = inputs[0].dimension
    source = inputs[0].source

    for inp in inputs:
        all_attributes.extend(inp.attributes)
        # Use the first non-None dimension
        if not dimension and inp.dimension:
            dimension = inp.dimension
        # Track combined sources
        if inp.source != source:
            source = "human_llm"

    if not all_attributes:
        return self._empty_checklist(dimension=dimension)

    # Run the 5-stage pipeline
    validated_questions = self._run_pipeline(
        all_attributes, rubric, dimension or "quality"
    )

    # Apply max_questions limit if specified
    if max_questions and len(validated_questions) > max_questions:
        validated_questions = validated_questions[:max_questions]

    # Convert to ChecklistItems
    items = []
    for q in validated_questions:
        item = ChecklistItem(
            id=str(uuid.uuid4()),
            question=q,
            category=dimension,
            metadata={
                "dimension": dimension,
            },
        )
        items.append(item)

    return Checklist(
        id=str(uuid.uuid4()),
        items=items,
        source_method="interacteval",
        generation_level="corpus",
        metadata={
            "dimension": dimension,
            "attribute_count": len(all_attributes),
            "source": source,
            "rubric": rubric[:200] if rubric else None,  # Truncate for metadata
        },
    )