feat: WIP Classification runs as container on localhost - Enhance use…

… case Jupyter Notebook to include guidance for

TASK: IL-421

src/examples/issue_classification_user_journey.ipynb

@@ -43,6 +43,8 @@
    "outputs": [],
    "source": [
     "### Helper methods ###\n",
+    "\n",
+    "\n",
     "def display_histograms(\n",
     "    expected_labels_histogram: dict[str, int],\n",
     "    predicted_labels_histogram: dict[str, int],\n",
@@ -727,6 +729,124 @@
     "\n",
     "Feel free to further play around and improve our classification example. "
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import http\n",
+    "import os\n",
+    "from http import HTTPStatus\n",
+    "from typing import Annotated, Sequence\n",
+    "\n",
+    "from aleph_alpha_client import Client\n",
+    "from dotenv import load_dotenv\n",
+    "from fastapi import Depends, FastAPI, HTTPException, Request, Response\n",
+    "from fastapi.datastructures import URL\n",
+    "\n",
+    "from intelligence_layer.connectors import AlephAlphaClientProtocol\n",
+    "from intelligence_layer.core import LuminousControlModel, NoOpTracer, Task\n",
+    "from intelligence_layer.use_cases import (\n",
+    "    ClassifyInput,\n",
+    "    PromptBasedClassify,\n",
+    "    SingleLabelClassifyOutput,\n",
+    ")\n",
+    "\n",
+    "# Minimal FastAPI app ##########################################################\n",
+    "\n",
+    "app = FastAPI()\n",
+    "\n",
+    "\n",
+    "@app.get(\"/\")\n",
+    "def root() -> Response:\n",
+    "    return Response(content=\"Classification Service\", status_code=HTTPStatus.OK)\n",
+    "\n",
+    "\n",
+    "# Authentication ###############################################################\n",
+    "\n",
+    "\n",
+    "class AuthService:\n",
+    "    def is_valid_token(self, token: str, permissions: Sequence[str], url: URL) -> bool:\n",
+    "        # Add your authentication logic here\n",
+    "        print(f\"Checking permission for route: {url.path}\")\n",
+    "        return True\n",
+    "\n",
+    "\n",
+    "class PermissionChecker:\n",
+    "    def __init__(self, permissions: Sequence[str] = []):\n",
+    "        self.permissions = permissions\n",
+    "\n",
+    "    def __call__(\n",
+    "        self,\n",
+    "        request: Request,\n",
+    "        auth_service: Annotated[AuthService, Depends(AuthService)],\n",
+    "    ) -> None:\n",
+    "        token = request.headers.get(\"Authorization\") or \"\"\n",
+    "        try:\n",
+    "            if not auth_service.is_valid_token(token, self.permissions, request.url):\n",
+    "                raise HTTPException(HTTPStatus.UNAUTHORIZED)\n",
+    "        except RuntimeError:\n",
+    "            raise HTTPException(HTTPStatus.INTERNAL_SERVER_ERROR)\n",
+    "\n",
+    "\n",
+    "permission_checker_for_user = PermissionChecker([\"User\"])\n",
+    "\n",
+    "\n",
+    "# Intelligence Layer Task ######################################################\n",
+    "\n",
+    "PROMPT = \"\"\"Identify the department that would be responsible for handling the given request.\n",
+    "Reply with only the department name.\"\"\"\n",
+    "\n",
+    "load_dotenv()\n",
+    "\n",
+    "\n",
+    "def client() -> Client:\n",
+    "    return Client(\n",
+    "        token=os.environ[\"AA_TOKEN\"],\n",
+    "        host=os.getenv(\"AA_CLIENT_BASE_URL\", \"https://api.aleph-alpha.com\"),\n",
+    "    )\n",
+    "\n",
+    "\n",
+    "def default_model(\n",
+    "    app_client: Annotated[AlephAlphaClientProtocol, Depends(client)],\n",
+    ") -> LuminousControlModel:\n",
+    "    return LuminousControlModel(\"luminous-supreme-control\", client=app_client)\n",
+    "\n",
+    "\n",
+    "def classification_task(\n",
+    "    model: Annotated[LuminousControlModel, Depends(default_model)],\n",
+    ") -> PromptBasedClassify:\n",
+    "    return PromptBasedClassify(instruction=PROMPT, model=model)\n",
+    "\n",
+    "\n",
+    "@app.post(\n",
+    "    \"/classify\",\n",
+    "    # dependencies=[Depends(PermissionChecker([\"User\"]))],\n",
+    "    status_code=http.HTTPStatus.OK,\n",
+    ")\n",
+    "def classification_task_route(\n",
+    "    input: ClassifyInput,\n",
+    "    task: Annotated[\n",
+    "        Task[ClassifyInput, SingleLabelClassifyOutput], Depends(classification_task)\n",
+    "    ],\n",
+    ") -> SingleLabelClassifyOutput:\n",
+    "    return task.run(input, NoOpTracer())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import nest_asyncio\n",
+    "import uvicorn\n",
+    "\n",
+    "nest_asyncio.apply()\n",
+    "uvicorn.run(app, port=8000)"
+   ]
   }
  ],
  "metadata": {

src/examples/issue_classification_user_journey/Dockerfile

@@ -0,0 +1,31 @@
+FROM ubuntu:latest as builder
+
+RUN apt-get update && apt-get upgrade -y \
+    && apt-get install -y python3 python3-pip python3-venv git
+
+RUN mkdir /app
+COPY requirements.txt /app/requirements.txt
+
+RUN python3 -m venv /app/venv
+ENV PATH="/app/venv/bin:$PATH"
+
+RUN echo $(cat /run/secrets/GITHUB_TOKEN)
+
+RUN pip install --upgrade pip
+RUN --mount=type=secret,id=GITHUB_TOKEN \
+    GITHUB_TOKEN=$(cat /run/secrets/GITHUB_TOKEN) pip install -r /app/requirements.txt
+
+
+FROM ubuntu:latest as runtime
+
+# Delete apt package lists immediately to save ~45MB. This has to be done in the same RUN command,
+# otherwise the data is already persisted in another layer.
+RUN apt-get update && apt-get upgrade -y \
+    && apt-get install -y python3 \
+    && rm -r /var/lib/apt/lists/*
+
+COPY --from=builder /app /app
+COPY main.py /app/main.py
+
+ENV PATH="/app/venv/bin:$PATH"
+ENTRYPOINT [ "uvicorn", "app.main:app" ]

src/examples/issue_classification_user_journey/main.py

@@ -0,0 +1,97 @@
+import http
+import os
+from http import HTTPStatus
+from typing import Annotated, Sequence
+
+from aleph_alpha_client import Client
+from dotenv import load_dotenv
+from fastapi import Depends, FastAPI, HTTPException, Request, Response
+from fastapi.datastructures import URL
+
+from intelligence_layer.connectors import AlephAlphaClientProtocol
+from intelligence_layer.core import LuminousControlModel, NoOpTracer, Task
+from intelligence_layer.use_cases import (
+    ClassifyInput,
+    PromptBasedClassify,
+    SingleLabelClassifyOutput,
+)
+
+# Minimal FastAPI app ##########################################################
+
+app = FastAPI()
+
+
+@app.get("/")
+def root() -> Response:
+    return Response(content="Classification Service", status_code=HTTPStatus.OK)
+
+
+# Authentication ###############################################################
+
+
+class AuthService:
+    def is_valid_token(self, token: str, permissions: Sequence[str], url: URL) -> bool:
+        # Add your authentication logic here
+        print(f"Checking permission for route: {url.path}")
+        return True
+
+
+class PermissionChecker:
+    def __init__(self, permissions: Sequence[str] = []):
+        self.permissions = permissions
+
+    def __call__(
+        self,
+        request: Request,
+        auth_service: Annotated[AuthService, Depends(AuthService)],
+    ) -> None:
+        token = request.headers.get("Authorization") or ""
+        try:
+            if not auth_service.is_valid_token(token, self.permissions, request.url):
+                raise HTTPException(HTTPStatus.UNAUTHORIZED)
+        except RuntimeError:
+            raise HTTPException(HTTPStatus.INTERNAL_SERVER_ERROR)
+
+
+permission_checker_for_user = PermissionChecker(["User"])
+
+
+# Intelligence Layer Task ######################################################
+
+PROMPT = """Identify the department that would be responsible for handling the given request.
+Reply with only the department name."""
+
+load_dotenv()
+
+
+def client() -> Client:
+    return Client(
+        token=os.environ["AA_TOKEN"],
+        host=os.getenv("AA_CLIENT_BASE_URL", "https://api.aleph-alpha.com"),
+    )
+
+
+def default_model(
+    app_client: Annotated[AlephAlphaClientProtocol, Depends(client)],
+) -> LuminousControlModel:
+    return LuminousControlModel("luminous-supreme-control", client=app_client)
+
+
+def classification_task(
+    model: Annotated[LuminousControlModel, Depends(default_model)],
+) -> PromptBasedClassify:
+    return PromptBasedClassify(instruction=PROMPT, model=model)
+
+
+@app.post(
+    "/classify",
+    # dependencies=[Depends(PermissionChecker(["User"]))],
+    status_code=http.HTTPStatus.OK,
+)
+def classification_task_route(
+    input: ClassifyInput,
+    task: Annotated[
+        Task[ClassifyInput, SingleLabelClassifyOutput], Depends(classification_task)
+    ],
+) -> SingleLabelClassifyOutput:
+    return task.run(input, NoOpTracer())