Refactored validation code in bonus/tossup interface.

src/components/quizbowl/bonus.py

@@ -6,7 +6,7 @@ import pandas as pd
 from datasets import Dataset
 from loguru import logger
 
-from app_configs import UNSELECTED_PIPELINE_NAME
+from app_configs import CONFIGS, UNSELECTED_PIPELINE_NAME
 from components import commons
 from components.model_pipeline.model_pipeline import PipelineInterface, PipelineState, PipelineUIState
 from components.typed_dicts import PipelineStateDict
@@ -15,13 +15,8 @@ from submission import submit
 from workflows.qb_agents import QuizBowlBonusAgent
 from workflows.structs import ModelStep, Workflow
 
-from . import populate
-from .plotting import (
-    create_bonus_confidence_plot,
-    create_bonus_html,
-    create_scatter_pyplot,
-    update_tossup_plot,
-)
+from . import populate, validation
+from .plotting import create_bonus_confidence_plot, create_bonus_html
 from .utils import evaluate_prediction
 
 
@@ -58,58 +53,6 @@ def initialize_eval_interface(example: dict, model_outputs: list[dict]):
         return f"<div>Error initializing interface: {str(e)}</div>", pd.DataFrame(), "{}"
 
 
-def validate_workflow(workflow: Workflow):
-    """Validate that a workflow is properly configured for the bonus task."""
-    if not workflow.steps:
-        raise ValueError("Workflow must have at least one step")
-
-    # Ensure all steps are properly configured
-    for step_id, step in workflow.steps.items():
-        validate_model_step(step)
-
-    # Check that the workflow has the correct structure
-    input_vars = set(workflow.inputs)
-    if "leadin" not in input_vars or "part" not in input_vars:
-        raise ValueError("Workflow must have 'leadin' and 'part' as inputs")
-
-    output_vars = set(workflow.outputs)
-    if not all(var in output_vars for var in ["answer", "confidence", "explanation"]):
-        raise ValueError("Workflow must produce 'answer', 'confidence', and 'explanation' as outputs")
-
-
-def validate_model_step(model_step: ModelStep):
-    """Validate that a model step is properly configured for the bonus task."""
-    # Check required fields
-    if not model_step.model or not model_step.provider:
-        raise ValueError("Model step must have both model and provider specified")
-
-    if model_step.call_type != "llm":
-        raise ValueError("Model step must have call_type 'llm'")
-
-    # Validate temperature for LLM steps
-    if model_step.temperature is None:
-        raise ValueError("Temperature must be specified for LLM model steps")
-
-    if not (0.0 <= model_step.temperature <= 1.0):
-        raise ValueError(f"Temperature must be between 0.0 and 1.0, got {model_step.temperature}")
-
-    # Validate input fields
-    input_field_names = {field.name for field in model_step.input_fields}
-    if "leadin" not in input_field_names or "part" not in input_field_names:
-        raise ValueError("Model step must have 'leadin' and 'part' input fields")
-
-    # Validate output fields
-    output_field_names = {field.name for field in model_step.output_fields}
-    required_outputs = {"answer", "confidence", "explanation"}
-    if not all(out in output_field_names for out in required_outputs):
-        raise ValueError("Model step must have all required output fields: answer, confidence, explanation")
-
-    # Validate confidence output field is of type float
-    for field in model_step.output_fields:
-        if field.name == "confidence" and field.type != "float":
-            raise ValueError("The 'confidence' output field must be of type 'float'")
-
-
 class BonusInterface:
     """Gradio interface for the Bonus mode."""
 
@@ -128,11 +71,12 @@ class BonusInterface:
         with gr.Row(elem_classes="bonus-header-row form-inline"):
             self.pipeline_selector = commons.get_pipeline_selector([])
             self.load_btn = gr.Button("⬇️ Import Pipeline", variant="secondary")
+        self.import_error_display = gr.HTML(label="Import Error", elem_id="import-error-display", visible=False)
         self.pipeline_interface = PipelineInterface(
             self.app,
             workflow,
-            simple=simple,
             model_options=list(self.model_options.keys()),
+            config=self.defaults,
         )
 
     def _render_qb_interface(self):
