src/documentation/quickstart_task.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from collections.abc import Iterable\n",
    "from statistics import mean\n",
    "\n",
    "from aleph_alpha_client import Prompt\n",
    "from dotenv import load_dotenv\n",
    "from pydantic import BaseModel\n",
    "\n",
    "from intelligence_layer.core import (\n",
    "    AlephAlphaModel,\n",
    "    CompleteInput,\n",
    "    InMemoryTracer,\n",
    "    NoOpTracer,\n",
    "    Task,\n",
    "    TaskSpan,\n",
    ")\n",
    "from intelligence_layer.evaluation import (\n",
    "    AggregationLogic,\n",
    "    Aggregator,\n",
    "    Evaluator,\n",
    "    Example,\n",
    "    InMemoryAggregationRepository,\n",
    "    InMemoryDatasetRepository,\n",
    "    InMemoryEvaluationRepository,\n",
    "    InMemoryRunRepository,\n",
    "    Runner,\n",
    "    SingleOutputEvaluationLogic,\n",
    ")\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setting up your own custom task\n",
    "\n",
    "If the available task methodologies are not suitable for your use case, this guide explains how to set up your own task from scratch.\n",
    "Using the task interface gives us the added benefit of getting built-in input and output logging and gives us the possibility of using the evaluation framework.\n",
    "\n",
    "For the purpose of this tutorial, we will set up a simple keyword extraction task.\n",
    "To do so, we will leverage `luminous-base` and a few-shot prompt to generate matching keywords for variable input texts.\n",
    "Next, we will build an evaluator to check how well our extractor performs.\n",
    "\n",
    "## Initial task setup\n",
    "\n",
    "Let's start with the interface of any generic task. The full `Task` interface can be found here: [../intelligence_layer/task.py](../intelligence_layer/task.py).\n",
    "However, to initially set up a `Task`, there are only a few parts relevant to us. For now, we shall only care about the following part of the interface:\n",
    "\n",
    "```python\n",
    "Input = TypeVar(\"Input\", bound=PydanticSerializable)\n",
    "Output = TypeVar(\"Output\", bound=PydanticSerializable)\n",
    "\n",
    "class Task(ABC, Generic[Input, Output]):\n",
    "    @abstractmethod\n",
    "    def do_run(self, input: Input, task_span: TaskSpan) -> Output:\n",
    "        \"\"\"Executes the process for this use-case.\"\"\"\n",
    "        ...\n",
    "```\n",
    "\n",
    "For every task, we have to define an `Input`, an `Output` and how we would like to run it. Since these can vary so much, we make no assumptions about a `Task`'s implementation. \n",
    "We only require both input and output to be `PydanticSerializable`. The best way to guarantee this is to make them pydantic `BaseModel`s. For our keyword extraction task, we will define `Input` and `Output` as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class KeywordExtractionInput(BaseModel):\n",
    "    \"\"\"This is the text we will extract keywords from.\"\"\"\n",
    "\n",
    "    text: str\n",
    "\n",
    "\n",
    "class KeywordExtractionOutput(BaseModel):\n",
    "    \"\"\"The matching set of keywords we aim to extract.\"\"\"\n",
    "\n",
    "    keywords: frozenset[str]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have our input and output defined, we  will implement the actual task.\n",
    "\n",
    "The steps that the task consists of are:\n",
    "- Create a `Prompt` using the input text.\n",
    "- Have `luminous-base` complete the prompt.\n",
    "- Extract keywords from said completion.\n",
    "\n",
    "When a task is executed, we offer the possibility to log all intermediate steps and outputs.\n",
    "This is crucial because large language models are inherently probabilistic.\n",
    "Therefore, we might get unexpected answers.\n",
    "This logging allows us to check the results afterwards and find out what went wrong.\n",
    "\n",
    "For this, we shall inject an `InMemoryTracer` into the task. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class KeywordExtractionTask(Task[KeywordExtractionInput, KeywordExtractionOutput]):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"Identify matching keywords for each text.\n",
    "###\n",
    "Text: The \"Whiskey War\" is an ongoing conflict between Denmark and Canada over ownership of Hans Island. The dispute began in 1973, when Denmark and Canada reached an agreement on Greenland's borders. However, no settlement regarding Hans Island could be reached by the time the treaty was signed. Since then both countries have used peaceful means - such as planting their national flag or burying liquor - to draw attention to the disagreement.\n",
    "Keywords: Conflict, Whiskey War, Denmark, Canada, Treaty, Flag, Liquor\n",
    "###\n",
    "Text: I really like pizza and sushi.\n",
    "Keywords: Pizza, Sushi\n",
    "###\n",
    "Text: NASA launched the Discovery program to explore the solar system. It comprises a series of expeditions that have continued from the program's launch in the 1990s to the present day. In the course of the 16 expeditions launched so far, the Moon, Mars, Mercury and Venus, among others, have been explored. Unlike other space programs, the Discovery program places particular emphasis on cost efficiency, true to the motto: \"faster, better, cheaper\".\n",
    "Keywords: Space program, NASA, Expedition, Cost efficiency, Moon, Mars, Mercury, Venus\n",
    "###\n",
    "Text: {text}\n",
    "Keywords:\"\"\"\n",
    "    MODEL: str = \"luminous-base\"\n",
    "\n",
    "    def __init__(self, model: AlephAlphaModel | None = None) -> None:\n",
    "        super().__init__()\n",
    "        self._model = model if model else AlephAlphaModel(name=\"luminous-base\")\n",
    "\n",
    "    def _create_complete_input(self, text: str) -> Prompt:\n",
    "        prompt = Prompt.from_text(self.PROMPT_TEMPLATE.format(text=text))\n",
    "        # 'stop_sequences' is a list of strings which will stop generation if they're generated.\n",
    "        model_input = CompleteInput(\n",
    "            prompt=prompt,\n",
    "            stop_sequences=[\"\\n\", \"###\"],\n",
    "            frequency_penalty=0.25,\n",
    "            model=self._model.name,\n",
    "        )\n",
    "        return model_input\n",
    "\n",
    "    def do_run(\n",
    "        self, input: KeywordExtractionInput, task_span: TaskSpan\n",
    "    ) -> KeywordExtractionOutput:\n",
    "        completion_input = self._create_complete_input(input.text)\n",
    "        completion = self._model.complete(completion_input, task_span)\n",
    "        return KeywordExtractionOutput(\n",
    "            keywords=set(\n",
    "                k.strip().lower() for k in completion.completion.split(\",\") if k.strip()\n",
    "            )\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we can run this `KeywordExtractionTask` like so:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task = KeywordExtractionTask()\n",
    "text = \"Computer vision describes the processing of an image by a machine using external devices (e.g., a scanner) into a digital description of that image for further processing. An example of this is optical character recognition (OCR), the recognition and processing of images containing text. Further processing and final classification of the image is often done using artificial intelligence methods. The goal of this field is to enable computers to process visual tasks that were previously reserved for humans.\"\n",
    "\n",
    "tracer = InMemoryTracer()\n",
    "output = task.run(KeywordExtractionInput(text=text), tracer)\n",
    "\n",
    "print(output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looks great!\n",
    "\n",
    "## Evaluation\n",
    "\n",
    "Now that our task is set up, we can start evaluating its performance.\n",
    "\n",
    "For this, we will have to set up an evaluator. The evaluator requires an `EvaluationLogic` and an `AggregationLogic` object.  \n",
    "The logic objects are responsible for how single examples are evaluated and how a list of examples are aggregated. \n",
    "How these single examples are put together is the job of the `Evaluator`. This typically does not need to be changed and can just be used.\n",
    "\n",
    "```python\n",
    "class EvaluationLogic(ABC, Generic[Input, Output, ExpectedOutput, Evaluation]):\n",
    "    @abstractmethod\n",
    "    def do_evaluate(\n",
    "        self,\n",
    "        example: Example[Input, ExpectedOutput],\n",
    "        *output: SuccessfulExampleOutput[Output],\n",
    "    ) -> Evaluation:\n",
    "        ...\n",
    "\n",
    "class AggregationLogic(ABC, Generic[Evaluation, AggregatedEvaluation]):\n",
    "    @abstractmethod\n",
    "    def aggregate(self, evaluations: Iterable[Evaluation]) -> AggregatedEvaluation:\n",
    "        ...\n",
    "```\n",
    "\n",
    "Notice that, just like our `Task`, the `EvaluationLogic` takes an `Input`. This input is the same as our task input.\n",
    "However, we don't just want to run a task; we also want to evaluate the result. \n",
    "Therefore, our evaluation logic also depends on some `ExpectedOutput`, as well as `Evaluation`.\n",
    "We will come back to the `AggregatedEvaluation` of the `AggregationLogic` at a later stage.\n",
    "\n",
    "Let's build an evaluation that can check the performance of our keyword extraction methodology. For this, we need four things:\n",
    "- An implementation of the task to be run (we suggest supplying this in the `Evaluator`'s `__init__`)\n",
    "- An interface for our `ExpectedOutput`\n",
    "- Some `Evaluation`, i.e., the output of the `do_evaluate` method\n",
    "- An implementation of the `do_evaluate` function in form of an `EvaluationLogic`.\n",
    "\n",
    "In our case, we will measure the performance of our keyword extraction by calculating the proportion of correctly generated keywords compared to all expected keywords. \n",
    "This is also known as the \"true positive rate\". \n",
    "To calculate this, our evaluate function will need a set of the expected keywords.\n",
    "Also, we will add the missing keywords and keywords that are generated that we don't expect. \n",
    "This way, we can see how our task performs for a specific example, and we can check for unexpected results.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class KeywordExtractionExpectedOutput(BaseModel):\n",
    "    \"\"\"This is the expected output for an example run. This is used to compare the output of the task with.\n",
    "\n",
    "    We will be evaluating our keyword extraction based on the expected keywords.\n",
    "    \"\"\"\n",
    "\n",
    "    keywords: frozenset[str]\n",
    "\n",
    "\n",
    "class KeywordExtractionEvaluation(BaseModel):\n",
    "    \"\"\"This is the interface for the metrics that are generated for each evaluation case.\"\"\"\n",
    "\n",
    "    true_positive_rate: float\n",
    "    true_positives: frozenset[str]\n",
    "    false_positives: frozenset[str]\n",
    "    false_negatives: frozenset[str]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Accordingly, our evaluate function will take a `KeywordExtractionInput`, and run the task with this.\n",
    "Next, we shall compare the generated output with the `KeywordExtractionExpectedOutput` to create the `KeywordExtractionEvaluation`.\n",
    "\n",
    "```python\n",
    "def do_evaluate(\n",
    "    self,\n",
    "    input: KeywordExtractionInput,\n",
    "    output: KeywordExtractionOutput,\n",
    "    expected_output: KeywordExtractionExpectedOutput,\n",
    ") -> KeywordExtractionEvaluation:\n",
    "    true_positives = output.keywords & expected_output.keywords\n",
    "    false_positives = output.keywords - expected_output.keywords\n",
    "    false_negatives = expected_output.keywords - output.keywords\n",
    "    return KeywordExtractionEvaluation(\n",
    "        true_positive_rate=len(true_positives) / len(expected_output.keywords),\n",
    "        true_positives=true_positives,\n",
    "        false_positives=false_positives,\n",
    "        false_negatives=false_negatives,\n",
    "    )\n",
    "```\n",
    "\n",
    "However, to quantitatively evaluate the performance of a task, we will need to run many different examples and calculate the metrics for each. \n",
    "To do this, we can use the `eval_and_aggregate_runs` function provided by the `Evaluator` base class. This takes a dataset, runs all the examples, and aggregates the metrics generated from the evaluation.\n",
    "\n",
    "To set this up, we will first need to create an interface for the `AggregatedEvaluation` and implement the `aggregate` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"This is the interface for the aggregated metrics that are generated from running a number of examples\"\"\"\n",
    "\n",
    "\n",
    "class KeywordExtractionAggregatedEvaluation(BaseModel):\n",
    "    average_true_positive_rate: float"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have all parts in place, let's run our task which will produce the results for evaluation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_repository = InMemoryDatasetRepository()\n",
    "run_repository = InMemoryRunRepository()\n",
    "\n",
    "runner = Runner(task, dataset_repository, run_repository, \"keyword-extraction\")\n",
    "model_input = KeywordExtractionInput(text=\"This is a text about dolphins and sharks.\")\n",
    "expected_output = KeywordExtractionExpectedOutput(keywords=[\"dolphins\", \"sharks\"])\n",
    "\n",
    "single_example_dataset = dataset_repository.create_dataset(\n",
    "    examples=[Example(input=model_input, expected_output=expected_output)],\n",
    "    dataset_name=\"quickstart-task-single-example-dataset\",\n",
    ").id\n",
    "\n",
    "run_overview = runner.run_dataset(single_example_dataset, NoOpTracer())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, let's build an evaluator.\n",
    "For this, we need to implement a method doing the actual evaluation in a `EvaluationLogic` class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class KeywordExtractionEvaluationLogic(\n",
    "    SingleOutputEvaluationLogic[\n",
    "        KeywordExtractionInput,\n",
    "        KeywordExtractionOutput,\n",
    "        KeywordExtractionExpectedOutput,\n",
    "        KeywordExtractionEvaluation,\n",
    "    ]\n",
    "):\n",
    "    def do_evaluate_single_output(\n",
    "        self,\n",
    "        example: Example[KeywordExtractionInput, KeywordExtractionOutput],\n",
    "        output: KeywordExtractionExpectedOutput,\n",
    "    ) -> KeywordExtractionEvaluation:\n",
    "        true_positives = output.keywords & example.expected_output.keywords\n",
    "        false_positives = output.keywords - example.expected_output.keywords\n",
    "        false_negatives = example.expected_output.keywords - output.keywords\n",
    "        return KeywordExtractionEvaluation(\n",
    "            true_positive_rate=len(true_positives) / len(output.keywords),\n",
    "            true_positives=true_positives,\n",
    "            false_positives=false_positives,\n",
    "            false_negatives=false_negatives,\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And now, we can create an evaluator and run it on our data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluation_repository = InMemoryEvaluationRepository()\n",
    "evaluation_logic = KeywordExtractionEvaluationLogic()\n",
    "evaluator = Evaluator(\n",
    "    dataset_repository,\n",
    "    run_repository,\n",
    "    evaluation_repository,\n",
    "    \"keyword-extraction\",\n",
    "    evaluation_logic,\n",
    ")\n",
    "\n",
    "evaluation_overview = evaluator.evaluate_runs(run_overview.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To aggregate the evaluation results, we have to implement a method doing this in an `AggregationLogic` class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class KeywordExtractionAggregationLogic(\n",
    "    AggregationLogic[\n",
    "        KeywordExtractionEvaluation,\n",
    "        KeywordExtractionAggregatedEvaluation,\n",
    "    ]\n",
    "):\n",
    "    def aggregate(\n",
    "        self, evaluations: Iterable[KeywordExtractionEvaluation]\n",
    "    ) -> KeywordExtractionAggregatedEvaluation:\n",
    "        evaluation_list = list(evaluations)\n",
    "        true_positive_rate = (\n",
    "            mean(evaluation.true_positive_rate for evaluation in evaluation_list)\n",
    "            if evaluation_list\n",
    "            else 0\n",
    "        )\n",
    "        return KeywordExtractionAggregatedEvaluation(\n",
    "            average_true_positive_rate=true_positive_rate\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now create an aggregator and generate evaluation statistics from the previously generated evaluation results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "aggregation_repository = InMemoryAggregationRepository()\n",
    "aggregation_logic = KeywordExtractionAggregationLogic()\n",
    "aggregator = Aggregator(\n",
    "    evaluation_repository,\n",
    "    aggregation_repository,\n",
    "    \"keyword-extraction\",\n",
    "    aggregation_logic,\n",
    ")\n",
    "\n",
    "aggregation_overview = aggregator.aggregate_evaluation(evaluation_overview.id)\n",
    "\n",
    "print(\"Statistics: \", aggregation_overview.statistics)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have implemented all required methods, let's run a dataset with some more examples."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_id = dataset_repository.create_dataset(\n",
    "    examples=[\n",
    "        Example(input=model_input, expected_output=expected_output),\n",
    "        Example(\n",
    "            input=KeywordExtractionInput(\n",
    "                text=\"Clinical psychology is an integration of human science, behavioral science, theory, and clinical knowledge for the purpose of understanding, preventing, and relieving psychologically-based distress or dysfunction and to promote subjective well-being and personal development.\"\n",
    "            ),\n",
    "            expected_output=KeywordExtractionExpectedOutput(\n",
    "                keywords={\"clinical psychology\", \"well-being\", \"personal development\"}\n",
    "            ),\n",
    "        ),\n",
    "        Example(\n",
    "            input=KeywordExtractionInput(\n",
    "                text=\"Prospect theory is a theory of behavioral economics, judgment and decision making that was developed by Daniel Kahneman and Amos Tversky in 1979.[1] The theory was cited in the decision to award Kahneman the 2002 Nobel Memorial Prize in Economics.[2]Based on results from controlled studies, it describes how individuals assess their loss and gain perspectives in an asymmetric manner (see loss aversion).\"\n",
    "            ),\n",
    "            expected_output=KeywordExtractionExpectedOutput(\n",
    "                keywords={\n",
    "                    \"prospect theory\",\n",
    "                    \"behavioural economics\",\n",
    "                    \"decision making\",\n",
    "                    \"losses and gains\",\n",
    "                }\n",
    "            ),\n",
    "        ),\n",
    "    ],\n",
    "    dataset_name=\"human-evaluation-multiple-examples-dataset\",\n",
    ").id\n",
    "\n",
    "run = runner.run_dataset(dataset_id)\n",
    "evaluation_overview = evaluator.evaluate_runs(run.id)\n",
    "aggregation_overview = aggregator.aggregate_evaluation(evaluation_overview.id)\n",
    "\n",
    "print(aggregation_overview)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have now run our first evaluation on this tiny dataset.\n",
    "Let's take a more detailed look at the debug log of one of the example runs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "examples = list(\n",
    "    dataset_repository.examples(\n",
    "        dataset_id, evaluator.input_type(), evaluator.expected_output_type()\n",
    "    )\n",
    ")\n",
    "print(examples[1].input.text)\n",
    "examples.sort(key=lambda x: x.input.text)\n",
    "last_example_result = run_repository.example_tracer(\n",
    "    next(iter(aggregation_overview.run_overviews())).id, examples[1].id\n",
    ")\n",
    "last_example_result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's inspect this debug log from top to bottom to try and figure out what happened here.\n",
    "\n",
    "1. **Input**: This corresponds to the `Input` we supplied for our task. In this case, it's just the text of the provided example.\n",
    "\n",
    "2. **Completion request**: The request sent to the Aleph Alpha API. Here you can see the formatted prompt.\n",
    "\n",
    "3. **The output of the `CompletionTask`**: This is the original completion created by the API.\n",
    "\n",
    "4. **The output of our `KeywordExtractionTask`**: The output of our task. Here this is just a list of stripped, lowercase keywords.\n",
    "\n",
    "5. **Metrics**: Several metrics generated by our `KeywordExtractionTaskEvaluationLogic`.\n",
    "\n",
    "Let's have a look at the evaluation results.\n",
    "Here, we can see that the model returned \"behavi*o*ral economics\" as a keyword.\n",
    "However, in the `false_negatives`, we can see that we did indeed expect this phrase, but with a different spelling: \"behavi*ou*ral economics\".\n",
    "Thus, the debug log helped us easily identify this misalignment between our dataset and the model's generation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "last_example_result = evaluation_repository.example_evaluation(\n",
    "    next(iter(aggregation_overview.evaluation_overviews)).id,\n",
    "    examples[1].id,\n",
    "    KeywordExtractionEvaluation,\n",
    ")\n",
    "print(last_example_result.result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, we predicted \"behavioural economics\" but expected \"behavioral economics\"...\n",
    "\n",
    "**What does this tell us?**\n",
    "\n",
    "Why did the British \"ou\" and the American \"o\" go to therapy?\n",
    "\n",
    "They had behavioural differences!"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}