Added from_url method to fetch evaluation annotation from the api.

src/ragas/dataset_schema.py

@@ -9,15 +9,25 @@
 from uuid import UUID
 
 import numpy as np
+import requests
 from datasets import Dataset as HFDataset
 from pydantic import BaseModel, field_validator
 
 from ragas.callbacks import ChainRunEncoder, parse_run_traces
 from ragas.cost import CostCallbackHandler
 from ragas.exceptions import UploadException
 from ragas.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
-from ragas.sdk import RAGAS_API_URL, RAGAS_APP_URL, upload_packet
+from ragas.sdk import (
+    upload_packet,
+    RAGAS_API_SOURCE,
+    get_app_token,
+    check_api_response,
+    build_evaluation_app_url,
+    get_api_url,
+    get_app_url,
+)
 from ragas.utils import safe_nanmean
+from ragas._version import __version__
 
 if t.TYPE_CHECKING:
     from pathlib import Path
@@ -508,7 +518,10 @@ def total_cost(
             cost_per_input_token, cost_per_output_token, per_model_costs
         )
 
-    def upload(self, base_url: str = RAGAS_API_URL, verbose: bool = True) -> str:
+    def upload(
+        self,
+        verbose: bool = True,
+    ) -> str:
         from datetime import datetime, timezone
 
         timestamp = datetime.now(timezone.utc).isoformat()
@@ -526,18 +539,16 @@ def upload(self, base_url: str = RAGAS_API_URL, verbose: bool = True) -> str:
         response = upload_packet(
             path="/alignment/evaluation",
             data_json_string=packet,
-            base_url=base_url,
         )
 
         # check status codes
-        evaluation_endpoint = (
-            f"{RAGAS_APP_URL}/dashboard/alignment/evaluation/{root_trace.run_id}"
-        )
+        app_url = get_app_url()
+        evaluation_app_url = build_evaluation_app_url(app_url, root_trace.run_id)
         if response.status_code == 409:
             # this evalution already exists
             if verbose:
-                print(f"Evaluation run already exists. View at {evaluation_endpoint}")
-            return evaluation_endpoint
+                print(f"Evaluation run already exists. View at {evaluation_app_url}")
+            return evaluation_app_url
         elif response.status_code != 200:
             # any other error
             raise UploadException(
@@ -546,8 +557,8 @@ def upload(self, base_url: str = RAGAS_API_URL, verbose: bool = True) -> str:
             )
 
         if verbose:
-            print(f"Evaluation results uploaded! View at {evaluation_endpoint}")
-        return evaluation_endpoint
+            print(f"Evaluation results uploaded! View at {evaluation_app_url}")
+        return evaluation_app_url
 
 
 class PromptAnnotation(BaseModel):
@@ -577,8 +588,24 @@ def __getitem__(self, key):
         return SingleMetricAnnotation(name=key, samples=self.root[key])
 
     @classmethod
-    def from_json(cls, path, metric_name: t.Optional[str]) -> "MetricAnnotation":
-        dataset = json.load(open(path))
+    def _process_dataset(
+        cls, dataset: dict, metric_name: t.Optional[str]
+    ) -> "MetricAnnotation":
+        """
+        Process raw dataset into MetricAnnotation format
+
+        Parameters
+        ----------
+        dataset : dict
+            Raw dataset to process
+        metric_name : str, optional
+            Name of the specific metric to filter
+
+        Returns
+        -------
+        MetricAnnotation
+            Processed annotation data
+        """
         if metric_name is not None and metric_name not in dataset:
             raise ValueError(f"Split {metric_name} not found in the dataset.")
 
@@ -590,6 +617,81 @@ def from_json(cls, path, metric_name: t.Optional[str]) -> "MetricAnnotation":
             }
         )
 
