Add user input validation to pipeline interfaces error display on pipeline change.

src/components/model_pipeline/model_pipeline.py

@@ -10,13 +10,14 @@ from components.model_pipeline.state_manager import (
     PipelineState,
     PipelineStateManager,
     PipelineUIState,
+    PipelineValidator,
     TossupPipelineState,
     TossupPipelineStateManager,
 )
 from components.model_step.model_step import ModelStepComponent
 from components.utils import make_state
 from workflows.structs import ModelStep, TossupWorkflow, Workflow
-from workflows.validators import WorkflowValidator
+from workflows.validators import WorkflowValidationError, WorkflowValidator
 
 from .state_manager import get_output_panel_state
 
@@ -33,6 +34,7 @@ class PipelineInterface:
         ui_state: PipelineUIState | None = None,
         model_options: list[str] = None,
         config: dict = {},
+        validator: PipelineValidator | None = None,
     ):
         self.app = app
         self.model_options = model_options
@@ -50,10 +52,10 @@ class PipelineInterface:
 
         if isinstance(workflow, TossupWorkflow):
             pipeline_state = TossupPipelineState(workflow=workflow, ui_state=ui_state)
-            self.sm = TossupPipelineStateManager()
+            self.sm = TossupPipelineStateManager(validator)
         else:
             pipeline_state = PipelineState(workflow=workflow, ui_state=ui_state)
-            self.sm = PipelineStateManager()
+            self.sm = PipelineStateManager(validator)
         self.pipeline_state = make_state(pipeline_state.model_dump())
 
         def get_aux_states(pipeline_state_dict: td.PipelineStateDict):
@@ -169,7 +171,11 @@ class PipelineInterface:
         """Validate the workflow."""
         try:
             state = self.sm.make_pipeline_state(state_dict)
-            WorkflowValidator().validate(state.workflow)
+            validator = WorkflowValidator(
+                max_temperature=self.config.get("max_temperature", 10),
+            )
+            if not validator.validate(state.workflow):
+                raise WorkflowValidationError(validator.errors)
         except ValueError as e:
             logger.exception(e)
             state_dict_str = yaml.dump(state_dict, default_flow_style=False, indent=2)
@@ -244,6 +250,7 @@ class PipelineInterface:
     def _render_pipeline_preview(self):
         export_btn = gr.Button("Export Pipeline", elem_classes="export-button", visible=False)
         # components.append(export_btn)
+        self.error_display = gr.HTML(label="Error", elem_id="pipeline-preview-error-display", visible=False)
 
         # Add a code box to display the workflow JSON
         # with gr.Column(elem_classes="workflow-json-container"):
@@ -262,7 +269,7 @@ class PipelineInterface:
         self.config_output.blur(
             fn=self.sm.update_workflow_from_code,
             inputs=[self.config_output, self.pipeline_change],
-            outputs=[self.pipeline_state, self.pipeline_change],
+            outputs=[self.pipeline_state, self.pipeline_change, self.error_display],
         )
 
         # Connect the export button to show the workflow JSON
@@ -326,6 +333,6 @@ class PipelineInterface:
         ).success(
             fn=self.sm.get_formatted_config,
             inputs=[self.pipeline_state, gr.State("yaml")],
-            outputs=[self.config_output],
+            outputs=[self.config_output, self.error_display],
             js=js,
         )

src/components/model_pipeline/state_manager.py

@@ -1,7 +1,12 @@
+# %%
 import json
+from abc import ABC, abstractmethod
 from typing import Literal
 
+import gradio as gr
 import yaml
+from loguru import logger
+from pydantic import BaseModel, ValidationError
 
 from app_configs import UNSELECTED_VAR_NAME
 from components import typed_dicts as td
@@ -22,24 +27,49 @@ def get_output_panel_state(workflow: Workflow) -> dict:
     return state
 
 
