generators

generators

Checklist generators.

ChecklistGenerator

Bases: ABC

Base class for all checklist generators.

Source code in autochecklist/generators/base.py

class ChecklistGenerator(ABC):
    """Base class for all checklist generators."""

    def __init__(
        self,
        model: Optional[str] = None,
        temperature: Optional[float] = None,
        max_tokens: int = 2048,
        api_key: Optional[str] = None,
        provider: Optional[str] = None,
        base_url: Optional[str] = None,
        client: Any = None,
        api_format: Optional[str] = None,
        reasoning_effort: Optional[str] = None,
    ):
        config = get_config()
        self.model = model or config.generator_model.model_id
        self.temperature = temperature if temperature is not None else config.generator_model.temperature
        self.max_tokens = max_tokens
        self.api_key = api_key
        self._client = client
        self._provider = provider or config.generator_model.provider or "openrouter"
        self._base_url = base_url
        self._api_format = api_format or "chat"
        self.reasoning_effort = reasoning_effort

    @property
    @abstractmethod
    def generation_level(self) -> str:
        """Return 'instance' or 'corpus'."""
        pass

    @property
    @abstractmethod
    def method_name(self) -> str:
        """Return the method name (e.g., 'tick', 'rlcf')."""
        pass

    @abstractmethod
    def generate(self, **kwargs: Any) -> Checklist:
        """Generate a checklist."""
        pass

    def generate_stream(self, **kwargs: Any) -> Iterator[str]:
        """Stream checklist generation (for UI).

        Default implementation just yields the final result.
        Override for true streaming support.
        """
        checklist = self.generate(**kwargs)
        yield checklist.to_text()

    def _get_or_create_client(self) -> Any:
        """Get injected client or create one from provider settings."""
        if self._client is not None:
            return self._client
        from ..providers.factory import get_client
        return get_client(
            provider=self._provider,
            api_key=self.api_key,
            base_url=self._base_url,
            model=self.model,
            api_format=self._api_format,
        )

    def _call_model(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        response_format: Optional[dict] = None,
    ) -> str:
        """Call the LLM and return the response text.

        Args:
            prompt: The user prompt text.
            system_prompt: Optional system prompt.
            response_format: Optional OpenAI-compatible response_format dict
                for structured JSON output. When provided, the call is attempted
                with ``response_format`` first; if the provider does not support
                it, the call is retried without it (fallback to schema-in-prompt).

        Returns:
            The model's response text.
        """
        messages: List[Dict[str, str]] = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        kwargs: Dict[str, Any] = {}
        if response_format is not None:
            kwargs["response_format"] = response_format
        if self.reasoning_effort is not None:
            kwargs["reasoning_effort"] = self.reasoning_effort

        client = self._get_or_create_client()

        try:
            response = client.chat_completion(
                model=self.model,
                messages=messages,
                temperature=self.temperature,
                max_tokens=self.max_tokens,
                **kwargs,
            )
        except (ValueError, KeyError, TypeError) as e:
            # Schema/parsing errors from response_format — fall back to schema-in-prompt
            if response_format is not None:
                logger.warning(
                    "Structured output failed (%s), retrying without "
                    "response_format (fallback to schema-in-prompt).",
                    e,
                )
                fallback_kwargs = {k: v for k, v in kwargs.items() if k != "response_format"}
                response = client.chat_completion(
                    model=self.model,
                    messages=messages,
                    temperature=self.temperature,
                    max_tokens=self.max_tokens,
                    **fallback_kwargs,
                )
            else:
                raise
        except Exception as e:
            # For HTTP errors (auth, rate limit, server), only fallback on 400 (bad schema)
            import httpx

            if (
                response_format is not None
                and isinstance(e, httpx.HTTPStatusError)
                and e.response.status_code == 400
            ):
                logger.warning(
                    "Structured output failed (%s), retrying without "
                    "response_format (fallback to schema-in-prompt).",
                    e,
                )
                fallback_kwargs = {k: v for k, v in kwargs.items() if k != "response_format"}
                response = client.chat_completion(
                    model=self.model,
                    messages=messages,
                    temperature=self.temperature,
                    max_tokens=self.max_tokens,
                    **fallback_kwargs,
                )
            else:
                raise

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

