File size: 4,531 Bytes
14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe 5b7c1e6 14040fe |
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
from fastapi import FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
import pandas as pd
from filtered_search_engine import SmartRecommender
from reranker import Reranker
from intent_classifier import IntentClassifier
from keyword_boosting_layer import apply_keyword_boost
from cache_manager import CacheManager
# ------------------------------
# Initialize App
# ------------------------------
app = FastAPI(
title="Salahkar AI Recommender",
description="Smart cultural, heritage & food recommendation engine for BharatVerse",
version="1.0.0"
)
from fastapi.staticfiles import StaticFiles
# CORS Support (allows frontend browser access)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount images folder to serve static files with Cache-Control
class CachedStaticFiles(StaticFiles):
def file_response(self, *args, **kwargs):
response = super().file_response(*args, **kwargs)
response.headers["Cache-Control"] = "public, max-age=31536000"
return response
app.mount("/images", CachedStaticFiles(directory="images"), name="images")
# ------------------------------
# Load Core Components Once
# ------------------------------
print("📌 Loading dataset...")
df = pd.read_csv("salahkar_enhanced.csv")
# Optimization: Create O(1) lookup map
name_to_row = df.set_index("name").to_dict('index')
print("📌 Loading smart recommendation engine...")
engine = SmartRecommender()
print("📌 Loading reranker model...")
reranker = Reranker()
print("📌 Loading intent recognizer...")
intent_detector = IntentClassifier()
print("📌 Initializing Cache Manager...")
cache = CacheManager(capacity=200, ttl_seconds=3600)
print("🚀 Salahkar AI Ready!")
# ------------------------------
# Routes
# ------------------------------
@app.get("/")
def root():
return {
"message": "🇮🇳 Welcome to Salahkar AI – BharatVerse Intelligent Recommendation System",
"usage": "/recommend?query=your text"
}
@app.get("/recommend")
def get_recommendation(query: str = Query(..., description="User's search text"), k: int = 7):
print(f"\n🔍 User Query: {query}")
# 0️⃣ Check Cache
cached_response = cache.get(query)
if cached_response:
return cached_response
# 1️⃣ Detect intent
detected_intent = intent_detector.predict_intent(query)
print(f"🧠 Intent Detected: {detected_intent}")
# 2️⃣ FAISS + Filter Search
# engine.recommend returns (results_list, intent)
# Optimization: Pass detected_intent to avoid re-running classification
rec_results, _ = engine.recommend(query, k=k, intent=detected_intent)
# 3️⃣ Prepare results for reranker
prepared = []
for item in rec_results:
name = item["name"]
domain = item["domain"]
category = item["category"]
region = item["region"]
score = item["score"]
# Optimization: O(1) lookup instead of O(N) dataframe filter
row = name_to_row.get(name)
if not row:
continue
prepared.append({
"name": name,
"domain": domain,
"category": category,
"region": region,
"embedding_score": float(score),
"text": row["search_embedding_text"],
"image": row["image_file"]
})
# 4️⃣ Re-rank using cross encoder
reranked_results = reranker.rerank(query, prepared)
# 5️⃣ Apply keyword boosting
final_results = apply_keyword_boost(query, reranked_results)
# 6️⃣ Format response for frontend
response = [
{
"name": item["name"],
"category": item["category"],
"domain": item["domain"],
"region": item["region"],
"score": float(item["final_score"]),
"image": f"/images/{item['image']}" if item.get("image") else None
}
for item in final_results[:k]
]
final_response = {
"query": query,
"intent": detected_intent,
"results": response
}
# 7️⃣ Save to Cache
cache.set(query, final_response)
return final_response
# -------------------------------------------
# Run (Ignored by HuggingFace — needed only for local testing)
# -------------------------------------------
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |