| from celery import shared_task | |
| from .data_quality import DataQualityRunner | |
| from .inference_deployer import deploy_to_huggingface | |
| from .models import * | |
| from .monitoring import run_monitors_for_endpoint | |
| from .pipeline_engine import PipelineExecutor | |
| def deploy_inference_endpoint(self, endpoint_id: str): | |
| endpoint = InferenceEndpoint.objects.get(id=endpoint_id) | |
| endpoint.status = "deploying" | |
| endpoint.save() | |
| try: | |
| url = deploy_to_huggingface(endpoint.model_version, endpoint.config) | |
| endpoint.endpoint_url = url | |
| endpoint.status = "running" | |
| endpoint.save() | |
| except Exception as e: | |
| endpoint.status = "failed" | |
| endpoint.save() | |
| raise self.retry(exc=e) | |
| def execute_pipeline(self, run_id: str): | |
| run = PipelineRun.objects.get(id=run_id) | |
| executor = PipelineExecutor(run) | |
| executor.execute() | |
| def run_data_quality_checks(): | |
| for rule in DataQualityRule.objects.filter(is_active=True): | |
| DataQualityRunner.run_check(rule, rule.config.get("dataset_path", "")) | |
| def run_ctcd_pipeline(self, run_id: str): | |
| run = CTCDRun.objects.get(id=run_id) | |
| # 1. Train | |
| run.status = "training" | |
| run.save() | |
| from experiment.models import Experiment, TrainingRun | |
| exp = Experiment.objects.create(name=f"CTCD: {run.pipeline.name}") | |
| training_run = TrainingRun.objects.create( | |
| experiment=exp, config=run.pipeline.training_config, status="queued" | |
| ) | |
| from experiment.tasks import run_training_task | |
| run_training_task.delay(str(training_run.id)) | |
| # 2. Evaluate | |
| # 3. Deploy | |
| # (simplified) | |
| def run_distillation(self, job_id: str): | |
| job = DistillationJob.objects.get(id=job_id) | |
| # Execute distillation using framework | |
| from framework.distillation import distill_model | |
| output = distill_model( | |
| job.teacher_model.artifact_path, | |
| job.student_model_config, | |
| job.compression_config, | |
| ) | |
| # Save output model version | |
| from model_versioning.services import create_version | |
| version = create_version( | |
| run=job.teacher_model.run, | |
| metrics=output["metrics"], | |
| artifact_dir=output["path"], | |
| ) | |
| job.output_model = version | |
| job.status = "completed" | |
| job.save() | |
| def launch_multi_node_training(self, training_id: str): | |
| training = MultiNodeTraining.objects.get(id=training_id) | |
| # Use federation cluster to distribute training | |
| # Launch across nodes | |
| def monitor_all_endpoints(): | |
| for endpoint in InferenceEndpoint.objects.filter(status="running"): | |
| run_monitors_for_endpoint(endpoint) | |
Xet Storage Details
- Size:
- 2.81 kB
- Xet hash:
- 53a694c0f0e88bb6ee702f201d452bcc781241a5c35f0d6c691d75cf07bce6bb
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.