+def strict_model_validate(model_cls: type[BaseModel], data: dict):
+    # Dynamically create a subclass with extra='forbid'
+    class_name = model_cls.__name__
+    strict_class_name = f"Strict{class_name}"
+
+    strict_class = type(
+        strict_class_name,
+        (model_cls,),
+        {"model_config": {**getattr(model_cls, "model_config", {}), "extra": "forbid"}},
+    )
+
+    return strict_class.model_validate(data)
+
+
+class PipelineValidator(ABC):
+    """Abstract base class for pipeline validators."""
+
+    @abstractmethod
+    def __call__(self, workflow: Workflow):
+        """
+        Validate the workflow.
+
+        Args:
+            workflow: The workflow to validate.
+
+        Raises:
+            ValueError: If the workflow is invalid.
+        """
+        pass
+
+
 class PipelineStateManager:
     """Manages a pipeline of multiple steps."""
 
+    pipeline_state_cls = PipelineState
+    workflow_cls = Workflow
+
+    def __init__(self, validator: PipelineValidator | None = None):
+        self.validator = validator
+
     def make_pipeline_state(self, state_dict: td.PipelineStateDict) -> PipelineState:
         """Make a state from a state dictionary."""
-        return PipelineState(**state_dict)
-
-    def get_formatted_config(self, state_dict: td.PipelineStateDict, format: Literal["json", "yaml"] = "yaml") -> str:
-        """Get the full pipeline configuration."""
-        state = self.make_pipeline_state(state_dict)
-        config = state.workflow.model_dump(exclude_defaults=True)
-        if isinstance(state.workflow, TossupWorkflow):
-            buzzer_config = state.workflow.buzzer.model_dump(exclude_defaults=False)
-            config["buzzer"] = buzzer_config
-        if format == "yaml":
-            return yaml.dump(config, default_flow_style=False, sort_keys=False, indent=4)
-        else:
-            return json.dumps(config, indent=4, sort_keys=False)
+        return self.pipeline_state_cls(**state_dict)
 
     def add_step(
         self, state_dict: td.PipelineStateDict, pipeline_change: bool, position: int = -1, name=""
@@ -102,7 +132,7 @@ class PipelineStateManager:
             produced_variable = None
         """Update the output variables for a step."""
         state = self.make_pipeline_state(state_dict)
-        state.workflow.outputs[target] = produced_variable
+        state = state.update_output_variable(target, produced_variable)
         return state.model_dump()
 
     def update_model_step_ui(
@@ -117,53 +147,101 @@ class PipelineStateManager:
         """Get all variables from all steps."""
         return self.make_pipeline_state(state_dict)
 
-    def parse_yaml_workflow(self, yaml_str: str) -> Workflow:
+    def parse_yaml_workflow(self, yaml_str: str, strict: bool = True) -> Workflow:
         """Parse a YAML workflow."""
         workflow = yaml.safe_load(yaml_str)
-        return Workflow(**workflow)
+        try:
+            if strict:
+                return strict_model_validate(self.workflow_cls, workflow)
+            else:
+                return self.workflow_cls.model_validate(workflow)
+        except ValidationError as e:
+            new_exception = ValidationError.from_exception_data(
+                e.title.removeprefix("Strict"), e.errors(), input_type="json"
+            )
+            raise new_exception from e
+
+    def _handle_pipeline_parsing_error(self, e: Exception) -> str:
+        """Format error messages for pipeline parsing errors with consistent styling."""
+        error_template = """
+        <div class="md" style='color: #FF0000; background-color: #FFEEEE; padding: 10px; border-radius: 5px; border-left: 4px solid #FF0000;'>
+            <strong style='color: #FF0000;'>{error_type}:</strong> <br>
+            <div class="code-wrap">
+                <pre><code>{error_message}</code></pre>
+            </div>
+            {help_text}
+        </div>
+        """
+        logger.exception(e)
+        if isinstance(e, yaml.YAMLError):
+            error_type = "Invalid YAML Error"
+            help_text = "Refer to the <a href='https://spacelift.io/blog/yaml#basic-yaml-syntax' target='_blank'>YAML schema</a> for correct formatting."
+        elif isinstance(e, ValidationError):
+            error_type = "Pipeline Parsing Error"
+            help_text = "Refer to the <a href='https://mgor.info' target='_blank'>documentation</a> for the correct pipeline schema."
+        elif isinstance(e, ValueError):
+            error_type = "Pipeline Validation Error"
+            help_text = "Refer to the <a href='https://mgor.info' target='_blank'>documentation</a> for the correct pipeline schema."
+        else:
+            error_type = "Unexpected Error"
+            help_text = "Please report this issue to us at <a href='https://github.com/maharshi95/QANTA25/issues' target='_blank'>GitHub Issues</a>."
+
+        return error_template.format(error_type=error_type, error_message=str(e), help_text=help_text)
 
-    def update_workflow_from_code(self, yaml_str: str) -> td.PipelineStateDict:
+    def get_formatted_config(
+        self, state_dict: td.PipelineStateDict, format: Literal["json", "yaml"] = "yaml"
+    ) -> tuple[str, dict]:
+        """Get the full pipeline configuration."""
+        try:
+            state = self.make_pipeline_state(state_dict)
+            config = state.workflow.model_dump(exclude_defaults=True)
+            if isinstance(state.workflow, TossupWorkflow):
+                buzzer_config = state.workflow.buzzer.model_dump(exclude_defaults=False)
+                config["buzzer"] = buzzer_config
+            if format == "yaml":
+                config_str = yaml.dump(config, default_flow_style=False, sort_keys=False, indent=4)
+            else:
+                config_str = json.dumps(config, indent=4, sort_keys=False)
+            return config_str, gr.update(visible=False)
+        except Exception as e:
+            error_message = self._handle_pipeline_parsing_error(e)
+            return gr.skip(), gr.update(value=error_message, visible=True)
+
+    def update_workflow_from_code(self, yaml_str: str, change_state: bool) -> tuple[td.PipelineStateDict, bool, dict]:
         """Update a workflow from a YAML string."""
-        workflow = self.parse_yaml_workflow(yaml_str)
-        return PipelineState.from_workflow(workflow).model_dump()
+        try:
+            workflow = self.parse_yaml_workflow(yaml_str, strict=True)
+            self.validator and self.validator(workflow)
+            state = self.pipeline_state_cls.from_workflow(workflow)
+            return state.model_dump(), not change_state, gr.update(visible=False)
+        except Exception as e:
+            error_message = self._handle_pipeline_parsing_error(e)
+            return gr.skip(), gr.skip(), gr.update(value=error_message, visible=True)
 
 
 class TossupPipelineStateManager(PipelineStateManager):
     """Manages a tossup pipeline state."""
 
-    def make_pipeline_state(self, state_dict: td.PipelineStateDict) -> TossupPipelineState:
-        """Make a state from a state dictionary."""
-        return TossupPipelineState(**state_dict)
+    pipeline_state_cls = TossupPipelineState
+    workflow_cls = TossupWorkflow
 
-    def parse_yaml_workflow(self, yaml_str: str) -> TossupWorkflow:
-        """Parse a YAML workflow."""
-        workflow = yaml.safe_load(yaml_str)
-        return TossupWorkflow(**workflow)
+    def make_pipeline_state(self, state_dict: td.PipelineStateDict) -> TossupPipelineState:
+        return super().make_pipeline_state(state_dict)
 
-    def update_workflow_from_code(self, yaml_str: str, change_state: bool) -> tuple[td.PipelineStateDict, bool]:
-        """Update a workflow from a YAML string."""
-        workflow = self.parse_yaml_workflow(yaml_str)
-        return TossupPipelineState.from_workflow(workflow).model_dump(), not change_state
+    def update_workflow_from_code(
+        self, yaml_str: str, change_state: bool
+    ) -> tuple[td.TossupPipelineStateDict, bool, dict]:
+        return super().update_workflow_from_code(yaml_str, change_state)
 
     def update_model_step_state(
         self, state_dict: td.TossupPipelineStateDict, model_step: ModelStep, ui_state: ModelStepUIState
     ) -> td.TossupPipelineStateDict:
-        """Update a particular model step in the pipeline."""
-        state = self.make_pipeline_state(state_dict)
-        state = state.update_step(model_step, ui_state)
-        state.workflow = state.workflow.refresh_buzzer()
-        return state.model_dump()
+        return super().update_model_step_state(state_dict, model_step, ui_state)
 
     def update_output_variables(
         self, state_dict: td.TossupPipelineStateDict, target: str, produced_variable: str
     ) -> td.TossupPipelineStateDict:
-        if produced_variable == UNSELECTED_VAR_NAME:
-            produced_variable = None
-        """Update the output variables for a step."""
-        state = self.make_pipeline_state(state_dict)
-        state.workflow.outputs[target] = produced_variable
-        state.workflow = state.workflow.refresh_buzzer()
-        return state.model_dump()
+        return super().update_output_variables(state_dict, target, produced_variable)
 
     def update_buzzer(
         self,

src/components/model_pipeline/tossup_pipeline.py

@@ -1,5 +1,4 @@
 import gradio as gr
-import numpy as np
 from loguru import logger
 
 from app_configs import AVAILABLE_MODELS, UNSELECTED_VAR_NAME
@@ -9,7 +8,8 @@ from components.typed_dicts import TossupPipelineStateDict
 from display.formatting import tiny_styled_warning
 from workflows.structs import Buzzer, TossupWorkflow
 
-from .model_pipeline import PipelineInterface, PipelineState, PipelineUIState
+from .model_pipeline import PipelineInterface
+from .state_manager import PipelineUIState, PipelineValidator
 
 
 def toggleable_slider(
@@ -40,8 +40,9 @@ class TossupPipelineInterface(PipelineInterface):
         ui_state: PipelineUIState | None = None,
         model_options: list[str] = None,
         config: dict = {},
+        validator: PipelineValidator | None = None,
     ):
-        super().__init__(app, workflow, ui_state, model_options, config)
+        super().__init__(app, workflow, ui_state, model_options, config, validator)
 
         self.buzzer_state = gr.State(workflow.buzzer.model_dump())
 

src/components/quizbowl/bonus.py

@@ -19,6 +19,7 @@ from workflows.qb_agents import QuizBowlBonusAgent
 from . import populate, validation
 from .plotting import create_bonus_confidence_plot, create_bonus_html
 from .utils import evaluate_prediction
+from .validation import UserInputWorkflowValidator
 
 
 def process_bonus_results(results: list[dict]) -> pd.DataFrame:
@@ -105,6 +106,7 @@ class BonusInterface:
             ui_state=pipeline_state.ui_state,
             model_options=list(self.model_options.keys()),
             config=self.defaults,
+            validator=UserInputWorkflowValidator("bonus"),
         )
 
     def _render_qb_interface(self):

src/components/quizbowl/tossup.py

@@ -25,6 +25,7 @@ from .plotting import (
     prepare_tossup_results_df,
 )
 from .utils import evaluate_prediction
+from .validation import UserInputWorkflowValidator
 
 # TODO: Error handling on run tossup and evaluate tossup and show correct messages
 # TODO: ^^ Same for Bonus
@@ -135,7 +136,7 @@ class TossupInterface:
         self.output_state = gr.State(value={})
         self.render()
 
-    # ------------------------------------- LOAD PIPELINE STATE FROM BROWSER STATE -------------------------------------
+    # ------------------------------------- LOAD PIPELINE STATE FROM BROWSER STATE ------------------------------------
 
     def load_presaved_pipeline_state(self, browser_state: dict, pipeline_change: bool):
         logger.debug(f"Loading presaved pipeline state from browser state:\n{json.dumps(browser_state, indent=4)}")
@@ -165,6 +166,7 @@ class TossupInterface:
             ui_state=pipeline_state.ui_state,
             model_options=list(self.model_options.keys()),
             config=self.defaults,
+            validator=UserInputWorkflowValidator("tossup"),
         )
 
     def _render_qb_interface(self):

src/components/quizbowl/validation.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from app_configs import CONFIGS
 from components.structs import PipelineState, TossupPipelineState
 from components.typed_dicts import PipelineStateDict, TossupPipelineStateDict
@@ -53,3 +55,31 @@ def validate_bonus_workflow(pipeline_state_dict: PipelineStateDict):
         CONFIGS["bonus"]["required_output_vars"],
     )
     return pipeline_state
