src/documentation/how_tos/how_to_implement_a_task.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dotenv import load_dotenv\n",
    "from pydantic import BaseModel\n",
    "\n",
    "from intelligence_layer.core import (\n",
    "    CompleteInput,\n",
    "    ControlModel,\n",
    "    NoOpTracer,\n",
    "    Pharia1ChatModel,\n",
    "    Task,\n",
    "    TaskSpan,\n",
    ")\n",
    "from intelligence_layer.examples import SingleChunkQa, SingleChunkQaInput\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How to implement a task using an Aleph Alpha model\n",
    "0. Define the task (see [here](./how_to_define_a_task.ipynb))\n",
    "1. Decide which model best suits your use case (for a list of Aleph Alpha control models see [here](https://aleph-alpha-intelligence-layer.readthedocs-hosted.com/en/latest/intelligence_layer.core.html#intelligence_layer.core.LuminousControlModel))\n",
    "2. Create a `Task` subclass\n",
    "   1. Pass the Model to the constructor \n",
    "   2. Implement your domain logic in `do_run()`\n",
    "      1. Generate a `Prompt`. Examples for generating prompts are `ControlModel.to_instruct_prompt()`, `PromptTemplate.to_rich_prompt()` or `Prompt.from_text()`\n",
    "         - We recommend to not write custom `PromptTemplate`s. If necessary, look up the documentation of `PromptTemplate`\n",
    "      2. Run the model with the prompt\n",
    "      3. Map the prompt output to the task output class\n",
    "3. Run and test it"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 0 - Define the input and output types for your task\n",
    "\n",
    "\n",
    "class TellAJokeTaskInput(BaseModel):\n",
    "    topic: str\n",
    "\n",
    "\n",
    "class TellAJokeTaskOutput(BaseModel):\n",
    "    joke: str\n",
    "\n",
    "\n",
    "# Step 1 - we want a control model but do not care otherwise. Therefore we use the default. For our case, the Chat models also work.\n",
    "model_to_use = Pharia1ChatModel()\n",
    "\n",
    "\n",
    "# Step 2\n",
    "class TellAJokeTask(Task[TellAJokeTaskInput, TellAJokeTaskOutput]):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"Tell me a joke about the following topic:\"\"\"\n",
    "\n",
    "    # Step 2.1\n",
    "    def __init__(self, model: ControlModel | None = None) -> None:\n",
    "        self._model = model if model else Pharia1ChatModel()\n",
    "\n",
    "    # Step 2.2\n",
    "    def do_run(\n",
    "        self, input: TellAJokeTaskInput, task_span: TaskSpan\n",
    "    ) -> TellAJokeTaskOutput:\n",
    "        # Step 2.2.1\n",
    "        prompt = self._model.to_instruct_prompt(self.PROMPT_TEMPLATE, input.topic)\n",
    "        completion_input = CompleteInput(prompt=prompt)\n",
    "        # Step 2.2.2\n",
    "        completion = self._model.complete(completion_input, task_span)\n",
    "        return TellAJokeTaskOutput(joke=completion.completions[0].completion)\n",
    "\n",
    "\n",
    "TellAJokeTask(model=model_to_use).run(\n",
    "    TellAJokeTaskInput(topic=\"Software Engineers\"), NoOpTracer()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How to use subtasks in your task\n",
    " -  Follow [How to implement a task using an Aleph Alpha model](#how-to-implement-a-task-using-an-aleph-alpha-model) to create a task and replace the `Model` class with a subtask of your choosing."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from intelligence_layer.core.model import LuminousControlModel\n",
    "\n",
    "\n",
    "class PeopleExtractorInput(BaseModel):\n",
    "    text_passage: str\n",
    "\n",
    "\n",
    "class PeopleExtractorOutput(BaseModel):\n",
    "    answer: str | None\n",
    "\n",
    "\n",
    "class PeopleExtractor(Task[PeopleExtractorInput, PeopleExtractorOutput]):\n",
    "    QUESTION: str = \"\"\"Who are the people involved in the text?\"\"\"\n",
    "\n",
    "    # Step 2.1 - pass the task into the init function\n",
    "    def __init__(self, task: SingleChunkQa | None = None) -> None:\n",
    "        self._task = task if task else SingleChunkQa()\n",
    "\n",
    "    def do_run(\n",
    "        self, input: PeopleExtractorInput, task_span: TaskSpan\n",
    "    ) -> PeopleExtractorOutput:\n",
    "        # Step 2.2.1 - create the required input for the task\n",
    "        question_input = SingleChunkQaInput(\n",
    "            chunk=input.text_passage, question=self.QUESTION\n",
    "        )\n",
    "        # Step 2.2.2 - use the task in the run\n",
    "        completion = self._task.run(\n",
    "            question_input,\n",
    "            task_span,\n",
    "        )\n",
    "        return PeopleExtractorOutput(answer=completion.answer)\n",
    "\n",
    "\n",
    "task_input = PeopleExtractorInput(\n",
    "    text_passage=\"Peter ate Sarahs Lunch, their teacher Mr. Meyers was very angry with him.'\"\n",
    ")\n",
    "PeopleExtractor(task=SingleChunkQa(model=LuminousControlModel())).run(\n",
    "    task_input, NoOpTracer()\n",
    ").answer"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "intelligence-layer-LP3DLT23-py3.12",
   "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
}