Let's build a Rest API using FastAPI and Copliot!

 

🧞Let's build a REST API using FastAPI, Copilot!

https://github.com/Hippy-Happy/DassuL

GitHub - Hippy-Happy/DassuL

 

I’m gonna use this code for bulid the api today.

To simply explain this document, It was the NLP project from Korea which can detect hatespeech or discrimination word.

I already made the code with Flask API. But today we’ll try another option FastAPI with Copilot!!

 

And one thing I wanna explain about this whole situation, using copilot instead of doing it myself, is that I was unable to install development software tools with Melbourne free Library computers at that moment..

 

 

Try 1

I applied this sentence to Copilot and then it works.

 

and basic codes etc… It’s not the code I wanted. so let’s try again.

 

Try 2

Downloaded the raw file of training sheets and then directly give it to Copilot.

And then It was impossible for using downloaded files with Library computers….😢

 

Try 3

I gave the file name in the github to Copilot, and then it works.

This is the code.

main.py

from fastapi import FastAPI
from routers import transformers

app = FastAPI()

app.include_router(transformers.router)

@app.get("/")
def read_root():
    return {"Hello": "World"}

 

 

routers/transformers.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from models import transformers
from schemas import transformers as transformers_schema
from crud import transformers as transformers_crud
from database import get_db

router = APIRouter()

@router.get("/transformers", response_model=List[transformers_schema.Transformer])
def read_transformers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    transformers = transformers_crud.get_transformers(db, skip=skip, limit=limit)
    return transformers

@router.post("/transformers", response_model=transformers_schema.Transformer)
def create_transformer(transformer: transformers_schema.TransformerCreate, db: Session = Depends(get_db)):
    return transformers_crud.create_transformer(db=db, transformer=transformer)

 

 

models/transformers.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from models import transformers
from schemas import transformers as transformers_schema
from crud import transformers as transformers_crud
from database import get_db

router = APIRouter()

@router.get("/transformers", response_model=List[transformers_schema.Transformer])
def read_transformers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    transformers = transformers_crud.get_transformers(db, skip=skip, limit=limit)
    return transformers

@router.post("/transformers", response_model=transformers_schema.Transformer)
def create_transformer(transformer: transformers_schema.TransformerCreate, db: Session = Depends(get_db)):
    return transformers_crud.create_transformer(db=db, transformer=transformer)

 

 

schemas/transformers.py

from pydantic import BaseModel

class TransformerBase(BaseModel):
    name: str
    accuracy: float
    model_path: str

class TransformerCreate(TransformerBase):
    pass

class Transformer(TransformerBase):
    id: int

    class Config:
        orm_mode = True

 

 

crud/transformers.py

from pydantic import BaseModel

class TransformerBase(BaseModel):
    name: str
    accuracy: float
    model_path: str

class TransformerCreate(TransformerBase):
    pass

class Transformer(TransformerBase):
    id: int

    class Config:
        orm_mode = True

 

 

services/transformers.py

# Add any service logic or utility functions related to your transformers here

def train_transformer(model_path: str):
    # Your training logic from the notebook goes here
    pass

def evaluate_transformer(model_path: str):
    # Your evaluation logic from the notebook goes here
    pass

 

 

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

 

 

And then Copilot suggested me these questions about the code.

 

basically what I needed in this code for using this API practically.

first I questioned of dependencies.

 

Then I wanted to add the error function into the code.

So many files pop up at this part, I questioned it again.

Added http_exception_handler and custom_exception_handler for handling exceptions.

main.py

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from routers import transformers

app = FastAPI()

app.include_router(transformers.router)

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={"message": f"Oops! Something went wrong with {exc.name}."},
    )

Added conditions to raise CustomException and HTTPException.

routers/transformers.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from models import transformers
from schemas import transformers as transformers_schema
from crud import transformers as transformers_crud
from database import get_db
+ from main import CustomException

router = APIRouter()

@router.get("/transformers", response_model=List[transformers_schema.Transformer])
def read_transformers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
+    if limit > 100:
+        raise CustomException(name="Transformer List Limit")
    transformers = transformers_crud.get_transformers(db, skip=skip, limit=limit)
