From d0a2b066669d11b35453c7233c8221229e3b7482 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A1s=20Schmelczer?= Date: Sat, 28 May 2022 15:15:34 +0200 Subject: [PATCH] Add new features --- examples/simple/deploy.ipynb | 4 +- great_ai/setup.cfg | 1 + great_ai/src/great_ai/__main__.py | 208 ++++++++++++--- great_ai/src/great_ai/great_ai/__init__.py | 1 - great_ai/src/great_ai/great_ai/constants.py | 1 + .../great_ai/great_ai/context/configure.py | 44 ++-- .../src/great_ai/great_ai/context/context.py | 4 +- .../great_ai/dashboard/create_dash_app.py | 50 ++-- .../src/great_ai/great_ai/deploy/great_ai.py | 244 ++++++++++++------ .../src/great_ai/great_ai/helper/__init__.py | 4 +- .../assert_function_is_not_finalised.py | 15 ++ .../great_ai/helper/freeze_arguments.py | 25 ++ .../helper/{get_args.py => get_arguments.py} | 13 +- .../src/great_ai/great_ai/models/use_model.py | 20 +- .../great_ai/parameters/log_metric.py | 10 +- .../great_ai/great_ai/parameters/parameter.py | 9 +- .../persistence/parallel_tinydb_driver.py | 2 +- .../persistence/persistence_driver.py | 2 +- .../great_ai/tracing/tracing_context.py | 6 +- .../src/great_ai/great_ai/views/__init__.py | 1 + .../great_ai/great_ai/views/api_metadata.py | 7 + .../great_ai/views/function_metadata.py | 3 +- great_ai/src/great_ai/open_s3/large_file.py | 17 +- great_ai/src/great_ai/parse_arguments.py | 8 - 24 files changed, 485 insertions(+), 214 deletions(-) create mode 100644 great_ai/src/great_ai/great_ai/helper/assert_function_is_not_finalised.py create mode 100644 great_ai/src/great_ai/great_ai/helper/freeze_arguments.py rename great_ai/src/great_ai/great_ai/helper/{get_args.py => get_arguments.py} (65%) create mode 100644 great_ai/src/great_ai/great_ai/views/api_metadata.py diff --git a/examples/simple/deploy.ipynb b/examples/simple/deploy.ipynb index ec3f1e2..5dacff9 100644 --- a/examples/simple/deploy.ipynb +++ b/examples/simple/deploy.ipynb @@ -133,11 +133,13 @@ } ], "source": [ - "result = predict_domain(\"\"\"\n", + "result = predict_domain(\n", + " \"\"\"\n", " State-of-the-art methods for zero-shot visual recognition formulate learning as a joint embedding problem of images and side information. In these formulations the current best complement to visual features are attributes: manually encoded vectors describing shared characteristics among categories. Despite good performance, attributes have limitations: (1) finer-grained recognition requires commensurately more, and (2) attributes do not provide a natural language interface. We propose to overcome these limitations by training neural language models from scratch; i.e. without pre-training and only consuming words and characters. Our proposed models train end-to-end to align with the fine-grained and category-specific content of images. Natural language provides a flexible and compact way of encoding only the salient visual aspects for distinguishing categories. By training on raw text, our model can do inference on raw text as well, providing humans a familiar mode both for annotation and retrieval. Our model achieves strong performance on zero-shot text-based image retrieval and significantly outperforms the attribute-based state-of-the-art for zero-shot classification on the CaltechUCSD Birds 200-2011 dataset. \"\"\"\n", ")\n", "\n", "from pprint import pprint\n", + "\n", "pprint(result.dict(), width=120)" ] } diff --git a/great_ai/setup.cfg b/great_ai/setup.cfg index cfbcd95..1a071a4 100644 --- a/great_ai/setup.cfg +++ b/great_ai/setup.cfg @@ -43,6 +43,7 @@ install_requires = plotly >= 5.8.0 dash >= 2.4.0 uvicorn[standard] >= 0.17.0 + watchdog >= 2.1.0 [options.package_data] * = *.json, *.yaml, *.yml, *.css diff --git a/great_ai/src/great_ai/__main__.py b/great_ai/src/great_ai/__main__.py index ff65916..44ee8f8 100644 --- a/great_ai/src/great_ai/__main__.py +++ b/great_ai/src/great_ai/__main__.py @@ -1,71 +1,193 @@ #!/usr/bin/env python3 +import logging import re -from importlib import import_module +import time +from importlib import import_module, reload +from pathlib import Path +from typing import Optional import uvicorn -from uvicorn.config import LOGGING_CONFIG +from uvicorn.config import LOGGING_CONFIG, Config +from uvicorn.subprocess import get_subprocess +from uvicorn.supervisors.basereload import BaseReload +from watchdog.events import FileSystemEvent, PatternMatchingEventHandler +from watchdog.observers import Observer -from .great_ai.context import get_context -from .great_ai.exceptions import MissingArgumentError +from .great_ai.context.configure import _is_in_production_mode +from .great_ai.deploy import GreatAI +from .great_ai.exceptions import ArgumentValidationError, MissingArgumentError from .parse_arguments import parse_arguments from .utilities.logger import get_logger logger = get_logger("GreatAI-Server") +GREAT_AI_LOGGING_CONFIG = { + **LOGGING_CONFIG, + "formatters": { + "default": { + "()": "great_ai.logger.CustomFormatter", + "fmt": "%(asctime)s | %(levelname)8s | %(message)s", + }, + "access": { + "()": "great_ai.logger.CustomFormatter", + "fmt": "%(asctime)s | %(levelname)8s | %(message)s", # noqa: E501 + }, + }, +} + + def main() -> None: args = parse_arguments() + should_auto_reload = not _is_in_production_mode(logger=None) - file_name = re.sub(r"\.py$", "", args.file_name) - function_name = args.function_name + if args.workers > 1 and should_auto_reload: + raise ArgumentValidationError( + "Cannot use auto-reload with multiple workers: set the `--workers=1` CLI argument," + + "or set the ENVIRONMENT environment variable to `production`." + ) - module = import_module(file_name) - - if not function_name: - logger.warning("Argument function_name not provided, trying to guess it") - - if not function_name and "app" in module.__dict__: - function_name = "app" - - if not function_name and file_name in module.__dict__: - function_name = file_name - - if function_name: - logger.warning(f"Found `{function_name}` as the value of function_name") - else: - raise MissingArgumentError("Argument function_name could not be guessed") - - app_name = f"{file_name}:{function_name}" - logger.info(f"Starting uvicorn server with app={app_name}") - - uvicorn.run( - app_name, + common_config = dict( host=args.host, port=args.port, timeout_keep_alive=args.timeout_keep_alive, workers=args.workers, - reload=not get_context().is_production, - log_config={ - **LOGGING_CONFIG, - "formatters": { - "default": { - "()": "great_ai.logger.CustomFormatter", - "fmt": "%(asctime)s | %(levelname)8s | %(message)s", - }, - "access": { - "()": "great_ai.logger.CustomFormatter", - "fmt": "%(asctime)s | %(levelname)8s | %(message)s", # noqa: E501 - }, - }, - }, + server_header=False, + reload=False, + log_config=GREAT_AI_LOGGING_CONFIG, ) + if not should_auto_reload: + file_name = get_script_name(args.file_name) + app = find_app(file_name) + + logger.info(f"Starting uvicorn server with app={app}") + + uvicorn.run(app, **common_config) # this will never return + + class EventHandler(PatternMatchingEventHandler): + def __init__(self) -> None: + super().__init__(patterns=["*.py", "*.ipynb"], ignore_patterns=["__*.py"]) + self.server: Optional[GreatAIReload] = None + self.restart() + + def on_closed(self, event: FileSystemEvent) -> None: + logger.warning(f"File {event.src_path} has triggered a restart") + self.restart() + + def restart(self) -> None: + file_name = get_script_name(args.file_name) + app = find_app(file_name) + if app is None: + logger.warning("Auto-reloading skipped") + return + + self.stop_server() + + config = Config(app, **common_config) + socket = config.bind_socket() + self.server = GreatAIReload( + config, target=uvicorn.Server(config=config).run, sockets=[socket] + ) + self.server.startup() + + def stop_server(self) -> None: + if self.server: + self.server.shutdown() + + restart_handler = EventHandler() + observer = Observer() + observer.schedule(restart_handler, path=".", recursive=True) + observer.start() + + try: + while True: + time.sleep(50) + finally: + observer.stop() + restart_handler.stop_server() + if args.file_name.endswith(".ipynb"): + Path(get_script_name_of_notebook(args.file_name)).unlink(missing_ok=True) + observer.join() + + +def get_script_name(file_name_argument: str) -> str: + if file_name_argument.endswith(".ipynb"): + logger.info("Converting notebook to Python script") + try: + from nbconvert import PythonExporter + + exporter = PythonExporter() + content, _ = exporter.from_filename(file_name_argument) + file_name_argument = get_script_name_of_notebook(file_name_argument) + with open(file_name_argument, "w", encoding="utf-8") as f: + f.write(content) + except ImportError: + raise ImportError( + "Install `nbconvert` to be able to use Jupyter notebook files or use a regular Python file instead" + ) + + return re.sub(r"\.(py|ipynb)$", "", file_name_argument) + + +def get_script_name_of_notebook(notebook_name: str) -> str: + base_name = re.sub(r"\.ipynb$", "", notebook_name) + return f"__{base_name}__.py" + + +module = None + + +def find_app(file_name: str) -> Optional[str]: + global module + + logging.disable(logging.CRITICAL) + try: + if module is None: + module = import_module(file_name) + else: + module = reload(module) + except Exception: + logging.disable(logging.NOTSET) + logger.exception("Could not load file because of an exception: fix your code") + return None + finally: + logging.disable(logging.NOTSET) + + for name, value in module.__dict__.items(): + if isinstance(value, GreatAI): + app_name = name + + if app_name: + logger.info(f"Found `{app_name}` to be the GreatAI app ") + else: + raise MissingArgumentError( + "GreatAI app could not be found, make sure to use `@GreatAI.deploy` on your prediction function" + ) + + return f"{file_name}:{app_name}.app" + + +class GreatAIReload(BaseReload): + def startup(self) -> None: + self.process = get_subprocess( + config=self.config, target=self.target, sockets=self.sockets + ) + self.process.start() + + def shutdown(self) -> None: + self.process.terminate() + self.process.join() + + for sock in self.sockets: + sock.close() + if __name__ == "__main__": try: main() - except (MissingArgumentError, ModuleNotFoundError) as e: - logger.error(e) except KeyboardInterrupt: exit() + except Exception as e: + logger.error(e) diff --git a/great_ai/src/great_ai/great_ai/__init__.py b/great_ai/src/great_ai/great_ai/__init__.py index 6bda650..22649c0 100644 --- a/great_ai/src/great_ai/great_ai/__init__.py +++ b/great_ai/src/great_ai/great_ai/__init__.py @@ -1,6 +1,5 @@ from .context import configure from .deploy import GreatAI -from .exceptions import ArgumentValidationError, MissingArgumentError from .models import save_model, use_model from .output_models import ClassificationOutput, RegressionOutput from .parameters import log_metric, parameter diff --git a/great_ai/src/great_ai/great_ai/constants.py b/great_ai/src/great_ai/great_ai/constants.py index 805a91f..4b0c623 100644 --- a/great_ai/src/great_ai/great_ai/constants.py +++ b/great_ai/src/great_ai/great_ai/constants.py @@ -1,3 +1,4 @@ ENV_VAR_KEY = "ENVIRONMENT" PRODUCTION_KEY = "production" DEFAULT_TRACING_DB_FILENAME = "tracing_database.json" +METRICS_PATH = "/metrics" diff --git a/great_ai/src/great_ai/great_ai/context/configure.py b/great_ai/src/great_ai/great_ai/context/configure.py index 89ad750..9ac946d 100644 --- a/great_ai/src/great_ai/great_ai/context/configure.py +++ b/great_ai/src/great_ai/great_ai/context/configure.py @@ -1,7 +1,8 @@ import os import random -from logging import INFO, Logger +from logging import DEBUG, Logger from pathlib import Path +from typing import Optional import great_ai.great_ai.context.context as context from great_ai.open_s3 import LargeFile @@ -12,12 +13,15 @@ from ..persistence import ParallelTinyDbDriver, PersistenceDriver def configure( - log_level: int = INFO, + version: str = "0.0.1", + log_level: int = DEBUG, s3_config: Path = Path("s3.ini"), seed: int = 42, persistence_driver: PersistenceDriver = ParallelTinyDbDriver( Path(DEFAULT_TRACING_DB_FILENAME) ), + should_log_exception_stack: Optional[bool] = None, + prediction_cache_size: int = 512, ) -> None: logger = get_logger("great_ai", level=log_level) @@ -27,9 +31,7 @@ def configure( + 'Make sure to call "configure()" before importing your application code.' ) - is_production = _is_in_production_mode( - logger=logger, - ) + is_production = _is_in_production_mode(logger=logger) _initialize_large_file(s3_config, logger=logger) _set_seed(seed) @@ -39,34 +41,38 @@ def configure( ) context._context = context.Context( - metrics_path="/metrics", + version=version, persistence=persistence_driver, is_production=is_production, logger=logger, + should_log_exception_stack=not is_production + if should_log_exception_stack is None + else should_log_exception_stack, + prediction_cache_size=prediction_cache_size, ) logger.info("Options: configured ✅") -def _is_in_production_mode(logger: Logger) -> bool: +def _is_in_production_mode(logger: Optional[Logger]) -> bool: environment = os.environ.get(ENV_VAR_KEY) if environment is None: - logger.info( - f"Environment variable {ENV_VAR_KEY} is not set, defaulting to development mode" - ) + if logger: + logger.warning( + f"Environment variable {ENV_VAR_KEY} is not set, defaulting to development mode ‼️" + ) is_production = False else: is_production = environment.lower() == PRODUCTION_KEY - if not is_production: - logger.info( - f"Value of {ENV_VAR_KEY} is `{environment}` which is not equal to `{PRODUCTION_KEY}`" - ) - - if is_production: - logger.info("Running in production mode ✅") - else: - logger.warning("Running in development mode ‼️") + if logger: + if not is_production: + logger.info( + f"Value of {ENV_VAR_KEY} is `{environment}` which is not equal to `{PRODUCTION_KEY}`" + + "defaulting to development mode ‼️" + ) + else: + logger.info("Running in production mode ✅") return is_production diff --git a/great_ai/src/great_ai/great_ai/context/context.py b/great_ai/src/great_ai/great_ai/context/context.py index 0033e41..e3998a0 100644 --- a/great_ai/src/great_ai/great_ai/context/context.py +++ b/great_ai/src/great_ai/great_ai/context/context.py @@ -7,10 +7,12 @@ from ..persistence import PersistenceDriver class Context(BaseModel): - metrics_path: str + version: str persistence: PersistenceDriver is_production: bool logger: Logger + should_log_exception_stack: bool + prediction_cache_size: int class Config: arbitrary_types_allowed = True diff --git a/great_ai/src/great_ai/great_ai/dashboard/create_dash_app.py b/great_ai/src/great_ai/great_ai/dashboard/create_dash_app.py index a94174d..103de91 100644 --- a/great_ai/src/great_ai/great_ai/dashboard/create_dash_app.py +++ b/great_ai/src/great_ai/great_ai/dashboard/create_dash_app.py @@ -9,6 +9,7 @@ from flask import Flask from great_ai.utilities.unique import unique +from ..constants import METRICS_PATH from ..context import get_context from ..helper import snake_case_to_text, text_to_hex_color from ..views import SortBy @@ -23,7 +24,7 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask: flask_app = Flask(__name__) app = Dash( function_name, - requests_pathname_prefix=get_context().metrics_path + "/", + requests_pathname_prefix=METRICS_PATH + "/", server=flask_app, title=snake_case_to_text(function_name), update_title=None, @@ -103,10 +104,11 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask: @app.callback( Output(execution_time_histogram, "figure"), + Output(parallel_coords, "figure"), Input(table, "filter_query"), Input(interval, "n_intervals"), ) - def update_execution_times(filter: str, _n_intervals: int) -> go.Figure: + def update_charts(filter: str, n_intervals: int) -> go.Figure: conjunctive_filters = [ get_filter_from_datatable(f) for f in filter.split(" && ") ] @@ -117,7 +119,7 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask: ) if not rows: - return go.Figure() + return go.Figure(), go.Figure() df = pd.DataFrame(rows) @@ -136,36 +138,18 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask: margin=dict(l=0, r=0, b=0, t=0, pad=0), ) - return fig - - @app.callback( - Output(parallel_coords, "figure"), - Input(table, "filter_query"), - Input(interval, "n_intervals"), - ) - def update_parallel_coords(filter: str, _n_intervals: int) -> go.Figure: - conjunctive_filters = [ - get_filter_from_datatable(f) for f in filter.split(" && ") - ] - non_null_conjunctive_filters = [f for f in conjunctive_filters if f is not None] - - rows = get_context().persistence.query( - conjunctive_filters=non_null_conjunctive_filters - ) - - if not rows: - return go.Figure() - - df = pd.DataFrame(rows) - return go.Figure( - go.Parcoords( - dimensions=[ - get_dimension_descriptor(df, c) - for c in df.columns - if c not in {"id", "created", "output"} - ], - line_color=accent_color, - ) + return ( + fig, + go.Figure( + go.Parcoords( + dimensions=[ + get_dimension_descriptor(df, c) + for c in df.columns + if c not in {"id", "created", "output"} + ], + line_color=accent_color, + ) + ), ) return flask_app diff --git a/great_ai/src/great_ai/great_ai/deploy/great_ai.py b/great_ai/src/great_ai/great_ai/deploy/great_ai.py index 5b1c7b9..8ce7f42 100644 --- a/great_ai/src/great_ai/great_ai/deploy/great_ai.py +++ b/great_ai/src/great_ai/great_ai/deploy/great_ai.py @@ -1,5 +1,5 @@ import inspect -from functools import partial +from functools import lru_cache, partial, wraps from pathlib import Path from typing import ( Any, @@ -14,7 +14,7 @@ from typing import ( cast, ) -from fastapi import FastAPI, HTTPException, status +from fastapi import APIRouter, FastAPI, HTTPException, status from fastapi.middleware.wsgi import WSGIMiddleware from fastapi.openapi.docs import get_swagger_ui_html from fastapi.responses import RedirectResponse @@ -22,47 +22,59 @@ from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, create_model from starlette.responses import HTMLResponse -from great_ai.great_ai.helper.use_http_exceptions import use_http_exceptions from great_ai.utilities.parallel_map import parallel_map +from ..constants import METRICS_PATH from ..context import get_context from ..dashboard import create_dash_app -from ..helper import get_function_metadata_store, snake_case_to_text +from ..helper import ( + freeze_arguments, + get_function_metadata_store, + snake_case_to_text, + use_http_exceptions, +) from ..parameters import automatically_decorate_parameters from ..tracing import TracingContext -from ..views import EvaluationFeedbackRequest, HealthCheckResponse, Query, Trace +from ..views import ( + ApiMetadata, + EvaluationFeedbackRequest, + HealthCheckResponse, + Query, + Trace, +) PATH = Path(__file__).parent.resolve() -class GreatAI(FastAPI): - def __init__(self, func: Callable[..., Any], *args: Any, **kwargs: Any): +class GreatAI: + def __init__(self, func: Callable[..., Any]): self._func = automatically_decorate_parameters(func) + self._func = freeze_arguments( + lru_cache(get_context().prediction_cache_size)(self._func) + ) - schema = self._get_schema() + get_function_metadata_store(self._func).is_finalised = True + wraps(func)(self) - def process_single(input_value: schema) -> Trace: # type: ignore - with TracingContext() as t: - result = self._func(**cast(BaseModel, input_value).dict()) - output = t.finalise(output=result) - return output - - self.process_single = process_single - - super().__init__( - *args, - title=snake_case_to_text(func.__name__), + self.app = FastAPI( + title=self.name, + version=self.version, description=self.documentation, docs_url=None, - version=get_function_metadata_store(func).version, redoc_url=None, - **kwargs, ) + def __call__(self, *args: Any, **kwargs: Any) -> Trace: + with TracingContext() as t: + result = self._func(*args, **kwargs) + output = t.finalise(output=result) + return output + @staticmethod def deploy( func: Optional[Callable[..., Any]] = None, *, + disable_rest_api: bool = False, disable_docs: bool = False, disable_metrics: bool = False, ) -> Union[Callable[[Callable[..., Any]], "GreatAI"], "GreatAI"]: @@ -71,14 +83,20 @@ class GreatAI(FastAPI): Callable[..., Any], partial( GreatAI.deploy, + disable_http=disable_rest_api, disable_docs=disable_docs, disable_metrics=disable_metrics, ), ) - return GreatAI(func)._bootstrap_rest_api( - disable_docs=disable_docs, disable_metrics=disable_metrics - ) + instance = GreatAI(func) + + if not disable_rest_api: + instance._bootstrap_rest_api( + disable_docs=disable_docs, disable_metrics=disable_metrics + ) + + return instance def process_batch( self, @@ -89,15 +107,68 @@ class GreatAI(FastAPI): concurrency = 1 get_context().logger.warning("Concurrency is ignored") - return parallel_map(self.process_single, batch, concurrency=concurrency) + return parallel_map(self, batch, concurrency=concurrency) + + @property + def name(self) -> str: + return snake_case_to_text(self._func.__name__) + + @property + def version(self) -> str: + return f"{get_context().version}+{get_function_metadata_store(self._func).model_versions}" @property def documentation(self) -> str: return ( - f"GreatAI wrapper for interacting with the '{self._func.__name__}' function.\n" - + (self._func.__doc__ or "") + f"GreatAI wrapper for interacting with the `{self._func.__name__}` function.\n\n" + + ( + "\n".join( + line.strip() + for line in (self._func.__doc__ or "").split("\n") + if line.strip() + ) + ) ) + def _bootstrap_rest_api(self, disable_docs: bool, disable_metrics: bool) -> None: + self._bootstrap_prediction_endpoints() + self._bootstrap_feedback_endpoints() + self._bootstrap_meta_endpoints() + + if not disable_docs: + self._bootstrap_docs_endpoints() + + if not disable_metrics: + self._bootstrap_metrics_endpoints() + + def _bootstrap_prediction_endpoints(self) -> None: + router = APIRouter( + prefix="/predictions", + tags=["predictions"], + ) + + schema = self._get_schema() + + @router.post("/", status_code=status.HTTP_200_OK, response_model=Trace) + @use_http_exceptions + def predict(input_value: schema) -> Trace: # type: ignore + return self(**cast(BaseModel, input_value).dict()) + + @router.get( + "/:prediction_id", response_model=Trace, status_code=status.HTTP_200_OK + ) + def get_prediction(prediction_id: str) -> Trace: + result = get_context().persistence.get_trace(prediction_id) + if result is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) + return result + + @router.delete("/:prediction_id", status_code=status.HTTP_204_NO_CONTENT) + def delete_prediction(prediction_id: str) -> None: + get_context().persistence.delete_trace(prediction_id) + + self.app.include_router(router) + def _get_schema(self) -> Type[BaseModel]: signature = inspect.signature(self._func) parameters = { @@ -112,61 +183,80 @@ class GreatAI(FastAPI): schema: Type[BaseModel] = create_model("InputModel", **parameters) # type: ignore return schema - def _bootstrap_rest_api( - self, disable_docs: bool, disable_metrics: bool - ) -> "GreatAI": - self.post("/evaluations", status_code=status.HTTP_200_OK, response_model=Trace)( - use_http_exceptions(self.process_single) + def _bootstrap_feedback_endpoints(self) -> None: + router = APIRouter( + prefix="/predictions/:prediction_id/feedback", + tags=["feedback"], ) - @self.get("/evaluations/:evaluation_id", status_code=status.HTTP_200_OK) - def get_evaluation(evaluation_id: str) -> Trace: - result = get_context().persistence.get_trace(evaluation_id) - if result is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) - return result + @router.put("/", status_code=status.HTTP_202_ACCEPTED) + def set_feedback(prediction_id: str, input: EvaluationFeedbackRequest) -> None: + get_context().persistence.add_feedback(prediction_id, input.evaluation) - @self.post( - "/evaluations/:evaluation_id/feedback", status_code=status.HTTP_202_ACCEPTED + @router.get("/", status_code=status.HTTP_200_OK) + def get_feedback(prediction_id: str) -> Any: + return get_context().persistence.get_feedback(prediction_id) + + @router.delete("/", status_code=status.HTTP_200_OK) + def delete_feedback(prediction_id: str) -> Any: + get_context().persistence.delete_feedback(prediction_id) + + self.app.include_router(router) + + def _bootstrap_meta_endpoints(self) -> None: + router = APIRouter( + tags=["meta"], ) - def give_feedback(evaluation_id: str, input: EvaluationFeedbackRequest) -> None: - get_context().persistence.add_evaluation(evaluation_id, input.evaluation) - if not disable_docs: - - @self.get("/docs", include_in_schema=False) - def custom_swagger_ui_html() -> HTMLResponse: - return get_swagger_ui_html(openapi_url="openapi.json", title=self.title) - - @self.get("/docs/index.html", include_in_schema=False) - def redirect_to_docs() -> RedirectResponse: - return RedirectResponse("/docs") - - if not disable_metrics: - dash_app = create_dash_app(self._func.__name__, self.documentation) - self.mount(get_context().metrics_path, WSGIMiddleware(dash_app)) - - @self.get("/", include_in_schema=False) - def redirect_to_entrypoint() -> RedirectResponse: - return RedirectResponse("/metrics") - - self.mount( - "/assets", - StaticFiles(directory=PATH / "../dashboard/assets"), - name="static", - ) - - @self.post("/query", status_code=status.HTTP_200_OK) - def query_metrics(query: Query) -> List[Dict[str, Any]]: - return get_context().persistence.query( - conjunctive_filters=query.filter, - sort_by=query.sort, - skip=query.skip, - take=query.take, - ) - - @self.get("/health", status_code=status.HTTP_200_OK) + @router.get("/health", status_code=status.HTTP_200_OK) def check_health() -> HealthCheckResponse: return HealthCheckResponse(is_healthy=True) - return self + @router.get( + "/version", response_model=ApiMetadata, status_code=status.HTTP_200_OK + ) + def get_version() -> ApiMetadata: + return ApiMetadata( + name=self.name, version=self.version, documentation=self.documentation + ) + + self.app.include_router(router) + + def _bootstrap_docs_endpoints(self) -> None: + @self.app.get("/docs", include_in_schema=False) + def custom_swagger_ui_html() -> HTMLResponse: + return get_swagger_ui_html(openapi_url="openapi.json", title=self.app.title) + + @self.app.get("/docs/index.html", include_in_schema=False) + def redirect_to_docs() -> RedirectResponse: + return RedirectResponse("/docs") + + def _bootstrap_metrics_endpoints(self) -> None: + dash_app = create_dash_app(self._func.__name__, self.documentation) + self.app.mount(METRICS_PATH, WSGIMiddleware(dash_app)) + + @self.app.get("/", include_in_schema=False) + def redirect_to_entrypoint() -> RedirectResponse: + return RedirectResponse("/metrics") + + self.app.mount( + "/assets", + StaticFiles(directory=PATH / "../dashboard/assets"), + name="static", + ) + + router = APIRouter( + prefix="/metrics", + tags=["metrics"], + ) + + @router.post("/query", status_code=status.HTTP_200_OK) + def query_metrics(query: Query) -> List[Dict[str, Any]]: + return get_context().persistence.query( + conjunctive_filters=query.filter, + sort_by=query.sort, + skip=query.skip, + take=query.take, + ) + + self.app.include_router(router) diff --git a/great_ai/src/great_ai/great_ai/helper/__init__.py b/great_ai/src/great_ai/great_ai/helper/__init__.py index bde98e2..3a80359 100644 --- a/great_ai/src/great_ai/great_ai/helper/__init__.py +++ b/great_ai/src/great_ai/great_ai/helper/__init__.py @@ -1,4 +1,6 @@ -from .get_args import get_args +from .assert_function_is_not_finalised import assert_function_is_not_finalised +from .freeze_arguments import freeze_arguments +from .get_arguments import get_arguments from .get_function_metadata_store import get_function_metadata_store from .snake_case_to_text import snake_case_to_text from .strip_lines import strip_lines diff --git a/great_ai/src/great_ai/great_ai/helper/assert_function_is_not_finalised.py b/great_ai/src/great_ai/great_ai/helper/assert_function_is_not_finalised.py new file mode 100644 index 0000000..b2110c2 --- /dev/null +++ b/great_ai/src/great_ai/great_ai/helper/assert_function_is_not_finalised.py @@ -0,0 +1,15 @@ +from typing import Any, Callable + +from ..context import get_context +from .get_function_metadata_store import get_function_metadata_store + + +def assert_function_is_not_finalised(func: Callable[..., Any]) -> None: + error_message = ( + "The outer-most (first) decorator has to be `@GreatAI.deploy`. " + + f"In the case of `{func.__name__}`, it is not: fix this by moving `@GreatAI.deploy` to the top." + ) + + if get_function_metadata_store(func).is_finalised: + get_context().logger.error(error_message) + exit(-1) diff --git a/great_ai/src/great_ai/great_ai/helper/freeze_arguments.py b/great_ai/src/great_ai/great_ai/helper/freeze_arguments.py new file mode 100644 index 0000000..3847ea5 --- /dev/null +++ b/great_ai/src/great_ai/great_ai/helper/freeze_arguments.py @@ -0,0 +1,25 @@ +from functools import wraps +from typing import Any, Callable, Dict, List + + +class FrozenDict(dict): + def __hash__(self) -> int: # type: ignore + return hash(frozenset(self.items())) + + +def freeze_arguments(func: Callable[..., Any]) -> Callable[..., Any]: + """Transform mutable dictionary + Into immutable + Useful to be compatible with cache + source: https://stackoverflow.com/questions/6358481/using-functools-lru-cache-with-dictionary-arguments + """ + + @wraps(func) + def wrapper(*args: List[Any], **kwargs: Dict[str, Any]) -> Any: + args = tuple(FrozenDict(arg) if isinstance(arg, dict) else arg for arg in args) + kwargs = { + k: FrozenDict(v) if isinstance(v, dict) else v for k, v in kwargs.items() + } + return func(*args, **kwargs) + + return wrapper diff --git a/great_ai/src/great_ai/great_ai/helper/get_args.py b/great_ai/src/great_ai/great_ai/helper/get_arguments.py similarity index 65% rename from great_ai/src/great_ai/great_ai/helper/get_args.py rename to great_ai/src/great_ai/great_ai/helper/get_arguments.py index 41ddcbd..91fb44c 100644 --- a/great_ai/src/great_ai/great_ai/helper/get_args.py +++ b/great_ai/src/great_ai/great_ai/helper/get_arguments.py @@ -2,14 +2,23 @@ import inspect from typing import Any, Callable, Dict, Mapping, Sequence -def get_args( +def get_arguments( func: Callable[..., Any], args: Sequence[Any], kwargs: Mapping[str, Any] ) -> Dict[str, Any]: """Return mapping from parameter names to actual argument values""" + signature = inspect.signature(func) + + defaults = { + p.name: p.default + for p in signature.parameters.values() + if p.default != inspect._empty + } + filter_keys = [ param.name for param in signature.parameters.values() if param.kind == param.POSITIONAL_OR_KEYWORD ] - return {**dict(zip(filter_keys, args)), **kwargs} + + return {**defaults, **dict(zip(filter_keys, args)), **kwargs} diff --git a/great_ai/src/great_ai/great_ai/models/use_model.py b/great_ai/src/great_ai/great_ai/models/use_model.py index 8743cea..6329d18 100644 --- a/great_ai/src/great_ai/great_ai/models/use_model.py +++ b/great_ai/src/great_ai/great_ai/models/use_model.py @@ -1,7 +1,7 @@ from functools import wraps from typing import Any, Callable, Dict, List, Literal, Union -from ..helper import get_function_metadata_store +from ..helper import assert_function_is_not_finalised, get_function_metadata_store from ..tracing import TracingContext from ..views import Model from .load_model import load_model @@ -14,7 +14,9 @@ def use_model( return_path: bool = False, model_kwarg_name: str = "model", ) -> Callable[..., Any]: - assert isinstance(version, int) or version == "latest" + assert ( + isinstance(version, int) or version == "latest" + ), "Only integers or the string literal `latest` is allowed as version" model, actual_version = load_model( key=key, @@ -23,17 +25,19 @@ def use_model( ) def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + assert_function_is_not_finalised(func) + store = get_function_metadata_store(func) store.model_parameter_names.append(model_kwarg_name) - if store.version: - store.version += "|" - store.version += f"{key}:{actual_version}" + if store.model_versions: + store.model_versions += "." + store.model_versions += f"{key}-v{actual_version}" @wraps(func) def wrapper(*args: List[Any], **kwargs: Dict[str, Any]) -> Any: - context = TracingContext.get_current_context() - if context: - context.log_model(Model(key=key, version=actual_version)) + tracing_context = TracingContext.get_current_context() + if tracing_context: + tracing_context.log_model(Model(key=key, version=actual_version)) return func(*args, **kwargs, **{model_kwarg_name: model}) return wrapper diff --git a/great_ai/src/great_ai/great_ai/parameters/log_metric.py b/great_ai/src/great_ai/great_ai/parameters/log_metric.py index 4a8263d..6cdf6c7 100644 --- a/great_ai/src/great_ai/great_ai/parameters/log_metric.py +++ b/great_ai/src/great_ai/great_ai/parameters/log_metric.py @@ -1,12 +1,16 @@ import inspect from typing import Any +from great_ai.great_ai.context.get_context import get_context + from ..tracing import TracingContext def log_metric(argument_name: str, value: Any) -> None: - context = TracingContext.get_current_context() + tracing_context = TracingContext.get_current_context() caller = inspect.stack()[1].function actual_name = f"metric:{caller}:{argument_name}" - if context: - context.log_value(name=actual_name, value=value) + if tracing_context: + tracing_context.log_value(name=actual_name, value=value) + + get_context().logger.info(f"{actual_name}={value}") diff --git a/great_ai/src/great_ai/great_ai/parameters/parameter.py b/great_ai/src/great_ai/great_ai/parameters/parameter.py index 8fd3d97..39b00d8 100644 --- a/great_ai/src/great_ai/great_ai/parameters/parameter.py +++ b/great_ai/src/great_ai/great_ai/parameters/parameter.py @@ -2,7 +2,11 @@ from functools import wraps from typing import Any, Callable, Dict from ..exceptions import ArgumentValidationError -from ..helper import get_args, get_function_metadata_store +from ..helper import ( + assert_function_is_not_finalised, + get_arguments, + get_function_metadata_store, +) from ..tracing import TracingContext @@ -14,12 +18,13 @@ def parameter( ) -> Callable[..., Any]: def decorator(func: Callable[..., Any]) -> Callable[..., Any]: get_function_metadata_store(func).input_parameter_names.append(parameter_name) + assert_function_is_not_finalised(func) actual_name = f"arg:{func.__name__}:{parameter_name}" @wraps(func) def wrapper(*args: Any, **kwargs: Dict[str, Any]) -> Any: - arguments = get_args(func, args, kwargs) + arguments = get_arguments(func, args, kwargs) argument = arguments[parameter_name] expected_type = func.__annotations__.get(parameter_name) diff --git a/great_ai/src/great_ai/great_ai/persistence/parallel_tinydb_driver.py b/great_ai/src/great_ai/great_ai/persistence/parallel_tinydb_driver.py index bd0976c..4ddce9f 100644 --- a/great_ai/src/great_ai/great_ai/persistence/parallel_tinydb_driver.py +++ b/great_ai/src/great_ai/great_ai/persistence/parallel_tinydb_driver.py @@ -24,7 +24,7 @@ class ParallelTinyDbDriver(PersistenceDriver): def save_trace(self, trace: Trace) -> str: return self._safe_execute(lambda db: db.insert(trace.dict())) - def add_evaluation(self, id: str, evaluation: Any) -> None: + def add_feedback(self, id: str, evaluation: Any) -> None: self._safe_execute( lambda db: db.update( fields={"evaluation": evaluation}, diff --git a/great_ai/src/great_ai/great_ai/persistence/persistence_driver.py b/great_ai/src/great_ai/great_ai/persistence/persistence_driver.py index d7137f2..76ea71a 100644 --- a/great_ai/src/great_ai/great_ai/persistence/persistence_driver.py +++ b/great_ai/src/great_ai/great_ai/persistence/persistence_driver.py @@ -12,7 +12,7 @@ class PersistenceDriver(ABC): pass @abstractmethod - def add_evaluation(self, id: str, evaluation: Any) -> None: + def add_feedback(self, id: str, evaluation: Any) -> None: pass @abstractmethod diff --git a/great_ai/src/great_ai/great_ai/tracing/tracing_context.py b/great_ai/src/great_ai/great_ai/tracing/tracing_context.py index 6af2035..afca76b 100644 --- a/great_ai/src/great_ai/great_ai/tracing/tracing_context.py +++ b/great_ai/src/great_ai/great_ai/tracing/tracing_context.py @@ -61,12 +61,12 @@ class TracingContext: if exception is not None and type is not None: self.finalise(exception=exception) - if get_context().is_production: + if get_context().should_log_exception_stack: + get_context().logger.exception("Could not finish operation") + else: get_context().logger.error( f"Could not finish operation because of {type.__name__}: {exception}" ) - else: - get_context().logger.exception("Could not finish operation") assert self._trace is not None get_context().persistence.save_trace(self._trace) diff --git a/great_ai/src/great_ai/great_ai/views/__init__.py b/great_ai/src/great_ai/great_ai/views/__init__.py index 82c77d9..35ddb50 100644 --- a/great_ai/src/great_ai/great_ai/views/__init__.py +++ b/great_ai/src/great_ai/great_ai/views/__init__.py @@ -1,3 +1,4 @@ +from .api_metadata import ApiMetadata from .evaluation_feedback_request import EvaluationFeedbackRequest from .filter import Filter from .function_metadata import FunctionMetadata diff --git a/great_ai/src/great_ai/great_ai/views/api_metadata.py b/great_ai/src/great_ai/great_ai/views/api_metadata.py new file mode 100644 index 0000000..1a73b56 --- /dev/null +++ b/great_ai/src/great_ai/great_ai/views/api_metadata.py @@ -0,0 +1,7 @@ +from pydantic import BaseModel + + +class ApiMetadata(BaseModel): + name: str + version: str + documentation: str diff --git a/great_ai/src/great_ai/great_ai/views/function_metadata.py b/great_ai/src/great_ai/great_ai/views/function_metadata.py index b2f7782..97dc4b9 100644 --- a/great_ai/src/great_ai/great_ai/views/function_metadata.py +++ b/great_ai/src/great_ai/great_ai/views/function_metadata.py @@ -6,4 +6,5 @@ from pydantic import BaseModel class FunctionMetadata(BaseModel): input_parameter_names: List[str] = [] model_parameter_names: List[str] = [] - version: str = "" + model_versions: str = "" + is_finalised: bool = False diff --git a/great_ai/src/great_ai/open_s3/large_file.py b/great_ai/src/great_ai/open_s3/large_file.py index 0957798..271d994 100644 --- a/great_ai/src/great_ai/open_s3/large_file.py +++ b/great_ai/src/great_ai/open_s3/large_file.py @@ -278,18 +278,13 @@ class LargeFile: self._versions = sorted(versions, key=self._get_version_from_key) - if self._versions: - logger.info(f"Found versions: {self.version_ids}") - else: - logger.info("No versions found") - def _fetch_versions_from_cache(self) -> List[Path]: - logger.info(f"Fetching offline versions of {self._name}") + logger.debug(f"Fetching offline versions of {self._name}") return list(self.cache_path.glob(f"{self._local_name}-*")) def _fetch_versions_from_s3(self) -> List[str]: - logger.info(f"Fetching online versions of {self._name}") + logger.debug(f"Fetching online versions of {self._name}") found_objects = self._client.list_objects_v2( Bucket=self.bucket_name, Prefix=self._name @@ -318,11 +313,15 @@ class LargeFile: if self._version is None: self._version = self.version_ids[-1] - logger.info(f"Latest version of {self._name} is {self._version}") + logger.info( + f"Latest version of {self._name} is {self._version} " + + f"(from versions: {', '.join((str(v) for v in self.version_ids))})" + ) elif self._version not in self.version_ids: raise FileNotFoundError( - f"File {self._name} not found with version {self._version}. Available versions: {self.version_ids}" + f"File {self._name} not found with version {self._version}. " + + f"(from versions: {', '.join((str(v) for v in self.version_ids))})" ) else: raise ValueError("Unsupported file mode.") diff --git a/great_ai/src/great_ai/parse_arguments.py b/great_ai/src/great_ai/parse_arguments.py index c68f8b2..1ca111c 100644 --- a/great_ai/src/great_ai/parse_arguments.py +++ b/great_ai/src/great_ai/parse_arguments.py @@ -12,14 +12,6 @@ def parse_arguments() -> Namespace: help="the name of the file containing your to-be-served function such as `main.py`\n", ) - parser.add_argument( - "--function_name", - type=str, - help="name of your inference function, defaults to the base name of the filename or the literal `app`", - default="", - required=False, - ) - parser.add_argument( "--host", type=str,