Maharshi Gor
Separated playground and evaluation models.
f94c32c
raw
history blame
4.66 kB
from typing import Literal
from app_configs import CONFIGS, PLAYGROUND_MODELS, SUBMISSION_MODELS
from display.formatting import styled_error
from shared.workflows.errors import ProviderAPIError, WorkflowExecutionError
from shared.workflows.structs import TossupWorkflow, Workflow
from shared.workflows.validators import ValidationErrorType, WorkflowValidationError, WorkflowValidator
class UnsupportedModelError(Exception):
"""Exception for unsupported model errors."""
def __init__(self, model_name: str, playground: bool = False):
self.playground = playground
self.model_name = model_name
def __str__(self):
if self.playground:
return f"Model '{self.model_name}' is not supported in playground mode. Please use one of the supported playground models: {PLAYGROUND_MODELS}.\n But, you can use the full suite for submission."
return (
f"Model '{self.model_name}' is not supported. Please use one of the supported models: {SUBMISSION_MODELS}."
)
def create_error_message(e: Exception) -> str:
"""Create an error message for a given exception."""
if isinstance(e, UnsupportedModelError):
return styled_error(str(e))
if isinstance(e, ProviderAPIError):
return styled_error(
f"Our {e.provider} models are currently experiencing issues. Please try again later. \n\nIf the problem persists, please contact support."
)
elif isinstance(e, WorkflowExecutionError):
return styled_error(
f"Workflow execution failed: {e}. Please try again later. \n\nIf the problem persists, please contact support."
)
elif isinstance(e, ValueError):
return styled_error(
f"Invalid input -- {e}. Please try again. \n\nIf the problem persists, please contact support."
)
else:
return styled_error("An unexpected error occurred. Please contact support.")
class InterfaceWorkflowValidator:
def __init__(self, mode: Literal["tossup", "bonus"], editing: bool = False):
self.mode = mode
self.required_input_vars = CONFIGS[mode]["required_input_vars"]
self.required_output_vars = CONFIGS[mode]["required_output_vars"]
self.editing = editing
def __call__(self, workflow: TossupWorkflow | Workflow, playground: bool = False):
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}"
)
# Validate the workflow
allowed_models = PLAYGROUND_MODELS if playground else SUBMISSION_MODELS
self.validator = WorkflowValidator(allowed_model_names=allowed_models)
try:
self.validator.validate(workflow, allow_empty=self.editing)
except WorkflowValidationError as e:
if e.errors and e.errors[0].error_type == ValidationErrorType.UNSUPPORTED_MODEL:
step_id = e.errors[0].step_id
model_name = workflow.steps[step_id].get_full_model_name()
raise UnsupportedModelError(model_name, playground=playground)
error_msg_total = f"Found {len(e.errors)} errors in the workflow:\n"
error_msg_list = [f"- {err.message}" for err in e.errors]
error_msg = error_msg_total + "\n".join(error_msg_list)
raise ValueError(error_msg)
def validate_state_dict(self, state_dict: dict, playground: bool = False):
"""Validate a state dictionary."""
if self.mode == "tossup":
workflow = TossupWorkflow(**state_dict["workflow"])
else:
workflow = Workflow(**state_dict["workflow"])
self(workflow, playground=playground)
return workflow