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

Add exception logging

Browse source
This commit is contained in:
Andras Schmelczer 2022-05-26 21:00:14 +02:00
parent f756df2e50
commit 5c92a12b4a
No known key found for this signature in database
GPG key ID: 39260B5B0614A13E
20 changed files with 144 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

0
examples/simple/helpers.py → examples/complex/helpers.py
View file

7
examples/complex/main_service.py
View file

@ -1,7 +0,0 @@
from great_ai import configure, create_service
configure(development_mode_override=True)
from predict_domain import predict_domain
app = create_service(predict_domain)

71
examples/complex/predict_domain.py
View file

@ -1,39 +1,26 @@
import re
from typing import Dict, Iterable, List
from great_ai import log_argument, log_metric, use_model
from great_ai.utilities.clean import clean
from pydantic import BaseModel
from great_ai import GreatAI, use_model, ClassificationOutput
from sklearn.pipeline import Pipeline
from preprocess import preprocess
class DomainPrediction(BaseModel):
domain: str
probability: float
explanation: List[str]
from helpers import lemmatize, preprocess
@GreatAI.deploy
@use_model("small-domain-prediction-v2", version="latest")
@log_argument("text", validator=lambda t: len(t) > 0)
def predict_domain(
text: str, model: Pipeline, cut_off_probability: float = 0.2
) -> List[DomainPrediction]:
assert 0 <= cut_off_probability <= 1
def predict_domain(text: str, model: Pipeline, target_confidence: float = 20) -> List[ClassificationOutput]:
"""
Predict the scientific domain of the input text.
Return labels until their sum likelihood is larger than cut_off_probability.
Return labels until their sum likelihood is larger than target_confidence.
"""
log_metric("text_length", len(text))
assert 0 <= target_confidence <= 100, "invalid argument"
cleaned = clean(text, convert_to_ascii=True)
text = re.sub(r"[^a-zA-Z0-9]", " ", cleaned)
text = preprocess(text)
feature_names = model.named_steps["vectorizer"].get_feature_names_out()
token_mapping = {preprocess(original): original for original in text.split(" ")}
token_mapping = {lemmatize(original): original for original in text.split(" ")}
feature_names = [
token_mapping.get(name)
for name in model.named_steps["vectorizer"].get_feature_names_out()
]
features = model.named_steps["vectorizer"].transform(
[" ".join(token_mapping.keys())]
@ -41,42 +28,38 @@ def predict_domain(
prediction = model.named_steps["classifier"].predict_proba(features)[0]
best_classes = sorted(enumerate(prediction), key=lambda v: v[1], reverse=True)
results: List[DomainPrediction] = []
results: List[ClassificationOutput] = []
for class_index, probability in best_classes:
weights = model.named_steps["classifier"].feature_log_prob_[class_index]
domain = model.named_steps["classifier"].classes_[class_index]
results.append(
DomainPrediction(
domain=domain,
probability=round(probability * 100),
ClassificationOutput(
label=domain,
confidence=round(probability * 100),
explanation=_get_explanation(
feature_names=feature_names,
features=features.A[0],
weights=weights,
token_mapping=token_mapping,
counts=features.A[0],
words=feature_names,
),
)
)
if sum(r.probability for r in results) >= cut_off_probability * 100:
if sum(r.confidence for r in results) >= target_confidence:
break
return results
def _get_explanation(
feature_names: Iterable[str],
features: Iterable[float],
weights: Iterable[float],
token_mapping: Dict[str, str],
counts: Iterable[float],
words: Iterable[str],
) -> List[str]:
influential = [
(weight, name)
for weight, value, name in zip(weights, features, feature_names)
if value
]
most_influential = sorted((
(weight, word)
for weight, count, word in zip(weights, counts, words)
if count > 0
), reverse=True)[:5]
most_influential = sorted(influential, reverse=True)[:5]
return [token_mapping[name] for _, name in most_influential]
return [word for _, word in most_influential]

8
examples/complex/preprocess.py
View file

@ -1,8 +0,0 @@
import re
from great_ai.utilities.lemmatize_text import lemmatize_text
def preprocess(text: str) -> str:
lemmas = [re.sub(r"\d[\d.,]*", "NUM", lemma) for lemma in lemmatize_text(text)]
return " ".join(lemmas)

0
examples/simple/train.ipynb → examples/complex/train.ipynb
View file

12
examples/simple/README.md
View file

@ -1,12 +0,0 @@
# Train Domain classifier on the [semantic scholar dataset](https://api.semanticscholar.org/corpus)
## Upload the dataset (or a part of it) to shared infrastructure
```sh
mkdir ss-data && cd ss-data
wget https://s3-us-west-2.amazonaws.com/ai2-s2-research-public/open-corpus/2022-02-01/manifest.txt
wget -B https://s3-us-west-2.amazonaws.com/ai2-s2-research-public/open-corpus/2022-02-01/ -i manifest.txt
cd -
python3 -m great_ai.open_s3 --secrets s3.ini --push ss-data
rm -rf ss-data
```

77
examples/simple/predict_domain.py
View file

@ -1,77 +0,0 @@
from typing import Dict, Iterable, List
from great_ai import GreatAI, use_model
from pydantic import BaseModel
from sklearn.pipeline import Pipeline
from helpers import lemmatize, preprocess
class DomainPrediction(BaseModel):
domain: str
probability: float
explanation: List[str]
@GreatAI.deploy()
@use_model("small-domain-prediction-v2", version="latest")
def predict_domain(
text: str, model: Pipeline, cut_off_probability: float = 0.2
) -> List[DomainPrediction]:
"""
Predict the scientific domain of the input text.
Return labels until their sum likelihood is larger than cut_off_probability.
"""
assert 0 <= cut_off_probability <= 1
text = preprocess(text)
feature_names = model.named_steps["vectorizer"].get_feature_names_out()
token_mapping = {lemmatize(original): original for original in text.split(" ")}
features = model.named_steps["vectorizer"].transform(
[" ".join(token_mapping.keys())]
)
prediction = model.named_steps["classifier"].predict_proba(features)[0]
best_classes = sorted(enumerate(prediction), key=lambda v: v[1], reverse=True)
results: List[DomainPrediction] = []
for class_index, probability in best_classes:
weights = model.named_steps["classifier"].feature_log_prob_[class_index]
domain = model.named_steps["classifier"].classes_[class_index]
results.append(
DomainPrediction(
domain=domain,
probability=round(probability * 100),
explanation=_get_explanation(
feature_names=feature_names,
features=features.A[0],
weights=weights,
token_mapping=token_mapping,
),
)
)
if sum(r.probability for r in results) >= cut_off_probability * 100:
break
return results
def _get_explanation(
feature_names: Iterable[str],
features: Iterable[float],
weights: Iterable[float],
token_mapping: Dict[str, str],
) -> List[str]:
influential = [
(weight, name)
for weight, value, name in zip(weights, features, feature_names)
if value
]
most_influential = sorted(influential, reverse=True)[:5]
return [token_mapping[name] for _, name in most_influential]