Repository: mlflow/mlflow

Issue:
Remove useless checkout in team-review workflow
### Summary

The `team-review.yml` workflow doesn't depend on files in this repository. We don't need to checkout.

```diff
diff --git a/.github/workflows/team-review.yml b/.github/workflows/team-review.yml
index 46a250dbb..e78a76c11 100644
--- a/.github/workflows/team-review.yml
+++ b/.github/workflows/team-review.yml
@@ -9,7 +9,6 @@ jobs:
     runs-on: ubuntu-latest
     if: ${{ github.event.requested_reviewer.login == 'mlflow-automation'}}
     steps:
-      - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
           github-token: ${{ secrets.MLFLOW_AUTOMATION_TOKEN }}
```

### Notes

- Make sure to open a PR from a **non-master** branch.
- Sign off the commit using the `-s` flag when making a commit:

  ```sh
  git commit -s -m "..."
           # ^^ make sure to use this
  ```

- Include `#{issue_number}` (e.g. `#123`) in the PR description when opening a PR.


Discussion:
github-actions[bot]: <!-- documentation preview -->


Documentation preview for 4e748dcb884bfd43c55d4cde8722b84181d51233 will be available when [this CircleCI job](https://circleci.com/gh/mlflow/mlflow/113315)
completes successfully.

- [Top page](https://output.circle-artifacts.com/output/job/ccbde15a-a66a-4e65-a89b-fc5df1fa9685/artifacts/0/docs/build/html/index.html)
- [Changed pages](https://output.circle-artifacts.com/output/job/ccbde15a-a66a-4e65-a89b-fc5df1fa9685/artifacts/0/docs/build/html/diff.html)

<details>
<summary>More info</summary>

- Ignore this comment if this PR does not change the documentation.
- It takes a few minutes for the preview to be available.
- The preview is updated when a new commit is pushed to this PR.
- This comment was created by https://github.com/mlflow/mlflow/actions/runs/9747393214.

</details>

github-actions[bot]: @amanjam Thank you for the contribution! Could you fix the following issue(s)?

#### &#x26a0; Invalid PR template

This PR does not appear to have been filed using the MLflow PR template. Please copy the PR template from [here](https://raw.githubusercontent.com/mlflow/mlflow/master/.github/pull_request_template.md) and fill it out.
amanjam: ### Related Issues/PRs
#12536 




### What changes are proposed in this pull request?

<!-- Please fill in changes proposed in this PR. -->

### How is this PR tested?

- [ ] Existing unit/integration tests
- [ ] New unit/integration tests
- [ ] Manual tests

<!-- Attach code, screenshot, video used for manual testing here. -->

### Does this PR require documentation update?

- [ ] No. You can skip the rest of this section.
- [ ] Yes. I've updated:
  - [ ] Examples
  - [ ] API references
  - [ ] Instructions

### Release Notes

#### Is this a user-facing change?

- [ ] No. You can skip the rest of this section.
- [ ] Yes. Give a description of this change to be included in the release notes for MLflow users.

<!-- Details in 1-2 sentences. You can just refer to another PR with a description if this PR is part of a larger change. -->

#### What component(s), interfaces, languages, and integrations does this PR affect?

Components

- [ ] `area/artifacts`: Artifact stores and artifact logging
- [ ] `area/build`: Build and test infrastructure for MLflow
- [ ] `area/deployments`: MLflow Deployments client APIs, server, and third-party Deployments integrations
- [ ] `area/docs`: MLflow documentation pages
- [ ] `area/examples`: Example code
- [ ] `area/model-registry`: Model Registry service, APIs, and the fluent client calls for Model Registry
- [ ] `area/models`: MLmodel format, model serialization/deserialization, flavors
- [ ] `area/recipes`: Recipes, Recipe APIs, Recipe configs, Recipe Templates
- [ ] `area/projects`: MLproject format, project running backends
- [ ] `area/scoring`: MLflow Model server, model deployment tools, Spark UDFs
- [ ] `area/server-infra`: MLflow Tracking server backend
- [ ] `area/tracking`: Tracking Service, tracking client APIs, autologging

Interface

- [ ] `area/uiux`: Front-end, user experience, plotting, JavaScript, JavaScript dev server
- [ ] `area/docker`: Docker use across MLflow's components, such as MLflow Projects and MLflow Models
- [ ] `area/sqlalchemy`: Use of SQLAlchemy in the Tracking Service or Model Registry
- [ ] `area/windows`: Windows support

Language

- [ ] `language/r`: R APIs and clients
- [ ] `language/java`: Java APIs and clients
- [ ] `language/new`: Proposals for new client languages

Integrations

- [ ] `integrations/azure`: Azure and Azure ML integrations
- [ ] `integrations/sagemaker`: SageMaker integrations
- [ ] `integrations/databricks`: Databricks integrations

<!--
Insert an empty named anchor here to allow jumping to this section with a fragment URL
(e.g. https://github.com/mlflow/mlflow/pull/123#user-content-release-note-category).
Note that GitHub prefixes anchor names in markdown with "user-content-".
-->

<a name="release-note-category"></a>

#### How should the PR be classified in the release notes? Choose one:

- [ ] `rn/none` - No description will be included. The PR will be mentioned only by the PR number in the "Small Bugfixes and Documentation Updates" section
- [ ] `rn/breaking-change` - The PR will be mentioned in the "Breaking Changes" section
- [ ] `rn/feature` - A new user-facing feature worth mentioning in the release notes
- [ ] `rn/bug-fix` - A user-facing bug fix worth mentioning in the release notes
- [ ] `rn/documentation` - A user-facing documentation change worth mentioning in the release notes

#### Should this PR be included in the next patch release?

`Yes` should be selected for bug fixes, documentation updates, and other small changes. `No` should be selected for new features and larger changes. If you're unsure about the release classification of this PR, leave this unchecked to let the maintainers decide.

<details>
<summary>What is a minor/patch release?</summary>

- Minor release: a release that increments the second part of the version number (e.g., 1.2.0 -> 1.3.0).
  Bug fixes, doc updates and new features usually go into minor releases.
- Patch release: a release that increments the third part of the version number (e.g., 1.2.0 -> 1.2.1).
  Bug fixes and doc updates usually go into patch releases.

</details>

<!-- patch -->

- [ ] Yes (this PR will be cherry-picked and included in the next patch release)
- [ ] No (this PR will be included in the next minor release)

Patch:
diff --git a/.github/workflows/autoformat.yml b/.github/workflows/autoformat.yml
index 8c0208ab4..0d1812bdd 100644
--- a/.github/workflows/autoformat.yml
+++ b/.github/workflows/autoformat.yml
@@ -35,7 +35,6 @@ jobs:
         id: judge
         uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             core.debug(JSON.stringify(context, null, 2));
             const autoformat = require('./.github/workflows/autoformat.js');
@@ -203,7 +202,6 @@ jobs:
       - name: Update status
         uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const needs = ${{ toJson(needs) }};
             const head_sha = '${{ needs.check-comment.outputs.head_sha }}'
diff --git a/.github/workflows/cancel.yml b/.github/workflows/cancel.yml
index 3b70f1fbb..57ccea95a 100644
--- a/.github/workflows/cancel.yml
+++ b/.github/workflows/cancel.yml
@@ -19,7 +19,6 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const script = require(
               `${process.env.GITHUB_WORKSPACE}/.github/workflows/cancel.js`
diff --git a/.github/workflows/patch.yml b/.github/workflows/patch.yml
index cf6406835..c79647187 100644
--- a/.github/workflows/patch.yml
+++ b/.github/workflows/patch.yml
@@ -25,7 +25,6 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const script = require('./.github/workflows/patch.js');
             await script({ github, context, core });
diff --git a/.github/workflows/protect.yml b/.github/workflows/protect.yml
index 08fe16f5f..b15e814ec 100644
--- a/.github/workflows/protect.yml
+++ b/.github/workflows/protect.yml
@@ -26,7 +26,6 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const script = require('./.github/workflows/protect.js');
             await script({ github, context });
diff --git a/.github/workflows/release-note.yml b/.github/workflows/release-note.yml
index 1cebdc915..ea391029d 100644
--- a/.github/workflows/release-note.yml
+++ b/.github/workflows/release-note.yml
@@ -39,7 +39,6 @@ jobs:
       - name: validate-labeled
         uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             // https://github.com/actions/github-script#run-a-separate-file
             const script = require("./.github/workflows/release-note.js");
@@ -56,7 +55,6 @@ jobs:
       - name: post-merge
         uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             // https://github.com/actions/github-script#run-a-separate-file
             const script = require("./.github/workflows/release-note.js");
diff --git a/.github/workflows/rerun.yml b/.github/workflows/rerun.yml
index eaa195f98..e99bf13be 100644
--- a/.github/workflows/rerun.yml
+++ b/.github/workflows/rerun.yml
@@ -24,7 +24,6 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const rerun = require(`${process.env.GITHUB_WORKSPACE}/.github/workflows/rerun.js`);
             await rerun({ context, github, workflow_id: "cross-version-tests.yml" });
diff --git a/.github/workflows/team-review.yml b/.github/workflows/team-review.yml
index 46a250dbb..e78a76c11 100644
--- a/.github/workflows/team-review.yml
+++ b/.github/workflows/team-review.yml
@@ -9,7 +9,6 @@ jobs:
     runs-on: ubuntu-latest
     if: ${{ github.event.requested_reviewer.login == 'mlflow-automation'}}
     steps:
-      - uses: actions/checkout@v4
       - uses: actions/github-script@v7
         with:
           github-token: ${{ secrets.MLFLOW_AUTOMATION_TOKEN }}
diff --git a/.github/workflows/warn-on-editing-requirements.yml b/.github/workflows/warn-on-editing-requirements.yml
index d6a89ad4c..e659bf870 100644
--- a/.github/workflows/warn-on-editing-requirements.yml
+++ b/.github/workflows/warn-on-editing-requirements.yml
@@ -14,7 +14,6 @@ jobs:
       - name: Fail without core maintainer approval
         uses: actions/github-script@v7
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const prNumber = context.payload.pull_request.number;
             const repo = context.repo;
diff --git a/mlflow/entities/assessment.py b/mlflow/entities/assessment.py
index e08196ce9..0e5fff07a 100644
--- a/mlflow/entities/assessment.py
+++ b/mlflow/entities/assessment.py
@@ -132,20 +132,6 @@ class Assessment(_MlflowObject):
             return self.to_dictionary() == __o.to_dictionary()
         return False
 
-    def get_value_type(self) -> str:
-        """
-        The type of the assessment value.
-
-        Returns:
-            str: The type of the assessment value.
-        """
-        if self.boolean_value is not None:
-            return "boolean"
-        elif self.numeric_value is not None:
-            return "numeric"
-        elif self.string_value is not None:
-            return "string"
-
     def to_dictionary(self) -> Dict[str, Any]:
         return {
             "evaluation_id": self.evaluation_id,
diff --git a/mlflow/entities/evaluation.py b/mlflow/entities/evaluation.py
new file mode 100644
index 000000000..dbd6a23d8
--- /dev/null
+++ b/mlflow/entities/evaluation.py
@@ -0,0 +1,188 @@
+from typing import Any, Dict, List, Optional
+
+from mlflow.entities._mlflow_object import _MlflowObject
+from mlflow.entities.assessment import Assessment
+from mlflow.entities.evaluation_tag import EvaluationTag  # Assuming EvaluationTag is in this module
+from mlflow.entities.metric import Metric
+
+
+class Evaluation(_MlflowObject):
+    """
+    Evaluation result data, including inputs, outputs, targets, assessments, and more.
+    """
+
+    def __init__(
+        self,
+        evaluation_id: str,
+        run_id: str,
+        inputs_id: str,
+        inputs: Dict[str, Any],
+        outputs: Optional[Dict[str, Any]] = None,
+        request_id: Optional[str] = None,
+        targets: Optional[Dict[str, Any]] = None,
+        error_code: Optional[str] = None,
+        error_message: Optional[str] = None,
+        assessments: Optional[List[Assessment]] = None,
+        metrics: Optional[List[Metric]] = None,
+        tags: Optional[List[EvaluationTag]] = None,
+    ):
+        """
+        Construct a new mlflow.entities.Evaluation instance.
+
+        Args:
+            evaluation_id: A unique identifier for the evaluation.
+            run_id: The ID of the MLflow Run containing the Evaluation.
+            inputs_id: A unique identifier for the input names and values for evaluation.
+            inputs: Input names and values for evaluation.
+            outputs: Outputs obtained during inference.
+            request_id: The ID of an MLflow Trace corresponding to the inputs and outputs.
+            targets: Expected values that the model should produce during inference.
+            error_code: An error code representing any issues encountered during the evaluation.
+            error_message: A descriptive error message representing any issues encountered during
+                the evaluation.
+            assessments: Assessments for the evaluation.
+            metrics: Objective numerical metrics for the evaluation, e.g., "number of input tokens",
+                "number of output tokens".
+            tags: List of tags associated with the evaluation.
+        """
+        self._evaluation_id = evaluation_id
+        self._run_id = run_id
+        self._inputs_id = inputs_id
+        self._inputs = inputs
+        self._outputs = outputs
+        self._request_id = request_id
+        self._targets = targets
+        self._error_code = error_code
+        self._error_message = error_message
+        self._assessments = assessments
+        self._metrics = metrics
+        self._tags = tags
+
+    @property
+    def evaluation_id(self) -> str:
+        """The evaluation ID."""
+        return self._evaluation_id
+
+    @property
+    def run_id(self) -> str:
+        """The ID of the MLflow Run containing the evaluation"""
+        return self._run_id
+
+    @property
+    def inputs_id(self) -> str:
+        """The evaluation inputs ID."""
+        return self._inputs_id
+
+    @property
+    def inputs(self) -> Dict[str, Any]:
+        """The evaluation inputs."""
+        return self._inputs
+
+    @property
+    def outputs(self) -> Optional[Dict[str, Any]]:
+        """The evaluation outputs."""
+        return self._outputs
+
+    @property
+    def request_id(self) -> Optional[str]:
+        """The evaluation request ID."""
+        return self._request_id
+
+    @property
+    def targets(self) -> Optional[Dict[str, Any]]:
+        """The evaluation targets."""
+        return self._targets
+
+    @property
+    def error_code(self) -> Optional[str]:
+        """The evaluation error code."""
+        return self._error_code
+
+    @property
+    def error_message(self) -> Optional[str]:
+        """The evaluation error message."""
+        return self._error_message
+
+    @property
+    def assessments(self) -> Optional[List[Assessment]]:
+        """The evaluation assessments."""
+        return self._assessments
+
+    @property
+    def metrics(self) -> Optional[List[Metric]]:
+        """The evaluation metrics."""
+        return self._metrics
+
+    @property
+    def tags(self) -> Optional[List[EvaluationTag]]:
+        """The evaluation tags."""
+        return self._tags
+
+    def __eq__(self, __o):
+        if isinstance(__o, self.__class__):
+            return self.to_dictionary() == __o.to_dictionary()
+        return False
+
+    def to_dictionary(self) -> Dict[str, Any]:
+        """
+        Convert the Evaluation object to a dictionary.
+
+        Returns:
+            dict: The Evaluation object represented as a dictionary.
+        """
+        evaluation_dict = {
+            "evaluation_id": self.evaluation_id,
+            "run_id": self.run_id,
+            "inputs_id": self.inputs_id,
+            "inputs": self.inputs,
+            "outputs": self.outputs,
+            "request_id": self.request_id,
+            "targets": self.targets,
+            "error_code": self.error_code,
+            "error_message": self.error_message,
+            "assessments": [assess.to_dictionary() for assess in self.assessments]
+            if self.assessments
+            else None,
+            "metrics": [metric.to_dictionary() for metric in self.metrics]
+            if self.metrics
+            else None,
+            "tags": [tag.to_dictionary() for tag in self.tags] if self.tags else None,
+        }
+        return {k: v for k, v in evaluation_dict.items() if v is not None}
+
+    @classmethod
+    def from_dictionary(cls, evaluation_dict: Dict[str, Any]):
+        """
+        Create an Evaluation object from a dictionary.
+
+        Args:
+            evaluation_dict (dict): Dictionary containing evaluation information.
+
+        Returns:
+            Evaluation: The Evaluation object created from the dictionary.
+        """
+        assessments = None
+        if "assessments" in evaluation_dict:
+            assessments = [
+                Assessment.from_dictionary(assess) for assess in evaluation_dict["assessments"]
+            ]
+        metrics = None
+        if "metrics" in evaluation_dict:
+            metrics = [Metric.from_dictionary(metric) for metric in evaluation_dict["metrics"]]
+        tags = None
+        if "tags" in evaluation_dict:
+            tags = [EvaluationTag(tag["key"], tag["value"]) for tag in evaluation_dict["tags"]]
+        return cls(
+            evaluation_id=evaluation_dict["evaluation_id"],
+            run_id=evaluation_dict["run_id"],
+            inputs_id=evaluation_dict["inputs_id"],
+            inputs=evaluation_dict["inputs"],
+            outputs=evaluation_dict.get("outputs"),
+            request_id=evaluation_dict.get("request_id"),
+            targets=evaluation_dict.get("targets"),
+            error_code=evaluation_dict.get("error_code"),
+            error_message=evaluation_dict.get("error_message"),
+            assessments=assessments,
+            metrics=metrics,
+            tags=tags,
+        )
diff --git a/mlflow/evaluation/__init__.py b/mlflow/evaluation/__init__.py
new file mode 100644
index 000000000..919020737
--- /dev/null
+++ b/mlflow/evaluation/__init__.py
@@ -0,0 +1,12 @@
+from mlflow.entities.assessment_source import AssessmentSource, AssessmentSourceType
+from mlflow.evaluation.assessment import Assessment
+from mlflow.evaluation.evaluation import Evaluation
+from mlflow.evaluation.fluent import log_evaluations
+
+__all__ = [
+    "Assessment",
+    "AssessmentSource",
+    "AssessmentSourceType",
+    "Evaluation",
+    "log_evaluations",
+]
diff --git a/mlflow/evaluation/assessment.py b/mlflow/evaluation/assessment.py
new file mode 100644
index 000000000..2ca433a16
--- /dev/null
+++ b/mlflow/evaluation/assessment.py
@@ -0,0 +1,155 @@
+import numbers
+import time
+from typing import Any, Dict, Optional, Union
+
+from mlflow.entities._mlflow_object import _MlflowObject
+from mlflow.entities.assessment import Assessment as AssessmentEntity
+from mlflow.entities.assessment_source import AssessmentSource
+from mlflow.exceptions import MlflowException
+from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE
+
+
+class Assessment(_MlflowObject):
+    """
+    Assessment data associated with an evaluation result.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        source: AssessmentSource,
+        value: Optional[Union[bool, float, str]] = None,
+        rationale: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        error_code: Optional[str] = None,
+        error_message: Optional[str] = None,
+    ):
+        """Construct a new Assessment instance.
+
+        Args:
+            name: The name of the piece of assessment.
+            source: The source of the assessment (AssessmentSource instance).
+            value: The value of the assessment. This can be a boolean, numeric, or string value.
+            rationale: The rationale / justification for the value.
+            metadata: Additional metadata for the assessment, e.g. the index of the chunk in the
+                      retrieved documents that the assessment applies to.
+            error_code: An error code representing any issues encountered during the assessment.
+            error_message: A descriptive error message representing any issues encountered during
+                the assessment.
+        """
+        if (value is None) == (error_code is None):
+            raise MlflowException(
+                "Exactly one of value or error_code must be specified for an assessment.",
+                INVALID_PARAMETER_VALUE,
+            )
+
+        if value is not None and error_message is not None:
+            raise MlflowException(
+                "error_message cannot be specified when value is specified.",
+                INVALID_PARAMETER_VALUE,
+            )
+
+        self._name = name
+        self._source = source
+        self._value = value
+        self._rationale = rationale
+        self._metadata = metadata or {}
+        self._error_code = error_code
+        self._error_message = error_message
+
+        self._boolean_value = None
+        self._numeric_value = None
+        self._string_value = None
+        if isinstance(value, bool):
+            self._boolean_value = value
+        elif isinstance(value, numbers.Number):
+            self._numeric_value = float(value)
+        elif value is not None:
+            self._string_value = str(value)
+
+    @property
+    def name(self) -> str:
+        """The name of the assessment."""
+        return self._name
+
+    @property
+    def value(self) -> Union[bool, float, str]:
+        """The assessment value."""
+        return self._value
+
+    @property
+    def rationale(self) -> Optional[str]:
+        """The rationale / justification for the assessment."""
+        return self._rationale
+
+    @property
+    def source(self) -> AssessmentSource:
+        """The source of the assessment."""
+        return self._source
+
+    @property
+    def metadata(self) -> Dict[str, Any]:
+        """The metadata associated with the assessment."""
+        return self._metadata
+
+    @property
+    def error_code(self) -> Optional[str]:
+        """The error code."""
+        return self._error_code
+
+    @property
+    def error_message(self) -> Optional[str]:
+        """The error message."""
+        return self._error_message
+
+    def __eq__(self, __o):
+        if isinstance(__o, self.__class__):
+            return self.to_dictionary() == __o.to_dictionary()
+        return False
+
+    def to_dictionary(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "source": self.source.to_dictionary(),
+            "value": self.value,
+            "rationale": self.rationale,
+            "metadata": self.metadata,
+            "error_code": self.error_code,
+            "error_message": self.error_message,
+        }
+
+    @classmethod
+    def from_dictionary(cls, assessment_dict: Dict[str, Any]) -> "Assessment":
+        """
+        Create an Assessment object from a dictionary.
+
+        Args:
+            assessment_dict (dict): Dictionary containing assessment information.
+
+        Returns:
+            Assessment: The Assessment object created from the dictionary.
+        """
+        return cls(
+            name=assessment_dict["name"],
+            source=AssessmentSource.from_dictionary(assessment_dict["source"]),
+            value=assessment_dict.get("value"),
+            rationale=assessment_dict.get("rationale"),
+            metadata=assessment_dict.get("metadata"),
+            error_code=assessment_dict.get("error_code"),
+            error_message=assessment_dict.get("error_message"),
+        )
+
+    def _to_entity(self, evaluation_id: str) -> AssessmentEntity:
+        return AssessmentEntity(
+            evaluation_id=evaluation_id,
+            name=self._name,
+            source=self._source,
+            timestamp=int(time.time() * 1000),
+            boolean_value=self._boolean_value,
+            numeric_value=self._numeric_value,
+            string_value=self._string_value,
+            rationale=self._rationale,
+            metadata=self._metadata,
+            error_code=self._error_code,
+            error_message=self._error_message,
+        )
diff --git a/mlflow/evaluation/evaluation.py b/mlflow/evaluation/evaluation.py
new file mode 100644
index 000000000..0aca3286b
--- /dev/null
+++ b/mlflow/evaluation/evaluation.py
@@ -0,0 +1,217 @@
+import hashlib
+import json
+from typing import Any, Dict, List, Optional, Union
+
+from mlflow.entities._mlflow_object import _MlflowObject
+from mlflow.entities.evaluation import Evaluation as EvaluationEntity
+from mlflow.entities.evaluation_tag import EvaluationTag  # Assuming EvaluationTag is in this module
+from mlflow.entities.metric import Metric
+from mlflow.evaluation.assessment import Assessment
+from mlflow.tracing.utils import TraceJSONEncoder
+
+
+class Evaluation(_MlflowObject):
+    """
+    Evaluation result data.
+    """
+
+    def __init__(
+        self,
+        inputs: Dict[str, Any],
+        outputs: Optional[Dict[str, Any]] = None,
+        inputs_id: Optional[str] = None,
+        request_id: Optional[str] = None,
+        targets: Optional[Dict[str, Any]] = None,
+        error_code: Optional[str] = None,
+        error_message: Optional[str] = None,
+        assessments: Optional[List[Assessment]] = None,
+        metrics: Optional[Union[Dict[str, float], List[Metric]]] = None,
+        tags: Optional[Dict[str, str]] = None,
+    ):
+        """
+        Construct a new Evaluation instance.
+
+        Args:
+            inputs: Input names and values for evaluation.
+            outputs: Outputs obtained during inference.
+            inputs_id: A unique identifier for the input names and values for evaluation.
+            request_id: The ID of an MLflow Trace corresponding to the inputs and outputs.
+            targets: Expected values that the model should produce during inference.
+            error_code: An error code representing any issues encountered during the evaluation.
+            error_message: A descriptive error message representing any issues encountered during
+                the evaluation.
+            assessments: Assessments for the evaluation.
+            metrics: Objective numerical metrics for the evaluation, e.g., "number of input tokens",
+                "number of output tokens".
+            tags: Dictionary of tags associated with the evaluation.
+        """
+        if isinstance(metrics, dict):
+            metrics = [
+                Metric(key=key, value=value, timestamp=0, step=0) for key, value in metrics.items()
+            ]
+        if isinstance(tags, dict):
+            tags = [EvaluationTag(key=str(key), value=str(value)) for key, value in tags.items()]
+
+        self._inputs = inputs
+        self._outputs = outputs
+        self._inputs_id = inputs_id or _generate_inputs_id(inputs)
+        self._request_id = request_id
+        self._targets = targets
+        self._error_code = error_code
+        self._error_message = error_message
+        self._assessments = assessments
+        self._metrics = metrics
+        self._tags = tags
+
+    @property
+    def inputs_id(self) -> str:
+        """The evaluation inputs ID."""
+        return self._inputs_id
+
+    @property
+    def inputs(self) -> Dict[str, Any]:
+        """The evaluation inputs."""
+        return self._inputs
+
+    @property
+    def outputs(self) -> Optional[Dict[str, Any]]:
+        """The evaluation outputs."""
+        return self._outputs
+
+    @property
+    def request_id(self) -> Optional[str]:
+        """The evaluation request ID."""
+        return self._request_id
+
+    @property
+    def targets(self) -> Optional[Dict[str, Any]]:
+        """The evaluation targets."""
+        return self._targets
+
+    @property
+    def error_code(self) -> Optional[str]:
+        """The evaluation error code."""
+        return self._error_code
+
+    @property
+    def error_message(self) -> Optional[str]:
+        """The evaluation error message."""
+        return self._error_message
+
+    @property
+    def assessments(self) -> Optional[List[Assessment]]:
+        """The evaluation assessments."""
+        return self._assessments
+
+    @property
+    def metrics(self) -> Optional[List[Metric]]:
+        """The evaluation metrics."""
+        return self._metrics
+
+    @property
+    def tags(self) -> Optional[Dict[str, str]]:
+        """The evaluation tags."""
+        return self._tags
+
+    def __eq__(self, __o):
+        if isinstance(__o, self.__class__):
+            return self.to_dictionary() == __o.to_dictionary()
+        return False
+
+    def _to_entity(self, run_id: str, evaluation_id: str) -> EvaluationEntity:
+        """
+        Convert the Evaluation object to an EvaluationEntity object.
+
+        Returns:
+            EvaluationEntity: An EvaluationEntity object.
+        """
+        return EvaluationEntity(
+            evaluation_id=evaluation_id,
+            run_id=run_id,
+            inputs_id=self.inputs_id,
+            inputs=self.inputs,
+            outputs=self.outputs,
+            request_id=self.request_id,
+            targets=self.targets,
+            error_code=self.error_code,
+            error_message=self.error_message,
+            assessments=[assess._to_entity(evaluation_id) for assess in self.assessments]
+            if self.assessments
+            else None,
+            metrics=self.metrics,
+            tags=self.tags,
+        )
+
+    def to_dictionary(self) -> Dict[str, Any]:
+        """
+        Convert the Evaluation object to a dictionary.
+
+        Returns:
+            dict: The Evaluation object represented as a dictionary.
+        """
+        evaluation_dict = {
+            "inputs_id": self.inputs_id,
+            "inputs": self.inputs,
+            "outputs": self.outputs,
+            "request_id": self.request_id,
+            "targets": self.targets,
+            "error_code": self.error_code,
+            "error_message": self.error_message,
+            "assessments": [assess.to_dictionary() for assess in self.assessments]
+            if self.assessments
+            else None,
+            "metrics": [metric.to_dictionary() for metric in self.metrics]
+            if self.metrics
+            else None,
+            "tags": [tag.to_dictionary() for tag in self.tags] if self.tags else None,
+        }
+        return {k: v for k, v in evaluation_dict.items() if v is not None}
+
+    @classmethod
+    def from_dictionary(cls, evaluation_dict: Dict[str, Any]):
+        """
+        Create an Evaluation object from a dictionary.
+
+        Args:
+            evaluation_dict (dict): Dictionary containing evaluation information.
+
+        Returns:
+            Evaluation: The Evaluation object created from the dictionary.
+        """
+        assessments = None
+        if "assessments" in evaluation_dict:
+            assessments = [
+                Assessment.from_dictionary(assess) for assess in evaluation_dict["assessments"]
+            ]
+        metrics = None
+        if "metrics" in evaluation_dict:
+            metrics = [Metric.from_dictionary(metric) for metric in evaluation_dict["metrics"]]
+        tags = None
+        if "tags" in evaluation_dict:
+            tags = [EvaluationTag(tag["key"], tag["value"]) for tag in evaluation_dict["tags"]]
+        return cls(
+            inputs_id=evaluation_dict["inputs_id"],
+            inputs=evaluation_dict["inputs"],
+            outputs=evaluation_dict.get("outputs"),
+            request_id=evaluation_dict.get("request_id"),
+            targets=evaluation_dict.get("targets"),
+            error_code=evaluation_dict.get("error_code"),
+            error_message=evaluation_dict.get("error_message"),
+            assessments=assessments,
+            metrics=metrics,
+            tags=tags,
+        )
+
+
+def _generate_inputs_id(inputs: Dict[str, Any]) -> str:
+    """
+    Generates a unique identifier for the inputs.
+
+    Args:
+        inputs (Dict[str, Any]): Input fields used by the model to compute outputs.
+
+    Returns:
+        str: A unique identifier for the inputs.
+    """
+    inputs_json = json.dumps(inputs, sort_keys=True, cls=TraceJSONEncoder)
+    return hashlib.sha256(inputs_json.encode("utf-8")).hexdigest()
diff --git a/mlflow/evaluation/fluent.py b/mlflow/evaluation/fluent.py
new file mode 100644
index 000000000..4b179b616
--- /dev/null
+++ b/mlflow/evaluation/fluent.py
@@ -0,0 +1,42 @@
+import uuid
+from typing import List, Optional
+
+from mlflow.entities.evaluation import Evaluation as EvaluationEntity
+from mlflow.evaluation.evaluation import Evaluation
+from mlflow.evaluation.utils import evaluations_to_dataframes
+from mlflow.tracking.client import MlflowClient
+from mlflow.tracking.fluent import _get_or_start_run
+
+
+def log_evaluations(
+    *, evaluations: List[Evaluation], run_id: Optional[str] = None
+) -> List[EvaluationEntity]:
+    """
+    Logs one or more evaluations to an MLflow Run.
+
+    Args:
+      evaluations (List[Evaluation]): List of one or more MLflow Evaluation objects.
+      run_id (Optional[str]): ID of the MLflow Run to log the evaluation. If unspecified, the
+          current active run is used, or a new run is started.
+
+    Returns:
+      List[EvaluationEntity]: The logged Evaluation objects.
+    """
+    run_id = run_id if run_id is not None else _get_or_start_run().info.run_id
+    if not evaluations:
+        return []
+
+    client = MlflowClient()
+    evaluation_entities = [
+        evaluation._to_entity(run_id=run_id, evaluation_id=uuid.uuid4().hex)
+        for evaluation in evaluations
+    ]
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes(
+        evaluation_entities
+    )
+    client.log_table(run_id=run_id, data=evaluations_df, artifact_file="_evaluations.json")
+    client.log_table(run_id=run_id, data=metrics_df, artifact_file="_metrics.json")
+    client.log_table(run_id=run_id, data=assessments_df, artifact_file="_assessments.json")
+    client.log_table(run_id=run_id, data=tags_df, artifact_file="_tags.json")
+
+    return evaluation_entities
diff --git a/mlflow/evaluation/utils.py b/mlflow/evaluation/utils.py
new file mode 100644
index 000000000..e907f6496
--- /dev/null
+++ b/mlflow/evaluation/utils.py
@@ -0,0 +1,191 @@
+from typing import Dict, List, Tuple
+
+import pandas as pd
+
+from mlflow.entities.evaluation import Evaluation as EvaluationEntity
+
+
+def evaluations_to_dataframes(
+    evaluations: List[EvaluationEntity],
+) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    """
+    Converts a list of Evaluation entities to four separate DataFrames: one for main evaluation
+    data (excluding assessments and metrics), one for metrics, one for assessments, and one for
+    tags.
+
+    Args:
+        evaluations (List[Evaluation]): List of Evaluation entities.
+
+    Returns:
+        Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]: A tuple of four DataFrames
+        containing evaluation data, metrics data, assessments data, and tags data.
+    """
+    evaluations_data = []
+    metrics_data = []
+    assessments_data = []
+    tags_data = []
+
+    for evaluation in evaluations:
+        eval_dict = evaluation.to_dictionary()
+
+        # Extract assessment and metrics
+        assessments_list = eval_dict.pop("assessments", [])
+        metrics_list = eval_dict.pop("metrics", [])
+        tags_list = eval_dict.pop("tags", [])
+
+        evaluations_data.append(eval_dict)
+
+        for metric_dict in metrics_list:
+            metric_dict["evaluation_id"] = eval_dict["evaluation_id"]
+            # Remove 'step' key if it exists, since it is not valid for evaluation metrics
+            metric_dict.pop("step", None)
+            metrics_data.append(metric_dict)
+
+        for assess_dict in assessments_list:
+            assess_dict["evaluation_id"] = eval_dict["evaluation_id"]
+            assessments_data.append(assess_dict)
+
+        for tag_dict in tags_list:
+            tag_dict["evaluation_id"] = eval_dict["evaluation_id"]
+            tags_data.append(tag_dict)
+
+    evaluations_df = (
+        _apply_schema_to_dataframe(
+            pd.DataFrame(evaluations_data), _get_evaluations_dataframe_schema()
+        )
+        if evaluations_data
+        else _get_empty_evaluations_dataframe()
+    )
+    metrics_df = (
+        _apply_schema_to_dataframe(pd.DataFrame(metrics_data), _get_metrics_dataframe_schema())
+        if metrics_data
+        else _get_empty_metrics_dataframe()
+    )
+    assessments_df = (
+        _apply_schema_to_dataframe(
+            pd.DataFrame(assessments_data), _get_assessments_dataframe_schema()
+        )
+        if assessments_data
+        else _get_empty_assessments_dataframe()
+    )
+    tags_df = (
+        _apply_schema_to_dataframe(pd.DataFrame(tags_data), _get_tags_dataframe_schema())
+        if tags_data
+        else _get_empty_tags_dataframe()
+    )
+
+    return evaluations_df, metrics_df, assessments_df, tags_df
+
+
+def _get_evaluations_dataframe_schema() -> Dict[str, str]:
+    """
+    Returns the pandas schema for the evaluation DataFrame.
+    """
+    return {
+        "evaluation_id": "string",
+        "run_id": "string",
+        "inputs_id": "string",
+        "inputs": "object",
+        "outputs": "object",
+        "request_id": "object",
+        "targets": "object",
+        "error_code": "object",
+        "error_message": "object",
+    }
+
+
+def _get_empty_evaluations_dataframe() -> pd.DataFrame:
+    """
+    Creates an empty DataFrame with columns for evaluation data.
+    """
+    schema = _get_evaluations_dataframe_schema()
+    df = pd.DataFrame(columns=schema.keys())
+    return _apply_schema_to_dataframe(df, schema)
+
+
+def _get_assessments_dataframe_schema() -> Dict[str, str]:
+    """
+    Returns the pandas schema for the assessments DataFrame.
+    """
+    return {
+        "evaluation_id": "string",
+        "name": "string",
+        "source": "object",
+        "timestamp": "int64",
+        "boolean_value": "object",
+        "numeric_value": "object",
+        "string_value": "object",
+        "rationale": "object",
+        "metadata": "object",
+        "error_code": "object",
+        "error_message": "object",
+    }
+
+
+def _get_empty_assessments_dataframe() -> pd.DataFrame:
+    """
+    Creates an empty DataFrame with columns for evaluation assessments data.
+    """
+    schema = _get_assessments_dataframe_schema()
+    df = pd.DataFrame(columns=schema.keys())
+    return _apply_schema_to_dataframe(df, schema)
+
+
+def _get_metrics_dataframe_schema() -> Dict[str, str]:
+    """
+    Returns the pandas schema for the metrics DataFrame.
+    """
+    return {
+        "evaluation_id": "string",
+        "key": "string",
+        "value": "float64",
+        "timestamp": "int64",
+    }
+
+
+def _get_empty_metrics_dataframe() -> pd.DataFrame:
+    """
+    Creates an empty DataFrame with columns for evaluation metric data.
+    """
+    schema = _get_metrics_dataframe_schema()
+    df = pd.DataFrame(columns=schema.keys())
+    return _apply_schema_to_dataframe(df, schema)
+
+
+def _get_tags_dataframe_schema() -> Dict[str, str]:
+    """
+    Returns the pandas schema for the tags DataFrame.
+    """
+    return {
+        "evaluation_id": "string",
+        "key": "string",
+        "value": "string",
+    }
+
+
+def _get_empty_tags_dataframe() -> pd.DataFrame:
+    """
+    Creates an empty DataFrame with columns for evaluation tags data.
+    """
+    schema = _get_tags_dataframe_schema()
+    df = pd.DataFrame(columns=schema.keys())
+    return _apply_schema_to_dataframe(df, schema)
+
+
+def _apply_schema_to_dataframe(df: pd.DataFrame, schema: Dict[str, str]) -> pd.DataFrame:
+    """
+    Applies a schema to a DataFrame.
+
+    Args:
+        df (pd.DataFrame): DataFrame to apply the schema to.
+        schema (Dict[str, Any]): Schema to apply.
+
+    Returns:
+        pd.DataFrame: DataFrame with schema applied.
+    """
+    for column in df.columns:
+        df[column] = df[column].astype(schema[column])
+    # By default, null values are represented as `pd.NA` in pandas when reading a dataframe from
+    # JSON. However, MLflow entities use `None` to represent null values. Accordingly, we convert
+    # instances of pd.NA to None so that DataFrame rows can be parsed as MLflow entities
+    return df.replace(pd.NA, None)


Test patch:
diff --git a/.github/workflows/cross-version-test-runner.yml b/.github/workflows/cross-version-test-runner.yml
index 9d5afa786..1f9b228f8 100644
--- a/.github/workflows/cross-version-test-runner.yml
+++ b/.github/workflows/cross-version-test-runner.yml
@@ -24,7 +24,6 @@ jobs:
         id: get-ref
         with:
           result-encoding: string
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const runner = require('./.github/workflows/cross-version-test-runner.js');
             await runner.main({ context, github });
diff --git a/.github/workflows/test-package-build.yml b/.github/workflows/test-package-build.yml
index 100bc3ef7..ac5677703 100644
--- a/.github/workflows/test-package-build.yml
+++ b/.github/workflows/test-package-build.yml
@@ -117,7 +117,6 @@ jobs:
         env:
           WHEEL_SIZE: ${{ steps.build-dist.outputs.wheel-size }}
         with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const { owner, repo } = context.repo;
 
diff --git a/tests/entities/test_evaluation.py b/tests/entities/test_evaluation.py
new file mode 100644
index 000000000..ce5e0bdc4
--- /dev/null
+++ b/tests/entities/test_evaluation.py
@@ -0,0 +1,185 @@
+from mlflow.entities import Metric
+from mlflow.entities.assessment import Assessment
+from mlflow.entities.assessment_source import AssessmentSource
+from mlflow.entities.evaluation import Evaluation
+from mlflow.entities.evaluation_tag import EvaluationTag
+
+
+def test_evaluation_equality():
+    source_1 = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    metric_1 = Metric(key="metric1", value=1.1, timestamp=123, step=0)
+    tag_1 = EvaluationTag(key="tag1", value="value1")
+
+    # Valid evaluations
+    evaluation_1 = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"prediction": 0.5},
+        request_id="req1",
+        targets={"actual": 0.6},
+        assessments=[
+            Assessment(
+                evaluation_id="eval1",
+                name="relevance",
+                source=source_1,
+                timestamp=123456789,
+                numeric_value=0.9,
+            )
+        ],
+        metrics=[metric_1],
+        tags=[tag_1],
+        error_code="E001",
+        error_message="An error occurred",
+    )
+    evaluation_2 = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"prediction": 0.5},
+        request_id="req1",
+        targets={"actual": 0.6},
+        assessments=[
+            Assessment(
+                evaluation_id="eval1",
+                name="relevance",
+                source=source_1,
+                timestamp=123456789,
+                numeric_value=0.9,
+            )
+        ],
+        metrics=[metric_1],
+        tags=[tag_1],
+        error_code="E001",
+        error_message="An error occurred",
+    )
+    evaluation_3 = Evaluation(
+        evaluation_id="eval2",
+        run_id="run2",
+        inputs_id="inputs2",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"prediction": 0.5},
+        request_id="req2",
+        targets={"actual": 0.7},
+        assessments=[
+            Assessment(
+                evaluation_id="eval2",
+                name="relevance",
+                source=source_1,
+                timestamp=123456789,
+                numeric_value=0.8,
+            )
+        ],
+        metrics=[Metric(key="metric1", value=1.2, timestamp=123, step=0)],
+        tags=[EvaluationTag(key="tag2", value="value2")],
+        error_code="E002",
+        error_message="Another error occurred",
+    )
+
+    assert evaluation_1 == evaluation_2  # Same evaluation data
+    assert evaluation_1 != evaluation_3  # Different evaluation data
+
+
+def test_evaluation_properties():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    metric = Metric(key="metric1", value=1.1, timestamp=123, step=0)
+    tag = EvaluationTag(key="tag1", value="value1")
+    assessment = Assessment(
+        evaluation_id="eval1",
+        name="relevance",
+        source=source,
+        timestamp=123456789,
+        numeric_value=0.9,
+        rationale="Rationale text",
+        metadata={"key1": "value1"},
+    )
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"prediction": 0.5},
+        request_id="req1",
+        targets={"actual": 0.6},
+        assessments=[assessment],
+        metrics=[metric],
+        tags=[tag],
+        error_code="E001",
+        error_message="An error occurred",
+    )
+
+    assert evaluation.evaluation_id == "eval1"
+    assert evaluation.run_id == "run1"
+    assert evaluation.inputs_id == "inputs1"
+    assert evaluation.inputs == {"feature1": 1.0, "feature2": 2.0}
+    assert evaluation.outputs == {"prediction": 0.5}
+    assert evaluation.request_id == "req1"
+    assert evaluation.targets == {"actual": 0.6}
+    assert evaluation.error_code == "E001"
+    assert evaluation.error_message == "An error occurred"
+    assert evaluation.assessments == [assessment]
+    assert evaluation.metrics == [metric]
+    assert evaluation.tags == [tag]
+
+
+def test_evaluation_to_from_dictionary():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    metric = Metric(key="metric1", value=1.1, timestamp=123, step=0)
+    tag = EvaluationTag(key="tag1", value="value1")
+    assessment = Assessment(
+        evaluation_id="eval1",
+        name="relevance",
+        source=source,
+        timestamp=123456789,
+        numeric_value=0.9,
+        rationale="Rationale text",
+        metadata={"key1": "value1"},
+    )
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"prediction": 0.5},
+        request_id="req1",
+        targets={"actual": 0.6},
+        assessments=[assessment],
+        metrics=[metric],
+        tags=[tag],
+        error_code="E001",
+        error_message="An error occurred",
+    )
+    evaluation_dict = evaluation.to_dictionary()
+
+    expected_dict = {
+        "evaluation_id": "eval1",
+        "run_id": "run1",
+        "inputs_id": "inputs1",
+        "inputs": {"feature1": 1.0, "feature2": 2.0},
+        "outputs": {"prediction": 0.5},
+        "request_id": "req1",
+        "targets": {"actual": 0.6},
+        "assessments": [assessment.to_dictionary()],
+        "metrics": [metric.to_dictionary()],
+        "tags": [tag.to_dictionary()],
+        "error_code": "E001",
+        "error_message": "An error occurred",
+    }
+    assert evaluation_dict == expected_dict
+
+    recreated_evaluation = Evaluation.from_dictionary(evaluation_dict)
+    assert recreated_evaluation == evaluation
+
+
+def test_evaluation_construction_with_minimal_required_fields():
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+    )
+    evaluation_dict = evaluation.to_dictionary()
+    recreated_evaluation = Evaluation.from_dictionary(evaluation_dict)
+    assert recreated_evaluation == evaluation
diff --git a/tests/entities/test_evaluation_tag.py b/tests/entities/test_evaluation_tag.py
index 80502610b..d9d9d889a 100644
--- a/tests/entities/test_evaluation_tag.py
+++ b/tests/entities/test_evaluation_tag.py
@@ -36,7 +36,6 @@ def test_evaluation_tag_to_from_dictionary():
 
 
 def test_evaluation_tag_key_value_validation():
-    # Valid cases
     # Valid cases
     EvaluationTag(key="tag1", value="value1")
     EvaluationTag(key="tag2", value="value2")
diff --git a/tests/evaluate/logging/__init__.py b/tests/evaluate/logging/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/evaluate/logging/test_assessment.py b/tests/evaluate/logging/test_assessment.py
new file mode 100644
index 000000000..ed11a6a87
--- /dev/null
+++ b/tests/evaluate/logging/test_assessment.py
@@ -0,0 +1,314 @@
+from unittest.mock import patch
+
+import pytest
+
+from mlflow.entities.assessment_source import AssessmentSource
+from mlflow.evaluation import Assessment
+from mlflow.exceptions import MlflowException
+
+
+def test_assessment_equality():
+    source_1 = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    source_2 = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    source_3 = AssessmentSource(source_type="AI_JUDGE", source_id="ai_1")
+
+    # Valid assessments
+    assessment_1 = Assessment(
+        name="relevance",
+        source=source_1,
+        value=0.9,
+    )
+    assessment_2 = Assessment(
+        name="relevance",
+        source=source_2,
+        value=0.9,
+    )
+    assessment_3 = Assessment(
+        name="relevance",
+        source=source_1,
+        value=0.8,
+    )
+    assessment_4 = Assessment(
+        name="relevance",
+        source=source_3,
+        value=0.9,
+    )
+    assessment_5 = Assessment(
+        name="relevance",
+        source=source_1,
+        error_code="E002",
+        error_message="A different error occurred.",
+    )
+    assessment_6 = Assessment(
+        name="relevance",
+        source=source_1,
+        error_code="E001",
+        error_message="Another error message.",
+    )
+
+    # Same name, source, and value
+    assert assessment_1 == assessment_2
+    assert assessment_1 != assessment_3  # Different value
+    assert assessment_1 != assessment_4  # Different source
+    assert assessment_1 != assessment_5  # One has value, other has error_code
+    assert assessment_5 != assessment_6  # Different error_code
+
+
+def test_assessment_properties():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=0.9,
+        rationale="Rationale text",
+        metadata={"key1": "value1"},
+    )
+
+    assert assessment.name == "relevance"
+    assert assessment.source == source
+    assert assessment.value == 0.9
+    assert assessment.rationale == "Rationale text"
+    assert assessment.metadata == {"key1": "value1"}
+    assert assessment.error_code is None
+    assert assessment.error_message is None
+
+
+def test_assessment_to_from_dictionary():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=0.9,
+        rationale="Rationale text",
+        metadata={"key1": "value1"},
+    )
+    assessment_dict = assessment.to_dictionary()
+
+    expected_dict = {
+        "name": "relevance",
+        "source": source.to_dictionary(),
+        "value": 0.9,
+        "rationale": "Rationale text",
+        "metadata": {"key1": "value1"},
+        "error_code": None,
+        "error_message": None,
+    }
+    assert assessment_dict == expected_dict
+
+    recreated_assessment = Assessment.from_dictionary(assessment_dict)
+    assert recreated_assessment == assessment
+
+
+def test_assessment_value_validation():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+
+    # Valid cases
+    try:
+        Assessment(
+            name="relevance",
+            source=source,
+            value=True,
+        )
+        Assessment(
+            name="relevance",
+            source=source,
+            value=0.9,
+        )
+        Assessment(
+            name="relevance",
+            source=source,
+            value="value",
+        )
+        Assessment(
+            name="relevance",
+            source=source,
+            error_code="E001",
+            error_message="Error",
+        )
+    except MlflowException:
+        pytest.fail("Valid value raised exception")
+
+    # Invalid case: more than one value type specified
+    with pytest.raises(
+        MlflowException,
+        match="Exactly one of value or error_code must be specified for an assessment.",
+    ):
+        Assessment(
+            name="relevance",
+            source=source,
+            value=True,
+            error_code="E002",
+        )
+
+    # Invalid case: no value type specified
+    with pytest.raises(
+        MlflowException,
+        match="Exactly one of value or error_code must be specified for an assessment.",
+    ):
+        Assessment(
+            name="relevance",
+            source=source,
+        )
+
+    # Invalid case: error_message specified with another value type
+    with pytest.raises(
+        MlflowException,
+        match="error_message cannot be specified when value is specified.",
+    ):
+        Assessment(
+            name="relevance",
+            source=source,
+            value=0.9,
+            error_message="An error occurred",
+        )
+
+    with pytest.raises(
+        MlflowException,
+        match="error_message cannot be specified when value is specified.",
+    ):
+        Assessment(
+            name="relevance",
+            source=source,
+            value="value",
+            error_message="An error occurred",
+        )
+
+    with pytest.raises(
+        MlflowException,
+        match="error_message cannot be specified when value is specified.",
+    ):
+        Assessment(
+            name="relevance",
+            source=source,
+            value=True,
+            error_message="An error occurred",
+        )
+
+
+def test_assessment_to_entity():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=0.9,
+        rationale="Rationale text",
+        metadata={"key1": "value1"},
+    )
+
+    evaluation_id = "evaluation_1"
+    with patch("time.time", return_value=1234567890):
+        assessment_entity = assessment._to_entity(evaluation_id)
+
+    assert assessment_entity.evaluation_id == evaluation_id
+    assert assessment_entity.name == assessment.name
+    assert assessment_entity.source == assessment.source
+    assert assessment_entity.boolean_value is None
+    assert assessment_entity.numeric_value == assessment.value
+    assert assessment_entity.string_value is None
+    assert assessment_entity.rationale == assessment.rationale
+    assert assessment_entity.metadata == assessment.metadata
+    assert assessment_entity.error_code is None
+    assert assessment_entity.error_message is None
+    assert assessment_entity.timestamp == 1234567890000  # Mocked timestamp
+
+
+def test_assessment_to_entity_with_error():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        error_code="E001",
+        error_message="An error occurred",
+    )
+
+    evaluation_id = "evaluation_1"
+    with patch("time.time", return_value=1234567890):
+        assessment_entity = assessment._to_entity(evaluation_id)
+
+    assert assessment_entity.evaluation_id == evaluation_id
+    assert assessment_entity.name == assessment.name
+    assert assessment_entity.source == assessment.source
+    assert assessment_entity.boolean_value is None
+    assert assessment_entity.numeric_value is None
+    assert assessment_entity.string_value is None
+    assert assessment_entity.rationale == assessment.rationale
+    assert assessment_entity.metadata == assessment.metadata
+    assert assessment_entity.error_code == assessment.error_code
+    assert assessment_entity.error_message == assessment.error_message
+    assert assessment_entity.timestamp == 1234567890000  # Mocked timestamp
+
+
+def test_assessment_to_entity_with_boolean_value():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=True,
+    )
+
+    evaluation_id = "evaluation_1"
+    with patch("time.time", return_value=1234567890):
+        assessment_entity = assessment._to_entity(evaluation_id)
+
+    assert assessment_entity.evaluation_id == evaluation_id
+    assert assessment_entity.name == assessment.name
+    assert assessment_entity.source == assessment.source
+    assert assessment_entity.boolean_value == assessment.value
+    assert assessment_entity.numeric_value is None
+    assert assessment_entity.string_value is None
+    assert assessment_entity.rationale == assessment.rationale
+    assert assessment_entity.metadata == assessment.metadata
+    assert assessment_entity.error_code is None
+    assert assessment_entity.error_message is None
+    assert assessment_entity.timestamp == 1234567890000  # Mocked timestamp
+
+
+def test_assessment_to_entity_with_string_value():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value="string value",
+    )
+
+    evaluation_id = "evaluation_1"
+    with patch("time.time", return_value=1234567890):
+        assessment_entity = assessment._to_entity(evaluation_id)
+
+    assert assessment_entity.evaluation_id == evaluation_id
+    assert assessment_entity.name == assessment.name
+    assert assessment_entity.source == assessment.source
+    assert assessment_entity.boolean_value is None
+    assert assessment_entity.numeric_value is None
+    assert assessment_entity.string_value == assessment.value
+    assert assessment_entity.rationale == assessment.rationale
+    assert assessment_entity.metadata == assessment.metadata
+    assert assessment_entity.error_code is None
+    assert assessment_entity.error_message is None
+    assert assessment_entity.timestamp == 1234567890000  # Mocked timestamp
+
+
+def test_assessment_with_full_metadata():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=0.9,
+        rationale="Detailed rationale",
+        metadata={"key1": "value1", "key2": "value2"},
+    )
+
+    assert assessment.metadata == {"key1": "value1", "key2": "value2"}
+    assert assessment.rationale == "Detailed rationale"
+
+
+def test_assessment_without_metadata():
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        name="relevance",
+        source=source,
+        value=0.9,
+    )
+
+    assert assessment.metadata == {}
+    assert assessment.rationale is None
diff --git a/tests/evaluate/logging/test_evaluation.py b/tests/evaluate/logging/test_evaluation.py
new file mode 100644
index 000000000..746ada957
--- /dev/null
+++ b/tests/evaluate/logging/test_evaluation.py
@@ -0,0 +1,244 @@
+from unittest.mock import patch
+
+from mlflow.entities import Metric
+from mlflow.entities.assessment_source import AssessmentSource
+from mlflow.entities.evaluation_tag import EvaluationTag
+from mlflow.evaluation import Assessment, Evaluation
+
+
+def test_evaluation_equality():
+    inputs = {"feature1": 1.0, "feature2": 2.0}
+    outputs = {"prediction": 0.5}
+    assessments = [
+        Assessment(
+            name="assessment1",
+            source=AssessmentSource(source_type="HUMAN", source_id="user_1"),
+            value=0.9,
+        )
+    ]
+    metrics = [Metric(key="metric1", value=1.1, timestamp=0, step=0)]
+    tags = {"tag1": "value1", "tag2": "value2"}
+
+    evaluation_1 = Evaluation(
+        inputs=inputs,
+        outputs=outputs,
+        assessments=assessments,
+        metrics=metrics,
+        tags=tags,
+    )
+    evaluation_2 = Evaluation(
+        inputs=inputs,
+        outputs=outputs,
+        assessments=assessments,
+        metrics=metrics,
+        tags=tags,
+    )
+    assert evaluation_1 == evaluation_2
+
+    evaluation_3 = Evaluation(
+        inputs={"feature1": 3.0, "feature2": 4.0},
+        outputs=outputs,
+    )
+    assert evaluation_1 != evaluation_3
+
+
+def test_evaluation_properties():
+    inputs = {"feature1": 1.0, "feature2": 2.0}
+    outputs = {"prediction": 0.5}
+    assessments = [
+        Assessment(
+            name="assessment1",
+            source=AssessmentSource(source_type="HUMAN", source_id="user_1"),
+            value=0.9,
+        )
+    ]
+    metrics = [Metric(key="metric1", value=1.1, timestamp=0, step=0)]
+    tags = {"tag1": "value1", "tag2": "value2"}
+
+    evaluation = Evaluation(
+        inputs=inputs,
+        outputs=outputs,
+        assessments=assessments,
+        metrics=metrics,
+        tags=tags,
+        request_id="req1",
+        targets={"target1": 1.0},
+        error_code="E001",
+        error_message="An error occurred",
+    )
+
+    assert evaluation.inputs == inputs
+    assert evaluation.outputs == outputs
+    assert evaluation.assessments == assessments
+    assert evaluation.metrics == metrics
+    assert evaluation.tags == [
+        EvaluationTag(key="tag1", value="value1"),
+        EvaluationTag(key="tag2", value="value2"),
+    ]
+    assert evaluation.request_id == "req1"
+    assert evaluation.targets == {"target1": 1.0}
+    assert evaluation.error_code == "E001"
+    assert evaluation.error_message == "An error occurred"
+
+
+def test_evaluation_to_from_dictionary():
+    inputs = {"feature1": 1.0, "feature2": 2.0}
+    outputs = {"prediction": 0.5}
+    assessments = [
+        Assessment(
+            name="assessment1",
+            source=AssessmentSource(source_type="HUMAN", source_id="user_1"),
+            value=0.9,
+        )
+    ]
+    metrics = [Metric(key="metric1", value=1.1, timestamp=0, step=0)]
+    tags = {"tag1": "value1", "tag2": "value2"}
+
+    evaluation = Evaluation(
+        inputs=inputs,
+        outputs=outputs,
+        assessments=assessments,
+        metrics=metrics,
+        tags=tags,
+        request_id="req1",
+        targets={"target1": 1.0},
+        error_code="E001",
+        error_message="An error occurred",
+    )
+    evaluation_dict = evaluation.to_dictionary()
+
+    expected_dict = {
+        "inputs_id": evaluation.inputs_id,
+        "inputs": inputs,
+        "outputs": outputs,
+        "request_id": "req1",
+        "targets": {"target1": 1.0},
+        "error_code": "E001",
+        "error_message": "An error occurred",
+        "assessments": [assessment.to_dictionary() for assessment in assessments],
+        "metrics": [metric.to_dictionary() for metric in metrics],
+        "tags": [tag.to_dictionary() for tag in evaluation.tags],
+    }
+    assert evaluation_dict == expected_dict
+
+    recreated_evaluation = Evaluation.from_dictionary(evaluation_dict)
+    assert recreated_evaluation == evaluation
+
+
+@patch("time.time", return_value=1234567890)
+def test_evaluation_to_entity(mock_time):
+    inputs = {"feature1": 1.0, "feature2": 2.0}
+    outputs = {"prediction": 0.5}
+    assessments = [
+        Assessment(
+            name="assessment1",
+            source=AssessmentSource(source_type="HUMAN", source_id="user_1"),
+            value=0.9,
+        )
+    ]
+    metrics = [Metric(key="metric1", value=1.1, timestamp=0, step=0)]
+    tags = {"tag1": "value1", "tag2": "value2"}
+
+    evaluation = Evaluation(
+        inputs=inputs,
+        outputs=outputs,
+        assessments=assessments,
+        metrics=metrics,
+        tags=tags,
+        request_id="req1",
+        targets={"target1": 1.0},
+        error_code="E001",
+        error_message="An error occurred",
+    )
+
+    entity = evaluation._to_entity(run_id="run1", evaluation_id="eval1")
+    assert entity.evaluation_id == "eval1"
+    assert entity.run_id == "run1"
+    assert entity.inputs_id == evaluation.inputs_id
+    assert entity.inputs == inputs
+    assert entity.outputs == outputs
+    assert entity.request_id == "req1"
+    assert entity.targets == {"target1": 1.0}
+    assert entity.error_code == "E001"
+    assert entity.error_message == "An error occurred"
+    assert entity.assessments == [a._to_entity("eval1") for a in assessments]
+    assert entity.metrics == metrics
+    assert entity.tags == [
+        EvaluationTag(key="tag1", value="value1"),
+        EvaluationTag(key="tag2", value="value2"),
+    ]
+
+
+def test_evaluation_inputs_id_uniqueness():
+    # Define a few different input objects
+    inputs_1 = {"feature1": 1.0, "feature2": 2.0}
+    inputs_2 = {"feature1": 1.0, "feature2": 2.0}  # Same as inputs_1
+    inputs_3 = {"feature1": 3.0, "feature2": 4.0}  # Different from inputs_1 and inputs_2
+    inputs_4 = {"feature1": "value1", "feature2": "value2"}
+    inputs_5 = {"feature1": "value1", "feature2": "value2"}  # Same as inputs_4
+    inputs_6 = {"feature1": "value3", "feature2": "value4"}  # Different from inputs_4 and inputs_5
+
+    # Create Evaluation objects
+    evaluation_1 = Evaluation(inputs=inputs_1)
+    evaluation_2 = Evaluation(inputs=inputs_2)
+    evaluation_3 = Evaluation(inputs=inputs_3)
+    evaluation_4 = Evaluation(inputs=inputs_4)
+    evaluation_5 = Evaluation(inputs=inputs_5)
+    evaluation_6 = Evaluation(inputs=inputs_6)
+
+    # Verify that inputs_id is the same for equivalent inputs
+    assert evaluation_1.inputs_id == evaluation_2.inputs_id
+    assert evaluation_4.inputs_id == evaluation_5.inputs_id
+
+    # Verify that inputs_id is different for different inputs
+    assert evaluation_1.inputs_id != evaluation_3.inputs_id
+    assert evaluation_1.inputs_id != evaluation_4.inputs_id
+    assert evaluation_1.inputs_id != evaluation_6.inputs_id
+    assert evaluation_4.inputs_id != evaluation_6.inputs_id
+
+    # Additional verification for unique inputs_id generation
+    inputs_7 = {"feature1": 5.0, "feature2": 6.0}
+    inputs_8 = {"feature1": 7.0, "feature2": 8.0}
+    evaluation_7 = Evaluation(inputs=inputs_7)
+    evaluation_8 = Evaluation(inputs=inputs_8)
+
+    assert evaluation_7.inputs_id != evaluation_8.inputs_id
+
+    # Ensure different orders of the same inputs result in the same inputs_id
+    inputs_9 = {"feature2": 2.0, "feature1": 1.0}  # Same values as inputs_1, but different order
+    evaluation_9 = Evaluation(inputs=inputs_9)
+
+    assert evaluation_1.inputs_id == evaluation_9.inputs_id
+
+
+def test_evaluation_with_non_json_serializable_inputs():
+    class NonSerializable:
+        def __init__(self, value):
+            self.value = value
+
+        def __str__(self):
+            return f"NonSerializable(value={self.value})"
+
+    # Define non-JSON-serializable inputs
+    inputs_1 = {"feature1": NonSerializable(1), "feature2": NonSerializable(2)}
+    inputs_2 = {"feature1": NonSerializable(1), "feature2": NonSerializable(2)}  # Same as inputs_1
+    inputs_3 = {
+        "feature1": NonSerializable(3),
+        "feature2": NonSerializable(4),
+    }  # Different from inputs_1
+
+    # Create Evaluation objects
+    evaluation_1 = Evaluation(inputs=inputs_1)
+    evaluation_2 = Evaluation(inputs=inputs_2)
+    evaluation_3 = Evaluation(inputs=inputs_3)
+
+    # Verify that inputs_id is the same for equivalent inputs
+    assert evaluation_1.inputs_id == evaluation_2.inputs_id
+
+    # Verify that inputs_id is different for different inputs
+    assert evaluation_1.inputs_id != evaluation_3.inputs_id
+
+    # Verify that inputs_id is generated
+    assert evaluation_1.inputs_id is not None
+    assert evaluation_2.inputs_id is not None
+    assert evaluation_3.inputs_id is not None
diff --git a/tests/evaluate/logging/test_fluent.py b/tests/evaluate/logging/test_fluent.py
new file mode 100644
index 000000000..fe056c1d1
--- /dev/null
+++ b/tests/evaluate/logging/test_fluent.py
@@ -0,0 +1,220 @@
+import pytest
+
+import mlflow
+from mlflow.entities import Metric
+from mlflow.entities.assessment_source import AssessmentSource, AssessmentSourceType
+from mlflow.entities.evaluation_tag import EvaluationTag
+from mlflow.evaluation import Assessment, Evaluation, log_evaluations
+
+from tests.evaluate.logging.utils import get_evaluation
+
+
+@pytest.fixture
+def end_run_at_test_end():
+    yield
+    mlflow.end_run()
+
+
+def test_log_evaluations_with_minimal_params_succeeds():
+    inputs1 = {"feature1": 1.0, "feature2": 2.0}
+    outputs1 = {"prediction": 0.5}
+
+    inputs2 = {"feature3": 3.0, "feature4": 4.0}
+    outputs2 = {"prediction": 0.8}
+
+    with mlflow.start_run():
+        # Create evaluation objects
+        evaluation1 = Evaluation(inputs=inputs1, outputs=outputs1)
+        evaluation2 = Evaluation(inputs=inputs2, outputs=outputs2)
+
+        # Log the evaluations
+        logged_evaluations = log_evaluations(evaluations=[evaluation1, evaluation2])
+        assert len(logged_evaluations) == 2
+
+        for logged_evaluation, expected_evaluation in zip(
+            logged_evaluations, [evaluation1, evaluation2]
+        ):
+            assert logged_evaluation.inputs == expected_evaluation.inputs
+            assert logged_evaluation.outputs == expected_evaluation.outputs
+            retrieved_evaluation = get_evaluation(
+                evaluation_id=logged_evaluation.evaluation_id,
+                run_id=mlflow.active_run().info.run_id,
+            )
+            assert retrieved_evaluation is not None
+            assert retrieved_evaluation.inputs == logged_evaluation.inputs
+            assert retrieved_evaluation.outputs == logged_evaluation.outputs
+
+
+def test_log_evaluations_with_all_params():
+    evaluations_data = [
+        (
+            {"feature1": 1.0, "feature2": 2.0},
+            {"prediction": 0.5},
+            {"actual": 1.0},
+            [
+                {
+                    "name": "assessment1",
+                    "value": 1.0,
+                    "source": {
+                        "source_type": "HUMAN",
+                        "source_id": "user_1",
+                        "metadata": {"sourcekey1": "sourcevalue1"},
+                    },
+                },
+                {
+                    "name": "assessment2",
+                    "value": 0.84,
+                    "source": {
+                        "source_type": "HUMAN",
+                        "source_id": "user_1",
+                        "metadata": {"sourcekey2": "sourcevalue2"},
+                    },
+                },
+            ],
+            [
+                Metric(key="metric1", value=1.4, timestamp=1717047609503, step=0),
+                Metric(key="metric2", value=1.2, timestamp=1717047609504, step=0),
+            ],
+            {"tag1": "value1", "tag2": "value2"},
+        ),
+        (
+            {"feature1": "text1", "feature2": "text2"},
+            {"prediction": "output_text"},
+            {"actual": "expected_text"},
+            [
+                Assessment(
+                    name="accuracy",
+                    value=0.8,
+                    source=AssessmentSource(
+                        source_type=AssessmentSourceType.HUMAN,
+                        source_id="user-1",
+                        metadata={"sourcekey3": "sourcevalue3"},
+                    ),
+                )
+            ],
+            {"metric1": 0.8, "metric2": 0.84},
+            {"tag3": "value3", "tag4": "value4"},
+        ),
+    ]
+
+    inputs_id = "unique-inputs-id"
+    request_id = "unique-request-id"
+
+    with mlflow.start_run() as run:
+        run_id = run.info.run_id
+
+        evaluations = []
+        for inputs, outputs, targets, assessments, metrics, tags in evaluations_data:
+            if isinstance(assessments[0], dict):
+                assessments = [Assessment.from_dictionary(assessment) for assessment in assessments]
+
+            if isinstance(metrics, dict):
+                metrics = [
+                    Metric(key=key, value=value, timestamp=0, step=0)
+                    for key, value in metrics.items()
+                ]
+
+            evaluation = Evaluation(
+                inputs=inputs,
+                outputs=outputs,
+                inputs_id=inputs_id,
+                request_id=request_id,
+                targets=targets,
+                assessments=assessments,
+                metrics=metrics,
+                tags=tags,
+            )
+            evaluations.append(evaluation)
+
+        # Log the evaluations
+        logged_evaluations = log_evaluations(evaluations=evaluations, run_id=run_id)
+
+        for logged_evaluation, (inputs, outputs, targets, assessments, metrics, tags) in zip(
+            logged_evaluations, evaluations_data
+        ):
+            # Assert the fields of the logged evaluation
+            assert logged_evaluation.inputs == inputs
+            assert logged_evaluation.outputs == outputs
+            assert logged_evaluation.inputs_id == inputs_id
+            assert logged_evaluation.request_id == request_id
+            assert logged_evaluation.targets == targets
+
+            logged_metrics = (
+                {metric.key: metric.value for metric in logged_evaluation.metrics}
+                if isinstance(metrics, list) and isinstance(metrics[0], Metric)
+                else metrics
+            )
+            assert {
+                metric.key: metric.value for metric in logged_evaluation.metrics
+            } == logged_metrics
+
+            logged_tags = (
+                {tag.key: tag.value for tag in logged_evaluation.tags}
+                if isinstance(tags, list) and isinstance(tags[0], EvaluationTag)
+                else tags
+            )
+            assert {tag.key: tag.value for tag in logged_evaluation.tags} == logged_tags
+
+            assessment_entities = [
+                Assessment.from_dictionary(assessment)._to_entity(
+                    evaluation_id=logged_evaluation.evaluation_id
+                )
+                if isinstance(assessment, dict)
+                else assessment._to_entity(evaluation_id=logged_evaluation.evaluation_id)
+                for assessment in assessments
+            ]
+
+            for logged_assessment, assessment_entity in zip(
+                logged_evaluation.assessments, assessment_entities
+            ):
+                assert logged_assessment.name == assessment_entity.name
+                assert logged_assessment.boolean_value == assessment_entity.boolean_value
+                assert logged_assessment.numeric_value == assessment_entity.numeric_value
+                assert logged_assessment.string_value == assessment_entity.string_value
+                assert logged_assessment.metadata == assessment_entity.metadata
+                assert logged_assessment.source == assessment_entity.source
+
+            retrieved_evaluation = get_evaluation(
+                evaluation_id=logged_evaluation.evaluation_id, run_id=run_id
+            )
+            assert logged_evaluation == retrieved_evaluation
+
+
+def test_log_evaluations_starts_run_if_not_started(end_run_at_test_end):
+    inputs = {"feature1": 1.0, "feature2": {"nested_feature": 2.0}}
+    outputs = {"prediction": 0.5}
+
+    # Ensure there is no active run
+    if mlflow.active_run() is not None:
+        mlflow.end_run()
+
+    # Log evaluation without explicitly starting a run
+    logged_evaluation = log_evaluations(evaluations=[Evaluation(inputs=inputs, outputs=outputs)])[0]
+
+    # Verify that a run has been started
+    active_run = mlflow.active_run()
+    assert active_run is not None, "Expected a run to be started automatically."
+
+    # Retrieve the evaluation using the run ID
+    retrieved_evaluation = get_evaluation(
+        evaluation_id=logged_evaluation.evaluation_id, run_id=active_run.info.run_id
+    )
+    assert retrieved_evaluation == logged_evaluation
+
+
+def test_evaluation_module_exposes_relevant_apis_for_logging():
+    import mlflow.evaluation
+
+    assert hasattr(mlflow.evaluation, "log_evaluations")
+    assert hasattr(mlflow.evaluation, "Evaluation")
+    assert hasattr(mlflow.evaluation, "Assessment")
+    assert hasattr(mlflow.evaluation, "AssessmentSource")
+    assert hasattr(mlflow.evaluation, "AssessmentSourceType")
+
+
+def test_log_evaluations_works_properly_with_empty_evaluations_list():
+    with mlflow.start_run():
+        log_evaluations(evaluations=[])
+
+        artifacts = mlflow.MlflowClient().list_artifacts(mlflow.active_run().info.run_id)
+        assert len(artifacts) == 0
diff --git a/tests/evaluate/logging/test_utils.py b/tests/evaluate/logging/test_utils.py
new file mode 100644
index 000000000..21074b367
--- /dev/null
+++ b/tests/evaluate/logging/test_utils.py
@@ -0,0 +1,259 @@
+from mlflow.entities import Metric
+from mlflow.entities.assessment import Assessment
+from mlflow.entities.assessment_source import AssessmentSource
+from mlflow.entities.evaluation import Evaluation
+from mlflow.entities.evaluation_tag import EvaluationTag
+from mlflow.evaluation.utils import evaluations_to_dataframes
+
+
+def test_evaluations_to_dataframes_basic():
+    # Setup an evaluation with minimal data
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the evaluations DataFrame
+    assert len(evaluations_df) == 1
+    assert evaluations_df["evaluation_id"].iloc[0] == "eval1"
+    assert evaluations_df["run_id"].iloc[0] == "run1"
+    assert evaluations_df["inputs_id"].iloc[0] == "inputs1"
+    assert evaluations_df["inputs"].iloc[0] == {"feature1": 1.0, "feature2": 2.0}
+
+    # Check that the other DataFrames are empty
+    assert metrics_df.empty
+    assert assessments_df.empty
+    assert tags_df.empty
+
+
+def test_evaluations_to_dataframes_full_data():
+    # Setup an evaluation with full data
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment = Assessment(
+        evaluation_id="eval1",
+        name="accuracy",
+        source=source,
+        timestamp=123456789,
+        numeric_value=0.95,
+        rationale="Good performance",
+    )
+    metric = Metric(key="metric1", value=0.9, timestamp=1234567890, step=0)
+    tag = EvaluationTag(key="tag1", value="value1")
+
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        outputs={"output1": 0.5},
+        request_id="request1",
+        targets={"target1": 0.6},
+        error_code="E001",
+        error_message="An error occurred",
+        assessments=[assessment],
+        metrics=[metric],
+        tags=[tag],
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the evaluations DataFrame
+    assert len(evaluations_df) == 1
+    assert evaluations_df["evaluation_id"].iloc[0] == "eval1"
+    assert evaluations_df["run_id"].iloc[0] == "run1"
+    assert evaluations_df["inputs_id"].iloc[0] == "inputs1"
+    assert evaluations_df["inputs"].iloc[0] == {"feature1": 1.0, "feature2": 2.0}
+    assert evaluations_df["outputs"].iloc[0] == {"output1": 0.5}
+    assert evaluations_df["request_id"].iloc[0] == "request1"
+    assert evaluations_df["targets"].iloc[0] == {"target1": 0.6}
+    assert evaluations_df["error_code"].iloc[0] == "E001"
+    assert evaluations_df["error_message"].iloc[0] == "An error occurred"
+
+    # Check the metrics DataFrame
+    assert len(metrics_df) == 1
+    assert metrics_df["evaluation_id"].iloc[0] == "eval1"
+    assert metrics_df["key"].iloc[0] == "metric1"
+    assert metrics_df["value"].iloc[0] == 0.9
+    assert metrics_df["timestamp"].iloc[0] == 1234567890
+
+    # Check the assessments DataFrame
+    assert len(assessments_df) == 1
+    assert assessments_df["evaluation_id"].iloc[0] == "eval1"
+    assert assessments_df["name"].iloc[0] == "accuracy"
+    assert assessments_df["source"].iloc[0] == source.to_dictionary()
+    assert assessments_df["boolean_value"].iloc[0] is None
+    assert assessments_df["numeric_value"].iloc[0] == 0.95
+    assert assessments_df["string_value"].iloc[0] is None
+    assert assessments_df["rationale"].iloc[0] == "Good performance"
+    assert assessments_df["error_code"].iloc[0] is None
+    assert assessments_df["error_message"].iloc[0] is None
+
+    # Check the tags DataFrame
+    assert len(tags_df) == 1
+    assert tags_df["evaluation_id"].iloc[0] == "eval1"
+    assert tags_df["key"].iloc[0] == "tag1"
+    assert tags_df["value"].iloc[0] == "value1"
+
+
+def test_evaluations_to_dataframes_empty():
+    # Empty evaluations list
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([])
+
+    # Verify that the DataFrames are empty
+    assert evaluations_df.empty
+    assert metrics_df.empty
+    assert assessments_df.empty
+    assert tags_df.empty
+
+    # Verify the column names of the empty DataFrames
+    expected_evaluation_columns = [
+        "evaluation_id",
+        "run_id",
+        "inputs_id",
+        "inputs",
+        "outputs",
+        "request_id",
+        "targets",
+        "error_code",
+        "error_message",
+    ]
+    expected_metrics_columns = ["evaluation_id", "key", "value", "timestamp"]
+    expected_assessments_columns = [
+        "evaluation_id",
+        "name",
+        "source",
+        "timestamp",
+        "boolean_value",
+        "numeric_value",
+        "string_value",
+        "rationale",
+        "metadata",
+        "error_code",
+        "error_message",
+    ]
+    expected_tags_columns = ["evaluation_id", "key", "value"]
+
+    assert list(evaluations_df.columns) == expected_evaluation_columns
+    assert list(metrics_df.columns) == expected_metrics_columns
+    assert list(assessments_df.columns) == expected_assessments_columns
+    assert list(tags_df.columns) == expected_tags_columns
+
+
+def test_evaluations_to_dataframes_basic():
+    # Setup an evaluation with minimal data
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the evaluations DataFrame
+    assert len(evaluations_df) == 1
+    assert evaluations_df["evaluation_id"].iloc[0] == "eval1"
+    assert evaluations_df["run_id"].iloc[0] == "run1"
+    assert evaluations_df["inputs_id"].iloc[0] == "inputs1"
+    assert evaluations_df["inputs"].iloc[0] == {"feature1": 1.0, "feature2": 2.0}
+
+    # Check that the othe
+
+
+def test_evaluations_to_dataframes_different_assessments():
+    # Different types of assessments in evaluations
+    source = AssessmentSource(source_type="HUMAN", source_id="user_1")
+    assessment_1 = Assessment(
+        evaluation_id="eval1",
+        name="accuracy",
+        source=source,
+        timestamp=123456789,
+        numeric_value=0.95,
+        rationale="Good performance",
+    )
+    assessment_2 = Assessment(
+        evaluation_id="eval1",
+        name="precision",
+        source=source,
+        timestamp=123456789,
+        numeric_value=0.85,
+        rationale="Reasonable performance",
+    )
+
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        assessments=[assessment_1, assessment_2],
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the assessments DataFrame
+    assert len(assessments_df) == 2
+    assert assessments_df["evaluation_id"].iloc[0] == "eval1"
+    assert assessments_df["name"].iloc[0] == "accuracy"
+    assert assessments_df["numeric_value"].iloc[0] == 0.95
+
+    assert assessments_df["evaluation_id"].iloc[1] == "eval1"
+    assert assessments_df["name"].iloc[1] == "precision"
+    assert assessments_df["numeric_value"].iloc[1] == 0.85
+
+
+def test_evaluations_to_dataframes_different_metrics():
+    # Different types of metrics in evaluations
+    metric_1 = Metric(key="metric1", value=0.9, timestamp=1234567890, step=0)
+    metric_2 = Metric(key="metric2", value=0.8, timestamp=1234567891, step=0)
+
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        metrics=[metric_1, metric_2],
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the metrics DataFrame
+    assert len(metrics_df) == 2
+    assert metrics_df["evaluation_id"].iloc[0] == "eval1"
+    assert metrics_df["key"].iloc[0] == "metric1"
+    assert metrics_df["value"].iloc[0] == 0.9
+    assert metrics_df["timestamp"].iloc[0] == 1234567890
+
+    assert metrics_df["evaluation_id"].iloc[1] == "eval1"
+    assert metrics_df["key"].iloc[1] == "metric2"
+    assert metrics_df["value"].iloc[1] == 0.8
+    assert metrics_df["timestamp"].iloc[1] == 1234567891
+
+
+def test_evaluations_to_dataframes_different_tags():
+    # Different tags in evaluations
+    tag1 = EvaluationTag(key="tag1", value="value1")
+    tag2 = EvaluationTag(key="tag2", value="value2")
+
+    evaluation = Evaluation(
+        evaluation_id="eval1",
+        run_id="run1",
+        inputs_id="inputs1",
+        inputs={"feature1": 1.0, "feature2": 2.0},
+        tags=[tag1, tag2],
+    )
+
+    evaluations_df, metrics_df, assessments_df, tags_df = evaluations_to_dataframes([evaluation])
+
+    # Check the tags DataFrame
+    assert len(tags_df) == 2
+    assert tags_df["evaluation_id"].iloc[0] == "eval1"
+    assert tags_df["key"].iloc[0] == "tag1"
+    assert tags_df["value"].iloc[0] == "value1"
+
+    assert tags_df["evaluation_id"].iloc[1] == "eval1"
+    assert tags_df["key"].iloc[1] == "tag2"
+    assert tags_df["value"].iloc[1] == "value2"
diff --git a/tests/evaluate/logging/utils.py b/tests/evaluate/logging/utils.py
new file mode 100644
index 000000000..2974b2850
--- /dev/null
+++ b/tests/evaluate/logging/utils.py
@@ -0,0 +1,228 @@
+from typing import List
+
+import pandas as pd
+
+from mlflow.entities.evaluation import Evaluation as EvaluationEntity
+from mlflow.evaluation.utils import (
+    _get_assessments_dataframe_schema,
+    _get_evaluations_dataframe_schema,
+    _get_metrics_dataframe_schema,
+    _get_tags_dataframe_schema,
+)
+from mlflow.exceptions import MlflowException
+from mlflow.protos.databricks_pb2 import INTERNAL_ERROR, RESOURCE_DOES_NOT_EXIST
+from mlflow.tracking.client import MlflowClient
+
+
+def get_evaluation(*, run_id: str, evaluation_id: str) -> EvaluationEntity:
+    """
+    Retrieves an Evaluation object from an MLflow Run.
+
+    Args:
+        run_id (str): ID of the MLflow Run containing the evaluation.
+        evaluation_id (str): The ID of the evaluation.
+
+    Returns:
+        Evaluation: The Evaluation object.
+    """
+    client = MlflowClient()
+    if not _contains_evaluation_artifacts(client=client, run_id=run_id):
+        raise MlflowException(
+            "The specified run does not contain any evaluations. "
+            "Please log evaluations to the run before retrieving them.",
+            error_code=RESOURCE_DOES_NOT_EXIST,
+        )
+
+    evaluations_file = client.download_artifacts(run_id=run_id, path="_evaluations.json")
+    evaluations_df = _read_evaluations_dataframe(evaluations_file)
+
+    assessments_file = client.download_artifacts(run_id=run_id, path="_assessments.json")
+    assessments_df = _read_assessments_dataframe(assessments_file)
+
+    metrics_file = client.download_artifacts(run_id=run_id, path="_metrics.json")
+    metrics_df = _read_metrics_dataframe(metrics_file)
+
+    tags_file = client.download_artifacts(run_id=run_id, path="_tags.json")
+    tags_df = _read_tags_dataframe(tags_file)
+
+    return _get_evaluation_from_dataframes(
+        run_id=run_id,
+        evaluation_id=evaluation_id,
+        evaluations_df=evaluations_df,
+        metrics_df=metrics_df,
+        assessments_df=assessments_df,
+        tags_df=tags_df,
+    )
+
+
+def _contains_evaluation_artifacts(*, client: MlflowClient, run_id: str) -> bool:
+    return {"_evaluations.json", "_metrics.json", "_assessments.json", "_tags.json"}.issubset(
+        {file.path for file in client.list_artifacts(run_id)}
+    )
+
+
+def _read_evaluations_dataframe(path: str) -> pd.DataFrame:
+    """
+    Reads an evaluations DataFrame from a file.
+
+    Args:
+        path (str): Path to the file.
+
+    Returns:
+        pd.DataFrame: The evaluations DataFrame.
+    """
+    schema = _get_evaluations_dataframe_schema()
+    return pd.read_json(path, orient="split", dtype=schema, convert_dates=False).replace(
+        pd.NA, None
+    )
+
+
+def _read_assessments_dataframe(path: str) -> pd.DataFrame:
+    """
+    Reads an assessments DataFrame from a file.
+
+    Args:
+        path (str): Path to the file.
+
+    Returns:
+        pd.DataFrame: The assessments DataFrame.
+    """
+    schema = _get_assessments_dataframe_schema()
+    return pd.read_json(path, orient="split", dtype=schema, convert_dates=False).replace(
+        pd.NA, None
+    )
+
+
+def _read_metrics_dataframe(path: str) -> pd.DataFrame:
+    """
+    Reads a metrics DataFrame from a file.
+
+    Args:
+        path (str): Path to the file.
+
+    Returns:
+        pd.DataFrame: The metrics DataFrame.
+    """
+    schema = _get_metrics_dataframe_schema()
+    return pd.read_json(path, orient="split", dtype=schema, convert_dates=False).replace(
+        pd.NA, None
+    )
+
+
+def _read_tags_dataframe(path: str) -> pd.DataFrame:
+    """
+    Reads a tags DataFrame from a file.
+
+    Args:
+        path (str): Path to the file.
+
+    Returns:
+        pd.DataFrame: The tags DataFrame.
+    """
+    schema = _get_tags_dataframe_schema()
+    return pd.read_json(path, orient="split", dtype=schema, convert_dates=False).replace(
+        pd.NA, None
+    )
+
+
+def _get_evaluation_from_dataframes(
+    *,
+    run_id: str,
+    evaluation_id: str,
+    evaluations_df: pd.DataFrame,
+    metrics_df: pd.DataFrame,
+    assessments_df: pd.DataFrame,
+    tags_df: pd.DataFrame,
+) -> EvaluationEntity:
+    """
+    Parses an Evaluation object with the specified evaluation ID from the specified DataFrames.
+    """
+    evaluation_row = evaluations_df[evaluations_df["evaluation_id"] == evaluation_id]
+    if evaluation_row.empty:
+        raise MlflowException(
+            f"The specified evaluation ID '{evaluation_id}' does not exist in the run '{run_id}'.",
+            error_code=RESOURCE_DOES_NOT_EXIST,
+        )
+
+    evaluations: List[EvaluationEntity] = _dataframes_to_evaluations(
+        evaluations_df=evaluation_row,
+        metrics_df=metrics_df,
+        assessments_df=assessments_df,
+        tags_df=tags_df,
+    )
+    if len(evaluations) != 1:
+        raise MlflowException(
+            f"Expected to find a single evaluation with ID '{evaluation_id}', but found "
+            f"{len(evaluations)} evaluations.",
+            error_code=INTERNAL_ERROR,
+        )
+
+    return evaluations[0]
+
+
+def _dataframes_to_evaluations(
+    evaluations_df: pd.DataFrame,
+    metrics_df: pd.DataFrame,
+    assessments_df: pd.DataFrame,
+    tags_df: pd.DataFrame,
+) -> List[EvaluationEntity]:
+    """
+    Converts four separate DataFrames (main evaluation data, metrics, assessments, and tags) back
+    into a list of Evaluation entities.
+
+    Args:
+        evaluations_df (pd.DataFrame): DataFrame with the main evaluation data
+            (excluding assessment and metrics).
+        metrics_df (pd.DataFrame): DataFrame with metrics.
+        assessments_df (pd.DataFrame): DataFrame with assessments.
+        tags_df (pd.DataFrame): DataFrame with tags.
+
+    Returns:
+        List[EvaluationEntity]: A list of Evaluation entities created from the DataFrames.
+    """
+    # Group metrics and assessment by evaluation_id
+    metrics_by_eval = _group_dataframe_by_evaluation_id(metrics_df)
+    assessments_by_eval = _group_dataframe_by_evaluation_id(assessments_df)
+    tags_by_eval = _group_dataframe_by_evaluation_id(tags_df)
+
+    # Convert main DataFrame to list of dictionaries and create Evaluation objects
+    evaluations = []
+    for eval_dict in evaluations_df.to_dict(orient="records"):
+        evaluation_id = eval_dict["evaluation_id"]
+        eval_dict["metrics"] = [
+            {
+                "key": metric["key"],
+                "value": metric["value"],
+                "timestamp": metric["timestamp"],
+                # Evaluation metrics don't have steps, but we're reusing the MLflow Metric
+                # class to represent Evaluation metrics as entities in Python for now. Accordingly,
+                # we set the step to 0 in order to parse the evaluation metric as an MLflow Metric
+                # Python entity
+                "step": 0,
+                # Also discard the evaluation_id field from the evaluation metric, since this
+                # field is not part of the MLflow Metric Python entity
+            }
+            for metric in metrics_by_eval.get(evaluation_id, [])
+        ]
+        eval_dict["assessments"] = assessments_by_eval.get(evaluation_id, [])
+        eval_dict["tags"] = tags_by_eval.get(evaluation_id, [])
+        evaluations.append(EvaluationEntity.from_dictionary(eval_dict))
+
+    return evaluations
+
+
+def _group_dataframe_by_evaluation_id(df: pd.DataFrame):
+    """
+    Groups evaluation dataframe rows by 'evaluation_id'.
+
+    Args:
+        df (pd.DataFrame): DataFrame to group.
+
+    Returns:
+        Dict[str, List]: A dictionary with 'evaluation_id' as keys and lists of entity
+            dictionaries as values.
+    """
+    grouped = df.groupby("evaluation_id", group_keys=False).apply(
+        lambda x: x.to_dict(orient="records")
+    )
+    return grouped.to_dict()