generation_level abstractmethod property

Return 'instance' or 'corpus'.

method_name abstractmethod property

Return the method name (e.g., 'tick', 'rlcf').

generate(**kwargs) abstractmethod

Generate a checklist.

Source code in autochecklist/generators/base.py

@abstractmethod
def generate(self, **kwargs: Any) -> Checklist:
    """Generate a checklist."""
    pass

generate_stream(**kwargs)

Stream checklist generation (for UI).

Default implementation just yields the final result. Override for true streaming support.

Source code in autochecklist/generators/base.py

def generate_stream(self, **kwargs: Any) -> Iterator[str]:
    """Stream checklist generation (for UI).

    Default implementation just yields the final result.
    Override for true streaming support.
    """
    checklist = self.generate(**kwargs)
    yield checklist.to_text()

InstanceChecklistGenerator

Bases: ChecklistGenerator

Base for instance-level generators (one checklist per input).

Source code in autochecklist/generators/base.py

class InstanceChecklistGenerator(ChecklistGenerator):
    """Base for instance-level generators (one checklist per input)."""

    @property
    def generation_level(self) -> str:
        return "instance"

    @abstractmethod
    def generate(
        self,
        input: str,
        target: Optional[str] = None,
        reference: Optional[str] = None,
        **kwargs: Any,
    ) -> Checklist:
        """Generate checklist for an input.

        Args:
            input: The input/query/instruction text
            target: Optional target response to evaluate
            reference: Optional reference target for comparison
            **kwargs: Method-specific arguments

        Returns:
            Generated Checklist
        """
        pass

generate(input, target=None, reference=None, **kwargs) abstractmethod

Generate checklist for an input.

Parameters:

Name	Type	Description	Default
input	str	The input/query/instruction text	required
target	Optional[str]	Optional target response to evaluate	None
reference	Optional[str]	Optional reference target for comparison	None
**kwargs	Any	Method-specific arguments	{}

Returns:

Type	Description
Checklist	Generated Checklist

Source code in autochecklist/generators/base.py

@abstractmethod
def generate(
    self,
    input: str,
    target: Optional[str] = None,
    reference: Optional[str] = None,
    **kwargs: Any,
) -> Checklist:
    """Generate checklist for an input.

    Args:
        input: The input/query/instruction text
        target: Optional target response to evaluate
        reference: Optional reference target for comparison
        **kwargs: Method-specific arguments

    Returns:
        Generated Checklist
    """
    pass

CorpusChecklistGenerator

Bases: ChecklistGenerator

Base for corpus-level generators (one checklist for entire dataset).

Source code in autochecklist/generators/base.py

class CorpusChecklistGenerator(ChecklistGenerator):
    """Base for corpus-level generators (one checklist for entire dataset)."""

    @property
    def generation_level(self) -> str:
        return "corpus"

    @abstractmethod
    def generate(
        self,
        inputs: List[Dict[str, Any]],
        **kwargs: Any,
    ) -> Checklist:
        """Generate checklist from corpus inputs.

        Args:
            inputs: List of input items (feedback, dimensions, think-aloud, etc.)
            **kwargs: Method-specific arguments

        Returns:
            Generated Checklist
        """
        pass

generate(inputs, **kwargs) abstractmethod

Generate checklist from corpus inputs.

Parameters:

Name	Type	Description	Default
inputs	List[Dict[str, Any]]	List of input items (feedback, dimensions, think-aloud, etc.)	required
**kwargs	Any	Method-specific arguments	{}

Returns:

Type	Description
Checklist	Generated Checklist

Source code in autochecklist/generators/base.py

@abstractmethod
def generate(
    self,
    inputs: List[Dict[str, Any]],
    **kwargs: Any,
) -> Checklist:
    """Generate checklist from corpus inputs.

    Args:
        inputs: List of input items (feedback, dimensions, think-aloud, etc.)
        **kwargs: Method-specific arguments

    Returns:
        Generated Checklist
    """
    pass

