src/documentation/how_tos/how_to_human_evaluation_via_argilla.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import argilla\n",
    "from dotenv import load_dotenv\n",
    "from example_data import StoryTaskInput, StoryTaskOutput, argilla_example_data\n",
    "from pydantic import BaseModel\n",
    "\n",
    "from intelligence_layer.connectors import (\n",
    "    ArgillaEvaluation,\n",
    "    ArgillaWrapperClient,\n",
    "    RecordData,\n",
    ")\n",
    "from intelligence_layer.evaluation import (\n",
    "    ArgillaEvaluationLogic,\n",
    "    ArgillaEvaluator,\n",
    "    AsyncInMemoryEvaluationRepository,\n",
    "    Example,\n",
    "    RecordDataSequence,\n",
    "    SuccessfulExampleOutput,\n",
    ")\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How to evaluate with human evaluation via Argilla\n",
    "1. Initialize an Argilla client with the correct settings for your setup\n",
    "   - By default, the url and api key are read from the environment variables `ARGILLA_API_URL` and `ARGILLA_API_KEY`\n",
    "2. Choose an Argilla workspace and get its ID\n",
    "3. Create an `AsyncEvaluationRepository`\n",
    "4. Define new output type for the evaluation\n",
    "5. Implement an `ArgillaEvaluationLogic`\n",
    "   1. Create questions and fields to structure the data that will be displayed in Argilla\n",
    "   2. Implement `to_record` to convert the task input into an Argilla record\n",
    "   3. Implement `from_record` to convert the record back to an evaluation result\n",
    "6. Submit tasks to the Argilla instance by running the `ArgillaEvaluator`\n",
    "7. **Use the Argilla web platform to evaluate** \n",
    "8. Collect all labelled evaluations from Argilla\n",
    "   - Make sure to save the `EvaluationOverview.id`, as it is needed to retrieve the results later"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 0 - Create a dataset and run a task on it\n",
    "\n",
    "\n",
    "my_example_data = argilla_example_data()\n",
    "\n",
    "# Step 1\n",
    "\n",
    "client = ArgillaWrapperClient(\n",
    "    # api_url=\"your url here\",     # not necessary if ARGILLA_API_URL is set in environment\n",
    "    # api_key=\"your api key here\", # not necessary if ARGILLA_API_KEY is set in environment\n",
    ")\n",
    "\n",
    "\n",
    "# Step 2\n",
    "workspace_id = client.ensure_workspace_exists(\"my-workspace-name\")\n",
    "\n",
    "# Step 3\n",
    "evaluation_repository = (\n",
    "    AsyncInMemoryEvaluationRepository()\n",
    ")  # Use FileEvaluationRepository for persistent results\n",
    "\n",
    "\n",
    "# Step 4\n",
    "class FunnyOutputRating(BaseModel):\n",
    "    rating: int\n",
    "\n",
    "\n",
    "# Step 5\n",
    "\n",
    "\n",
    "class CustomArgillaEvaluationLogic(\n",
    "    ArgillaEvaluationLogic[\n",
    "        StoryTaskInput, StoryTaskOutput, None, FunnyOutputRating\n",
    "    ]  # No expected output, therefore \"None\"\n",
    "):\n",
    "    # Step 5.1\n",
    "    def __init__(self):\n",
    "        super().__init__(\n",
    "            questions=[\n",
    "                argilla.RatingQuestion(\n",
    "                    name=\"rating\",\n",
    "                    title=\"Funniness\",\n",
    "                    description=\"How funny do you think is the joke? Rate it from 1-5.\",\n",
    "                    values=list(range(1, 6)),\n",
    "                ),\n",
    "                argilla.LabelQuestion(\n",
    "                    name=\"on-topic\",\n",
    "                    title=\"Is the question on-topic?\",\n",
    "                    description=\"<<description on what on topic means>>\",\n",
    "                    labels={\"YES\": \"displayed-yes\", \"NO\": \"displayed-no\"},\n",
    "                ),\n",
    "            ],\n",
    "            fields={\n",
    "                \"input\": argilla.TextField(name=\"input\", title=\"Topic\"),\n",
    "                \"output\": argilla.TextField(name=\"output\", title=\"Joke\"),\n",
    "            },\n",
    "        )\n",
    "\n",
    "    # Step 5.2\n",
    "    def to_record(\n",
    "        self,\n",
    "        example: Example[StoryTaskInput, None],\n",
    "        *output: SuccessfulExampleOutput[StoryTaskOutput],\n",
    "    ) -> RecordDataSequence:\n",
    "        return RecordDataSequence(\n",
    "            records=[\n",
    "                RecordData(\n",
    "                    content={\n",
    "                        # labels as defined in Field.name\n",
    "                        self.fields[\"input\"].name: example.input.topic,\n",
    "                        self.fields[\"output\"].name: run_output.output.story,\n",
    "                    },\n",
    "                    example_id=example.id,\n",
    "                )\n",
    "                for run_output in output\n",
    "            ]\n",
    "        )\n",
    "\n",
    "    # Step 5.3\n",
    "    def from_record(self, argilla_evaluation: ArgillaEvaluation) -> FunnyOutputRating:\n",
    "        print(argilla_evaluation)\n",
    "        rating = (\n",
    "            argilla_evaluation.responses[\"rating\"]\n",
    "            if argilla_evaluation.responses[\"on-topic\"] == \"YES\"\n",
    "            else 0\n",
    "        )\n",
    "        return FunnyOutputRating(rating=rating)\n",
    "\n",
    "\n",
    "evaluation_logic = CustomArgillaEvaluationLogic()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# we skip this as we do not have a dataset or run in this example\n",
    "\n",
    "# Step 6\n",
    "\n",
    "evaluator = ArgillaEvaluator(\n",
    "    my_example_data.dataset_repository,\n",
    "    my_example_data.run_repository,\n",
    "    evaluation_repository,\n",
    "    description=\"My evaluation description\",\n",
    "    evaluation_logic=evaluation_logic,\n",
    "    argilla_client=client,\n",
    "    workspace_id=workspace_id,\n",
    ")\n",
    "partial_evaluation_overview = evaluator.submit(*my_example_data.run_ids)\n",
    "print(\"ID to retrieve results later: \", partial_evaluation_overview.id)\n",
    "\n",
    "# Step 7\n",
    "\n",
    "####################################\n",
    "# Evaluate via the Argilla UI here #\n",
    "####################################\n",
    "\n",
    "# Step 8\n",
    "evaluation_overview = evaluator.retrieve(partial_evaluation_overview.id)\n",
    "print(\"ID to retrieve the evaluation later: \", evaluation_overview.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# cleanup\n",
    "client._delete_dataset(partial_evaluation_overview.id)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "intelligence-layer-d3iSWYpm-py3.10",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}