@@ -177,6 +121,18 @@ class BonusInterface:
 
         self._setup_event_listeners()
 
+    def validate_workflow(self, state_dict: PipelineStateDict):
+        """Validate the workflow."""
+        try:
+            pipeline_state = PipelineState(**state_dict)
+            validation.validate_workflow(
+                pipeline_state.workflow,
+                required_input_vars=CONFIGS["bonus"]["required_input_vars"],
+                required_output_vars=CONFIGS["bonus"]["required_output_vars"],
+            )
+        except Exception as e:
+            raise gr.Error(f"Error validating workflow: {str(e)}")
+
     def get_new_question_html(self, question_id: int):
         """Get the HTML for a new question."""
         if question_id is None:
@@ -237,10 +193,10 @@ class BonusInterface:
     ) -> tuple[str, Any, Any]:
         """Run the agent in bonus mode."""
         try:
-            pipeline_state = PipelineState(**state_dict)
+            pipeline_state = validation.validate_bonus_workflow(state_dict)
             question_id = int(question_id - 1)
             if not self.ds or question_id < 0 or question_id >= len(self.ds):
-                return "Invalid question ID or dataset not loaded", None, None
+                raise gr.Error("Invalid question ID or dataset not loaded")
 
             example = self.ds[question_id]
             outputs = self.get_model_outputs(example, pipeline_state)
@@ -272,7 +228,7 @@ class BonusInterface:
     def evaluate(self, state_dict: PipelineStateDict, progress: gr.Progress = gr.Progress()):
         """Evaluate the bonus questions."""
         try:
-            pipeline_state = PipelineState(**state_dict)
+            pipeline_state = validation.validate_bonus_workflow(state_dict)
             # Validate inputs
             if not self.ds or not self.ds.num_rows:
                 return "No dataset loaded", None, None