+
+
+class UserInputWorkflowValidator:
+    def __init__(self, mode: Literal["tossup", "bonus"]):
+        self.required_input_vars = CONFIGS[mode]["required_input_vars"]
+        self.required_output_vars = CONFIGS[mode]["required_output_vars"]
+
+    def __call__(self, workflow: TossupWorkflow):
+        input_vars = set(workflow.inputs)
+        for req_var in self.required_input_vars:
+            if req_var not in input_vars:
+                default_str = "inputs:\n" + "\n".join([f"- {var}" for var in self.required_input_vars])
+                raise ValueError(
+                    f"Missing required input variable: '{req_var}'. "
+                    "\nDon't modify the 'inputs' field in the workflow. "
+                    "Please set it back to:"
+                    f"\n{default_str}"
+                )
+
+        output_vars = set(workflow.outputs)
+        for req_var in self.required_output_vars:
+            if req_var not in output_vars:
+                default_str = "[" + ", ".join([f"'{var}'" for var in self.required_output_vars]) + "]"
+                raise ValueError(
+                    f"Missing required output variable: '{req_var}'. "
+                    "\nDon't remove the keys from the 'outputs' field in the workflow. Only update their values."
+                    f"\nMake sure you have values set for all the outputs: {default_str}"
+                )

src/components/structs.py

@@ -143,6 +143,11 @@ class PipelineState(BaseModel):
             update["ui_state"] = self.ui_state.update_step(step.id, ui_state)
         return self.model_copy(update=update)
 
+    def update_output_variable(self, target: str, produced_variable: str) -> "PipelineState":
+        """Update the output variables for a step."""
+        self.workflow.outputs[target] = produced_variable
+        return self
+
     def get_available_variables(self, model_step_id: str | None = None) -> list[str]:
         """Get all variables from all steps."""
         available_variables = self.available_variables
@@ -170,3 +175,14 @@ class PipelineState(BaseModel):
 
 class TossupPipelineState(PipelineState):
     workflow: TossupWorkflow
+
+    def update_step(self, step: ModelStep, ui_state: ModelStepUIState | None = None) -> "TossupPipelineState":
+        """Update a step in the pipeline."""
+        state = super().update_step(step, ui_state)
+        state.workflow = state.workflow.refresh_buzzer()
+        return state
+
+    def update_output_variable(self, target: str, produced_variable: str) -> "TossupPipelineState":
+        state = super().update_output_variable(target, produced_variable)
+        state.workflow = state.workflow.refresh_buzzer()
+        return state

src/workflows/validators.py

@@ -13,7 +13,6 @@ SUPPORTED_TYPES = {"str", "int", "float", "bool", "list[str]", "list[int]", "lis
 MAX_FIELD_NAME_LENGTH = 50
 MAX_DESCRIPTION_LENGTH = 200
 MAX_SYSTEM_PROMPT_LENGTH = 4000
-MIN_TEMPERATURE = 0.0
 MAX_TEMPERATURE = 10.0
 
 
@@ -40,7 +39,7 @@ class ValidationError:
     field_name: Optional[str] = None
 
 
-class WorkflowValidationError(Exception):
+class WorkflowValidationError(ValueError):
     """Base class for workflow validation errors"""
 
     def __init__(self, errors: list[ValidationError]):
@@ -77,9 +76,18 @@ def create_step_dep_graph(workflow: Workflow) -> dict[str, set[str]]:
 class WorkflowValidator:
     """Validates workflows for correctness and consistency"""
 
-    def __init__(self):
+    def __init__(
+        self,
+        min_temperature: float = 0,
+        max_temperature: float = MAX_TEMPERATURE,
+        max_field_name_length: int = MAX_FIELD_NAME_LENGTH,
+        max_description_length: int = MAX_DESCRIPTION_LENGTH,
+        max_system_prompt_length: int = MAX_SYSTEM_PROMPT_LENGTH,
+    ):
         self.errors: list[ValidationError] = []
         self.workflow: Optional[Workflow] = None
+        self.min_temperature = min_temperature
+        self.max_temperature = max_temperature
 
     def validate(self, workflow: Workflow) -> bool:
         """Main validation entry point"""
@@ -272,7 +280,7 @@ class WorkflowValidator:
             self.errors.append(
                 ValidationError(
                     ValidationErrorType.NAMING,
-                    f"Invalid step ID format: {step.id}. Must be a valid Python identifier.",
+                    f"Invalid step ID format: {step.id}. Must be a valid identifier.",
                     step.id,
                 )
             )
@@ -286,11 +294,11 @@ class WorkflowValidator:
                 )
                 return False
 
-            if not MIN_TEMPERATURE <= step.temperature <= MAX_TEMPERATURE:
+            if not self.min_temperature <= step.temperature <= self.max_temperature:
                 self.errors.append(
                     ValidationError(
                         ValidationErrorType.RANGE,
-                        f"Temperature must be between {MIN_TEMPERATURE} and {MAX_TEMPERATURE}",
+                        f"Temperature must be between {self.min_temperature} and {self.max_temperature}",
                         step.id,
                     )
                 )
@@ -304,11 +312,11 @@ class WorkflowValidator:
                 )
                 return False
 
-            if len(step.system_prompt) > MAX_SYSTEM_PROMPT_LENGTH:
+            if len(step.system_prompt) > self.max_system_prompt_length:
                 self.errors.append(
                     ValidationError(
                         ValidationErrorType.LENGTH,
-                        f"System prompt exceeds maximum length of {MAX_SYSTEM_PROMPT_LENGTH} characters",
+                        f"System prompt exceeds maximum length of {self.max_system_prompt_length} characters",
                         step.id,
                     )
                 )
@@ -365,22 +373,22 @@ class WorkflowValidator:
             return False
 
         # Validate field name length
-        if len(field.name) > MAX_FIELD_NAME_LENGTH:
+        if len(field.name) > self.max_field_name_length:
             self.errors.append(
                 ValidationError(
                     ValidationErrorType.LENGTH,
-                    f"Field name exceeds maximum length of {MAX_FIELD_NAME_LENGTH} characters",
+                    f"Field name exceeds maximum length of {self.max_field_name_length} characters",
                     field_name=field.name,
                 )
             )
             return False
 
         # Validate description length