+    if not transformers:
+        raise HTTPException(status_code=404, detail="Transformers not found")
    return transformers

@router.post("/transformers", response_model=transformers_schema.Transformer)
def create_transformer(transformer: transformers_schema.TransformerCreate, db: Session = Depends(get_db)):
+    if not transformer.name:
+        raise CustomException(name="Transformer Name")
    return transformers_crud.create_transformer(db=db, transformer=transformer)

Try 4

app/main.py

from fastapi import FastAPI
from app.routes import predict, train, update, hyperparameter_tuning
from app.utils.error_handler import custom_error_handler
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# Include the routers from different endpoints
app.include_router(predict.router)
app.include_router(train.router)
app.include_router(update.router)
app.include_router(hyperparameter_tuning.router)

# Add error handler
app.add_exception_handler(RequestValidationError, custom_error_handler)
app.add_exception_handler(Exception, custom_error_handler)

 

app/utils/error_handler.py

from fastapi.responses import JSONResponse
from fastapi import Request

async def custom_error_handler(request: Request, exc):
    return JSONResponse(
        status_code=400,
        content={"message": f"Oops! Something went wrong: {str(exc)}"}
    )

 

 

app/utils/transformers_utils.py

from transformers import ElectraTokenizer, ElectraForSequenceClassification, Trainer, TrainingArguments
import torch
from datasets import load_metric, Dataset
import optuna

class TransformersModel:
    def __init__(self, model_name: str):
        self.tokenizer = ElectraTokenizer.from_pretrained(model_name)
        self.model = ElectraForSequenceClassification.from_pretrained(model_name)
        self.metric = load_metric("accuracy")

    def predict(self, text: str):
        inputs = self.tokenizer(text, return_tensors='pt')
        outputs = self.model(**inputs)
        logits = outputs.logits
        probabilities = torch.nn.functional.softmax(logits, dim=-1)
        predicted_class = torch.argmax(probabilities, dim=1).item()
        return {"class": predicted_class, "probabilities": probabilities.tolist()}

    def train(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str, training_args: dict):
        args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=training_args["num_train_epochs"],
            per_device_train_batch_size=training_args["per_device_train_batch_size"],
            per_device_eval_batch_size=training_args["per_device_eval_batch_size"],
            evaluation_strategy="steps",
            save_steps=10_000,
            eval_steps=500,
            logging_dir=f"{output_dir}/logs",
            learning_rate=training_args["learning_rate"],
            weight_decay=training_args["weight_decay"],
        )
        trainer = Trainer(
            model=self.model,
            args=args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            tokenizer=self.tokenizer,
            compute_metrics=self.compute_metrics,
        )
        trainer.train()
        trainer.save_model(output_dir)
        return {"status": "Training completed"}

    def update(self, new_train_texts: list, new_train_labels: list, output_dir: str):
        new_train_dataset = Dataset.from_dict({"text": new_train_texts, "label": new_train_labels})
        return self.train(new_train_dataset, new_train_dataset, output_dir, self.default_training_args())
    
    def hyperparameter_tuning(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str):
        def objective(trial):
            training_args = {
                "num_train_epochs": trial.suggest_int("num_train_epochs", 1, 5),
                "per_device_train_batch_size": trial.suggest_categorical("per_device_train_batch_size", [8, 16, 32]),
                "per_device_eval_batch_size": trial.suggest_categorical("per_device_eval_batch_size", [8, 16, 32]),
                "learning_rate": trial.suggest_loguniform("learning_rate", 1e-5, 1e-4),
                "weight_decay": trial.suggest_loguniform("weight_decay", 1e-4, 1e-2),
            }
            result = self.train(train_dataset, eval_dataset, output_dir, training_args)
            accuracy = result["accuracy"]
            return accuracy
        
        study = optuna.create_study(direction="maximize")
        study.optimize(objective, n_trials=20)
        return study.best_params
    
    def compute_metrics(self, p):
        preds = torch.argmax(p.predictions, dim=1)
        return self.metric.compute(predictions=preds, references=p.label_ids)

    def default_training_args(self):
        return {
            "num_train_epochs": 3,
            "per_device_train_batch_size": 8,
            "per_device_eval_batch_size": 8,
            "learning_rate": 2e-5,
            "weight_decay": 0.01,
        }

 