@@ -345,15 +301,11 @@ class BonusInterface:
         self.load_btn.click(
             fn=self.load_pipeline,
             inputs=[self.pipeline_selector, pipeline_change],
-            outputs=[self.pipeline_selector, pipeline_state, pipeline_change, self.error_display],
+            outputs=[self.pipeline_selector, pipeline_state, pipeline_change, self.import_error_display],
         )
         self.pipeline_interface.add_triggers_for_pipeline_export([pipeline_state.change], pipeline_state)
 
         self.run_btn.click(
-            self.pipeline_interface.validate_workflow,
-            inputs=[self.pipeline_interface.pipeline_state],
-            outputs=[],
-        ).success(
             self.single_run,
             inputs=[
                 self.qid_selector,

src/components/quizbowl/tossup.py

@@ -7,7 +7,7 @@ import pandas as pd
 from datasets import Dataset
 from loguru import logger
 
-from app_configs import UNSELECTED_PIPELINE_NAME
+from app_configs import CONFIGS, UNSELECTED_PIPELINE_NAME
 from components import commons
 from components.model_pipeline.model_pipeline import PipelineInterface, PipelineState, PipelineUIState
 from components.model_pipeline.tossup_pipeline import TossupPipelineInterface, TossupPipelineState
@@ -17,7 +17,7 @@ from submission import submit
 from workflows.qb_agents import QuizBowlTossupAgent, TossupResult
 from workflows.structs import ModelStep, TossupWorkflow
 
-from . import populate
+from . import populate, validation
 from .plotting import (
     create_scatter_pyplot,
     create_tossup_confidence_pyplot,
@@ -101,77 +101,6 @@ def process_tossup_results(results: list[dict], top_k_mode: bool = False) -> pd.
     )
 
 
-def validate_workflow(workflow: TossupWorkflow):
-    """
-    Validate that a workflow is properly configured for the tossup task.
-
-    Args:
-        workflow (TossupWorkflow): The workflow to validate
-
-    Raises:
-        ValueError: If the workflow is not properly configured
-    """
-    if not workflow.steps:
-        raise ValueError("Workflow must have at least one step")
-
-    # Ensure all steps are properly configured
-    for step_id, step in workflow.steps.items():
-        validate_model_step(step)
-
-    # Check that the workflow has the correct structure
-    input_vars = set(workflow.inputs)
-    if "question" not in input_vars:
-        raise ValueError("Workflow must have 'question' as an input")
-
-    output_vars = set(workflow.outputs)
-    if not any("answer" in out_var for out_var in output_vars):
-        raise ValueError("Workflow must produce an 'answer' as output")
-    if not any("confidence" in out_var for out_var in output_vars):
-        raise ValueError("Workflow must produce a 'confidence' score as output")
-
-
-def validate_model_step(model_step: ModelStep):
-    """
-    Validate that a model step is properly configured for the tossup task.
-
-    Args:
-        model_step (ModelStep): The model step to validate
-
-    Raises:
-        ValueError: If the model step is not properly configured
-    """
-    # Check required fields
-    if not model_step.model or not model_step.provider:
-        raise ValueError("Model step must have both model and provider specified")
-
-    if model_step.call_type != "llm":
-        raise ValueError("Model step must have call_type 'llm'")
-
-    # Validate temperature for LLM steps
-    if model_step.temperature is None:
-        raise ValueError("Temperature must be specified for LLM model steps")
-
-    if not (0.0 <= model_step.temperature <= 1.0):
-        raise ValueError(f"Temperature must be between 0.0 and 1.0, got {model_step.temperature}")
-
-    # Validate input fields
-    input_field_names = {field.name for field in model_step.input_fields}
-    if "question" not in input_field_names:
-        raise ValueError("Model step must have a 'question' input field")
-
-    # Validate output fields
-    output_field_names = {field.name for field in model_step.output_fields}
-    if "answer" not in output_field_names:
-        raise ValueError("Model step must have an 'answer' output field")
-    if "confidence" not in output_field_names:
-        raise ValueError("Model step must have a 'confidence' output field")
-
-    # Validate confidence output field is of type float
-    for field in model_step.output_fields:
-        if field.name == "confidence" and field.type != "float":
-            raise ValueError("The 'confidence' output field must be of type 'float'")
-
-
 class TossupInterface:
     """Gradio interface for the Tossup mode."""
 
@@ -190,12 +119,12 @@ class TossupInterface:
         with gr.Row(elem_classes="bonus-header-row form-inline"):
             self.pipeline_selector = commons.get_pipeline_selector([])
             self.load_btn = gr.Button("⬇️ Import Pipeline", variant="secondary")
+        self.import_error_display = gr.HTML(label="Import Error", elem_id="import-error-display", visible=False)
         self.pipeline_interface = TossupPipelineInterface(
             self.app,
             workflow,
-            simple=simple,
             model_options=list(self.model_options.keys()),
-            defaults=self.defaults,
+            config=self.defaults,
         )
 
     def _render_qb_interface(self):
@@ -251,14 +180,6 @@ class TossupInterface:
 
         self._setup_event_listeners()
 
-    def validate_workflow(self, state_dict: TossupPipelineStateDict):
-        """Validate the workflow."""
-        try:
-            pipeline_state = TossupPipelineState(**state_dict)
-            validate_workflow(pipeline_state.workflow)
-        except Exception as e:
-            raise gr.Error(f"Error validating workflow: {str(e)}")
-
     def get_new_question_html(self, question_id: int) -> str:
         """Get the HTML for a new question."""
         if question_id is None:
@@ -313,12 +234,12 @@ class TossupInterface:
     ) -> tuple[str, Any, Any]:
         """Run the agent in tossup mode with a system prompt."""
         try:
+            pipeline_state = validation.validate_tossup_workflow(state_dict)
             # Validate inputs
             question_id = int(question_id - 1)
             if not self.ds or question_id < 0 or question_id >= len(self.ds):
-                return "Invalid question ID or dataset not loaded", None, None
+                raise gr.Error("Invalid question ID or dataset not loaded")
             example = self.ds[question_id]
-            pipeline_state = TossupPipelineState(**state_dict)
             outputs = self.get_model_outputs(example, pipeline_state, early_stop)
 
             # Process results and prepare visualization data
@@ -352,7 +273,7 @@ class TossupInterface:
             # Validate inputs
             if not self.ds or not self.ds.num_rows:
                 return "No dataset loaded", None, None
-            pipeline_state = TossupPipelineState(**state_dict)
+            pipeline_state = validation.validate_tossup_workflow(state_dict)
             buzz_counts = 0
             correct_buzzes = 0
             token_positions = []
@@ -397,10 +318,14 @@ class TossupInterface:
         description: str,
         state_dict: TossupPipelineStateDict,
         profile: gr.OAuthProfile = None,
-    ):
+    ) -> str:
         """Submit the model output."""