DirectGenerator

Bases: InstanceChecklistGenerator

Generate checklists using a prompt template + structured JSON output.

Can be configured via pipeline presets (built-in methods) or custom prompts.

Parameters:

Name	Type	Description	Default
method_name	str	Pipeline preset name (e.g., "tick") or custom name. If a known preset, loads config from PIPELINE_PRESETS.	'custom'
custom_prompt	Optional[Union[str, Path]]	Custom prompt template. Pass a Path to load from file, or a str for raw prompt text. Overrides preset template.	None
response_schema	Optional[type]	Pydantic model for JSON validation. Default: ChecklistResponse.	None
format_name	Optional[str]	Format prompt file name (e.g., "checklist"). Default from preset.	None
max_items	int	Maximum checklist items to return.	10
min_items	int	Minimum expected items.	2
**kwargs	Any	Passed to InstanceChecklistGenerator (model, temperature, etc.)	{}

Source code in autochecklist/generators/instance_level/direct.py

class DirectGenerator(InstanceChecklistGenerator):
    """Generate checklists using a prompt template + structured JSON output.

    Can be configured via pipeline presets (built-in methods) or custom prompts.

    Args:
        method_name: Pipeline preset name (e.g., "tick") or custom name.
            If a known preset, loads config from PIPELINE_PRESETS.
        custom_prompt: Custom prompt template. Pass a Path to load from file,
            or a str for raw prompt text. Overrides preset template.
        response_schema: Pydantic model for JSON validation. Default: ChecklistResponse.
        format_name: Format prompt file name (e.g., "checklist"). Default from preset.
        max_items: Maximum checklist items to return.
        min_items: Minimum expected items.
        **kwargs: Passed to InstanceChecklistGenerator (model, temperature, etc.)
    """

    def __init__(
        self,
        method_name: str = "custom",
        custom_prompt: Optional[Union[str, Path]] = None,
        response_schema: Optional[type] = None,
        format_name: Optional[str] = None,
        max_items: int = 10,
        min_items: int = 2,
        **kwargs: Any,
    ):
        # Load preset defaults if this is a known method
        from .pipeline_presets import PIPELINE_PRESETS

        preset = PIPELINE_PRESETS.get(method_name, {})

        # Apply preset defaults, allowing kwargs to override
        if "temperature" not in kwargs and "temperature" in preset:
            kwargs["temperature"] = preset["temperature"]

        super().__init__(**kwargs)

        self._method_name = method_name
        self.max_items = preset.get("max_items", max_items)
        self.min_items = preset.get("min_items", min_items)

        is_custom_schema = response_schema is not None
        self._response_schema = response_schema or preset.get(
            "response_schema", ChecklistResponse
        )
        if format_name is not None:
            self._format_name = format_name
        elif is_custom_schema:
            self._format_name = None
        else:
            self._format_name = preset.get("format_name", "checklist")

        # Load template
        if custom_prompt is not None:
            if isinstance(custom_prompt, Path):
                template_text = custom_prompt.read_text(encoding="utf-8")
            else:
                template_text = custom_prompt
        elif preset:
            template_text = load_template(
                preset["template_dir"], preset["template_name"]
            )
        else:
            raise ValueError(
                f"Unknown method '{method_name}' and no custom_prompt provided"
            )

        self._template = PromptTemplate(template_text)

    @property
    def method_name(self) -> str:
        return self._method_name

    @property
    def prompt_text(self) -> str:
        """The raw prompt template text."""
        return self._template.template

    def generate(
        self,
        input: str,
        target: Optional[str] = None,
        reference: Optional[str] = None,
        history: str = "",
        **kwargs: Any,
    ) -> Checklist:
        """Generate checklist from input using template + structured output.

        Automatically detects which placeholders the template needs and passes
        only those. This allows the same class to handle TICK (input only),
        RocketEval (input + reference + history), RLCF-direct
        (input + reference), etc.
        """
        # Build format kwargs — only pass placeholders that exist in template
        format_kwargs: dict[str, str] = {"input": input}
        if "target" in self._template._placeholders and target is not None:
            format_kwargs["target"] = target
        if "reference" in self._template._placeholders:
            if reference is None:
                raise ValueError(
                    f"{self._method_name} requires a reference target."
                )
            format_kwargs["reference"] = reference
        if "history" in self._template._placeholders:
            format_kwargs["history"] = history

        # Load format instructions (skip for custom schemas)
        format_text = load_format(self._format_name) if self._format_name else ""

        # Inject format inline if template has {format_instructions} placeholder,
        # otherwise append after the prompt (default).
        if "format_instructions" in self._template._placeholders:
            format_kwargs["format_instructions"] = format_text
            full_prompt = self._template.format(**format_kwargs)
        else:
            prompt = self._template.format(**format_kwargs)
            full_prompt = prompt + "\n\n" + format_text

        # Call model with structured output
        response_format = to_response_format(
            self._response_schema, self._method_name
        )
        raw = self._call_model(full_prompt, response_format=response_format)

        # Parse structured response
        items = self._parse_structured(raw)

        return Checklist(
            items=items,
            source_method=self.method_name,
            generation_level=self.generation_level,
            input=input,
            metadata={"raw_response": raw},
        )

    def _parse_structured(self, raw: str) -> list[ChecklistItem]:
        """Parse JSON response using Pydantic schema.

        Primary path: json.loads() succeeds (structured output).
        Fallback path: extract_json() extracts JSON from raw text.

        Auto-detects the list field and item fields from the schema,
        supporting both built-in and custom response schemas.
        """
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            data = extract_json(raw)
        validated = self._response_schema.model_validate(data)

        # Find the list field (first List[BaseModel] field)
        item_list = self._get_item_list(validated)

        items = []
        for q in item_list[: self.max_items]:
            q_data = q.model_dump() if hasattr(q, "model_dump") else {}
            # Find question text: use 'question' field, or first str field
            question, question_key = self._get_question_text(q, q_data)
            weight = getattr(q, "weight", 100.0)
            category = getattr(q, "category", None)
            # Extra fields → metadata
            known = {question_key, "weight", "category"}
            extra = {k: v for k, v in q_data.items() if k not in known}
            items.append(
                ChecklistItem(
                    question=question,
                    weight=weight,
                    category=category,
                    metadata=extra if extra else {},
                )
            )
        return items

    @staticmethod
    def _get_item_list(validated: Any) -> list:
        """Extract the list of items from a validated response model."""
        # Try 'questions' first (built-in convention)
        if hasattr(validated, "questions"):
            return validated.questions
        # Auto-detect: first list attribute
        for field_name in type(validated).model_fields:
            value = getattr(validated, field_name)
            if isinstance(value, list):
                return value
        raise ValueError(
            f"Cannot find list field in {type(validated).__name__}. "
            "Schema must have a list field (e.g., 'questions', 'items')."
        )

    @staticmethod
    def _get_question_text(item: Any, item_data: dict) -> tuple[str, str]:
        """Extract question text and its field key from an item."""
        if isinstance(item, str):
            return item, "question"
        if hasattr(item, "question"):
            return item.question, "question"
        # Fall back to first str field
        for key, value in item_data.items():
            if isinstance(value, str):
                return value, key
        raise ValueError(
            f"Cannot find question text in {type(item).__name__}. "
            "Item must have a 'question' field or at least one str field."
        )

