FastAPI Template per LLM SaaS Parte 1 — Autenticazione e Caricamento File

Template FastAPI per LLM SaaS Parte 1 — Autenticazione e Caricamento File

FastAPI sta ottenendo molta popolarità nella comunità di sviluppatori backend Python grazie alla sua semplicità, natura asincrona e interfaccia utente Swagger nativa.

Tra i progetti open-source LLM più popolari su GitHub, Quivr è uno dei migliori e ha molti stelle (24.2k al momento della scrittura) e un codice ben strutturato. Prima di tutto, vorrei rendere omaggio a questo repository e ai suoi contributori per il loro ottimo lavoro nel creare un progetto di riferimento così valido per la comunità Python.

GitHub – StanGirard/quivr: 🧠 Your supercharged Second Brain 🧠 Your personal productivity…

🧠 Your supercharged Second Brain 🧠 Your personal productivity assistant per chattare con i tuoi file (PDF, CSV) & app…

github.com

Ci sono molte cose interessanti da esaminare in questo repository e, inoltre, vorremmo creare un template basato su questo repository per qualsiasi futuro caso d’uso LLM. Pertanto, ho deciso di suddividerlo in due articoli. In questo articolo, ci concentreremo su quanto segue:

  • Architettura ad alto livello
  • Autenticazione con Supabase e FastAPI
  • Caricamento file con Supabase

Nella parte 2, affronteremo:

  • Celery worker e message queue per processi a lungo termine
  • Plug-in pg-vector in Postgres

Nella parte 3, affronteremo:

  • FastAPI per ChatGPT, ad esempio streaming di payload
  • Pagamento con Stripe
  • Testing API
  • Template per qualsiasi futuro caso d’uso

Architettura ad alto livello

Fonte: disegno dell'autore

L’architettura del backend è composta da tre parti principali: Supabase DB, server backend FastAPI e server Celery. Celery viene utilizzato per attività di background a lungo termine, ad esempio l’inserimento di un grande documento PDF. Tra FastAPI e il server Celery, Redis viene utilizzato come message broker. La comunicazione tra FastAPI/Celery e Supabase avviene tramite il client Supabase (Python SDK)

Supabase Auth

Supabase è un’alternativa open-source a Firebase. È essenzialmente un database Postgres, ma ha altre funzioni integrate, ad esempio autenticazione, funzioni di bordo, storage di blob, pg-vector, ecc., che semplificano lo sviluppo rispetto all’utilizzo di un database Postgres da zero.