app/routes/predict.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.utils.transformers_utils import TransformersModel

router = APIRouter()

# Initialize the transformers model with Electra
model = TransformersModel(model_name='monologg/koelectra-base-v3-discriminator')

class TextData(BaseModel):
    text: str

@router.post("/predict")
async def predict(text_data: TextData):
    try:
        prediction = model.predict(text_data.text)
        return prediction
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

 

 

app/routes/train.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.utils.transformers_utils import TransformersModel
from datasets import Dataset

router = APIRouter()

# Initialize the transformers model (replace 'model_name' with your model)
model = TransformersModel(model_name='monologg/koelectra-base-v3-discriminator')

class TrainData(BaseModel):
    train_texts: list
    train_labels: list
    eval_texts: list
    eval_labels: list
    output_dir: str

@router.post("/train")
async def train_model(train_data: TrainData):
    try:
        train_dataset = Dataset.from_dict({"text": train_data.train_texts, "label": train_data.train_labels})
        eval_dataset = Dataset.from_dict({"text": train_data.eval_texts, "label": train_data.eval_labels})
        
        result = model.train(train_dataset, eval_dataset, train_data.output_dir, model.default_training_args())
        return result
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

 

 

app/routes/updates.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.utils.transformers_utils import TransformersModel

router = APIRouter()

# Initialize the transformers model (replace 'model_name' with your model)
model = TransformersModel(model_name='monologg/koelectra-base-v3-discriminator')

class UpdateData(BaseModel):
    new_train_texts: list
    new_train_labels: list
    output_dir: str

@router.post("/update")
async def update_model(update_data: UpdateData):
    try:
        result = model.update(update_data.new_train_texts, update_data.new_train_labels, update_data.output_dir)
        return result
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

 

 

app/routes/hyperparameter_tuning.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.utils.transformers_utils import TransformersModel
from datasets import Dataset

router = APIRouter()

# Initialize the transformers model with Electra
model = TransformersModel(model_name='monologg/koelectra-base-v3-discriminator')

class HyperparameterData(BaseModel):
    train_texts: list
    train_labels: list
    eval_texts: list
    eval_labels: list
    output_dir: str

@router.post("/hyperparameter_tuning")
async def hyperparameter_tuning(hyperparameter_data: HyperparameterData):
    try:
        train_dataset = Dataset.from_dict({"text": hyperparameter_data.train_texts, "label": hyperparameter_data.train_labels})
        eval_dataset = Dataset.from_dict({"text": hyperparameter_data.eval_texts, "label": hyperparameter_data.eval_labels})
        
        best_params = model.hyperparameter_tuning(train_dataset, eval_dataset, hyperparameter_data.output_dir)
        return {"best_params": best_params}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

 

 

I asked Copilot about the latest paper to approve it.

 

I asked the Copilot, With consideration of response speed of my api which paper is most suitable. And show me the result.

Considering the need to balance response speed with high performance in your FastAPI application, one of the most suitable advancements could be from the paper on RAG Foundry: Enhancing LLMs for Retrieval Augmented Generation.

Why RAG Foundry is Suitable:

1. Efficient Retrieval-Augmented Generation: RAG Foundry integrates retrieval mechanisms to augment the generation process. This can help in providing faster, contextually accurate responses by retrieving relevant information efficiently.
2. Asynchronous Retrieval: The paper's methodologies can be adapted to use asynchronous processing, which FastAPI supports natively. This ensures non-blocking operations, leading to faster response times.
3. Reduced Computational Load: By leveraging pre-retrieved relevant information, the model can generate responses with less computational overhead compared to processing everything from scratch.
4. Optimized for Real-Time Applications: The techniques are designed to enhance real-time applications, making them ideal for APIs that require quick turnaround times.