prompt_text property

The raw prompt template text.

generate(input, target=None, reference=None, history='', **kwargs)

Generate checklist from input using template + structured output.

Automatically detects which placeholders the template needs and passes only those. This allows the same class to handle TICK (input only), RocketEval (input + reference + history), RLCF-direct (input + reference), etc.

Source code in autochecklist/generators/instance_level/direct.py

def generate(
    self,
    input: str,
    target: Optional[str] = None,
    reference: Optional[str] = None,
    history: str = "",
    **kwargs: Any,
) -> Checklist:
    """Generate checklist from input using template + structured output.

    Automatically detects which placeholders the template needs and passes
    only those. This allows the same class to handle TICK (input only),
    RocketEval (input + reference + history), RLCF-direct
    (input + reference), etc.
    """
    # Build format kwargs — only pass placeholders that exist in template
    format_kwargs: dict[str, str] = {"input": input}
    if "target" in self._template._placeholders and target is not None:
        format_kwargs["target"] = target
    if "reference" in self._template._placeholders:
        if reference is None:
            raise ValueError(
                f"{self._method_name} requires a reference target."
            )
        format_kwargs["reference"] = reference
    if "history" in self._template._placeholders:
        format_kwargs["history"] = history

    # Load format instructions (skip for custom schemas)
    format_text = load_format(self._format_name) if self._format_name else ""

    # Inject format inline if template has {format_instructions} placeholder,
    # otherwise append after the prompt (default).
    if "format_instructions" in self._template._placeholders:
        format_kwargs["format_instructions"] = format_text
        full_prompt = self._template.format(**format_kwargs)
    else:
        prompt = self._template.format(**format_kwargs)
        full_prompt = prompt + "\n\n" + format_text

    # Call model with structured output
    response_format = to_response_format(
        self._response_schema, self._method_name
    )
    raw = self._call_model(full_prompt, response_format=response_format)

    # Parse structured response
    items = self._parse_structured(raw)

    return Checklist(
        items=items,
        source_method=self.method_name,
        generation_level=self.generation_level,
        input=input,
        metadata={"raw_response": raw},
    )

