examples/6.streaming_pipelines/streaming_pipelines.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Building streaming pipelines in Toloka\n",
    "\n",
    "Let's solve the following task: find the goods in the online-store by given image and aggange found results by relevance.\n",
    "\n",
    "It can be solved in 3 steps:\n",
    "* For given image find corresponding goods in the online shop;\n",
    "* Verfiy that the selected goods are correct;\n",
    "* Arrange found goods by relevance using side-by-side comparison.\n",
    "\n",
    "Each step is represented by Toloka pool. We should also connect those pools and move data between them.\n",
    "<img src=\"https://avatars.mds.yandex.net/get-direct/4119367/JkCWTSuLpAkabWwEJ2pn-Q/orig\" alt=\"Example pipeline steps\" width=\"600\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Call to action\n",
    "If you found some bugs or have a new feature idea, don't hesitate to [open a new issue on Github](https://github.com/Toloka/toloka-kit/issues/new/choose).\n",
    "Like our library and examples? Star [our repo on Github](https://github.com/Toloka/toloka-kit)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-08-26T15:33:19.731357Z",
     "start_time": "2021-08-26T15:33:17.615813Z"
    }
   },
   "outputs": [],
   "source": [
    "%%capture\n",
    "!pip install toloka-kit==0.1.26\n",
    "!pip install crowd-kit==1.0.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-08-26T15:33:21.245650Z",
     "start_time": "2021-08-26T15:33:21.228813Z"
    }
   },
   "outputs": [],
   "source": [
    "import logging\n",
    "import sys\n",
    "import getpass\n",
    "from toloka.client import TolokaClient\n",
    "\n",
    "logging.basicConfig(format='%(levelname)s - %(asctime)s - %(name)s: %(message)s',\n",
    "                    level=logging.INFO,\n",
    "                    stream=sys.stdout)\n",
    "client = toloka.TolokaClient(getpass.getpass('Enter your OAuth token: '), 'PRODUCTION') # Or switch to 'SANDBOX'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This example focuses on pools connections, so we don't pay much attention on projects and pools configuration here.\n",
    "Let's just load configuration from files stored on GitHub."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "import requests\n",
    "import os\n",
    "from toloka.client import Pool, Project, structure\n",
    "\n",
    "GITHUB_RAW = 'https://raw.githubusercontent.com'\n",
    "GITHUB_BASE_PATH = 'Toloka/toloka-kit/main/examples/6.streaming_pipelines'\n",
    "\n",
    "def _load_json_from_github(filename: str):\n",
    "    response = requests.get(os.path.join(GITHUB_RAW, GITHUB_BASE_PATH, filename))\n",
    "    response.raise_for_status()\n",
    "    return response.json()\n",
    "\n",
    "def create_project(filename: str) -> Project:\n",
    "    return client.create_project(_load_json_from_github(filename))\n",
    "\n",
    "def create_pool(filename: str, project_id: str, reward_per_assignment: float) -> Pool:\n",
    "    pool = structure(_load_json_from_github(filename), Pool)\n",
    "    pool.project_id = project_id\n",
    "    pool.reward_per_assignment = reward_per_assignment\n",
    "    pool.will_expire = datetime.datetime.now() + datetime.timedelta(days=3)\n",
    "    return client.create_pool(pool)\n",
    "\n",
    "find_items_project = create_project('find_items_project.json')\n",
    "find_items_pool = create_pool('find_items_pool.json', find_items_project.id, 0.08)\n",
    "\n",
    "verification_project = create_project('verification_project.json')\n",
    "verification_pool = create_pool('verification_pool.json', verification_project.id, 0.02)\n",
    "\n",
    "sbs_project = create_project('sbs_project.json')\n",
    "sbs_pool = create_pool('sbs_pool.json', sbs_project.id, 0.04)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some data flows may be implicitely implemented using pools quality control rules.\n",
    "\n",
    "Here, if some assignment is rejected, the overlap of the corresponding tasks increases, that results in new microtasks appearance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from toloka.client.actions import ChangeOverlap\n",
    "from toloka.client.collectors import AssignmentsAssessment\n",
    "from toloka.client.conditions import AssessmentEvent\n",
    "\n",
    "find_items_pool.quality_control.add_action(\n",
    "    collector=AssignmentsAssessment(),\n",
    "    conditions=[AssessmentEvent == AssessmentEvent.REJECT],\n",
    "    action=ChangeOverlap(delta=1, open_pool=True),\n",
    ")\n",
    "client.update_pool(find_items_pool.id, find_items_pool);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connections\n",
    "\n",
    "Now define each connection as a separate callable.\n",
    "\n",
    "Entire pipeline will be as follows:\n",
    "<img src=\"https://avatars.mds.yandex.net/get-direct/5220563/U4kLMb8FxINJsxXTWeQGew/orig\" alt=\"Example pipeline steps\" width=\"800\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import collections\n",
    "import itertools\n",
    "import pandas as pd\n",
    "from typing import List\n",
    "\n",
    "from toloka.client.task import Task\n",
    "from toloka.streaming.event import AssignmentEvent\n",
    "\n",
    "OVERLAP_FIND_ITEMS = 12\n",
    "OVERLAP_VERIFICATION = 3\n",
    "OVERLAP_SBS = 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handle_found_items(events: List[AssignmentEvent]) -> None:\n",
    "    \"\"\"find_items_pool -> verification_pool\"\"\"\n",
    "    verification_tasks = [\n",
    "        Task(\n",
    "            pool_id=verification_pool.id,\n",
    "            unavailable_for=[event.assignment.user_id],\n",
    "            overlap=OVERLAP_VERIFICATION,\n",
    "            input_values={\n",
    "                'image': task.input_values['image'],\n",
    "                'found_link': solution.output_values['found_link'],\n",
    "                'assignment_id': event.assignment.id\n",
    "            },\n",
    "        )\n",
    "        for event in events\n",
    "        for task, solution in zip(event.assignment.tasks, event.assignment.solutions)\n",
    "    ]\n",
    "    client.create_tasks(verification_tasks, open_pool=True)\n",
    "    logging.info('Verification tasks created count: %d', len(verification_tasks))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from crowdkit.aggregation import MajorityVote\n",
    "from toloka.client.exceptions import IncorrectActionsApiError\n",
    "\n",
    "\n",
    "class VerificationDoneHandler:\n",
    "    \"\"\"verification_pool -> find_items_pool back using quality control rule\"\"\"\n",
    "    def __init__(self, client: TolokaClient):\n",
    "        self.client = client\n",
    "        self.waiting = collections.defaultdict(list)\n",
    "\n",
    "    def __call__(self, events: List[AssignmentEvent]) -> None:\n",
    "        for event in events:\n",
    "            for task, solution in zip(event.assignment.tasks, event.assignment.solutions):\n",
    "                answer = (solution.output_values['result'], event.assignment.user_id)\n",
    "                self.waiting[task.input_values['assignment_id']].append(answer)\n",
    "\n",
    "        to_aggregate = []\n",
    "        for assignment_id, answers in self.waiting.items():\n",
    "            if len(answers) >= OVERLAP_VERIFICATION:\n",
    "                to_aggregate.extend((assignment_id, result, user_id) for result, user_id in answers)\n",
    "\n",
    "        if to_aggregate:\n",
    "            to_aggregate_df = pd.DataFrame(to_aggregate, columns=['task', 'label', 'worker'])\n",
    "            aggregated: pd.Series = MajorityVote().fit_predict(to_aggregate_df)\n",
    "            logging.info('Statuses to apply count: %s', collections.Counter(aggregated.values))\n",
    "\n",
    "            for assignment_id, result in aggregated.items():\n",
    "                try:\n",
    "                    if result == 'Yes':\n",
    "                        self.client.accept_assignment(assignment_id, 'Well done!')\n",
    "                    else:\n",
    "                        self.client.reject_assignment(assignment_id, 'Incorrect object.')\n",
    "                except IncorrectActionsApiError:  # You could have accepted or rejected it in the UI.\n",
    "                    logging.exception('Can\\'t set status %s at %s', result, assignment_id)\n",
    "                del self.waiting[assignment_id]\n",
    "\n",
    "        logging.info('Waiting for verification count: %d', len(self.waiting))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class AcceptedItemsToComparison:\n",
    "    \"\"\"find_items_pool -> sbs_pool\"\"\"\n",
    "    def __init__(self, client: TolokaClient):\n",
    "        self.client = client\n",
    "        self.waiting = collections.defaultdict(list)\n",
    "\n",
    "    def __call__(self, events: List[AssignmentEvent]) -> None:\n",
    "        for event in events:\n",
    "            for task, solution in zip(event.assignment.tasks, event.assignment.solutions):\n",
    "                self.waiting[task.input_values['image']].append(solution.output_values['found_link'])\n",
    "\n",
    "        to_sbs = [(image, found_links)\n",
    "                  for image, found_links in self.waiting.items()\n",
    "                  if len(found_links) >= OVERLAP_FIND_ITEMS]\n",
    "\n",
    "        if to_sbs:\n",
    "            logging.info('Got images ready for SbS count: %d', len(to_sbs))\n",
    "\n",
    "            sbs_tasks = []\n",
    "            for image, found_links in to_sbs:\n",
    "                for left_link, right_link in itertools.combinations(found_links, 2):\n",
    "                    input_values = {'image': image, 'left_link': left_link, 'right_link': right_link}\n",
    "                    sbs_tasks.append(Task(pool_id=sbs_pool.id, overlap=OVERLAP_SBS, input_values=input_values))\n",
    "\n",
    "            logging.info('SbS tasks to create count: %d', len(sbs_tasks))\n",
    "            self.client.create_tasks(sbs_tasks, open_pool=True)\n",
    "\n",
    "        for image, _ in to_sbs:\n",
    "            del self.waiting[image]\n",
    "        logging.info('Waiting for SbS count: %d', len(self.waiting))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from crowdkit.aggregation import BradleyTerry\n",
    "\n",
    "\n",
    "class HandleSbS:\n",
    "    \"\"\"sbs_pool results aggregation\"\"\"\n",
    "    def __init__(self, client: TolokaClient):\n",
    "        self.client = client\n",
    "        self.waiting = collections.defaultdict(list)\n",
    "        self.scores_by_image = {}\n",
    "\n",
    "    def __call__(self, events: List[AssignmentEvent]) -> None:\n",
    "        for event in events:\n",
    "            for task, solution in zip(event.assignment.tasks, event.assignment.solutions):\n",
    "                answer = {'image': task.input_values['image'],\n",
    "                          'worker': event.assignment.user_id,\n",
    "                          'left': task.input_values['left_link'],\n",
    "                          'right': task.input_values['right_link'],\n",
    "                          'label': solution.output_values['result']}\n",
    "                self.waiting[task.input_values['image']].append(answer)\n",
    "\n",
    "        for image, answers in list(self.waiting.items()):\n",
    "            if len(answers) >= OVERLAP_SBS:\n",
    "                scores = BradleyTerry(n_iter=100).fit_predict(pd.DataFrame(answers))\n",
    "                self.scores_by_image[image] = scores.sort_values(ascending=False)\n",
    "                del self.waiting[image]\n",
    "\n",
    "        logging.info('Waiting for SbS aggregation count: %d', len(self.waiting))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Putting it all together"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from toloka.streaming import AssignmentsObserver, Pipeline\n",
    "\n",
    "pipeline = Pipeline()\n",
    "found_items_observer = pipeline.register(AssignmentsObserver(client, find_items_pool.id))\n",
    "verification_observer = pipeline.register(AssignmentsObserver(client, verification_pool.id))\n",
    "sbs_observer = pipeline.register(AssignmentsObserver(client, sbs_pool.id))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "found_items_observer.on_submitted(handle_found_items)\n",
    "found_items_observer.on_accepted(AcceptedItemsToComparison(client))\n",
    "verification_observer.on_accepted(VerificationDoneHandler(client))\n",
    "sbs_handler = sbs_observer.on_accepted(HandleSbS(client))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create tasks for initial pool."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "images = [\n",
    "    'https://tlk.s3.yandex.net/wsdm2020/photos/8ca087fe33065d75327cafdb8720204b.jpg',\n",
    "    'https://tlk.s3.yandex.net/wsdm2020/photos/d0c9eb8737f48df5964d93b08ec0d758.jpg',\n",
    "    'https://tlk.s3.yandex.net/wsdm2020/photos/9245eed8aa1d1e6f5d5d39d00ab044c6.jpg',\n",
    "    'https://tlk.s3.yandex.net/wsdm2020/photos/0aff4fc1edbe6096a9a517092902627f.jpg',\n",
    "    'http://tolokaadmin.s3.yandex.net/demo/abb61898-c886-4e20-b7cd-c0d359ddbb9a',\n",
    "]\n",
    "tasks = [\n",
    "    Task(pool_id=find_items_pool.id, overlap=OVERLAP_FIND_ITEMS, input_values={'image': image})\n",
    "    for image in images\n",
    "]\n",
    "client.create_tasks(tasks, open_pool=True);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Google Colab is using a global event pool,\n",
    "# so in order to run our pipeline we have to apply nest_asyncio to create an inner pool\n",
    "if 'google.colab' in str(get_ipython()):\n",
    "    import nest_asyncio, asyncio\n",
    "    nest_asyncio.apply()\n",
    "    asyncio.get_event_loop().run_until_complete(pipeline.run())\n",
    "else:\n",
    "    await pipeline.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Display results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import Image, display\n",
    "\n",
    "for image, scores in sbs_handler.scores_by_image.items():\n",
    "    display(Image(url=image, height=200))\n",
    "    print(f'{scores.nlargest(1)}\\n')"
   ]
  }
 ],
 "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}