+    @classmethod
+    def from_json(cls, path: str, metric_name: t.Optional[str]) -> "MetricAnnotation":
+        """Load annotations from a JSON file"""
+        dataset = json.load(open(path))
+        return cls._process_dataset(dataset, metric_name)
+
+    @classmethod
+    def from_ragas_platform(
+        cls,
+        evaluation_result: EvaluationResult,
+        metric_name: t.Optional[str],
+    ) -> "MetricAnnotation":
+        """
+        Fetch annotations from a URL using the evaluation result's run_id
+
+        Parameters
+        ----------
+        evaluation_result : EvaluationResult
+            The evaluation result containing the run_id
+        metric_name : str, optional
+            Name of the specific metric to filter
+
+        Returns
+        -------
+        MetricAnnotation
+            Annotation data from the API
+
+        Raises
+        ------
+        ValueError
+            If no traces found, no root trace found, or no annotations available
+        """
+        # Find root trace run_id
+        if not evaluation_result.ragas_traces:
+            raise ValueError("No traces found in evaluation_result")
+
+        root_trace = [
+            trace
+            for trace in evaluation_result.ragas_traces.values()
+            if trace.parent_run_id is None
+        ]
+
+        if not root_trace:
+            raise ValueError("No root trace found in evaluation_result")
+
+        run_id = root_trace[0].run_id
+
+        endpoint = f"/api/v1/alignment/evaluation/annotation/{run_id}"
+
+        app_token = get_app_token()
+        base_url = get_api_url()
+        app_url = get_app_url()
+
+        response = requests.get(
+            f"{base_url}{endpoint}",
+            headers={
+                "Content-Type": "application/json",
+                "x-app-token": app_token,
+                "x-source": RAGAS_API_SOURCE,
+                "x-app-version": __version__,
+            },
+        )
+
+        check_api_response(response)
+        dataset = response.json()["data"]
+
+        if not dataset:
+            evaluation_url = build_evaluation_app_url(app_url, run_id)
+            raise ValueError(
+                f"No annotations found. Please annotate the Evaluation first then run this method. "
+                f"\nNote: you can annotate the evaluations using the Ragas app by going to {evaluation_url}"
+            )
+
+        return cls._process_dataset(dataset, metric_name)
+
     def __len__(self):
         return sum(len(value) for value in self.root.values())
 

src/ragas/sdk.py

@@ -24,8 +24,20 @@ def get_app_token() -> str:
     return app_token
 
 
-def upload_packet(path: str, data_json_string: str, base_url: str = RAGAS_API_URL):
+@lru_cache(maxsize=1)
+def get_api_url() -> str:
+    return os.environ.get("RAGAS_API_URL", RAGAS_API_URL)
+
+
+@lru_cache(maxsize=1)
+def get_app_url() -> str:
+    return os.environ.get("RAGAS_APP_URL", RAGAS_APP_URL)
+
+
+def upload_packet(path: str, data_json_string: str):
     app_token = get_app_token()
+    base_url = get_api_url()
+
     response = requests.post(
         f"{base_url}/api/v1{path}",
         data=data_json_string,
@@ -36,9 +48,31 @@ def upload_packet(path: str, data_json_string: str, base_url: str = RAGAS_API_UR
             "x-app-version": __version__,
         },
     )
+    check_api_response(response)
+    return response
+
+
+def check_api_response(response: requests.Response) -> None:
+    """
+    Check API response status and raise appropriate exceptions
+
+    Parameters
+    ----------
+    response : requests.Response
+        Response object from API request
+
+    Raises
+    ------
+    UploadException
+        If authentication fails or other API errors occur
+    """
     if response.status_code == 403:
         raise UploadException(
             status_code=response.status_code,
             message="AUTHENTICATION_ERROR: The app token is invalid. Please check your RAGAS_APP_TOKEN environment variable.",
         )
-    return response
+    response.raise_for_status()
+
+
+def build_evaluation_app_url(app_url: str, run_id: str) -> str:
+    return f"{app_url}/dashboard/alignment/evaluation/{run_id}"

src/ragas/testset/synthesizers/testset_schema.py

@@ -16,7 +16,7 @@
     SingleTurnSample,
 )
 from ragas.exceptions import UploadException
-from ragas.sdk import RAGAS_API_URL, RAGAS_APP_URL, upload_packet
+from ragas.sdk import upload_packet, get_app_url
 
 
 class TestsetSample(BaseSample):
@@ -136,14 +136,15 @@ def total_cost(
             cost_per_output_token=cost_per_output_token,
         )
 
-    def upload(self, base_url: str = RAGAS_API_URL, verbose: bool = True) -> str:
+    def upload(self, verbose: bool = True) -> str:
         packet = TestsetPacket(samples_original=self.samples, run_id=self.run_id)
         response = upload_packet(
             path="/alignment/testset",
             data_json_string=packet.model_dump_json(),
-            base_url=base_url,
         )
-        testset_endpoint = f"{RAGAS_APP_URL}/dashboard/alignment/testset/{self.run_id}"
+        app_url = get_app_url()
+
+        testset_endpoint = f"{app_url}/dashboard/alignment/testset/{self.run_id}"
         if response.status_code == 409:
             # this testset already exists
             if verbose: