Hardening and deployment¶
Part 3: Create production inference function
The blue boxes show the steps implemented in this notebook.
In Part 2, we trained our AI model. Now, it's time to create General Robust End-to-end Automated Trustworthy deployment.
Create the inference function¶
Next to the prediction, we also return the top-n most influential words based on their weights.
Local setup¶
Use Python 3.12 or 3.13 and install this checkout with uv pip install --editable '.[notebooks]'.
See Run notebooks locally in the repository README for the kernel, execution order,
and full-data options. Run related notebooks from the same working directory
to share the default .cache/ model store and tracing_database.json database.
In [1]:
Copied!
import re
import numpy as np
from sklearn.pipeline import Pipeline
from great_ai.utilities import clean
from great_ai import (
MultiLabelClassificationOutput,
ClassificationOutput,
GreatAI,
use_model,
parameter,
)
@GreatAI.create
@use_model("small-domain-prediction", version="latest")
@parameter("target_confidence", validate=lambda c: 0 <= c <= 100)
def predict_domain(
text: str, model: Pipeline, target_confidence: int = 50
) -> MultiLabelClassificationOutput:
"""
Predict the scientific domain of the input text.
Return labels until their sum likelihood is larger than `target_confidence`.
"""
preprocessed = re.sub(r"[^a-zA-Z\s]", "", clean(text, convert_to_ascii=True))
features = model.named_steps["vectorizer"].transform([preprocessed])
prediction = model.named_steps["classifier"].predict_proba(features)[0]
best_classes = sorted(enumerate(prediction), key=lambda v: v[1], reverse=True)
results = MultiLabelClassificationOutput()
for class_index, probability in best_classes:
results.labels.append(
get_label(
model=model,
features=features,
class_index=class_index,
probability=probability,
)
)
if sum(r.confidence for r in results.labels) >= target_confidence:
break
return results
def get_label(
model: Pipeline, features: np.ndarray, class_index: int, probability: float
) -> ClassificationOutput:
return ClassificationOutput(
label=model.named_steps["classifier"].classes_[class_index],
confidence=round(probability * 100),
explanation=[
word
for _, word in sorted(
(
(weight, word)
for weight, word, count in zip(
model.named_steps["classifier"].feature_log_prob_[class_index],
model.named_steps["vectorizer"].get_feature_names_out(),
features.toarray()[0],
)
if count > 0
),
reverse=True,
)
][:5],
)
import re
import numpy as np
from sklearn.pipeline import Pipeline
from great_ai.utilities import clean
from great_ai import (
MultiLabelClassificationOutput,
ClassificationOutput,
GreatAI,
use_model,
parameter,
)
@GreatAI.create
@use_model("small-domain-prediction", version="latest")
@parameter("target_confidence", validate=lambda c: 0 <= c <= 100)
def predict_domain(
text: str, model: Pipeline, target_confidence: int = 50
) -> MultiLabelClassificationOutput:
"""
Predict the scientific domain of the input text.
Return labels until their sum likelihood is larger than `target_confidence`.
"""
preprocessed = re.sub(r"[^a-zA-Z\s]", "", clean(text, convert_to_ascii=True))
features = model.named_steps["vectorizer"].transform([preprocessed])
prediction = model.named_steps["classifier"].predict_proba(features)[0]
best_classes = sorted(enumerate(prediction), key=lambda v: v[1], reverse=True)
results = MultiLabelClassificationOutput()
for class_index, probability in best_classes:
results.labels.append(
get_label(
model=model,
features=features,
class_index=class_index,
probability=probability,
)
)
if sum(r.confidence for r in results.labels) >= target_confidence:
break
return results
def get_label(
model: Pipeline, features: np.ndarray, class_index: int, probability: float
) -> ClassificationOutput:
return ClassificationOutput(
label=model.named_steps["classifier"].classes_[class_index],
confidence=round(probability * 100),
explanation=[
word
for _, word in sorted(
(
(weight, word)
for weight, word, count in zip(
model.named_steps["classifier"].feature_log_prob_[class_index],
model.named_steps["vectorizer"].get_feature_names_out(),
features.toarray()[0],
)
if count > 0
),
reverse=True,
)
][:5],
)
Environment variable ENVIRONMENT is not set, defaulting to development mode ‼️
Cannot find credentials files, defaulting to using ParallelTinyDbDriver
The selected tracing database (ParallelTinyDbDriver) is not recommended for production
Cannot find credentials files, defaulting to using LargeFileLocal
GreatAI (v0.1.11): configured ✅
🔩 tracing_database: ParallelTinyDbDriver
🔩 large_file_implementation: LargeFileLocal
🔩 is_production: False
🔩 should_log_exception_stack: True
🔩 prediction_cache_size: 512
🔩 dashboard_table_size: 50
You still need to check whether you follow all best practices before trusting your deployment.
> Find out more at https://se-ml.github.io/practices
Fetching cached versions of small-domain-prediction
Latest version of small-domain-prediction is 0 (from versions: 0)
File small-domain-prediction-0 found in cache
File small-domain-prediction-0 found in cache
Check accuracy on the test split¶
Anything under if __name__ == "__main__": will not be run when the script is executed by the great-ai CLI app. This, combined with query_ground_truth and the /traces/{trace_id}/feedback endpoint are ideal for creating a continuous-integration job for checking the quality of the model before deployment.
We keep each paper's full set of accepted labels fixed. Top-1 hit rate is the fraction of papers whose highest-ranked returned label belongs to that set (an empty prediction is a miss). It is an application-specific hit criterion, not ordinary multiclass accuracy.
We also report per-class and micro, macro, weighted and sample-averaged multilabel precision, recall and F1 for all returned labels, using the default target_confidence=50. The label vocabulary comes from the train/test annotations, independently of predictions. Each class's support is the number of test papers annotated with that class. A lower confidence target can return fewer labels and trade recall for precision. zero_division=0 assigns zero to undefined metrics.
The earlier report and confusion matrix used prediction-dependent targets. Those outputs have been removed; rerun this cell to generate the corrected report for your model.
In [2]:
Copied!
if __name__ == "__main__":
from great_ai import query_ground_truth
from sklearn import metrics
from sklearn.preprocessing import MultiLabelBinarizer
data = query_ground_truth(["test", "local-domain-example"])
if not data:
raise ValueError("No test samples are available for evaluation")
X = [d.input for d in data]
y_actual = [d.feedback for d in data]
# Retain the complete returned label set, in descending confidence order.
y_predicted = [
[label.label for label in d.output.labels]
for d in predict_domain.process_batch(
X, do_not_persist_traces=True, concurrency=2
)
]
top_1_hit_rate = sum(
bool(predicted) and predicted[0] in actual
for predicted, actual in zip(y_predicted, y_actual)
) / len(y_actual)
print(f"Top-1 hit rate (any accepted label): {top_1_hit_rate:.3f}")
# Fix the label vocabulary from annotations, including classes absent in test.
label_binarizer = MultiLabelBinarizer()
label_binarizer.fit(
[d.feedback for d in query_ground_truth(["train", "local-domain-example"])]
+ y_actual
)
unknown_labels = {label for labels in y_predicted for label in labels}.difference(
label_binarizer.classes_
)
if unknown_labels:
raise ValueError(f"Predictions contain unrecognised labels: {unknown_labels}")
y_true_multilabel = label_binarizer.transform(y_actual)
y_pred_multilabel = label_binarizer.transform(y_predicted)
print(
metrics.classification_report(
y_true_multilabel,
y_pred_multilabel,
target_names=label_binarizer.classes_,
zero_division=0,
)
)
if __name__ == "__main__":
from great_ai import query_ground_truth
from sklearn import metrics
from sklearn.preprocessing import MultiLabelBinarizer
data = query_ground_truth(["test", "local-domain-example"])
if not data:
raise ValueError("No test samples are available for evaluation")
X = [d.input for d in data]
y_actual = [d.feedback for d in data]
# Retain the complete returned label set, in descending confidence order.
y_predicted = [
[label.label for label in d.output.labels]
for d in predict_domain.process_batch(
X, do_not_persist_traces=True, concurrency=2
)
]
top_1_hit_rate = sum(
bool(predicted) and predicted[0] in actual
for predicted, actual in zip(y_predicted, y_actual)
) / len(y_actual)
print(f"Top-1 hit rate (any accepted label): {top_1_hit_rate:.3f}")
# Fix the label vocabulary from annotations, including classes absent in test.
label_binarizer = MultiLabelBinarizer()
label_binarizer.fit(
[d.feedback for d in query_ground_truth(["train", "local-domain-example"])]
+ y_actual
)
unknown_labels = {label for labels in y_predicted for label in labels}.difference(
label_binarizer.classes_
)
if unknown_labels:
raise ValueError(f"Predictions contain unrecognised labels: {unknown_labels}")
y_true_multilabel = label_binarizer.transform(y_actual)
y_pred_multilabel = label_binarizer.transform(y_predicted)
print(
metrics.classification_report(
y_true_multilabel,
y_pred_multilabel,
target_names=label_binarizer.classes_,
zero_division=0,
)
)
0%| | 0/84 [00:00<?, ?it/s]
1%| | 1/84 [00:00<00:21, 3.81it/s]
13%|█▎ | 11/84 [00:00<00:01, 36.69it/s]
25%|██▌ | 21/84 [00:00<00:01, 54.23it/s]
37%|███▋ | 31/84 [00:00<00:00, 63.01it/s]
49%|████▉ | 41/84 [00:00<00:00, 69.13it/s]
61%|██████ | 51/84 [00:00<00:00, 73.60it/s]
85%|████████▍ | 71/84 [00:01<00:00, 94.70it/s]
96%|█████████▋| 81/84 [00:01<00:00, 91.80it/s]
100%|██████████| 84/84 [00:01<00:00, 73.28it/s]
Top-1 hit rate (any accepted label): 0.405
precision recall f1-score support
business 0.17 0.78 0.28 9
economics 0.22 0.82 0.35 11
geography 0.38 0.93 0.54 14
medicine 0.38 0.86 0.52 14
politics 0.14 0.36 0.21 11
psychology 0.28 0.69 0.40 16
sociology 0.17 0.67 0.27 9
micro avg 0.25 0.74 0.37 84
macro avg 0.25 0.73 0.37 84
weighted avg 0.26 0.74 0.38 84
samples avg 0.27 0.74 0.39 84
In [3]:
Copied!
if __name__ == "__main__":
from fastapi.testclient import TestClient
with TestClient(predict_domain.app) as client:
response = client.post(
"/predict", json={"text": "Neural networks learn useful features."}
)
response.raise_for_status()
assert response.json()["output"]["labels"]
print(response.json()["output"])
if __name__ == "__main__":
from fastapi.testclient import TestClient
with TestClient(predict_domain.app) as client:
response = client.post(
"/predict", json={"text": "Neural networks learn useful features."}
)
response.raise_for_status()
assert response.json()["output"]["labels"]
print(response.json()["output"])
{'labels': [{'label': 'psychology', 'confidence': 17.0, 'explanation': ['neural', 'networks', 'features']}, {'label': 'medicine', 'confidence': 16.0, 'explanation': ['features', 'neural', 'networks']}, {'label': 'economics', 'confidence': 16.0, 'explanation': ['networks', 'neural', 'features']}, {'label': 'geography', 'confidence': 15.0, 'explanation': ['features', 'neural', 'networks']}]}