14 lines
463 B
Python
14 lines
463 B
Python
from celery import shared_task
|
|
|
|
from apps.modelhub.models import ModelInvocation
|
|
from apps.modelhub.services import execute_invocation
|
|
|
|
|
|
@shared_task(name="apps.modelhub.tasks.run_model_test")
|
|
def run_model_test(invocation_id: int) -> None:
|
|
invocation = ModelInvocation.objects.select_related("deployment__provider").get(
|
|
id=invocation_id
|
|
)
|
|
messages = invocation.input_snapshot.get("messages", [])
|
|
execute_invocation(invocation, messages)
|