Modern Moderation API Test
'''
def transform_predictions(model_choice, prediction_dict):
"""
OpenAI tarzı response formatı için tahminleri dönüştür.
Dönüştürülen alanlar: flagged, categories, category_scores, category_applied_input_types
"""
category_keys = [
"sexual", "sexual/minors", "harassment", "harassment/threatening",
"hate", "hate/threatening", "illicit", "illicit/violent",
"self-harm", "self-harm/intent", "self-harm/instructions",
"violence", "violence/graphic"
]
scores = {}
if model_choice == "unitaryai/detoxify-multilingual":
scores["sexual"] = prediction_dict.get("sexual_explicit", 0.0)
scores["harassment"] = max(prediction_dict.get("identity_attack", 0.0), prediction_dict.get("insult", 0.0))
scores["hate"] = prediction_dict.get("toxicity", 0.0)
scores["violence"] = max(prediction_dict.get("severe_toxicity", 0.0), prediction_dict.get("threat", 0.0))
for key in category_keys:
if key not in scores:
scores[key] = 0.0
else:
for key in category_keys:
scores[key] = prediction_dict.get(key, 0.0)
threshold = 0.5
bool_categories = {key: (scores[key] > threshold) for key in category_keys}
cat_applied_input_types = {key: (["text"] if scores[key] > 0 else []) for key in category_keys}
flagged = any(bool_categories.values())
return flagged, bool_categories, scores, cat_applied_input_types
@app.route('/')
def home():
return render_template_string(HTML_TEMPLATE)
@app.route('/v1/moderations', methods=['POST'])
def moderations():
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith("Bearer "):
return jsonify({"error": "Unauthorized"}), 401
provided_api_key = auth_header.split(" ")[1]
if provided_api_key != API_KEY:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
raw_input = data.get('input') or data.get('texts')
if raw_input is None:
return jsonify({"error": "Invalid input, expected 'input' or 'texts' field"}), 400
if isinstance(raw_input, str):
texts = [raw_input]
elif isinstance(raw_input, list):
texts = raw_input
else:
return jsonify({"error": "Invalid input format, expected string or list of strings"}), 400
if len(texts) > 10:
return jsonify({"error": "Too many input items. Maximum 10 allowed."}), 400
for text in texts:
if not isinstance(text, str) or len(text) > 100000:
return jsonify({"error": "Each input item must be a string with a maximum of 100k characters."}), 400
results = []
model_choice = data.get('model', 'unitaryai/detoxify-multilingual')
if model_choice == "koalaai/text-moderation":
for text in texts:
inputs = koala_tokenizer(text, return_tensors="pt")
outputs = koala_model(**inputs)
logits = outputs.logits
probabilities = torch.softmax(logits, dim=-1).squeeze().tolist()
if isinstance(probabilities, float):
probabilities = [probabilities]
labels = [koala_model.config.id2label[idx] for idx in range(len(probabilities))]
prediction = {label: prob for label, prob in zip(labels, probabilities)}
flagged, bool_categories, scores, cat_applied_input_types = transform_predictions(model_choice, prediction)
results.append({
"flagged": flagged,
"categories": bool_categories,
"category_scores": scores,
"category_applied_input_types": cat_applied_input_types
})
response_model = "koalaai/text-moderation"
else:
for text in texts:
pred = detoxify_model.predict([text])
prediction = {k: v[0] for k, v in pred.items()}
flagged, bool_categories, scores, cat_applied_input_types = transform_predictions(model_choice, prediction)
results.append({
"flagged": flagged,
"categories": bool_categories,
"category_scores": scores,
"category_applied_input_types": cat_applied_input_types
})
response_model = "unitaryai/detoxify-multilingual"
response_data = {
"id": "modr-" + uuid.uuid4().hex[:24],
"model": response_model,
"results": results,
"object": "moderation"
}
return jsonify(response_data)
if __name__ == '__main__':
port = int(os.getenv('PORT', 7860))
app.run(host='0.0.0.0', port=port, debug=True)