import gradio as gr
import json
import logging
from enum import Enum, auto
from typing import Protocol, List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import difflib
import pytest
from concurrent.futures import ThreadPoolExecutor
import asyncio
# Initialize logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class Config:
"""Configuration class for the agent system"""
rag_system_path: str
max_workers: int = 10
log_level: str = "INFO"
model_settings: Dict[str, Any] = field(default_factory=dict)
api_keys: Dict[str, str] = field(default_factory=dict)
design_system_config: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
"""Validate configuration after initialization"""
if not self.rag_system_path:
raise ValueError("RAG system path must be specified in config")
if not self.design_system_config:
self.design_system_config = {
'theme': 'light',
'responsive': True,
'accessibility_level': 'AAA'
}
class DesignSystem:
def __init__(self):
self.theme = {
'colors': {
'primary': '#007bff',
'secondary': '#6c757d',
'success': '#28a745',
'danger': '#dc3545',
'warning': '#ffc107',
'info': '#17a2b8',
'light': '#f8f9fa',
'dark': '#343a40'
},
'typography': {
'fontFamily': {
'primary': '"Helvetica Neue", Arial, sans-serif',
'secondary': 'Georgia, serif',
'monospace': 'Monaco, Consolas, monospace'
},
'fontSizes': {
'xs': '12px',
'sm': '14px',
'base': '16px',
'lg': '18px',
'xl': '20px',
'xxl': '24px',
'display': '32px'
},
'lineHeight': {
'tight': '1.25',
'normal': '1.5',
'relaxed': '1.75'
}
},
'spacing': {
'unit': '8px',
'scales': {
'xs': '0.25rem',
'sm': '0.5rem',
'md': '1rem',
'lg': '1.5rem',
'xl': '2rem'
}
},
'breakpoints': {
'mobile': '320px',
'tablet': '768px',
'desktop': '1024px',
'large': '1440px'
}
}
def get_color(self, color_name: str) -> str:
return self.theme['colors'].get(color_name)
def get_font_family(self, type_name: str) -> str:
return self.theme['typography']['fontFamily'].get(type_name)
def get_spacing(self, scale: str) -> str:
return self.theme['spacing']['scales'].get(scale)
class UIComponentLibrary:
def __init__(self, design_system: DesignSystem):
self.design_system = design_system
def get_component_template(self, component_type: str) -> Dict[str, Any]:
components = {
'button': {
'template': '''
''',
'styles': self._get_button_styles()
},
'card': {
'template': '''
''',
'styles': self._get_card_styles()
},
'input': {
'template': '''
{{helper_text}}
''',
'styles': self._get_input_styles()
}
}
return components.get(component_type)
def _get_button_styles(self) -> Dict[str, str]:
return {
'base': f'''
font-family: {self.design_system.get_font_family('primary')};
padding: {self.design_system.get_spacing('sm')} {self.design_system.get_spacing('md')};
border-radius: 4px;
border: none;
cursor: pointer;
transition: all 0.3s ease;
''',
'primary': f'background-color: {self.design_system.get_color("primary")};',
'secondary': f'background-color: {self.design_system.get_color("secondary")};'
}
def _get_card_styles(self) -> Dict[str, str]:
return {
'base': f'''
border-radius: 8px;
padding: {self.design_system.get_spacing('md')};
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
'''
}
def _get_input_styles(self) -> Dict[str, str]:
return {
'base': f'''
font-family: {self.design_system.get_font_family('primary')};
padding: {self.design_system.get_spacing('sm')};
border-radius: 4px;
border: 1px solid {self.design_system.get_color('secondary')};
'''
}
class WebDesignValidator:
def __init__(self):
self.accessibility_rules = {
'aria_labels': True,
'alt_texts': True,
'semantic_html': True,
'keyboard_navigation': True,
'color_contrast': True,
'focus_indicators': True
}
self.responsive_breakpoints = {
'mobile': '320px',
'tablet': '768px',
'desktop': '1024px',
'large': '1440px'
}
async def validate_design_principles(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
results = {
'accessibility': await self._check_accessibility(implementation),
'responsive_design': await self._check_responsive_design(implementation),
'color_contrast': await self._check_color_contrast(implementation),
'typography': await self._check_typography(implementation),
'performance': await self._check_performance(implementation)
}
return {
'validation_results': results,
'overall_status': self._calculate_overall_status(results),
'recommendations': self._generate_recommendations(results)
}
async def _check_accessibility(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
results = {
'passed': [],
'failed': [],
'warnings': []
}
for component in implementation.get('components', []):
for rule, enabled in self.accessibility_rules.items():
if enabled:
result = await self._validate_accessibility_rule(component, rule)
if result['status'] == 'passed':
results['passed'].append(f"{component['name']}: {rule}")
elif result['status'] == 'failed':
results['failed'].append({
'component': component['name'],
'rule': rule,
'message': result['message']
})
else:
results['warnings'].append({
'component': component['name'],
'rule': rule,
'message': result['message']
})
return results
async def _validate_accessibility_rule(self, component: Dict[str, Any], rule: str) -> Dict[str, Any]:
validators = {
'aria_labels': self._check_aria_labels,
'alt_texts': self._check_alt_texts,
'semantic_html': self._check_semantic_html,
'keyboard_navigation': self._check_keyboard_navigation,
'color_contrast': self._check_specific_color_contrast,
'focus_indicators': self._check_focus_indicators
}
validator = validators.get(rule)
if validator:
return await validator(component)
return {'status': 'warning', 'message': f'No validator found for {rule}'}
async def _check_responsive_design(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
results = {
'breakpoints_tested': [],
'layout_issues': [],
'recommendations': []
}
for breakpoint, size in self.responsive_breakpoints.items():
test_result = await self._test_breakpoint(implementation, breakpoint, size)
results['breakpoints_tested'].append({
'breakpoint': breakpoint,
'size': size,
'result': test_result
})
if test_result.get('issues'):
results['layout_issues'].extend(test_result['issues'])
results['recommendations'].extend(test_result['recommendations'])
return results
async def _test_breakpoint(self, implementation: Dict[str, Any], breakpoint: str, size: str) -> Dict[str, Any]:
# Simulate testing at different viewport sizes
return {
'breakpoint': breakpoint,
'size': size,
'layout_integrity': True,
'issues': [],
'recommendations': []
}
async def _check_color_contrast(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
results = {
'passed': [],
'failed': [],
'warnings': []
}
for component in implementation.get('components', []):
contrast_ratio = await self._calculate_contrast_ratio(
component.get('foreground_color'),
component.get('background_color')
)
if contrast_ratio >= 4.5: # WCAG AA standard
results['passed'].append({
'component': component['name'],
'ratio': contrast_ratio
})
else:
results['failed'].append({
'component': component['name'],
'ratio': contrast_ratio,
'recommendation': 'Increase contrast to at least 4.5:1'
})
return results
async def _check_typography(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
return {
'font_hierarchy': await self._check_font_hierarchy(implementation),
'line_height': await self._check_line_height(implementation),
'font_sizes': await self._check_font_sizes(implementation),
'recommendations': []
}
async def _check_performance(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
return {
'image_optimization': await self._check_image_optimization(implementation),
'css_optimization': await self._check_css_optimization(implementation),
'loading_time': await self._check_loading_time(implementation),
'recommendations': []
}
def _calculate_overall_status(self, results: Dict[str, Any]) -> str:
# Implementation of overall status calculation
failed_count = sum(len(category.get('failed', [])) for category in results.values())
warning_count = sum(len(category.get('warnings', [])) for category in results.values())
if failed_count == 0 and warning_count == 0:
return 'PASSED'
elif failed_count == 0:
return 'PASSED_WITH_WARNINGS'
else:
return 'FAILED'
def _generate_recommendations(self, results: Dict[str, Any]) -> List[str]:
recommendations = []
for category, result in results.items():
if 'failed' in result:
for failure in result['failed']:
recommendations.append(f"Fix {category}: {failure}")
if 'warnings' in result:
for warning in result['warnings']:
recommendations.append(f"Consider improving {category}: {warning}")
return recommendations
class AgentRole(Enum):
ARCHITECT = auto()
FRONTEND = auto()
BACKEND = auto()
DATABASE = auto()
TESTER = auto()
REVIEWER = auto()
DEPLOYER = auto()
DESIGNER = auto()
@dataclass
class AgentDecision:
agent: 'Agent'
decision: str
confidence: float
reasoning: str
timestamp: datetime = field(default_factory=datetime.now)
dependencies: List['AgentDecision'] = field(default_factory=list)
design_implications: Dict[str, Any] = field(default_factory=dict)
class Agent:
def __init__(self, role: AgentRole, design_system: DesignSystem, ui_library: UIComponentLibrary):
self.role = role
self.design_system = design_system
self.ui_library = ui_library
self.decisions: List[AgentDecision] = []
self.current_task: Optional[Dict[str, Any]] = None
self.autonomy_level: float = 5.0 # Scale of 0-10
self.validator = WebDesignValidator()
async def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
self.current_task = task
try:
if self.role == AgentRole.DESIGNER:
return await self._process_design_task(task)
elif self.role == AgentRole.FRONTEND:
return await self._process_frontend_task(task)
elif self.role == AgentRole.BACKEND:
return await self._process_backend_task(task)
else:
return await self._process_generic_task(task)
except Exception as e:
logger.error(f"Error processing task in {self.role}: {str(e)}")
return {'status': 'error', 'message': str(e)}
async def _process_design_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
design_requirements = task.get('design_requirements', {})
# Create design specifications
design_specs = {
'layout': await self._create_layout_specs(design_requirements),
'components': await self._create_component_specs(design_requirements),
'styling': await self._create_styling_specs(design_requirements),
'interactions': await self._create_interaction_specs(design_requirements)
}
# Validate design specifications
validation_results = await self.validator.validate_design_principles({
'specs': design_specs,
'requirements': design_requirements
})
if validation_results['overall_status'] == 'FAILED':
return {
'status': 'revision_needed',
'specs': design_specs,
'validation': validation_results,
'recommendations': validation_results['recommendations']
}
return {
'status': 'success',
'specs': design_specs,
'validation': validation_results
}
async def _create_layout_specs(self, requirements: Dict[str, Any]) -> Dict[str, Any]:
return {
'grid_system': {
'type': 'flexbox',
'columns': 12,
'breakpoints': self.design_system.theme['breakpoints']
},
'layout_type': requirements.get('layout_type', 'responsive'),
'containers': {
'max_width': '1200px',
'padding': self.design_system.get_spacing('md')
}
}
async def _create_component_specs(self, requirements: Dict[str, Any]) -> Dict[str, Any]:
components = {}
for component_type in requirements.get('components', []):
template = self.ui_library.get_component_template(component_type)
if template:
components[component_type] = {
'template': template['template'],
'styles': template['styles'],
'variants': self._generate_component_variants(component_type)
}
return components
async def _process_frontend_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
design_specs = task.get('design_specs', {})
implementation = {
'components': await self._implement_components(design_specs),
'styles': await self._implement_styles(design_specs),
'layouts': await self._implement_layouts(design_specs)
}
# Validate frontend implementation
validation_results = await self.validator.validate_design_principles(implementation)
return {
'status': 'success',
'implementation': implementation,
'validation': validation_results
}
async def _implement_components(self, design_specs: Dict[str, Any]) -> List[Dict[str, Any]]:
implemented_components = []
for component_name, specs in design_specs.get('components', {}).items():
implemented_components.append({
'name': component_name,
'implementation': self.ui_library.get_component_template(component_name),
'styles': specs.get('styles', {}),
'variants': specs.get('variants', [])
})
return implemented_components
class AgentSystem:
def __init__(self, config: Config):
self.config = config
self.design_system = DesignSystem()
self.ui_library = UIComponentLibrary(self.design_system)
self.validator = WebDesignValidator()
self.agents = self._initialize_agents()
self.executor = ThreadPoolExecutor(max_workers=config.max_workers)
self.current_project: Optional[Dict[str, Any]] = None
def _initialize_agents(self) -> Dict[AgentRole, Agent]:
return {
role: Agent(role, self.design_system, self.ui_library)
for role in AgentRole
}
async def process_request(self, description: str, context: Dict[str, Any]) -> Dict[str, Any]:
try:
self.current_project = {
'description': description,
'context': context,
'status': 'initializing',
'timestamp': datetime.now()
}
# Design Phase
design_result = await self._process_design_phase(context)
if design_result['status'] != 'success':
return design_result
# Implementation Phase
implementation_result = await self._process_implementation_phase(design_result['specs'])
if implementation_result['status'] != 'success':
return implementation_result
# Validation Phase
validation_result = await self._process_validation_phase(implementation_result['implementation'])
return {
'status': 'success',
'design': design_result,
'implementation': implementation_result,
'validation': validation_result
}
except Exception as e:
logger.error(f"Error processing request: {str(e)}")
return {'status': 'error', 'message': str(e)}
async def _process_design_phase(self, context: Dict[str, Any]) -> Dict[str, Any]:
design_agent = self.agents[AgentRole.DESIGNER]
return await design_agent.process_task({
'type': 'design',
'requirements': context.get('design_requirements', {}),
'constraints': context.get('constraints', {})
})
async def _process_implementation_phase(self, design_specs: Dict[str, Any]) -> Dict[str, Any]:
frontend_result = await self.agents[AgentRole.FRONTEND].process_task({
'type': 'implementation',
'design_specs': design_specs
})
backend_result = await self.agents[AgentRole.BACKEND].process_task({
'type': 'implementation',
'design_specs': design_specs
})
return {
'status': 'success',
'frontend': frontend_result,
'backend': backend_result
}
async def _process_validation_phase(self, implementation: Dict[str, Any]) -> Dict[str, Any]:
return await self.validator.validate_design_principles(implementation)
async def set_autonomy_level(self, level: float) -> None:
if not 0 <= level <= 10:
raise ValueError("Autonomy level must be between 0 and 10")
for agent in self.agents.values():
agent.autonomy_level = level
# Main execution and utility functions
async def create_gradio_interface(agent_system: AgentSystem):
"""Create the Gradio interface for the web design system"""
def process_design_request(description: str, requirements: str) -> Dict[str, Any]:
try:
context = json.loads(requirements) if requirements else {}
result = asyncio.run(agent_system.process_request(description, context))
return json.dumps(result, indent=2)
except Exception as e:
return f"Error: {str(e)}"
interface = gr.Interface(
fn=process_design_request,
inputs=[
gr.Textbox(label="Project Description", lines=3),
gr.Textbox(label="Design Requirements (JSON)", lines=5)
],
outputs=gr.JSON(label="Results"),
title="Web Design Agent System",
description="Enter your project details and design requirements to generate web design specifications and implementation."
)
return interface
class TestSuite:
"""Test suite for the web design system"""
def __init__(self, agent_system: AgentSystem):
self.agent_system = agent_system
self.test_cases = self._initialize_test_cases()
def _initialize_test_cases(self) -> List[Dict[str, Any]]:
return [
{
'name': 'basic_website',
'description': 'Create a basic website with homepage and contact form',
'context': {
'design_requirements': {
'theme': 'light',
'components': ['header', 'footer', 'contact_form'],
'responsive': True
}
}
},
{
'name': 'e_commerce',
'description': 'Create an e-commerce product page',
'context': {
'design_requirements': {
'theme': 'modern',
'components': ['product_card', 'shopping_cart', 'checkout_form'],
'responsive': True
}
}
}
]
async def run_tests(self) -> Dict[str, Any]:
results = {
'passed': [],
'failed': [],
'total_tests': len(self.test_cases),
'timestamp': datetime.now()
}
for test_case in self.test_cases:
try:
result = await self.agent_system.process_request(
test_case['description'],
test_case['context']
)
if result['status'] == 'success':
results['passed'].append({
'test_name': test_case['name'],
'result': result
})
else:
results['failed'].append({
'test_name': test_case['name'],
'error': result
})
except Exception as e:
results['failed'].append({
'test_name': test_case['name'],
'error': str(e)
})
return results
class ProjectManager:
"""Manages web design projects and their lifecycle"""
def __init__(self, agent_system: AgentSystem):
self.agent_system = agent_system
self.projects: Dict[str, Dict[str, Any]] = {}
self.active_project_id: Optional[str] = None
async def create_project(self, name: str, description: str, requirements: Dict[str, Any]) -> str:
project_id = f"proj_{len(self.projects) + 1}"
self.projects[project_id] = {
'name': name,
'description': description,
'requirements': requirements,
'status': 'created',
'created_at': datetime.now(),
'history': []
}
return project_id
async def process_project(self, project_id: str) -> Dict[str, Any]:
if project_id not in self.projects:
raise ValueError(f"Project {project_id} not found")
project = self.projects[project_id]
self.active_project_id = project_id
try:
result = await self.agent_system.process_request(
project['description'],
{'design_requirements': project['requirements']}
)
project['status'] = result['status']
project['history'].append({
'timestamp': datetime.now(),
'action': 'process',
'result': result
})
return result
except Exception as e:
project['status'] = 'error'
project['history'].append({
'timestamp': datetime.now(),
'action': 'process',
'error': str(e)
})
raise
def main():
"""Main entry point for the application"""
# Initialize configuration
config = Config(
rag_system_path="/path/to/rag",
max_workers=10,
log_level="INFO",
model_settings={
'temperature': 0.7,
'max_tokens': 1000
},
api_keys={},
design_system_config={
'theme': 'light',
'responsive': True,
'accessibility_level': 'AAA'
}
)
# Initialize agent system
agent_system = AgentSystem(config)
# Initialize project manager
project_manager = ProjectManager(agent_system)
# Initialize test suite
test_suite = TestSuite(agent_system)
# Create and launch Gradio interface
interface = asyncio.run(create_gradio_interface(agent_system))
interface.launch(share=True)
if __name__ == "__main__":
# Run tests before starting the application
config = Config(rag_system_path="/path/to/rag")
agent_system = AgentSystem(config)
test_suite = TestSuite(agent_system)
# Run tests
test_results = asyncio.run(test_suite.run_tests())
print("Test Results:", json.dumps(test_results, indent=2))
# Start the application
main()
# Pytest functions for testing
def test_agent_system():
config = Config(rag_system_path="/path/to/rag")
agent_system = AgentSystem(config)
# Test design system initialization
assert agent_system.design_system is not None
assert agent_system.ui_library is not None
# Test agent initialization
assert len(agent_system.agents) == len(AgentRole)
# Test validator initialization
assert agent_system.validator is not None
def test_design_validation():
config = Config(rag_system_path="/path/to/rag")
agent_system = AgentSystem(config)
implementation = {
'components': [
{
'name': 'button',
'foreground_color': '#FFFFFF',
'background_color': '#007BFF'
}
]
}
validation_result = asyncio.run(agent_system.validator.validate_design_principles(implementation))
assert validation_result is not None
assert 'validation_results' in validation_result
assert 'overall_status' in validation_result
def test_ui_component_library():
design_system = DesignSystem()
ui_library = UIComponentLibrary(design_system)
# Test component template retrieval
button_template = ui_library.get_component_template('button')
assert button_template is not None
assert 'template' in button_template
assert 'styles' in button_template