from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from sentence_transformers import SentenceTransformer from groq import Groq from firebase_admin import credentials, firestore from google.cloud.firestore_v1.base_query import FieldFilter import firebase_admin import numpy as np import os import json from dotenv import load_dotenv import uvicorn # 🔥 LOAD ENV load_dotenv() GROQ_API_KEY = os.getenv("GROQ_API_KEY") # 🔥 FIREBASE INIT firebase_creds = os.getenv("FIREBASE_CREDENTIALS") if firebase_creds: # Running on Hugging Face (Load from Secret) cred_dict = json.loads(firebase_creds) cred = credentials.Certificate(cred_dict) else: # Running locally (Load from file) cred = credentials.Certificate("serviceAccountKey.json") firebase_admin.initialize_app(cred) db = firestore.client() # 🔥 GROQ groq_client = Groq(api_key=GROQ_API_KEY) # 🔥 EMBEDDING MODEL model = SentenceTransformer('all-MiniLM-L6-v2') # 🔥 FASTAPI app = FastAPI() # 🔥 CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # 🔥 INTENT CLASSIFIER def classify_intent(question: str): q = question.lower().strip() greetings = ["hi", "hello", "hey", "hii", "yo"] if any(q == word or q.startswith(word) for word in greetings): return "greeting" decision_keywords = [ "should i", "what should i do", "choose", "decision", "which is better", "do i", "whether i should", "i should", "can i", "whether" ] if any(word in q for word in decision_keywords): return "decision" return "irrelevant" class DecisionEngine: # 🔥 EMBEDDING def embed(self, text): try: return model.encode(text).tolist() except Exception as e: print("Embedding Error:", e) return None # 🔥 GET PERSONALITY def get_personality(self, user_id): try: doc = db.collection("personality_profiles") \ .document(user_id) \ .get() if doc.exists: data = doc.to_dict() return data.get("trait_scores", {}) return {} except Exception as e: print("Personality Error:", e) return {} # 🔥 GET MEMORY (RAG) def get_memory(self, user_id, question): try: query_vec = self.embed(question) if query_vec is None: return "" query_vec = np.array(query_vec) docs = db.collection("chat_history") \ .where(filter=FieldFilter("userId", "==", user_id)) \ .stream() data = [doc.to_dict() for doc in docs] if not data: return "" scored = [] for row in data: emb = row.get("embedding") if emb is None: continue emb = np.array(emb, dtype=float) similarity = np.dot(query_vec, emb) / ( np.linalg.norm(query_vec) * np.linalg.norm(emb) ) scored.append((similarity, row)) if not scored: return "" scored.sort(reverse=True, key=lambda x: x[0]) top = scored[:3] memory = "" for _, row in top: memory += f"{row['message']} → {row['response']}\n" return memory except Exception as e: print("Memory Error:", e) return "" # 🔥 GENERATE AI RESPONSE def generate(self, question, personality, memory): prompt = f""" You are a strict decision-making AI. Your primary source for making a decision MUST be the User personality: {personality} You should lightly consider, but not strictly rely on Past behavior: {memory} Question: {question} Rules: - Give ONLY ONE decision - No "it depends" - No multiple options - Be confident - Base your decision primarily on the user's personality traits Format: Decision: Reason: """ try: completion = groq_client.chat.completions.create( model="llama-3.1-8b-instant", messages=[ { "role": "system", "content": "You are a decisive AI." }, { "role": "user", "content": prompt } ], temperature=0.3, max_tokens=512, top_p=1, stream=False ) return completion.choices[0].message.content except Exception as e: print("Groq Error:", e) return self.generate_fallback_openrouter(prompt) def generate_fallback_openrouter(self, prompt): try: import urllib.request openrouter_api_key = os.getenv("OPENROUTER_API_KEY") if not openrouter_api_key: return "AI failed: Groq error and no OpenRouter API key provided." url = "https://openrouter.ai/api/v1/chat/completions" headers = { "Authorization": f"Bearer {openrouter_api_key}", "Content-Type": "application/json", } data = { "model": "mistralai/mistral-7b-instruct:free", "messages": [ {"role": "system", "content": "You are a decisive AI."}, {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 512, "top_p": 1 } req = urllib.request.Request(url, headers=headers, data=json.dumps(data).encode('utf-8')) with urllib.request.urlopen(req) as response: result = json.loads(response.read().decode('utf-8')) return result['choices'][0]['message']['content'] except Exception as fallback_error: print("OpenRouter Fallback Error:", fallback_error) return "AI failed on both primary and fallback APIs." # 🔥 SAVE CHAT def save_chat(self, user_id, question, response): try: embedding = self.embed(question) db.collection("chat_history").add({ "userId": user_id, "message": question, "response": response, "embedding": embedding }) except Exception as e: print("Save Error:", e) # 🔥 ENGINE engine = DecisionEngine() # 🔥 MAIN API @app.post("/ask") def ask(data: dict): try: user_id = data.get("user_id") question = data.get("question") if not user_id or not question: return { "response": "Invalid input" } # 🔥 INTENT CHECK intent = classify_intent(question) if intent == "greeting": return { "response": "Hello 👋 Tell me what decision you want help with today." } if intent == "irrelevant": return { "response": "I only help with decision-making. Please ask something like 'Should I do this or that?'" } # 🔥 GET USER DATA personality = engine.get_personality(user_id) memory = engine.get_memory(user_id, question) # 🔥 GENERATE RESPONSE response = engine.generate( question, personality, memory ) # 🔥 SAVE CHAT engine.save_chat( user_id, question, response ) return { "response": response } except Exception as e: print("Server Error:", e) return { "response": str(e) } # 🔥 HEALTH CHECK @app.get("/") def home(): return { "message": "Thinkless AI Backend Running with Firebase 🚀" } # 🔥 RUN SERVER if __name__ == "__main__": uvicorn.run( "app:app", host="0.0.0.0", port=7860, reload=True )