Practical Integration

Here's how you might apply the principles from RAG Foundry to your existing FastAPI application:

1. Implement a Retriever for Augmenting Responses: Use a retriever to gather relevant data quickly.
2. Asynchronous Endpoints: Ensure all endpoints are asynchronous to handle concurrent requests efficiently.
3. Optimize Model Inference: Use optimized inference libraries or frameworks (e.g., ONNX Runtime) to speed up model predictions.

app/utils/transformers_utils.py

from transformers import RagTokenizer, RagRetriever, RAGForSequenceClassification
import torch
from datasets import load_metric, Dataset
import optuna

class TransformersModel:
    def __init__(self, model_name: str):
        self.tokenizer = RagTokenizer.from_pretrained(model_name)
        self.retriever = RagRetriever.from_pretrained(model_name)
        self.model = RAGForSequenceClassification.from_pretrained(model_name)
        self.metric = load_metric("accuracy")

    async def predict(self, text: str):
        inputs = self.tokenizer(text, return_tensors='pt')
        retrieved_docs = await self.retriever(inputs.input_ids)
        inputs['retrieved_docs'] = retrieved_docs
        outputs = self.model(**inputs)
        logits = outputs.logits
        probabilities = torch.nn.functional.softmax(logits, dim=-1)
        predicted_class = torch.argmax(probabilities, dim=1).item()
        return {"class": predicted_class, "probabilities": probabilities.tolist()}

    async def train(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str, training_args: dict):
        args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=training_args["num_train_epochs"],
            per_device_train_batch_size=training_args["per_device_train_batch_size"],
            per_device_eval_batch_size=training_args["per_device_eval_batch_size"],
            evaluation_strategy="steps",
            save_steps=10_000,
            eval_steps=500,
            logging_dir=f"{output_dir}/logs",
            learning_rate=training_args["learning_rate"],
            weight_decay=training_args["weight_decay"],
        )
        trainer = Trainer(
            model=self.model,
            args=args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            tokenizer=self.tokenizer,
            compute_metrics=self.compute_metrics,
        )
        await trainer.train()
        trainer.save_model(output_dir)
        return {"status": "Training completed"

    def update(self, new_train_texts: list, new_train_labels: list, output_dir: str):
        new_train_dataset = Dataset.from_dict({"text": new_train_texts, "label": new_train_labels})
        return self.train(new_train_dataset, new_train_dataset, output_dir, self.default_training_args())

    def hyperparameter_tuning(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str):
        def objective(trial):
            training_args = {
                "num_train_epochs": trial.suggest_int("num_train_epochs", 1, 5),
                "per_device_train_batch_size": trial.suggest_categorical("per_device_train_batch_size", [8, 16, 32]),
                "per_device_eval_batch_size": trial.suggest_categorical("per_device_eval_batch_size", [8, 16, 32]),
                "learning_rate": trial.suggest_loguniform("learning_rate", 1e-5, 1e-4),
                "weight_decay": trial.suggest_loguniform("weight_decay", 1e-4, 1e-2),
            }
            result = self.train(train_dataset, eval_dataset, output_dir, training_args)
            accuracy = result["accuracy"]
            return accuracy
        
        study = optuna.create_study(direction="maximize")
        study.optimize(objective, n_trials=20)
        return study.best_params

    def compute_metrics(self, p):
        preds = torch.argmax(p.predictions, dim=1)
        return self.metric.compute(predictions=preds, references=p.label_ids)

    def default_training_args(self):
        return {
            "num_train_epochs": 3,
            "per_device_train_batch_size": 8,
            "per_device_eval_batch_size": 8,
            "learning_rate": 2e-5,
            "weight_decay": 0.01,
        }

 

 

app/routes/predict.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.utils.transformers_utils import TransformersModel

router = APIRouter()

# Initialize the transformers model with RAG
model = TransformersModel(model_name='facebook/rag-token-nq')

class TextData(BaseModel):
    text: str

@router.post("/predict")
async def predict(text_data: TextData):
    try:
        prediction = await model.predict(text_data.text)
        return prediction
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))