chore(llmobs): projects support for dne (#14517)

adds support for projects
1. if a project name is provided in pull_dataset or create_dataset*
methods, the dataset will be searched for or created under the specified
project
2. otherwise, the default project is used, or the project name from the
`LLMObs.enable` call or the env var is used

with this snippet
```
LLMObs.enable(api_key=os.getenv("DD_API_KEY"), app_key=os.getenv("DD_APPLICATION_KEY"),  project_name="Onboarding", ml_app="Onboarding-ML-App")

dataset_name = "weatheraus-2025090900011"
dataset = LLMObs.create_dataset_from_csv(csv_path="/Users/gary.huang/Downloads/weatherAUS.csv", dataset_name=dataset_name, project_name="big-data-gh" , input_data_columns=["Date", "input", "Evaporation", "Sunshine"], expected_output_columns=["RainTomorrow"])

print(dataset.as_dataframe())

print(dataset.url)

time.sleep(30)

try:
    pds = LLMObs.pull_dataset(dataset_name=dataset_name)
except Exception as e:
    print(e)

pds = LLMObs.pull_dataset(dataset_name=dataset_name, project_name="big-data-gh")
print(pds.as_dataframe())

print(pds.url)
```
we get the following output
```
python test-project.py
3.15.0.dev13+gcb1068a93
        input_data expected_output input_data                    
              Date    RainTomorrow   Sunshine Evaporation   input
0       2008-12-01              No         NA          NA  Albury
1       2008-12-02              No         NA          NA  Albury
2       2008-12-03              No         NA          NA  Albury
3       2008-12-04              No         NA          NA  Albury
4       2008-12-05              No         NA          NA  Albury
...            ...             ...        ...         ...     ...
145455  2017-06-21              No         NA          NA   Uluru
145456  2017-06-22              No         NA          NA   Uluru
145457  2017-06-23              No         NA          NA   Uluru
145458  2017-06-24              No         NA          NA   Uluru
145459  2017-06-25              NA         NA          NA   Uluru

[145460 rows x 5 columns]
https://app.datadoghq.com/llm/datasets/7f81325e-dce1-49c7-b6bf-c75947490891
Dataset 'weatheraus-202509090928' not found in project Onboarding
        input_data expected_output input_data                    
              Date    RainTomorrow   Sunshine Evaporation   input
0       2008-12-05              No         NA          NA  Albury
1       2008-12-08              No         NA          NA  Albury
2       2008-12-10              No         NA          NA  Albury
3       2008-12-09             Yes         NA          NA  Albury
4       2008-12-12             Yes         NA          NA  Albury
...            ...             ...        ...         ...     ...
145455  2016-08-25              No         NA          NA   Uluru
145456  2016-11-20              No         NA          NA   Uluru
145457  2016-12-26             Yes         NA          NA   Uluru
145458  2016-12-31              No         NA          NA   Uluru
145459  2017-04-29              No         NA          NA   Uluru

[145460 rows x 5 columns]
https://app.datadoghq.com/llm/datasets/7f81325e-dce1-49c7-b6bf-c75947490891
```

the first pull hits an exception because the dataset wasn't found in the
default project from the enable call, and the 2nd pull succeeds, all as
expected

an alternate approach would be to initialize default project in
`enable`, but that messes with the tests everywhere that don't have
cassettes

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

---------

Co-authored-by: Sam Brenner <[email protected]>

Author: gary-huang

ddtrace/llmobs/_experiment.py

@@ -44,6 +44,11 @@
 DatasetRecordInputType = Dict[str, NonNoneJSONType]
 
 
+class Project(TypedDict):
+    name: str
+    _id: str
+
+
 class DatasetRecordRaw(TypedDict):
     input_data: DatasetRecordInputType
     expected_output: JSONType
@@ -106,13 +111,15 @@ class Dataset:
     def __init__(
         self,
         name: str,
+        project: Project,
         dataset_id: str,
         records: List[DatasetRecord],
         description: str,
         version: int,
         _dne_client: "LLMObsExperimentsClient",
     ) -> None:
         self.name = name
+        self.project = project
         self.description = description
         self._id = dataset_id
         self._version = version
@@ -335,8 +342,8 @@ def run(
             )
             return []
 
-        project_id = self._llmobs_instance._dne_client.project_create_or_get(self._project_name)
-        self._project_id = project_id
+        project = self._llmobs_instance._dne_client.project_create_or_get(self._project_name)
+        self._project_id = project.get("_id", "")
 
         experiment_id, experiment_run_name = self._llmobs_instance._dne_client.experiment_create(
             self.name,
@@ -416,6 +423,7 @@ def _run_task(self, jobs: int, raise_errors: bool = False, sample_size: Optional
             subset_name = "[Test subset of {} records] {}".format(sample_size, self._dataset.name)
             subset_dataset = Dataset(
                 name=subset_name,
+                project=self._dataset.project,
                 dataset_id=self._dataset._id,
                 records=subset_records,
                 description=self._dataset.description,

ddtrace/llmobs/_llmobs.py

@@ -92,6 +92,7 @@
 from ddtrace.llmobs._experiment import Experiment
 from ddtrace.llmobs._experiment import ExperimentConfigType
 from ddtrace.llmobs._experiment import JSONType
+from ddtrace.llmobs._experiment import Project
 from ddtrace.llmobs._utils import AnnotationContext
 from ddtrace.llmobs._utils import LinkTracker
 from ddtrace.llmobs._utils import _get_ml_app
@@ -212,6 +213,7 @@ def __init__(
             interval=float(os.getenv("_DD_LLMOBS_WRITER_INTERVAL", 1.0)),
             timeout=float(os.getenv("_DD_LLMOBS_WRITER_TIMEOUT", 5.0)),
             _app_key=self._app_key,
+            _default_project=Project(name=self._project_name, _id=""),
             is_agentless=True,  # agent proxy doesn't seem to work for experiments
         )
 
@@ -645,15 +647,21 @@ def enable(
             )
 
     @classmethod
-    def pull_dataset(cls, name: str) -> Dataset:
-        ds = cls._instance._dne_client.dataset_get_with_records(name)
+    def pull_dataset(cls, dataset_name: str, project_name: Optional[str] = None) -> Dataset:
+        ds = cls._instance._dne_client.dataset_get_with_records(dataset_name, (project_name or cls._project_name))
         return ds
 
     @classmethod
-    def create_dataset(cls, name: str, description: str = "", records: Optional[List[DatasetRecord]] = None) -> Dataset:
+    def create_dataset(
+        cls,
+        dataset_name: str,
+        project_name: Optional[str] = None,
+        description: str = "",
+        records: Optional[List[DatasetRecord]] = None,
+    ) -> Dataset:
         if records is None:
             records = []
-        ds = cls._instance._dne_client.dataset_create(name, description)
+        ds = cls._instance._dne_client.dataset_create(dataset_name, project_name, description)
         for r in records:
             ds.append(r)
         if len(records) > 0:
@@ -669,19 +677,20 @@ def create_dataset_from_csv(
         expected_output_columns: Optional[List[str]] = None,
         metadata_columns: Optional[List[str]] = None,
         csv_delimiter: str = ",",
-        description="",
+        description: str = "",
+        project_name: Optional[str] = None,
     ) -> Dataset:
         if expected_output_columns is None:
             expected_output_columns = []
         if metadata_columns is None:
             metadata_columns = []
-        ds = cls._instance._dne_client.dataset_create(dataset_name, description)
 
         # Store the original field size limit to restore it later
         original_field_size_limit = csv.field_size_limit()
 
         csv.field_size_limit(EXPERIMENT_CSV_FIELD_MAX_SIZE)  # 10mb
 
+        records = []
         try:
             with open(csv_path, mode="r") as csvfile:
                 content = csvfile.readline().strip()
@@ -708,7 +717,7 @@ def create_dataset_from_csv(
                     raise ValueError(f"Metadata columns not found in CSV header: {missing_metadata_columns}")
 
                 for row in rows:
-                    ds.append(
+                    records.append(
                         DatasetRecord(
                             input_data={col: row[col] for col in input_data_columns},
                             expected_output={col: row[col] for col in expected_output_columns},
@@ -721,6 +730,9 @@ def create_dataset_from_csv(
             # Always restore the original field size limit
             csv.field_size_limit(original_field_size_limit)
 
+        ds = cls._instance._dne_client.dataset_create(dataset_name, project_name, description)
+        for r in records:
+            ds.append(r)
         if len(ds) > 0:
             cls._instance._dne_client.dataset_bulk_upload(ds._id, ds._records)
         return ds

ddtrace/llmobs/_writer.py

@@ -49,6 +49,7 @@
 from ddtrace.llmobs._experiment import DatasetRecord
 from ddtrace.llmobs._experiment import DatasetRecordRaw
 from ddtrace.llmobs._experiment import JSONType
+from ddtrace.llmobs._experiment import Project
 from ddtrace.llmobs._experiment import UpdatableDatasetRecord
 from ddtrace.llmobs._utils import safe_json
 from ddtrace.settings._agent import config as agent_config
@@ -141,6 +142,7 @@ def __init__(
         _api_key: str = "",
         _app_key: str = "",
         _override_url: str = "",
+        _default_project: Project = Project(name="", _id=""),
     ) -> None:
         super(BaseLLMObsWriter, self).__init__(interval=interval)
         self._lock = forksafe.RLock()
@@ -151,6 +153,7 @@ def __init__(
         self._site: str = _site or config._dd_site
         self._app_key: str = _app_key
         self._override_url: str = _override_url or os.environ.get("DD_LLMOBS_OVERRIDE_ORIGIN", "")
+        self._default_project: Project = _default_project
 
         self._agentless: bool = is_agentless
         self._intake: str = self._override_url or (
@@ -371,23 +374,32 @@ def dataset_delete(self, dataset_id: str) -> None:
             raise ValueError(f"Failed to delete dataset {id}: {resp.get_json()}")
         return None
 
-    def dataset_create(self, name: str, description: str) -> Dataset:
-        path = "/api/unstable/llm-obs/v1/datasets"
+    def dataset_create(
+        self,
+        dataset_name: str,
+        project_name: Optional[str],
+        description: str,
+    ) -> Dataset:
+        project = self.project_create_or_get(project_name)
+        project_id = project.get("_id")
+        logger.debug("getting records with project ID %s for %s", project_id, project_name)
+
+        path = f"/api/unstable/llm-obs/v1/{project_id}/datasets"
         body: JSONType = {
             "data": {
                 "type": "datasets",
-                "attributes": {"name": name, "description": description},
+                "attributes": {"name": dataset_name, "description": description},
             }
         }
         resp = self.request("POST", path, body)
         if resp.status != 200:
-            raise ValueError(f"Failed to create dataset {name}: {resp.status} {resp.get_json()}")
+            raise ValueError(f"Failed to create dataset {dataset_name}: {resp.status} {resp.get_json()}")
         response_data = resp.get_json()
         dataset_id = response_data["data"]["id"]
         if dataset_id is None or dataset_id == "":
             raise ValueError(f"unexpected dataset state, invalid ID (is None: {dataset_id is None})")
         curr_version = response_data["data"]["attributes"]["current_version"]
-        return Dataset(name, dataset_id, [], description, curr_version, _dne_client=self)
+        return Dataset(dataset_name, project, dataset_id, [], description, curr_version, _dne_client=self)
 
     @staticmethod
     def _get_record_json(record: Union[UpdatableDatasetRecord, DatasetRecordRaw], is_update: bool) -> JSONType:
@@ -445,16 +457,22 @@ def dataset_batch_update(
         new_record_ids: List[str] = [r["id"] for r in data] if data else []
         return new_version, new_record_ids
 
-    def dataset_get_with_records(self, name: str) -> Dataset:
-        path = f"/api/unstable/llm-obs/v1/datasets?filter[name]={quote(name)}"
+    def dataset_get_with_records(self, dataset_name: str, project_name: Optional[str] = None) -> Dataset:
+        project = self.project_create_or_get(project_name)
+        project_id = project.get("_id")
+        logger.debug("getting records with project ID %s for %s", project_id, project_name)
+
+        path = f"/api/unstable/llm-obs/v1/{project_id}/datasets?filter[name]={quote(dataset_name)}"
         resp = self.request("GET", path)
         if resp.status != 200:
-            raise ValueError(f"Failed to pull dataset {name}: {resp.status}")
+            raise ValueError(
+                f"Failed to pull dataset {dataset_name} from project {project_name} (id={project_id}): {resp.status}"
+            )
 
         response_data = resp.get_json()
         data = response_data["data"]
         if not data:
-            raise ValueError(f"Dataset '{name}' not found")
+            raise ValueError(f"Dataset '{dataset_name}' not found in project {project_name}")
 
         curr_version = data[0]["attributes"]["current_version"]
         dataset_description = data[0]["attributes"].get("description", "")
@@ -469,7 +487,8 @@ def dataset_get_with_records(self, name: str) -> Dataset:
             resp = self.request("GET", list_path, timeout=self.LIST_RECORDS_TIMEOUT)
             if resp.status != 200:
                 raise ValueError(
-                    f"Failed to pull {page_num}th page of dataset records {name}: {resp.status} {resp.get_json()}"
+                    f"Failed to pull dataset records for {dataset_name}, page={page_num}: "
+                    f"{resp.status} {resp.get_json()}"
                 )
             records_data = resp.get_json()
 
@@ -490,7 +509,9 @@ def dataset_get_with_records(self, name: str) -> Dataset:
                 list_path = f"{list_base_path}?page[cursor]={next_cursor}"
                 logger.debug("next list records request path %s", list_path)
                 page_num += 1
-        return Dataset(name, dataset_id, class_records, dataset_description, curr_version, _dne_client=self)
+        return Dataset(
+            dataset_name, project, dataset_id, class_records, dataset_description, curr_version, _dne_client=self
+        )
 
     def dataset_bulk_upload(self, dataset_id: str, records: List[DatasetRecord]):
         with tempfile.NamedTemporaryFile(suffix=".csv") as tmp:
@@ -543,17 +564,42 @@ def dataset_bulk_upload(self, dataset_id: str, records: List[DatasetRecord]):
             raise ValueError(f"Failed to upload dataset from file: {resp.status} {resp.get_json()}")
         logger.debug("successfully uploaded with code %d", resp.status)
 
-    def project_create_or_get(self, name: str) -> str:
+    def project_create_or_get(self, name: Optional[str] = None) -> Project:
+        default_project_name = self._default_project["name"]
+        project_name = default_project_name
+
+        if not name:
+            if self._default_project.get("_id"):
+                # default project already initialized, use it
+                return self._default_project
+        else:
+            project_name = name
+
         path = "/api/unstable/llm-obs/v1/projects"
         resp = self.request(
             "POST",
             path,
-            body={"data": {"type": "projects", "attributes": {"name": name, "description": ""}}},
+            body={"data": {"type": "projects", "attributes": {"name": project_name, "description": ""}}},
         )
         if resp.status != 200:
-            raise ValueError(f"Failed to create project {name}: {resp.status} {resp.get_json()}")
+            raise ValueError(f"Failed to create project {project_name}: {resp.status} {resp.get_json()}")
         response_data = resp.get_json()
-        return response_data["data"]["id"]
+        project_id = response_data["data"]["id"]
+
+        if not project_id:
+            logger.error(
+                "got empty project ID for project %s in response, code=%d, resp=%s",
+                project_name,
+                resp.status,
+                resp.get_json(),
+            )
+
+        project = Project(name=project_name, _id=project_id)
+        # after the initial GET of the project ID, store it
+        if project_name == default_project_name:
+            self._default_project = project
+
+        return project
 
     def experiment_create(
         self,