-        pipeline_state = TossupPipelineState(**state_dict)
-        return submit.submit_model(model_name, description, pipeline_state.workflow, "tossup", profile)
+        try:
+            pipeline_state = validation.validate_tossup_workflow(state_dict)
+            return submit.submit_model(model_name, description, pipeline_state.workflow, "tossup", profile)
+        except Exception as e:
+            logger.exception(f"Error submitting model: {e.args}")
+            return styled_error(f"Error: {str(e)}")
 
     def _setup_event_listeners(self):
         gr.on(
@@ -421,15 +346,11 @@ class TossupInterface:
         self.load_btn.click(
             fn=self.load_pipeline,
             inputs=[self.pipeline_selector, pipeline_change],
-            outputs=[self.pipeline_selector, pipeline_state, pipeline_change, self.error_display],
+            outputs=[self.pipeline_selector, pipeline_state, pipeline_change, self.import_error_display],
         )
         self.pipeline_interface.add_triggers_for_pipeline_export([pipeline_state.change], pipeline_state)
 
         self.run_btn.click(
-            self.pipeline_interface.validate_workflow,
-            inputs=[self.pipeline_interface.pipeline_state],
-            outputs=[],
-        ).success(
             self.single_run,
             inputs=[
                 self.qid_selector,

src/components/quizbowl/validation.py

@@ -0,0 +1,55 @@
+from app_configs import CONFIGS
+from components.structs import PipelineState, TossupPipelineState
+from components.typed_dicts import PipelineStateDict, TossupPipelineStateDict
+from workflows.structs import TossupWorkflow, Workflow
+from workflows.validators import WorkflowValidator
+
+
+def validate_workflow(
+    workflow: TossupWorkflow | Workflow, required_input_vars: list[str], required_output_vars: list[str]
+):
+    """
+    Validate that a workflow is properly configured for the tossup task.
+
+    Args:
+        workflow (TossupWorkflow): The workflow to validate
+
+    Raises:
+        ValueError: If the workflow is not properly configured
+    """
+    if not workflow.steps:
+        raise ValueError("Workflow must have at least one step")
+
+    # Check that the workflow has the correct structure
+    input_vars = set(workflow.inputs)
+    for req_var in required_input_vars:
+        if req_var not in input_vars:
+            raise ValueError(f"Workflow must have '{req_var}' as an input")
+
+    output_vars = set(workflow.outputs)
+    for req_var in required_output_vars:
+        if req_var not in output_vars:
+            raise ValueError(f"Workflow must produce '{req_var}' as an output")
+
+    # Ensure all steps are properly configured
+    WorkflowValidator().validate(workflow)
+
+
+def validate_tossup_workflow(pipeline_state_dict: TossupPipelineStateDict) -> TossupPipelineState:
+    pipeline_state = TossupPipelineState(**pipeline_state_dict)
+    validate_workflow(
+        pipeline_state.workflow,
+        CONFIGS["tossup"]["required_input_vars"],
+        CONFIGS["tossup"]["required_output_vars"],
+    )
+    return pipeline_state
+
+
+def validate_bonus_workflow(pipeline_state_dict: PipelineStateDict):
+    pipeline_state = PipelineState(**pipeline_state_dict)
+    validate_workflow(
+        pipeline_state.workflow,
+        CONFIGS["bonus"]["required_input_vars"],
+        CONFIGS["bonus"]["required_output_vars"],
+    )
+    return pipeline_state