ContrastiveGenerator

Bases: DirectGenerator

Generate checklists by comparing candidate responses (RLCF candidate modes).

Extends DirectGenerator with candidate auto-generation. Candidates are generated by smaller models and included in the prompt for contrastive analysis.

Two modes: - rlcf_candidate: input + reference + candidates - rlcf_candidates_only: input + candidates (no reference)

Source code in autochecklist/generators/instance_level/contrastive.py

class ContrastiveGenerator(DirectGenerator):
    """Generate checklists by comparing candidate responses (RLCF candidate modes).

    Extends DirectGenerator with candidate auto-generation. Candidates are
    generated by smaller models and included in the prompt for contrastive
    analysis.

    Two modes:
    - rlcf_candidate: input + reference + candidates
    - rlcf_candidates_only: input + candidates (no reference)
    """

    def __init__(
        self,
        candidate_models: Optional[List[str]] = None,
        num_candidates: int = 4,
        generate_candidates: Optional[bool] = None,
        candidate_provider: Optional[str] = None,
        candidate_base_url: Optional[str] = None,
        candidate_api_key: Optional[str] = None,
        candidate_api_format: Optional[str] = None,
        **kwargs: Any,
    ):
        super().__init__(**kwargs)
        # Read generate_candidates from preset if not explicitly provided
        from .pipeline_presets import PIPELINE_PRESETS
        preset = PIPELINE_PRESETS.get(self._method_name, {})
        if generate_candidates is None:
            self.generate_candidates = preset.get("generate_candidates", True)
        else:
            self.generate_candidates = generate_candidates
        self.candidate_models = candidate_models
        self.num_candidates = num_candidates
        self._candidate_provider = candidate_provider
        self._candidate_base_url = candidate_base_url
        self._candidate_api_key = candidate_api_key
        self._candidate_api_format = candidate_api_format

    def generate(
        self,
        input: str,
        target: Optional[str] = None,
        reference: Optional[str] = None,
        candidates: Optional[Union[List[str], Dict[str, str]]] = None,
        **kwargs: Any,
    ) -> Checklist:
        """Generate checklist from input + candidates.

        Args:
            input: The instruction/query
            target: Alias for reference
            reference: Expert/reference target (optional for candidates_only)
            candidates: Candidate responses. Can be:
                - List[str]: multiple candidates (RLCF or listwise)
                - Dict with "chosen"/"rejected" keys (pairwise CRG)
                - None: auto-generated if candidate_models is set
            **kwargs: Additional arguments
        """
        # Get or generate candidates
        if candidates is None:
            if self.generate_candidates and self.candidate_models:
                candidates = self._generate_candidates(input)
            else:
                raise ValueError(
                    f"{self.method_name} requires 'candidates' argument."
                )

        # Delegate to _generate_with_candidates with raw candidates
        checklist = self._generate_with_candidates(
            input=input,
            candidates=candidates,
            reference=reference,
            **kwargs,
        )
        # Store raw candidates and count in metadata
        if isinstance(candidates, dict):
            checklist.metadata["candidates"] = list(candidates.values())
            checklist.metadata["num_candidates"] = 2
        else:
            checklist.metadata["candidates"] = candidates
            checklist.metadata["num_candidates"] = len(candidates)
        return checklist

    def _generate_with_candidates(
        self,
        input: str,
        candidates: Union[List[str], Dict[str, str]],
        reference: Optional[str] = None,
        **kwargs: Any,
    ) -> Checklist:
        """Build prompt with candidates and call model.

        Routes candidates to template placeholders based on type and template:
        - Dict → {chosen} + {rejected} placeholders (pairwise CRG)
        - List + {responses} placeholder → numbered Response blocks (listwise)
        - List + {candidates} placeholder → numbered Candidate blocks (RLCF)
        """
        placeholders = self._template._placeholders
        format_kwargs: dict[str, str] = {"input": input}

        # --- Route candidates to placeholders ---
        if isinstance(candidates, dict):
            # Pairwise: dict must have chosen+rejected, template must have those placeholders
            if "candidates" in placeholders:
                raise ValueError(
                    "Template has {candidates} placeholder but received dict candidates. "
                    "Use {chosen}/{rejected} placeholders for pairwise, or pass a list."
                )
            if not {"chosen", "rejected"} <= placeholders:
                raise ValueError(
                    "Template must have {chosen} and {rejected} placeholders for dict candidates."
                )
            if set(candidates.keys()) != {"chosen", "rejected"}:
                raise ValueError(
                    "Dict candidates must have exactly 'chosen' and 'rejected' keys, "
                    f"got: {set(candidates.keys())}"
                )
            format_kwargs["chosen"] = candidates["chosen"]
            format_kwargs["rejected"] = candidates["rejected"]
        else:
            # List candidates
            if "chosen" in placeholders or "rejected" in placeholders:
                raise ValueError(
                    "Template has {chosen}/{rejected} placeholders but received list candidates. "
                    "Pass a dict with 'chosen' and 'rejected' keys instead."
                )
            if "responses" in placeholders:
                format_kwargs["responses"] = self._format_ordered_responses(candidates)
            elif "candidates" in placeholders:
                format_kwargs["candidates"] = self._format_candidates(candidates)
            else:
                raise ValueError(
                    "Template must have {candidates} or {responses} placeholder for list candidates."
                )

        # --- Handle optional placeholders ---
        if "context" in placeholders:
            format_kwargs["context"] = kwargs.pop("context", "")

        if "reference" in placeholders:
            if reference is None:
                raise ValueError(
                    f"{self.method_name} requires a reference target."
                )
            format_kwargs["reference"] = reference

        # Load format instructions (skip for custom schemas)
        format_text = load_format(self._format_name) if self._format_name else ""

        # Inject format inline if template has {format_instructions} placeholder,
        # otherwise append after the prompt (default).
        if "format_instructions" in placeholders:
            format_kwargs["format_instructions"] = format_text
            full_prompt = self._template.format(**format_kwargs)
        else:
            prompt = self._template.format(**format_kwargs)
            full_prompt = prompt + "\n\n" + format_text

        response_format = to_response_format(
            self._response_schema, self._method_name
        )
        raw = self._call_model(full_prompt, response_format=response_format)
        items = self._parse_structured(raw)

        return Checklist(
            items=items,
            source_method=self.method_name,
            generation_level=self.generation_level,
            input=input,
            metadata={"raw_response": raw},
        )

    def _get_candidate_client(self) -> Any:
        """Get client for candidate generation.

        If any candidate_* provider param is set, creates a separate client.
        Otherwise falls back to the main client via _get_or_create_client().
        """
        if any([
            self._candidate_provider,
            self._candidate_base_url,
            self._candidate_api_key,
            self._candidate_api_format,
        ]):
            return get_client(
                provider=self._candidate_provider or self._provider,
                base_url=self._candidate_base_url,
                api_key=self._candidate_api_key,
                model=self.model,
                api_format=self._candidate_api_format,
            )
        return self._get_or_create_client()

    def _generate_candidates(self, input: str) -> List[str]:
        """Generate candidate responses using smaller models."""
        candidates = []
        client = self._get_candidate_client()

        if len(self.candidate_models) > 1:
            for model in self.candidate_models:
                resp = client.chat_completion(
                    model=model,
                    messages=[{"role": "user", "content": input}],
                    temperature=0.7,
                    max_tokens=1024,
                )
                candidates.append(resp["choices"][0]["message"]["content"])
        else:
            model = self.candidate_models[0]
            for _ in range(self.num_candidates):
                resp = client.chat_completion(
                    model=model,
                    messages=[{"role": "user", "content": input}],
                    temperature=0.9,
                    max_tokens=1024,
                )
                candidates.append(resp["choices"][0]["message"]["content"])

        return candidates

    def _format_ordered_responses(self, responses: List[str]) -> str:
        """Format responses as numbered Response blocks for listwise CRG."""
        formatted = []
        for i, response in enumerate(responses, 1):
            formatted.append(f"### Response {i}\n{response}")
        return "\n\n".join(formatted)

    def _format_candidates(self, candidates: List[str]) -> str:
        """Format candidate responses for prompt injection."""
        formatted = []
        for i, candidate in enumerate(candidates, 1):
            formatted.append(f"### Candidate {i}\n{candidate}")
        return "\n\n".join(formatted)