-        if len(field.description) > MAX_DESCRIPTION_LENGTH:
+        if len(field.description) > self.max_description_length:
             self.errors.append(
                 ValidationError(
                     ValidationErrorType.LENGTH,
-                    f"Description exceeds maximum length of {MAX_DESCRIPTION_LENGTH} characters",
+                    f"Description exceeds maximum length of {self.max_description_length} characters",
                     field_name=field.name,
                 )
             )
@@ -422,22 +430,22 @@ class WorkflowValidator:
             return False
 
         # Validate field name length
-        if len(field.name) > MAX_FIELD_NAME_LENGTH:
+        if len(field.name) > self.max_field_name_length:
             self.errors.append(
                 ValidationError(
                     ValidationErrorType.LENGTH,
-                    f"Field name exceeds maximum length of {MAX_FIELD_NAME_LENGTH} characters",
+                    f"Field name exceeds maximum length of {self.max_field_name_length} characters",
                     field_name=field.name,
                 )
             )
             return False
 
         # Validate description length
-        if len(field.description) > MAX_DESCRIPTION_LENGTH:
+        if len(field.description) > self.max_description_length:
             self.errors.append(
                 ValidationError(
                     ValidationErrorType.LENGTH,
-                    f"Description exceeds maximum length of {MAX_DESCRIPTION_LENGTH} characters",
+                    f"Description exceeds maximum length of {self.max_description_length} characters",
                     field_name=field.name,
                 )
             )
@@ -545,10 +553,6 @@ class WorkflowValidator:
 
     def _is_valid_identifier(self, name: str) -> bool:
         """Validates if a string is a valid Python identifier"""
-        if not name:
-            return False
-        if keyword.iskeyword(name):
-            return False
-        if not name.strip():  # Check for whitespace-only strings
-            return False
-        return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", name))
+        if name and name.strip():
+            return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", name))
+        return False