실행 및 아티팩트 추적

from typing import Dict, Optional

from google.cloud import aiplatform


def create_artifact_sample(
    schema_title: str,
    project: str,
    location: str,
    uri: Optional[str] = None,
    resource_id: Optional[str] = None,
    display_name: Optional[str] = None,
    schema_version: Optional[str] = None,
    description: Optional[str] = None,
    metadata: Optional[Dict] = None,
):
    artifact = aiplatform.Artifact.create(
        schema_title=schema_title,
        uri=uri,
        resource_id=resource_id,
        display_name=display_name,
        schema_version=schema_version,
        description=description,
        metadata=metadata,
        project=project,
        location=location,
    )
    return artifact

from typing import Any, Dict, List, Optional

from google.cloud import aiplatform


def start_execution_sample(
    schema_title: str,
    display_name: str,
    input_artifacts: List[aiplatform.Artifact],
    output_artifacts: List[aiplatform.Artifact],
    project: str,
    location: str,
    resource_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    schema_version: Optional[str] = None,
    resume: bool = False,
):
    aiplatform.init(project=project, location=location)

    with aiplatform.start_execution(
        schema_title=schema_title,
        display_name=display_name,
        resource_id=resource_id,
        metadata=metadata,
        schema_version=schema_version,
        resume=resume,
    ) as execution:
        execution.assign_input_artifacts(input_artifacts)
        execution.assign_output_artifacts(output_artifacts)
        return execution