from __future__ import annotations from typing import Any from django.core.exceptions import ValidationError from django.core.validators import URLValidator from django.db import IntegrityError, transaction from django.utils import timezone from apps.modelhub.crypto import CredentialEncryptionError, encrypt_secret from apps.modelhub.models import ModelDeployment, ModelProvider from apps.modelhub.structured_output import ( StructuredOutputError, validate_mode_for_capability, ) CHAT_CAPABILITIES = {"text_chat", "vision_chat"} CAPABILITIES = CHAT_CAPABILITIES | {"embedding", "rerank"} class ModelGovernanceError(Exception): def __init__(self, code: str, message: str, *, status: int = 400): super().__init__(message) self.code = code self.status = status def _clean_name(value: str, field_label: str) -> str: cleaned = value.strip() if not cleaned: raise ModelGovernanceError("invalid_model_config", f"{field_label}不能为空") return cleaned def validate_provider_values(values: dict[str, Any], provider: ModelProvider | None = None) -> dict: cleaned = dict(values) if "name" in cleaned: cleaned["name"] = _clean_name(cleaned["name"], "渠道名称") if "base_url" in cleaned: base_url = cleaned["base_url"].strip().rstrip("/") try: URLValidator(schemes=["http", "https"])(base_url) except ValidationError as error: raise ModelGovernanceError( "invalid_provider_url", "渠道地址必须是有效的 HTTP(S) URL" ) from error cleaned["base_url"] = base_url timeout = cleaned.get( "default_timeout_seconds", provider.default_timeout_seconds if provider else 60, ) if not 1 <= timeout <= 300: raise ModelGovernanceError( "invalid_provider_timeout", "请求超时必须在 1 到 300 秒之间" ) if cleaned.get("clear_api_key") and cleaned.get("api_key"): raise ModelGovernanceError( "credential_action_conflict", "不能同时替换和清除渠道凭证" ) return cleaned def validate_deployment_values( values: dict[str, Any], deployment: ModelDeployment | None = None, ) -> dict: cleaned = dict(values) for field, label in (("display_name", "显示名称"), ("model_name", "实际模型名")): if field in cleaned: cleaned[field] = _clean_name(cleaned[field], label) capability = cleaned.get( "capability", deployment.capability if deployment else "text_chat" ) if capability not in CAPABILITIES: raise ModelGovernanceError("invalid_model_capability", "未知的模型能力") if capability not in CHAT_CAPABILITIES: cleaned["structured_output_mode"] = "none" temperature = cleaned.get( "default_temperature", deployment.default_temperature if deployment else 0.2, ) if not 0 <= temperature <= 2: raise ModelGovernanceError( "invalid_model_temperature", "Temperature 必须在 0 到 2 之间" ) max_tokens = cleaned.get( "max_output_tokens", deployment.max_output_tokens if deployment else 2048 ) if not 1 <= max_tokens <= 32768: raise ModelGovernanceError( "invalid_model_max_tokens", "最大输出 Token 必须在 1 到 32768 之间" ) mode = cleaned.get( "structured_output_mode", deployment.structured_output_mode if deployment else "unconfigured", ) try: validate_mode_for_capability(capability, mode) except StructuredOutputError as error: raise ModelGovernanceError(error.code, str(error)) from error return cleaned def create_provider_config(values: dict[str, Any], user) -> ModelProvider: cleaned = validate_provider_values(values) api_key = cleaned.pop("api_key").strip() if not api_key: raise ModelGovernanceError( "credential_required", "创建渠道时必须录入有效凭证" ) cleaned.pop("clear_api_key", None) approved = cleaned.get("image_egress_approved", False) now = timezone.now() try: encrypted = encrypt_secret(api_key) return ModelProvider.objects.create( **cleaned, api_key_encrypted=encrypted, credential_updated_at=now, credential_updated_by=user, image_egress_approved_by=user if approved else None, image_egress_approved_at=now if approved else None, created_by=user, ) except CredentialEncryptionError as error: raise ModelGovernanceError("credential_encryption_failed", str(error)) from error except IntegrityError as error: raise ModelGovernanceError( "provider_name_conflict", "活跃渠道名称已存在", status=409 ) from error def update_provider_config( provider: ModelProvider, values: dict[str, Any], user, ) -> tuple[ModelProvider, dict[str, Any]]: if provider.archived_at: raise ModelGovernanceError( "model_config_archived", "归档渠道不允许编辑、测试或启用" ) cleaned = validate_provider_values(values, provider) api_key = cleaned.pop("api_key", None) clear_api_key = cleaned.pop("clear_api_key", False) approved = cleaned.pop("image_egress_approved", None) if ( cleaned.get("enabled") is True and not clear_api_key and not api_key and not provider.api_key_encrypted ): raise ModelGovernanceError( "credential_required", "启用渠道前必须录入有效凭证" ) base_url_changed = ( "base_url" in cleaned and cleaned["base_url"] != provider.base_url.rstrip("/") ) credential_action = "unchanged" now = timezone.now() try: if api_key: provider.api_key_encrypted = encrypt_secret(api_key) provider.credential_updated_at = now provider.credential_updated_by = user credential_action = "replaced" elif clear_api_key: provider.api_key_encrypted = "" provider.credential_updated_at = now provider.credential_updated_by = user provider.enabled = False credential_action = "cleared" for field, value in cleaned.items(): setattr(provider, field, value) if clear_api_key: provider.enabled = False if approved is not None: provider.image_egress_approved = approved provider.image_egress_approved_by = user if approved else None provider.image_egress_approved_at = now if approved else None provider.save() except CredentialEncryptionError as error: raise ModelGovernanceError("credential_encryption_failed", str(error)) from error except IntegrityError as error: raise ModelGovernanceError( "provider_name_conflict", "活跃渠道名称已存在", status=409 ) from error if base_url_changed: provider.deployments.filter( archived_at__isnull=True, capability__in=CHAT_CAPABILITIES, ).update( structured_output_test_status="untested", structured_output_verified_at=None, structured_output_test_message="渠道地址已变更,请重新验证结构化输出", ) return provider, { "fields": sorted([*cleaned, *(["image_egress_approved"] if approved is not None else [])]), "credential_action": credential_action, } def create_deployment_config(values: dict[str, Any]) -> ModelDeployment: cleaned = validate_deployment_values(values) provider = ModelProvider.objects.get(id=cleaned.pop("provider_id")) if provider.archived_at: raise ModelGovernanceError( "model_config_archived", "不能在已归档渠道下创建模型" ) try: return ModelDeployment.objects.create(provider=provider, **cleaned) except IntegrityError as error: raise ModelGovernanceError( "deployment_name_conflict", "同一活跃渠道下实际模型名已存在", status=409, ) from error def update_deployment_config( deployment: ModelDeployment, values: dict[str, Any], ) -> tuple[ModelDeployment, dict[str, Any]]: if deployment.archived_at or deployment.provider.archived_at: raise ModelGovernanceError( "model_config_archived", "归档模型不允许编辑、测试或启用" ) cleaned = validate_deployment_values(values, deployment) reset_test = any( field in cleaned and cleaned[field] != getattr(deployment, field) for field in {"model_name", "structured_output_mode", "capability"} ) for field, value in cleaned.items(): setattr(deployment, field, value) if reset_test: deployment.structured_output_test_status = "untested" deployment.structured_output_verified_at = None deployment.structured_output_test_message = ( "模型配置已变更,请重新验证结构化输出" if deployment.capability in CHAT_CAPABILITIES else "" ) try: deployment.save() except IntegrityError as error: raise ModelGovernanceError( "deployment_name_conflict", "同一活跃渠道下实际模型名已存在", status=409, ) from error return deployment, {"fields": sorted(cleaned), "structured_test_reset": reset_test} def deployment_impact(deployment: ModelDeployment) -> dict[str, int | str]: counts = { "agent_version_count": deployment.agent_versions.count(), "step_binding_count": deployment.agent_step_bindings.count(), "run_step_count": deployment.agent_run_steps.count(), "invocation_count": deployment.invocations.count(), } return { "suggested_action": "archive" if any(counts.values()) else "delete", "deployment_count": 1, **counts, } def provider_impact(provider: ModelProvider) -> dict[str, int | str]: from django.db.models import Count, Sum deployments = provider.deployments.annotate( agent_version_refs=Count("agent_versions", distinct=True), step_binding_refs=Count("agent_step_bindings", distinct=True), run_step_refs=Count("agent_run_steps", distinct=True), invocation_refs=Count("invocations", distinct=True), ) totals = deployments.aggregate( agent_version_count=Sum("agent_version_refs"), step_binding_count=Sum("step_binding_refs"), run_step_count=Sum("run_step_refs"), invocation_count=Sum("invocation_refs"), ) counts = {key: value or 0 for key, value in totals.items()} return { "suggested_action": "archive" if any(counts.values()) else "delete", "deployment_count": deployments.count(), **counts, } @transaction.atomic def remove_deployment(deployment_id: int, user) -> tuple[str, dict]: deployment = ( ModelDeployment.objects.select_for_update() .select_related("provider") .get(id=deployment_id) ) impact = deployment_impact(deployment) if deployment.archived_at: return "archived", impact if impact["suggested_action"] == "delete": deployment.delete() return "deleted", impact deployment.enabled = False deployment.archived_at = timezone.now() deployment.archived_by = user deployment.archived_via_provider = False deployment.save( update_fields=[ "enabled", "archived_at", "archived_by", "archived_via_provider", "updated_at", ] ) return "archived", impact @transaction.atomic def remove_provider(provider_id: int, user) -> tuple[str, dict]: provider = ModelProvider.objects.select_for_update().get(id=provider_id) list(provider.deployments.select_for_update()) impact = provider_impact(provider) if provider.archived_at: return "archived", impact if impact["suggested_action"] == "delete": provider.deployments.all().delete() provider.delete() return "deleted", impact now = timezone.now() provider.deployments.filter(archived_at__isnull=True).update( enabled=False, archived_at=now, archived_by=user, archived_via_provider=True, ) provider.enabled = False provider.archived_at = now provider.archived_by = user provider.api_key_encrypted = "" provider.credential_updated_at = now provider.credential_updated_by = user provider.image_egress_approved = False provider.image_egress_approved_by = None provider.image_egress_approved_at = None provider.save() return "archived", impact @transaction.atomic def restore_deployment(deployment_id: int) -> ModelDeployment: deployment = ( ModelDeployment.objects.select_for_update() .select_related("provider") .get(id=deployment_id) ) if deployment.archived_at is None: deployment.enabled = False deployment.save(update_fields=["enabled", "updated_at"]) return deployment if deployment.provider.archived_at: raise ModelGovernanceError( "provider_archived", "请先恢复模型所属渠道", status=409 ) if ModelDeployment.objects.filter( provider_id=deployment.provider_id, model_name=deployment.model_name, archived_at__isnull=True, ).exclude(id=deployment.id).exists(): raise ModelGovernanceError( "lifecycle_name_conflict", "恢复失败:活跃模型已复用该实际模型名", status=409 ) deployment.archived_at = None deployment.archived_by = None deployment.archived_via_provider = False deployment.enabled = False deployment.structured_output_test_status = "untested" deployment.structured_output_verified_at = None deployment.structured_output_test_message = ( "模型已恢复,请重新验证" if deployment.capability in CHAT_CAPABILITIES else "" ) if deployment.capability not in CHAT_CAPABILITIES: deployment.structured_output_mode = "none" deployment.save() return deployment @transaction.atomic def restore_provider(provider_id: int) -> ModelProvider: provider = ModelProvider.objects.select_for_update().get(id=provider_id) children = list(provider.deployments.select_for_update()) if ModelProvider.objects.filter( name=provider.name, archived_at__isnull=True ).exclude(id=provider.id).exists(): raise ModelGovernanceError( "lifecycle_name_conflict", "恢复失败:活跃渠道已复用该名称", status=409 ) for deployment in children: if not deployment.archived_via_provider: continue if ModelDeployment.objects.filter( provider_id=provider.id, model_name=deployment.model_name, archived_at__isnull=True, ).exclude(id=deployment.id).exists(): raise ModelGovernanceError( "lifecycle_name_conflict", f"恢复失败:模型名 {deployment.model_name} 已被复用", status=409, ) provider.archived_at = None provider.archived_by = None provider.enabled = False provider.save(update_fields=["archived_at", "archived_by", "enabled", "updated_at"]) restored_ids = list( provider.deployments.filter(archived_via_provider=True).values_list( "id", flat=True, ) ) restored_deployments = provider.deployments.filter(id__in=restored_ids) restored_deployments.update( archived_at=None, archived_by=None, archived_via_provider=False, enabled=False, structured_output_test_status="untested", structured_output_verified_at=None, ) restored_deployments.filter(capability__in=CHAT_CAPABILITIES).update( structured_output_test_message="渠道已恢复,请重新录入凭证并验证", ) restored_deployments.exclude(capability__in=CHAT_CAPABILITIES).update( structured_output_mode="none", structured_output_test_message="", ) return provider