andras/great-ai
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases 4 Packages Wiki Activity Actions

Add new features

Browse source
This commit is contained in:
Andras Schmelczer 2022-05-28 15:15:34 +02:00
parent 2d8e1c1758
commit 2c3f19e67c
24 changed files with 485 additions and 214 deletions
Download patch file Download diff file Expand all files Collapse all files

208
great_ai/src/great_ai/__main__.py
View file

@ -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)

1
great_ai/src/great_ai/great_ai/__init__.py
View file

@ -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

1
great_ai/src/great_ai/great_ai/constants.py
View file

@ -1,3 +1,4 @@
ENV_VAR_KEY = "ENVIRONMENT"
PRODUCTION_KEY = "production"
DEFAULT_TRACING_DB_FILENAME = "tracing_database.json"
METRICS_PATH = "/metrics"

44
great_ai/src/great_ai/great_ai/context/configure.py
View file

@ -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

4
great_ai/src/great_ai/great_ai/context/context.py
View file

@ -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

50
great_ai/src/great_ai/great_ai/dashboard/create_dash_app.py
View file

@ -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

244
great_ai/src/great_ai/great_ai/deploy/great_ai.py
View file

@ -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)

4
great_ai/src/great_ai/great_ai/helper/__init__.py
View file

@ -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

15
great_ai/src/great_ai/great_ai/helper/assert_function_is_not_finalised.py Normal file
View file

@ -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)

25
great_ai/src/great_ai/great_ai/helper/freeze_arguments.py Normal file
View file

@ -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

13
great_ai/src/great_ai/great_ai/helper/get_args.py → great_ai/src/great_ai/great_ai/helper/get_arguments.py
View file

@ -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}

20
great_ai/src/great_ai/great_ai/models/use_model.py
View file

@ -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

10
great_ai/src/great_ai/great_ai/parameters/log_metric.py
View file

@ -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}")

9
great_ai/src/great_ai/great_ai/parameters/parameter.py
View file

@ -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)

2
great_ai/src/great_ai/great_ai/persistence/parallel_tinydb_driver.py
View file

@ -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},

2
great_ai/src/great_ai/great_ai/persistence/persistence_driver.py
View file

@ -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

6
great_ai/src/great_ai/great_ai/tracing/tracing_context.py
View file

@ -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)

1
great_ai/src/great_ai/great_ai/views/__init__.py
View file

@ -1,3 +1,4 @@
from .api_metadata import ApiMetadata
from .evaluation_feedback_request import EvaluationFeedbackRequest
from .filter import Filter
from .function_metadata import FunctionMetadata

7
great_ai/src/great_ai/great_ai/views/api_metadata.py Normal file
View file

@ -0,0 +1,7 @@
from pydantic import BaseModel
class ApiMetadata(BaseModel):
name: str
version: str
documentation: str

3
great_ai/src/great_ai/great_ai/views/function_metadata.py
View file

@ -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

17
great_ai/src/great_ai/open_s3/large_file.py
View file

@ -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.")

8
great_ai/src/great_ai/parse_arguments.py
View file

@ -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,