175 lines
6.6 KiB
Python
Raw Normal View History

Support Component Validation API in AGS (#5503) <!-- Thank you for your contribution! Please review https://microsoft.github.io/autogen/docs/Contribute before opening a pull request. --> <!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. --> ## Why are these changes needed? It is useful to rapidly validate any changes to a team structure as teams are built either via drag and drop or by modifying the underlying spec You can now “validate” your team. The key ideas are as follows - Each team is based on some Component Config specification which is a pedantic model underneath. - Validation is 3 pronged based on a ValidatorService class - Data model validation (validate component schema) - Instantiation validation (validate component can be instantiated) - Provider validation, component_type validation (validate that provider exists and can be imported) - UX: each time a component is **loaded or saved**, it is automatically validated and any errors shown (via a server endpoint). This way, the developer immediately knows if updates to the configuration is wrong or has errors. > Note: this is different from actually running the component against a task. Currently you can run the entire team. In a separate PR we will implement ability to run/test other components. <img width="1360" alt="image" src="https://github.com/user-attachments/assets/d61095b7-0b07-463a-b4b2-5c50ded750f6" /> <img width="1368" alt="image" src="https://github.com/user-attachments/assets/09a1677e-76e8-44a4-9749-15c27457efbb" /> <!-- Please give a short summary of the change and the problem this solves. --> ## Related issue number Closes #4616 <!-- For example: "Closes #1234" --> ## Checks - [ ] I've included any doc changes needed for https://microsoft.github.io/autogen/. See https://microsoft.github.io/autogen/docs/Contribute#documentation to build and test documentation locally. - [ ] I've added tests (if relevant) corresponding to the changes introduced in this PR. - [ ] I've made sure all auto checks have passed.
2025-02-12 09:29:43 -08:00
# api/routes/validation.py
import importlib
from typing import Any, Dict, List, Optional
from autogen_core import ComponentModel, is_component_class
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter()
class ValidationRequest(BaseModel):
component: Dict[str, Any]
class ValidationError(BaseModel):
field: str
error: str
suggestion: Optional[str] = None
class ValidationResponse(BaseModel):
is_valid: bool
errors: List[ValidationError] = []
warnings: List[ValidationError] = []
class ValidationService:
@staticmethod
def validate_provider(provider: str) -> Optional[ValidationError]:
"""Validate that the provider exists and can be imported"""
try:
if provider in ["azure_openai_chat_completion_client", "AzureOpenAIChatCompletionClient"]:
provider = "autogen_ext.models.openai.AzureOpenAIChatCompletionClient"
elif provider in ["openai_chat_completion_client", "OpenAIChatCompletionClient"]:
provider = "autogen_ext.models.openai.OpenAIChatCompletionClient"
module_path, class_name = provider.rsplit(".", maxsplit=1)
module = importlib.import_module(module_path)
component_class = getattr(module, class_name)
if not is_component_class(component_class):
return ValidationError(
field="provider",
error=f"Class {provider} is not a valid component class",
suggestion="Ensure the class inherits from Component and implements required methods",
)
return None
except ImportError:
return ValidationError(
field="provider",
error=f"Could not import provider {provider}",
suggestion="Check that the provider module is installed and the path is correct",
)
except Exception as e:
return ValidationError(
field="provider",
error=f"Error validating provider: {str(e)}",
suggestion="Check the provider string format and class implementation",
)
@staticmethod
def validate_component_type(component: Dict[str, Any]) -> Optional[ValidationError]:
"""Validate the component type"""
if "component_type" not in component:
return ValidationError(
field="component_type",
error="Component type is missing",
suggestion="Add a component_type field to the component configuration",
)
return None
@staticmethod
def validate_config_schema(component: Dict[str, Any]) -> List[ValidationError]:
"""Validate the component configuration against its schema"""
errors = []
try:
# Convert to ComponentModel for initial validation
model = ComponentModel(**component)
# Get the component class
provider = model.provider
module_path, class_name = provider.rsplit(".", maxsplit=1)
module = importlib.import_module(module_path)
component_class = getattr(module, class_name)
# Validate against component's schema
if hasattr(component_class, "component_config_schema"):
try:
component_class.component_config_schema.model_validate(model.config)
except Exception as e:
errors.append(
ValidationError(
field="config",
error=f"Config validation failed: {str(e)}",
suggestion="Check that the config matches the component's schema",
)
)
else:
errors.append(
ValidationError(
field="config",
error="Component class missing config schema",
suggestion="Implement component_config_schema in the component class",
)
)
except Exception as e:
errors.append(
ValidationError(
field="config",
error=f"Schema validation error: {str(e)}",
suggestion="Check the component configuration format",
)
)
return errors
@staticmethod
def validate_instantiation(component: Dict[str, Any]) -> Optional[ValidationError]:
"""Validate that the component can be instantiated"""
try:
model = ComponentModel(**component)
# Attempt to load the component
module_path, class_name = model.provider.rsplit(".", maxsplit=1)
module = importlib.import_module(module_path)
component_class = getattr(module, class_name)
component_class.load_component(model)
return None
except Exception as e:
return ValidationError(
field="instantiation",
error=f"Failed to instantiate component: {str(e)}",
suggestion="Check that the component can be properly instantiated with the given config",
)
@classmethod
def validate(cls, component: Dict[str, Any]) -> ValidationResponse:
"""Validate a component configuration"""
errors = []
warnings = []
# Check provider
if provider_error := cls.validate_provider(component.get("provider", "")):
errors.append(provider_error)
# Check component type
if type_error := cls.validate_component_type(component):
errors.append(type_error)
# Validate schema
schema_errors = cls.validate_config_schema(component)
errors.extend(schema_errors)
# Only attempt instantiation if no errors so far
if not errors:
if inst_error := cls.validate_instantiation(component):
errors.append(inst_error)
# Check for version warnings
if "version" not in component:
warnings.append(
ValidationError(
field="version",
error="Component version not specified",
suggestion="Consider adding a version to ensure compatibility",
)
)
return ValidationResponse(is_valid=len(errors) == 0, errors=errors, warnings=warnings)
@router.post("/")
async def validate_component(request: ValidationRequest) -> ValidationResponse:
"""Validate a component configuration"""
return ValidationService.validate(request.component)