andras/great-ai
andras/great-ai
Watch
1
0
Fork
You've already forked great-ai
 0
Code Issues Pull requests 2 Projects Releases 4 Packages 1 Wiki Activity Actions

Integrate MongoDB

Browse source
This commit is contained in:
Andras Schmelczer 2022-06-25 13:01:44 +02:00
commit 79ddc5c7df
No known key found for this signature in database
GPG key ID: 0EA1BC97D0AB076E
30 changed files with 1003 additions and 748 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.vscode/settings.json vendored
View file

@ -13,6 +13,7 @@
"iloc",
"initialisation",
"initialised",
"initialising",
"inplace",
"ipynb",
"joblib",

80
examples/simple/data.ipynb
View file

@ -11,15 +11,6 @@
"> The blue boxes show the steps implemented in this notebook."
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"MAX_CHUNK_COUNT = 4"
]
},
{
"cell_type": "markdown",
"metadata": {},
@ -31,6 +22,15 @@
"In this case, we download the semantic scholar dataset from a public S3 bucket."
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"MAX_CHUNK_COUNT = 1"
]
},
{
"cell_type": "code",
"execution_count": 2,
@ -39,7 +39,7 @@
{
"data": {
"text/plain": [
"'Processing 4 out of the 6002 available chunks'"
"'Processing 1 out of the 6002 available chunks'"
]
},
"execution_count": 2,
@ -49,6 +49,7 @@
],
"source": [
"import urllib.request\n",
"from random import shuffle\n",
"\n",
"manifest = (\n",
" urllib.request.urlopen(\n",
@ -58,7 +59,9 @@
" .decode()\n",
") # a list of available chunks separated by '\\n' characters\n",
"\n",
"chunks = manifest.split()[:MAX_CHUNK_COUNT]\n",
"lines = manifest.split()\n",
"shuffle(lines)\n",
"chunks = lines[:MAX_CHUNK_COUNT]\n",
"\n",
"f\"Processing {len(chunks)} out of the {len(manifest.split())} available chunks\""
]
@ -83,23 +86,11 @@
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[38;5;226m2022-06-19 14:59:12,562 | WARNING | Limiting concurrency to 4 because there are only 4 chunks\u001b[0m\n",
"\u001b[38;5;39m2022-06-19 14:59:12,563 | INFO | Starting parallel map (concurrency: 4, chunk size: 1)\u001b[0m\n"
"\u001b[38;5;226m2022-06-25 11:20:01,955 | WARNING | Limiting concurrency to 1 because there are only 1 chunks\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:20:01,956 | INFO | Starting parallel map (concurrency: 1, chunk size: 1)\u001b[0m\n",
"\u001b[38;5;226m2022-06-25 11:20:01,956 | WARNING | Running in series, there is no reason for parallelism\u001b[0m\n",
"100%|██████████| 1/1 [04:02<00:00, 242.61s/it]\n"
]
},
{
"data": {
"application/vnd.jupyter.widget-view+json": {
"model_id": "ff8fc113515944cfa75127f4aba3d491",
"version_major": 2,
"version_minor": 0
},
"text/plain": [
" 0%| | 0/4 [00:00<?, ?it/s]"
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
@ -121,15 +112,13 @@
"\n",
" # Transform\n",
" return [\n",
" # Create pairs of `(text, [...domains])`\n",
" # The text is cleaned to remove PDF extraction, web scraping, and other common artifacts\n",
" (\n",
" clean(\n",
" f'{c[\"title\"]} {c[\"paperAbstract\"]} {c[\"journalName\"]} {c[\"venue\"]}',\n",
" convert_to_ascii=True,\n",
" ),\n",
" ), # The text is cleaned to remove PDF extraction, web scraping, and other common artifacts\n",
" c[\"fieldsOfStudy\"],\n",
" )\n",
" ) # Create pairs of `(text, [...domains])`\n",
" for c in chunk\n",
" if c[\"fieldsOfStudy\"] and is_english(predict_language(c[\"paperAbstract\"]))\n",
" ]\n",
@ -160,16 +149,15 @@
"\n",
"Upload the dataset (or a part of it) to a central repository using `great_ai.add_ground_truth`. This step automatically tags each datapoint with a split label according to the ratios we set. Additional tags can be also given.\n",
"\n",
"#### Use a different repository\n",
"#### Production-ready backend\n",
"\n",
"For the sake of simplicity, the tutorial uses the local hard drive (`great_ai.ParallelTinyDbDriver`) as the central repository.\n",
"This can be simply changed, for example, by the following snippet:\n",
"The MongoDB driver is automatically configured if `mongo.ini` exists with the following scheme:\n",
"\n",
"```python\n",
"from great_ai import configure, MongoDbDriver\n",
"\n",
"configure(tracing_database=MongoDbDriver('mongodb://localhost:27017_or_something_like_that'))\n",
"```"
"```ini\n",
"mongo_connection_string=mongodb://localhost:27017/\n",
"mongo_database=my_great_ai_db\n",
"```\n",
"> You can install MongoDB from [here](https://www.mongodb.com/docs/manual/installation) or [use it as a service](https://www.mongodb.com/cloud/atlas/register)\n"
]
},
{
@ -181,9 +169,17 @@
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[38;5;226m2022-06-19 15:03:30,300 | WARNING | Environment variable ENVIRONMENT is not set, defaulting to development mode ‼️\u001b[0m\n",
"\u001b[38;5;226m2022-06-19 15:03:30,301 | WARNING | The selected persistence driver (ParallelTinyDbDriver) is not recommended for production\u001b[0m\n",
"\u001b[38;5;39m2022-06-19 15:03:30,301 | INFO | Options: configured ✅\u001b[0m\n"
"\u001b[38;5;226m2022-06-25 11:24:04,668 | WARNING | Environment variable ENVIRONMENT is not set, defaulting to development mode ‼️\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,669 | INFO | Found credentials file (/data/projects/great-ai/examples/simple/mongo.ini), initialising MongodbDriver\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,670 | INFO | Found credentials file (/data/projects/great-ai/examples/simple/mongo.ini), initialising LargeFileMongo\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,671 | INFO | Settings: configured ✅\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,672 | INFO | 🔩 tracing_database: MongodbDriver\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,672 | INFO | 🔩 large_file_implementation: LargeFileMongo\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,673 | INFO | 🔩 is_production: False\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,673 | INFO | 🔩 should_log_exception_stack: True\u001b[0m\n",
"\u001b[38;5;39m2022-06-25 11:24:04,674 | INFO | 🔩 prediction_cache_size: 512\u001b[0m\n",
"\u001b[38;5;226m2022-06-25 11:24:04,674 | WARNING | You still need to check whether you follow all best practices before trusting your deployment.\u001b[0m\n",
"\u001b[38;5;226m2022-06-25 11:24:04,674 | WARNING | > Find out more at https://se-ml.github.io/practices/\u001b[0m\n"
]
}
],

151
examples/simple/deploy.ipynb
View file

File diff suppressed because one or more lines are too long

BIN
examples/simple/diagrams/ss-confusion.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 160 KiB

BIN
examples/simple/diagrams/ss-distribution.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 64 KiB

5
examples/simple/mongo.ini
View file

@ -1,3 +1,2 @@
[DEFAULT]
connection_string=mongodb://localhost:27017/
database=large_file_tests
mongo_connection_string=mongodb://localhost:27017/ # change this
mongo_database=great_ai_db2 # this will be automatically created

928
examples/simple/train.ipynb
View file

File diff suppressed because it is too large Load diff

1
great_ai/example_secrets.ini
View file

@ -1,4 +1,3 @@
[DEFAULT]
aws_region_name = your_region_like_eu-west-2
aws_access_key_id = YOUR_ACCESS_KEY_ID
aws_secret_access_key = YOUR_VERY_SECRET_ACCESS_KEY

1
great_ai/open_s3.md
View file

@ -78,7 +78,6 @@ The package can be used as a module from the command-line to give you more flexi
Create an .ini file (or use _~/.aws/credentials_). It may look like this:
```ini
[DEFAULT]
aws_region_name = your_region_like_eu-west-2
aws_access_key_id = YOUR_ACCESS_KEY_ID
aws_secret_access_key = YOUR_VERY_SECRET_ACCESS_KEY

1
great_ai/setup.cfg
View file

@ -20,7 +20,6 @@ packages = find:
include_package_data = True
python_requires = >=3.8
install_requires =
click < 8.1.0
unidecode >= 1.3.0
multiprocess >= 0.70.0.0
tqdm >= 4.0.0

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

@ -19,7 +19,7 @@ from .great_ai.context 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
from .utilities import get_logger
logger = get_logger(SERVER_NAME)

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

@ -7,5 +7,5 @@ from .output_models import (
RegressionOutput,
)
from .parameters import log_metric, parameter
from .persistence import MongoDbDriver, ParallelTinyDbDriver, TracingDatabaseDriver
from .tracing import add_ground_truth, query_ground_truth
from .persistence import MongodbDriver, ParallelTinyDbDriver, TracingDatabaseDriver
from .tracing import add_ground_truth, delete_ground_truth, query_ground_truth

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

@ -1,16 +1,18 @@
from pathlib import Path
from great_ai.large_file import LargeFileLocal, LargeFileMongo, LargeFileS3
from ..large_file import LargeFileMongo, LargeFileS3
from .persistence.mongodb_driver import MongodbDriver
ENV_VAR_KEY = "ENVIRONMENT"
PRODUCTION_KEY = "production"
DEFAULT_TRACING_DB_FILENAME = "tracing_database.json"
DASHBOARD_PATH = "/dashboard"
MONGO_CONFIG_PATHS = ["mongodb.ini", "mongo.ini", "mongo_db.ini", "mongo-db.ini"]
DEFAULT_TRACING_DATABASE_CONFIG_PATHS = {
MongodbDriver: MONGO_CONFIG_PATHS,
}
DEFAULT_LARGE_FILE_CONFIG_PATHS = {
LargeFileLocal: None,
LargeFileMongo: Path("mongodb.ini"),
LargeFileS3: Path("s3.ini"),
LargeFileS3: ["s3.ini", "b2.ini"],
LargeFileMongo: MONGO_CONFIG_PATHS,
}
GITHUB_LINK = "https://github.com/ScoutinScience/great-ai"
@ -25,4 +27,4 @@ ONLINE_TAG_NAME = "online"
SERVER_NAME = "GreatAI-Server"
SE4ML_WEBSITE = 'https://se-ml.github.io/practices/'
SE4ML_WEBSITE = "https://se-ml.github.io/practices/"

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

@ -7,11 +7,11 @@ from typing import Any, Dict, Optional, Type, cast
from pydantic import BaseModel
from great_ai.large_file import LargeFile, LargeFileLocal
from great_ai.utilities.logger import get_logger
import yaml
from great_ai.utilities import get_logger
from .constants import (
DEFAULT_LARGE_FILE_CONFIG_PATHS,
DEFAULT_TRACING_DB_FILENAME,
DEFAULT_TRACING_DATABASE_CONFIG_PATHS,
ENV_VAR_KEY,
PRODUCTION_KEY,
SE4ML_WEBSITE,
@ -26,6 +26,7 @@ class Context(BaseModel):
logger: Logger
should_log_exception_stack: bool
prediction_cache_size: int
dashboard_table_size: int
class Config:
arbitrary_types_allowed = True
@ -37,6 +38,7 @@ class Context(BaseModel):
"is_production": self.is_production,
"should_log_exception_stack": self.should_log_exception_stack,
"prediction_cache_size": self.prediction_cache_size,
"dashboard_table_size": self.dashboard_table_size,
}
@ -54,13 +56,12 @@ def configure(
*,
log_level: int = DEBUG,
seed: int = 42,
tracing_database: TracingDatabaseDriver = ParallelTinyDbDriver(
Path(DEFAULT_TRACING_DB_FILENAME)
),
large_file_implementation: Type[LargeFile] = LargeFileLocal,
tracing_database: Optional[Type[TracingDatabaseDriver]] = None,
large_file_implementation: Optional[Type[LargeFile]] = None,
should_log_exception_stack: Optional[bool] = None,
prediction_cache_size: int = 512,
disable_se4ml_banner: bool=False
disable_se4ml_banner: bool = False,
dashboard_table_size: int = 50,
) -> None:
global _context
logger = get_logger("great_ai", level=log_level)
@ -72,9 +73,11 @@ def configure(
)
is_production = _is_in_production_mode(logger=logger)
_initialize_large_file(large_file_implementation, logger=logger)
_set_seed(seed)
tracing_database = _initialize_tracing_database(tracing_database, logger=logger)()
if not tracing_database.is_production_ready:
if is_production:
logger.error(
@ -87,23 +90,27 @@ def configure(
_context = Context(
tracing_database=tracing_database,
large_file_implementation=large_file_implementation,
large_file_implementation=_initialize_large_file(
large_file_implementation, logger=logger
),
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,
dashboard_table_size=dashboard_table_size,
)
logger.info("Setting: configured ✅")
logger.info("Settings: configured ✅")
for k, v in get_context().to_flat_dict().items():
logger.info(f'🔩 {k}: {v}')
logger.info(f"🔩 {k}: {v}")
if not is_production and not disable_se4ml_banner:
logger.warning(f'You still need to check whether you follow all best practices so that you and others can trust your deployment.')
logger.warning(f'> Find out more at {SE4ML_WEBSITE}')
logger.warning(
"You still need to check whether you follow all best practices before trusting your deployment."
)
logger.warning(f"> Find out more at {SE4ML_WEBSITE}")
def _is_in_production_mode(logger: Optional[Logger]) -> bool:
@ -129,24 +136,48 @@ def _is_in_production_mode(logger: Optional[Logger]) -> bool:
return is_production
def _initialize_large_file(large_file: Type[LargeFile], logger: Logger) -> None:
path = DEFAULT_LARGE_FILE_CONFIG_PATHS[large_file]
if path is None:
return
def _initialize_tracing_database(
selected: Optional[Type[TracingDatabaseDriver]], logger: Logger
) -> Type[TracingDatabaseDriver]:
for tracing_driver, paths in DEFAULT_TRACING_DATABASE_CONFIG_PATHS.items():
if selected is None or selected == tracing_driver:
if tracing_driver.initialized:
logger.warning(
f"{tracing_driver.__name__} has been already configured: skipping initialisation"
)
return tracing_driver
for p in paths:
if Path(p).exists():
logger.info(
f"Found credentials file ({Path(p).absolute()}), initialising {tracing_driver.__name__}"
)
tracing_driver.configure_credentials_from_file(p)
return tracing_driver
logger.warning(
"Cannot find credentials files, defaulting to using ParallelTinyDbDriver"
)
return ParallelTinyDbDriver
if large_file.initialized:
logger.warning(
f"{large_file.__name__} has been already configured: skipping initialisation"
)
return
if path.exists():
large_file.configure_credentials_from_file(path)
logger.info(f"{large_file.__name__} initialised with config ({path.resolve()})")
else:
logger.warning(
f"Default {large_file.__name__} config ({path.resolve()}) not found, skipping {large_file.__name__} initialisation"
)
def _initialize_large_file(
selected: Optional[Type[LargeFile]], logger: Logger
) -> Type[LargeFile]:
for large_file, paths in DEFAULT_LARGE_FILE_CONFIG_PATHS.items():
if selected is None or selected == large_file:
if large_file.initialized:
logger.warning(
f"{large_file.__name__} has been already configured: skipping initialisation"
)
return large_file
for p in paths:
if Path(p).exists():
logger.info(
f"Found credentials file ({Path(p).absolute()}), initialising {large_file.__name__}"
)
large_file.configure_credentials_from_file(p)
return large_file
logger.warning("Cannot find credentials files, defaulting to using LargeFileLocal")
return LargeFileLocal
def _set_seed(seed: int) -> None:

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

@ -13,13 +13,12 @@ from typing import (
cast,
)
import yaml
from fastapi import APIRouter, FastAPI, status
from pydantic import BaseModel, create_model
from great_ai.great_ai.deploy.routes.bootstrap_dashboard import bootstrap_dashboard
from great_ai.great_ai.views.cache_statistics import CacheStatistics
from great_ai.utilities.parallel_map import parallel_map
from great_ai.utilities import parallel_map
from ..constants import DASHBOARD_PATH
from ..context import get_context
@ -40,6 +39,7 @@ from .routes import (
T = TypeVar("T")
class GreatAI(Generic[T]):
def __init__(self, func: Callable[..., Any], version: str):
func = automatically_decorate_parameters(func)
@ -70,7 +70,6 @@ class GreatAI(Generic[T]):
redoc_url=None,
)
@staticmethod
def deploy(
func: Optional[Callable[..., T]] = None,

4
great_ai/src/great_ai/great_ai/deploy/routes/bootstrap_trace_endpoints.py
View file

@ -20,6 +20,10 @@ def bootstrap_trace_endpoints(app: FastAPI) -> None:
) -> List[Trace]:
return get_context().tracing_database.query(
conjunctive_filters=query.filter,
conjunctive_tags=query.conjunctive_tags,
since=query.since,
until=query.until,
has_feedback=query.has_feedback,
sort_by=query.sort,
skip=skip,
take=take,

203
great_ai/src/great_ai/great_ai/deploy/routes/dashboard/create_dash_app.py
View file

@ -1,5 +1,5 @@
from math import ceil
from typing import Any, Dict, List, Sequence, Tuple
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import pandas as pd
import plotly.express as px
@ -8,12 +8,12 @@ from dash import Dash, dcc, html
from dash.dependencies import Input, Output
from flask import Flask
from great_ai.utilities.unique import unique
from great_ai.utilities import unique
from ....constants import DASHBOARD_PATH, ONLINE_TAG_NAME
from ....context import get_context
from ....helper import snake_case_to_text, text_to_hex_color
from ....views import SortBy
from ....views import SortBy, Trace
from .get_description import get_description
from .get_filter_from_datatable import get_filter_from_datatable
from .get_footer import get_footer
@ -105,19 +105,32 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask:
@app.callback(
Output(table, "data"),
Output(table, "page_count"),
Output(table, "columns"),
Output(traces_table_container, "style"),
Output(execution_time_histogram_container, "children"),
Output(parallel_coordinates, "figure"),
Output(parallel_coordinates, "style"),
Input(table, "page_current"),
Input(table, "page_size"),
Input(table, "sort_by"),
Input(table, "filter_query"),
Input(interval, "n_intervals"),
)
def update_table(
def update_page(
page_current: int,
page_size: int,
sort_by: List[SortBy],
sort_by: List[Dict[str, Union[str, int]]],
filter_query: str,
n_intervals: int,
) -> Tuple[List[Dict[str, Any]], int]:
) -> Tuple[
List[Dict[str, Any]],
int,
List[Dict[str, Sequence[str]]],
Dict[str, Any],
Any,
go.Figure,
Dict[str, Any],
]:
conjunctive_filters = (
[get_filter_from_datatable(f) for f in filter_query.split(" && ")]
if filter_query
@ -128,114 +141,98 @@ def create_dash_app(function_name: str, function_docs: str) -> Flask:
elements, count = get_context().tracing_database.query(
skip=page_current * page_size,
take=page_size,
conjunctive_tags=[ONLINE_TAG_NAME],
conjunctive_filters=non_null_conjunctive_filters,
sort_by=sort_by,
conjunctive_tags=[ONLINE_TAG_NAME],
sort_by=[SortBy.parse_obj(s) for s in sort_by],
)
columns, style = update_layout(elements[0] if elements else None)
execution_time_histogram, parallel_coords_fig, style = update_charts(
elements=elements, function_name=function_name, accent_color=accent_color
)
return (
[e.to_flat_dict() for e in elements],
[e.to_flat_dict(include_original=False) for e in elements],
max(1, ceil(count / page_size)),
)
@app.callback(
Output(table, "columns"),
Output(traces_table_container, "style"),
Input(interval, "n_intervals"),
)
def update_layout(
n_intervals: int,
) -> Tuple[List[Dict[str, Sequence[str]]], Dict[str, Any]]:
elements, count = get_context().tracing_database.query(
take=1, conjunctive_tags=[ONLINE_TAG_NAME]
)
if elements:
keys = list(elements[0].to_flat_dict().keys())
header_height = max(len(i.split(":")) for i in keys)
columns = [
{
"name": [""] * (header_height - len(k.split(":")))
+ k.replace("_flat", "").split(":"),
"id": k,
}
for k in keys
]
else:
columns = []
return (
columns,
{"display": "block" if count > 0 else "none"},
style,
execution_time_histogram,
parallel_coords_fig,
style,
)
@app.callback(
Output(execution_time_histogram_container, "children"),
Output(parallel_coordinates, "figure"),
Output(parallel_coordinates, "style"),
Input(table, "filter_query"),
Input(interval, "n_intervals"),
)
def update_charts(
filter_query: str, n_intervals: int
) -> Tuple[Any, go.Figure, Dict[str, Any]]:
conjunctive_filters = (
[get_filter_from_datatable(f) for f in filter_query.split(" && ")]
if filter_query
else []
)
non_null_conjunctive_filters = [f for f in conjunctive_filters if f is not None]
elements, count = get_context().tracing_database.query(
conjunctive_tags=[ONLINE_TAG_NAME],
conjunctive_filters=non_null_conjunctive_filters,
)
if not elements:
return (
html.Span(
f"No traces yet: call your function ({function_name}) to create one.",
className="placeholder",
),
go.Figure(),
{"display": "none"},
)
flat_elements = [e.to_flat_dict() for e in elements]
execution_time_histogram = dcc.Graph(config={"displaylogo": False})
df = pd.DataFrame(flat_elements)
fig = px.histogram(
df,
x="original_execution_time_ms",
labels={"original_execution_time_ms": "Execution time (ms)"},
nbins=20,
height=400,
log_y=True,
color_discrete_sequence=[accent_color],
)
fig.update_layout(
margin=dict(l=0, r=0, b=0, t=0, pad=0),
)
execution_time_histogram.figure = fig
parallel_coords_fig = go.Figure(
go.Parcoords(
dimensions=[
get_dimension_descriptor(df, c)
for c in df.columns
if c
not in {"trace_id", "created", "output", "exception", "feedback"}
and "_flat" not in c
],
line_color=accent_color,
)
)
return execution_time_histogram, parallel_coords_fig, {}
return app.server
def update_layout(
first_element: Optional[Trace],
) -> Tuple[List[Dict[str, Sequence[str]]], Dict[str, Any]]:
if first_element:
keys = list(first_element.to_flat_dict(include_original=False).keys())
header_height = max(len(i.split(":")) for i in keys)
columns = [
{
"name": [""] * (header_height - len(k.split(":")))
+ k.replace("_flat", "").split(":"),
"id": k,
}
for k in keys
]
else:
columns = []
return (
columns,
{"display": "none" if first_element is None else "block"},
)
def update_charts(
elements: List[Trace], function_name: str, accent_color: str
) -> Tuple[Any, go.Figure, Dict[str, Any]]:
if not elements:
return (
html.Span(
f"No traces yet: call your function ({function_name}) to create one.",
className="placeholder",
),
go.Figure(),
{"display": "none"},
)
flat_elements = [e.to_flat_dict(include_original=False) for e in elements]
execution_time_histogram = dcc.Graph(config={"displaylogo": False})
df = pd.DataFrame(flat_elements)
fig = px.histogram(
df,
x="original_execution_time_ms",
labels={"original_execution_time_ms": "Execution time (ms)"},
nbins=20,
height=400,
log_y=True,
color_discrete_sequence=[accent_color],
)
fig.update_layout(
margin=dict(l=0, r=0, b=0, t=0, pad=0),
)
execution_time_histogram.figure = fig
parallel_coords_fig = go.Figure(
go.Parcoords(
dimensions=[
get_dimension_descriptor(df, c)
for c in df.columns
if c not in {"trace_id", "created", "output", "exception", "feedback"}
and "_flat" not in c
],
line_color=accent_color,
)
)
return execution_time_histogram, parallel_coords_fig, {}
def get_dimension_descriptor(df: pd.DataFrame, column: str) -> Dict[str, Any]:
dimension: Dict[str, Any] = {
"label": snake_case_to_text(column),

5
great_ai/src/great_ai/great_ai/deploy/routes/dashboard/get_traces_table.py
View file

@ -1,10 +1,12 @@
from dash import dash_table
from great_ai.great_ai.context import get_context
def get_traces_table() -> dash_table.DataTable:
return dash_table.DataTable(
page_current=0,
page_size=20,
page_size=get_context().dashboard_table_size,
page_action="custom",
filter_action="custom",
sort_action="custom",
@ -24,7 +26,6 @@ def get_traces_table() -> dash_table.DataTable:
"background-color": "white",
"font-weight": "bold",
},
style_table={"max-height": "70vh", "overflow": "auto"},
merge_duplicate_headers=True,
style_cell_conditional=[
{"if": {"column_id": "output"}, "width": 1500},

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

@ -1,3 +1,3 @@
from .mongodb_driver import MongoDbDriver
from .mongodb_driver import MongodbDriver
from .parallel_tinydb_driver import ParallelTinyDbDriver
from .tracing_database_driver import TracingDatabaseDriver

134
great_ai/src/great_ai/great_ai/persistence/mongodb_driver.py
View file

@ -1,5 +1,137 @@
from datetime import datetime
from typing import Any, List, Mapping, Optional, Sequence, Tuple
from pymongo import MongoClient
from ..views import Filter, SortBy, Trace
from .tracing_database_driver import TracingDatabaseDriver
operator_mapping = {
"=": "$eq",
"!=": "$ne",
"<": "$lt",
"<=": "$lte",
">": "$gt",
">=": "$gte",
"contains": "$regex",
}
class MongoDbDriver(TracingDatabaseDriver):
class MongodbDriver(TracingDatabaseDriver):
is_production_ready = True
def __init__(self) -> None:
super().__init__()
if self.mongo_connection_string is None or self.mongo_database is None:
raise ValueError(
"Please configure the MongoDB access options by calling MongodbDriver.configure_credentials"
)
@classmethod
def configure_credentials( # type: ignore
cls,
*,
mongo_connection_string: str,
mongo_database: str,
**_: Mapping[str, Any],
) -> None:
cls.mongo_connection_string = mongo_connection_string
cls.mongo_database = mongo_database
super().configure_credentials()
def save(self, trace: Trace) -> str:
serialized = trace.to_flat_dict()
serialized["_id"] = trace.trace_id
with MongoClient(self.mongo_connection_string) as client:
return client[self.mongo_database].traces.insert_one(serialized)
def save_batch(self, documents: List[Trace]) -> List[str]:
serialized = [d.to_flat_dict() for d in documents]
for s in serialized:
s["_id"] = s["trace_id"]
with MongoClient(self.mongo_connection_string) as client:
return client[self.mongo_database].traces.insert_many(
serialized, ordered=False
)
def get(self, id: str) -> Optional[Trace]:
with MongoClient(self.mongo_connection_string) as client:
value = client[self.mongo_database].traces.find_one(id)
if value:
value = Trace.parse_obj(value)
return value
def _get_operator(self, filter: Filter) -> str:
if filter.operator == "contains" and not isinstance(filter.value, str):
return operator_mapping["="]
return operator_mapping[filter.operator]
def query(
self,
*,
skip: int = 0,
take: Optional[int] = None,
conjunctive_filters: Sequence[Filter] = [],
conjunctive_tags: Sequence[str] = [],
since: Optional[datetime] = None,
until: Optional[datetime] = None,
has_feedback: Optional[bool] = None,
sort_by: Sequence[SortBy] = [],
) -> Tuple[List[Trace], int]:
query = {
"filter": {
"$and": [{"tags": tag} for tag in conjunctive_tags]
+ [
{f.property: {self._get_operator(f): f.value}}
for f in conjunctive_filters
]
+ [{}]
},
"sort": [
(col.column_id, 1 if col.direction == "asc" else -1) for col in sort_by
],
}
if skip:
query["skip"] = skip
if take:
query["limit"] = take
if since:
query["filter"]["$and"].append({"created": {"$gte": since}})
if until:
query["filter"]["$and"].append({"created": {"$lte": until}})
if has_feedback is not None:
query["filter"]["$and"].append(
{"feedback": {"$ne": None}} if has_feedback else {"feedback": None}
)
with MongoClient(self.mongo_connection_string) as client:
values = client[self.mongo_database].traces.find(**query)
documents = [Trace.parse_obj(t) for t in values]
return documents, len(documents)
def update(self, id: str, new_version: Trace) -> None:
serialized = new_version.dict()
serialized["_id"] = new_version.trace_id
with MongoClient(self.mongo_connection_string) as client:
client[self.mongo_database].traces.update_one(id, new_version)
def delete(self, id: str) -> None:
with MongoClient(self.mongo_connection_string) as client:
client[self.mongo_database].traces.delete_one(id)
def delete_batch(self, ids: List[str]) -> List[str]:
delete_filter = {"_id": {"$in": ids}}
with MongoClient(self.mongo_connection_string) as client:
return client[self.mongo_database].traces.delete_many(delete_filter)

Some files were not shown because too many files have changed in this diff Show more