generate(input, target=None, reference=None, candidates=None, **kwargs)

Generate checklist from input + candidates.

Parameters:

Name	Type	Description	Default
input	str	The instruction/query	required
target	Optional[str]	Alias for reference	None
reference	Optional[str]	Expert/reference target (optional for candidates_only)	None
candidates	Optional[Union[List[str], Dict[str, str]]]	Candidate responses. Can be: - List[str]: multiple candidates (RLCF or listwise) - Dict with "chosen"/"rejected" keys (pairwise CRG) - None: auto-generated if candidate_models is set	None
**kwargs	Any	Additional arguments	{}

Source code in autochecklist/generators/instance_level/contrastive.py

def generate(
    self,
    input: str,
    target: Optional[str] = None,
    reference: Optional[str] = None,
    candidates: Optional[Union[List[str], Dict[str, str]]] = None,
    **kwargs: Any,
) -> Checklist:
    """Generate checklist from input + candidates.

    Args:
        input: The instruction/query
        target: Alias for reference
        reference: Expert/reference target (optional for candidates_only)
        candidates: Candidate responses. Can be:
            - List[str]: multiple candidates (RLCF or listwise)
            - Dict with "chosen"/"rejected" keys (pairwise CRG)
            - None: auto-generated if candidate_models is set
        **kwargs: Additional arguments
    """
    # Get or generate candidates
    if candidates is None:
        if self.generate_candidates and self.candidate_models:
            candidates = self._generate_candidates(input)
        else:
            raise ValueError(
                f"{self.method_name} requires 'candidates' argument."
            )

    # Delegate to _generate_with_candidates with raw candidates
    checklist = self._generate_with_candidates(
        input=input,
        candidates=candidates,
        reference=reference,
        **kwargs,
    )
    # Store raw candidates and count in metadata
    if isinstance(candidates, dict):
        checklist.metadata["candidates"] = list(candidates.values())
        checklist.metadata["num_candidates"] = 2
    else:
        checklist.metadata["candidates"] = candidates
        checklist.metadata["num_candidates"] = len(candidates)
    return checklist

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()

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
        },
    )