Con supabase auth, è possibile semplicemente chiamare le funzioni signUp() e signIn() dalla libreria client di supabase. Di seguito viene mostrato un esempio in JavaScript (fonte:https://supabase.com/docs/guides/auth/social-login) .

async function signUpNewUser() {  const { data, error } = await supabase.auth.signUp({    email: '[email protected]',    password: 'example-password',    options: {      redirectTo: 'https//example.com/welcome'    }  })}

async function signInWithEmail() {  const { data, error } = await supabase.auth.signInWithPassword({    email: '[email protected]',    password: 'example-password',    options: {      redirectTo: 'https//example.com/welcome'    }  })}

async function signOut() {  const { error } = await supabase.auth.signOut()}

Questo è il codice frontend, quindi cosa fare nel backend??

Buona domanda. Con l’interazione tra il tuo frontend e Supabase, Supabase crea effettivamente una tabella chiamata auth.users. Questa tabella si trova nella sezione Autenticazione nella dashboard di Supabase.

Fonte: screenshot dell'autore

Per eventuali tabelle future che richiedono riferimenti ad auth.users, è sufficiente fare quanto segue

CREATE TABLE IF NOT EXISTS user_daily_usage(    user_id UUID REFERENCES auth.users (id),    email TEXT,    date TEXT,    daily_requests_count INT,    PRIMARY KEY (user_id, date));

Poi dobbiamo autenticare l’utente per alcune API di backend. Se il frontend utilizza direttamente Supabase.auth, come autenticare le richieste dell’utente per altre chiamate API sul backend?

Per spiegare questo, è necessario capire come funzionano i JWT (JavaScript Web Token).

Fonte: disegno dell'autore

Puoi testare l’encoding e decodifica per JWT a https://jwt.io/. Una volta che l’utente si è registrato/acceso al server di autenticazione, otterrà un JWT. Quindi, se l’utente cerca di caricare il sito nuovamente entro un breve lasso di tempo (prima che il token scada), non sarà necessario inserire nuovamente la password.

Fonte: screenshot dell'autore

Per generare un JWT per l’utente, avrai bisogno del campo ‘sub’, che è l’ID utente (UUID assegnato automaticamente all’utente da auth.users) e dell’email utilizzata durante la registrazione.

Quindi, per decodificare il JWT, sia il server di autenticazione che il backend avranno bisogno della chiave segreta a 256 bit. Se utilizzi l’autenticazione di Supabase, viene chiamata ‘Anon key’ nel pannello di amministrazione. Sarà la stessa chiave che utilizzerai per decodificare il JWT sul backend.

Il modulo di autenticazione sul backend FastAPI sarebbe simile a questo:

import osfrom typing import Optionalfrom auth.jwt_token_handler import decode_access_token, verify_tokenfrom fastapi import Depends, HTTPException, Requestfrom fastapi.security import HTTPAuthorizationCredentials, HTTPBearerfrom models import UserIdentityclass AuthBearer(HTTPBearer):    def __init__(self, auto_error: bool = True):        super().__init__(auto_error=auto_error)    async def __call__(        self,        request: Request,    ):        credentials: Optional[HTTPAuthorizationCredentials] = await super().__call__(            request        )        self.check_scheme(credentials)        token = credentials.credentials  # pyright: ignore reportPrivateUsage=none        return await self.authenticate(            token,        )    def check_scheme(self, credentials):        if credentials and credentials.scheme != "Bearer":            raise HTTPException(status_code=401, detail="Il token deve essere Bearer")        elif not credentials:            raise HTTPException(                status_code=403, detail="Credenziali di autenticazione mancanti"            )    async def authenticate(        self,        token: str,    ) -> UserIdentity:        if os.environ.get("AUTHENTICATE") == "false":            return self.get_test_user()        elif verify_token(token):            return decode_access_token(token)        else:            raise HTTPException(status_code=401, detail="Token o chiave API non validi")    def get_test_user(self) -> UserIdentity:        return UserIdentity(            email="[email protected]", id="696dda89-d395-4601-af3d-e1c66de3df1a"  # type: ignore        )  # replace with test user informationdef get_current_user(user: UserIdentity = Depends(AuthBearer())) -> UserIdentity:    return user

import osfrom datetime import datetime, timedeltafrom typing import Optionalfrom jose import jwtfrom jose.exceptions import JWTErrorfrom models import UserIdentitySECRET_KEY = os.environ.get("JWT_SECRET_KEY")ALGORITHM = "HS256"if not SECRET_KEY:    raise ValueError("La variabile d'ambiente JWT_SECRET_KEY non è stata impostata")def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    to_encode = data.copy()    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwtdef decode_access_token(token: str) -> UserIdentity:    try:        payload = jwt.decode(            token, SECRET_KEY, algorithms=[ALGORITHM], options={"verify_aud": False}        )    except JWTError:        return None  # pyright: ignore reportPrivateUsage=none    return UserIdentity(        email=payload.get("email"),        id=payload.get("sub"),  # pyright: ignore reportPrivateUsage=none    )def verify_token(token: str):    payload = decode_access_token(token)    return payload is not None

Caricamento file con Supabase

Sarai in grado di chiamare direttamente la libreria client di Supabase per caricare i file. Una funzione utility può essere scritta come segue:

import jsonimport multiprocessing.get_loggerfrom langchain.pydantic_v1 import Fieldfrom langchain.schema import Documentfrom supabase.client import Client, create_clientimport osfrom dotenv import load_dotenvload_dotenv()logger = get_logger()def get_supabase_client() -> Client:    supabase_client: Client = create_client(        os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_SERVICE_KEY")    )    return supabase_clientdef upload_file_storage(file, file_identifier: str):    supabase_client: Client = get_supabase_client()    # res = supabase_client.storage.create_bucket("quivr")    response = None    try:        response = supabase_client.storage.from_(os.getenv("SUPABASE_BUCKET")).upload(file_identifier, file)        return response    except Exception as e:        logger.error(e)        raise e

Quindi una route per FastAPI

import osfrom typing import Optionalfrom uuid import UUIDfrom auth import AuthBearer, get_current_userfrom fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFilefrom logger import get_loggerfrom models import UserIdentity, UserUsagefrom repository.files.upload_file import upload_file_storagefrom repository.user_identity import get_user_identitylogger = get_logger(__name__)upload_router = APIRouter()@upload_router.get("/upload/healthz", tags=["Salute"])async def healthz():    return {"status": "ok"}@upload_router.post("/upload", dependencies=[Depends(AuthBearer())], tags=["Caricamento"])async def upload_file(    request: Request,    uploadFile: UploadFile,    chat_id: Optional[UUID] = Query(None, description="L'ID della chat"),    current_user: UserIdentity = Depends(get_current_user),):    file_content = await uploadFile.read()    filename_with_user_id = str(current_user.id) + "/" + str(uploadFile.filename)    try:        fileInStorage = upload_file_storage(file_content, filename_with_user_id)        logger.info(f"File {fileInStorage} caricato con successo")    except Exception as e:        if "La risorsa esiste già" in str(e):            raise HTTPException(                status_code=403,                detail=f"Il file {uploadFile.filename} esiste già nello storage.",            )        else:            raise HTTPException(                status_code=500, detail="Impossibile caricare il file nello storage."            )    return {"message": "È iniziato il processamento del file."}

Continua nella Parte 2…

Se non sei familiare con FastAPI, potrebbe essere un po’ difficoltoso capire. Ma alla fine della Parte 3, condividerò l’intero repository GitHub, sarà molto più chiaro. Rimani sintonizzato.