[ODSC-66852] Watch aqua FT job logs after creation (#1047)

Author: VipulMascarenhas

ads/aqua/finetuning/entities.py

@@ -122,6 +122,8 @@ class CreateFineTuningDetails(Serializable):
         The log group id for fine tuning job infrastructure.
     log_id: (str, optional). Defaults to `None`.
         The log id for fine tuning job infrastructure.
+    watch_logs: (bool, optional). Defaults to `False`.
+        The flag to watch the job run logs when a fine-tuning job is created.
     force_overwrite: (bool, optional). Defaults to `False`.
         Whether to force overwrite the existing file in object storage.
     freeform_tags: (dict, optional)
@@ -148,6 +150,7 @@ class CreateFineTuningDetails(Serializable):
     subnet_id: Optional[str] = None
     log_id: Optional[str] = None
     log_group_id: Optional[str] = None
+    watch_logs: Optional[bool] = False
     force_overwrite: Optional[bool] = False
     freeform_tags: Optional[dict] = None
     defined_tags: Optional[dict] = None

ads/aqua/finetuning/finetuning.py

@@ -4,6 +4,8 @@
 
 import json
 import os
+import time
+import traceback
 from typing import Dict
 
 from oci.data_science.models import (
@@ -149,6 +151,15 @@ def create(
                 f"Logging is required for fine tuning if replica is larger than {DEFAULT_FT_REPLICA}."
             )
 
+        if create_fine_tuning_details.watch_logs and not (
+            create_fine_tuning_details.log_id
+            and create_fine_tuning_details.log_group_id
+        ):
+            raise AquaValueError(
+                "Logging is required for fine tuning if watch_logs is set to True. "
+                "Please provide log_id and log_group_id with the request parameters."
+            )
+
         ft_parameters = self._get_finetuning_params(
             create_fine_tuning_details.ft_parameters
         )
@@ -422,6 +433,20 @@ def create(
             value=source.display_name,
         )
 
+        if create_fine_tuning_details.watch_logs:
+            logger.info(
+                f"Watching fine-tuning job run logs for {ft_job_run.id}. Press Ctrl+C to stop watching logs.\n"
+            )
+            try:
+                ft_job_run.watch()
+            except KeyboardInterrupt:
+                logger.info(f"\nStopped watching logs for {ft_job_run.id}.\n")
+                time.sleep(1)
+            except Exception:
+                logger.debug(
+                    f"Something unexpected occurred while watching logs.\n{traceback.format_exc()}"
+                )
+
         return AquaFineTuningSummary(
             id=ft_model.id,
             name=ft_model.display_name,

ads/cli.py

@@ -1,14 +1,15 @@
 #!/usr/bin/env python
-
 # Copyright (c) 2021, 2025 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
+import json
 import logging
 import sys
 import traceback
 import uuid
 
 import fire
+from pydantic import BaseModel
 
 from ads.common import logger
 
@@ -84,7 +85,13 @@ def serialize(data):
         The string representation of each dataclass object.
     """
     if isinstance(data, list):
-        [print(str(item)) for item in data]
+        for item in data:
+            if isinstance(item, BaseModel):
+                print(json.dumps(item.dict(), indent=4))
+            else:
+                print(str(item))
+    elif isinstance(data, BaseModel):
+        print(json.dumps(data.dict(), indent=4))
     else:
         print(str(data))
 

tests/unitary/with_extras/aqua/test_finetuning.py

@@ -22,6 +22,7 @@
 from ads.aqua.finetuning.constants import FineTuneCustomMetadata
 from ads.aqua.finetuning.entities import AquaFineTuningParams
 from ads.jobs.ads_job import Job
+from ads.jobs.builders.infrastructure.dsc_job import DataScienceJobRun
 from ads.model.datascience_model import DataScienceModel
 from ads.model.model_metadata import ModelCustomMetadata
 from ads.aqua.common.errors import AquaValueError
@@ -49,6 +50,12 @@ def tearDownClass(cls):
         reload(ads.aqua)
         reload(ads.aqua.finetuning.finetuning)
 
+    @parameterized.expand(
+        [
+            ("watch_logs", True),
+            ("watch_logs", False),
+        ]
+    )
     @patch.object(Job, "run")
     @patch("ads.jobs.ads_job.Job.name", new_callable=PropertyMock)
     @patch("ads.jobs.ads_job.Job.id", new_callable=PropertyMock)
@@ -60,6 +67,8 @@ def tearDownClass(cls):
     @patch.object(AquaApp, "get_source")
     def test_create_fine_tuning(
         self,
+        mock_watch_logs,
+        mock_watch_logs_called,
         mock_get_source,
         mock_mvs_create,
         mock_ds_model_create,
@@ -117,6 +126,7 @@ def test_create_fine_tuning(
         ft_job_run.id = "test_ft_job_run_id"
         ft_job_run.lifecycle_details = "Job run artifact execution in progress."
         ft_job_run.lifecycle_state = "IN_PROGRESS"
+        ft_job_run.watch = MagicMock()
         mock_job_run.return_value = ft_job_run
 
         self.app.ds_client.update_model = MagicMock()
@@ -144,7 +154,20 @@ def test_create_fine_tuning(
             defined_tags=ft_model_defined_tags,
         )
 
-        aqua_ft_summary = self.app.create(**create_aqua_ft_details)
+        inputs = {
+            **create_aqua_ft_details,
+            **{
+                mock_watch_logs: mock_watch_logs_called,
+                "log_id": "test_log_id",
+                "log_group_id": "test_log_group_id",
+            },
+        }
+        aqua_ft_summary = self.app.create(**inputs)
+
+        if mock_watch_logs_called:
+            ft_job_run.watch.assert_called()
+        else:
+            ft_job_run.watch.assert_not_called()
 
         assert aqua_ft_summary.to_dict() == {
             "console_url": f"https://cloud.oracle.com/data-science/models/{ft_model.id}?region={self.app.region}",