|
from transformers import AutoTokenizer, AutoModel |
|
import torch |
|
import torch.nn.functional as F |
|
|
|
def mean_pooling(model_output, attention_mask): |
|
token_embeddings = model_output[ |
|
0 |
|
] |
|
input_mask_expanded = ( |
|
attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float() |
|
) |
|
return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp( |
|
input_mask_expanded.sum(1), min=1e-9 |
|
) |
|
|
|
def cosine_similarity(u, v): |
|
return F.cosine_similarity(u, v, dim=1) |
|
|
|
|
|
def compare(text1, text2): |
|
|
|
sentences = [text1, text2] |
|
|
|
tokenizer = AutoTokenizer.from_pretrained("dmlls/all-mpnet-base-v2-negation") |
|
model = AutoModel.from_pretrained("dmlls/all-mpnet-base-v2-negation") |
|
|
|
encoded_input = tokenizer( |
|
sentences, padding=True, truncation=True, return_tensors="pt" |
|
) |
|
|
|
with torch.no_grad(): |
|
model_output = model(**encoded_input) |
|
|
|
sentence_embeddings = mean_pooling(model_output, encoded_input["attention_mask"]) |
|
|
|
sentence_embeddings = F.normalize(sentence_embeddings, p=2, dim=1) |
|
|
|
similarity_score = cosine_similarity( |
|
sentence_embeddings[0].unsqueeze(0), sentence_embeddings[1].unsqueeze(0) |
|
) |
|
return similarity_score.item() |
|
|
|
|
|
|
|
from fastapi import FastAPI |
|
|
|
app = FastAPI() |
|
|
|
@app.get("/") |
|
def greet_json(): |
|
return {"Hello": "World!"} |
|
|
|
|
|
|
|
from transformers import pipeline |
|
|
|
summarizer = pipeline("summarization", model="facebook/bart-large-cnn") |
|
|
|
def Summerized_Text(text): |
|
text = text.strip() |
|
a = summarizer(text, max_length=130, min_length=30, do_sample=False) |
|
print(a) |
|
return a[0]['summary_text'] |
|
|
|
|
|
|
|
from fastapi.responses import JSONResponse |
|
from pydantic import BaseModel |
|
from fastapi import FastAPI |
|
|
|
class StrRequest(BaseModel): |
|
text: str |
|
|
|
|
|
class CompareRequest(BaseModel): |
|
summary: str |
|
text: str |
|
|
|
|
|
@app.get("/api/check") |
|
def check_connection(): |
|
try: |
|
return JSONResponse( |
|
{"status": 200, "message": "Message Successfully Sent"}, status_code=200 |
|
) |
|
except Exception as e: |
|
print("Error => ", e) |
|
return JSONResponse({"status": 500, "message": str(e)}, status_code=500) |
|
|
|
|
|
@app.post("/api/summerized") |
|
async def get_summerized(request: StrRequest): |
|
try: |
|
print(request) |
|
text = request.text |
|
if not text: |
|
return JSONResponse( |
|
{"status": 422, "message": "Invalid Input"}, status_code=422 |
|
) |
|
summary = Summerized_Text(text) |
|
if "No abstract text." in summary: |
|
return JSONResponse( |
|
{"status": 500, "message": "No matching text found", "data": "None"} |
|
) |
|
|
|
if not summary: |
|
return JSONResponse( |
|
{"status": 500, "message": "No matching text found", "data": {}} |
|
) |
|
|
|
return JSONResponse( |
|
{"status": 200, "message": "Matching text found", "data": summary} |
|
) |
|
|
|
except Exception as e: |
|
print("Error => ", e) |
|
return JSONResponse({"status": 500, "message": str(e)}, status_code=500) |
|
|
|
|
|
@app.post("/api/compare") |
|
def compareTexts(request: CompareRequest): |
|
try: |
|
text = request.text |
|
summary = request.summary |
|
if not summary or not text: |
|
return JSONResponse( |
|
{"status": 422, "message": "Invalid Input"}, status_code=422 |
|
) |
|
value = compare(text, summary) |
|
return JSONResponse( |
|
{ |
|
"status": 200, |
|
"message": "Comparisons made", |
|
"value": value, |
|
"text": text, |
|
"summary": summary, |
|
} |
|
) |
|
except Exception as e: |
|
print("Error => ", e) |
|
return JSONResponse({"status": 500, "message": str(e)}, status_code=500) |
|
|
|
|
