Spaces:
Runtime error
Runtime error
File size: 2,947 Bytes
c1757e9 e48c60b c1757e9 d1c9cd0 c1757e9 d1c9cd0 c1757e9 e48c60b c1757e9 e48c60b c1757e9 c0c2a9f e48c60b c0c2a9f e48c60b c1757e9 c0c2a9f e48c60b c1757e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import os
import re
import pandas as pd
import faiss
import torch
import numpy as np
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sentence_transformers import SentenceTransformer
from huggingface_hub import snapshot_download
REPO_ID = "abhinavsunil/kitchenelite-recipe-model"
MODEL_CACHE = "/tmp/model_cache"
TOP_K = 5
app = FastAPI(title="KitchenElite Recipe Search API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods (GET, POST, etc.)
allow_headers=["*"], # Allows all headers
)
model = None
index = None
df = None
# ==============================
# STARTUP EVENT
# ==============================
@app.on_event("startup")
def load_assets():
global model, index, df
print("🚀 Downloading model repo snapshot...")
local_dir = snapshot_download(
repo_id=REPO_ID,
local_dir=MODEL_CACHE,
local_dir_use_symlinks=False
)
print("📦 Loading metadata...")
df = pd.read_parquet(os.path.join(local_dir, "metadata.parquet"))
print("📦 Loading FAISS index...")
index = faiss.read_index(os.path.join(local_dir, "recipes.index"))
print("📦 Loading SentenceTransformer model...")
model = SentenceTransformer(local_dir, device="cpu")
print("✅ All assets loaded successfully!")
# ==============================
# UTILITY FUNCTION
# ==============================
def clean_instructions(instruction_input):
if isinstance(instruction_input, str) and instruction_input.startswith('c("'):
content = re.search(r'c\("(.*)"\)', instruction_input)
if content:
return [
step.strip().strip('"')
for step in content.group(1).split('", "')
]
if isinstance(instruction_input, (list, np.ndarray)):
return list(instruction_input)
return [str(instruction_input)]
# ==============================
# ROUTES
# ==============================
@app.get("/")
def home():
return {"status": "KitchenElite API Running 🚀"}
@app.get("/search")
def search(query: str):
query_vector = model.encode([query])
faiss.normalize_L2(query_vector)
distances, indices = index.search(
query_vector.astype("float32"),
TOP_K
)
results = df.iloc[indices[0]]
output = []
for _, row in results.iterrows():
output.append({
"name": str(row["name"]),
"ingredients": (
list(row["ingredients"])
if isinstance(row["ingredients"], np.ndarray)
else row["ingredients"]
),
"calories": float(row["calories"]),
"protein": float(row["protein"]),
"instructions": clean_instructions(row["RecipeInstructions"])
})
return {
"query": query,
"results": output
}
|