import os from fastapi import FastAPI, HTTPException, Form, status from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import List, Literal, Optional, Tuple from dotenv import load_dotenv # --- LangChain & AI Components --- # Document handling: Load and split text data from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # Models: Google Gemini for Chat and Embeddings from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings # Vector Store: FAISS for efficient similarity search from langchain_community.vectorstores import FAISS # RAG Logic: Prompts and Retrieval Chain from langchain.prompts import PromptTemplate from langchain.chains import ConversationalRetrievalChain # --- Configuration --- # Load environment variables from .env file load_dotenv() # Retrieve Google API Key securely GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") # --- App Initialization --- # Initialize the FastAPI application with metadata app = FastAPI(title="Vio's AI Assistant API", version="1.0.0") # --- CORS Configuration --- # Enable CORS to allow external applications (frontend) to communicate with this API app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allow requests from any domain allow_credentials=True, allow_headers=["*"], allow_methods=["*"] ) # --- Request Schema --- # Define the data structure for incoming chat requests class ChatRequest(BaseModel): text: str # The user's input question language: Literal["English", "Indonesian"] = "English" # Preferred output language history: List[tuple[str, str]] = [] # Chat context passed from frontend (stateless) # Global variable for the Vector Store (Lazy Loading pattern) vectorstore = None def get_vectorstore(): global vectorstore # Check if the Vector Store is not loaded yet (Lazy Loading) if vectorstore is None: print("⏳ Loading Vector Store (Embeddings)...") try: # 1. Load Data: Read the JSON profile loader = TextLoader(file_path="data/silvio_profile.json", encoding="utf-8") doc = loader.load() # 2. Text Splitting: Break documents into chunks for better retrieval text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100) text_chunks = text_splitter.split_documents(doc) # 3. Embedding Model: Initialize Google Gemini Embeddings embeddings = GoogleGenerativeAIEmbeddings( model="models/gemini-embedding-001", google_api_key=GOOGLE_API_KEY ) # 4. Vector Store: Create FAISS index from chunks vectorstore = FAISS.from_documents(text_chunks, embedding=embeddings) print("✅ Vector Store Ready!") except FileNotFoundError: # Specific handling if the JSON file is missing print("❌ Error: Source file not found.") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Configuration Error: The profile data file (silvio_profile.json) is missing." ) except Exception as e: error_msg = str(e) print(f"❌ Critical Error loading model: {error_msg}") # Using 'if' logic to identify specific API errors if "401" in error_msg or "API key" in error_msg: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Authentication Failed: Invalid Google Gemini API Key." ) elif "429" in error_msg or "Quota" in error_msg: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, # 503 is standard for Rate Limits detail="Service Unavailable: Google API Quota exceeded. Please try again later." ) else: # Fallback for any other unexpected errors raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal Server Error: Failed to initialize AI Model. Please contact the administrator." ) return vectorstore def get_qa_chain(vs, language: str): try: # 1. Initialize LLM: Setup Google Gemini with low temperature for factual consistency llm = ChatGoogleGenerativeAI( model="gemini-2.5-flash", temperature=0.3, google_api_key=GOOGLE_API_KEY ) # 2. Define Persona: Inject the user's language choice into the strict system prompt prompt_template = f""" You are the AI Portfolio Assistant for **Silvio Christian Joe** (Vio). Represent him professionally to recruiters and developers. Use the following pieces of context to answer the user's question. ### INSTRUCTIONS: 1. **LANGUAGE PRIORITY:** The user is speaking in **{language}**. You MUST answer strictly in **{language}**. 2. **ELABORATE & ENGAGE:** Be detailed and professional. 3. **SPECIALIZATION:** Highlight expertise in **NLP, Data Science, and Tabular Data**. 4. **NO HALLUCINATIONS:** If the answer is not in the context, say (in {language}): "I don't have that information. Contact Vio via LinkedIn or Email." Context: {{context}} Question: {{question}} Detailed Answer (in {language}): """ QA_PROMPT = PromptTemplate.from_template(prompt_template) # 3. Build RAG Chain: Connect the LLM, Vector Store Retriever, and Custom Prompt chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=vs.as_retriever(), return_source_documents=True, combine_docs_chain_kwargs={"prompt": QA_PROMPT} ) return chain except Exception as e: error_msg = str(e) print(f"❌ Error creating QA Chain: {error_msg}") # 1. Handle Authentication/API Key Issues if "401" in error_msg or "API key" in error_msg: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Authentication Failed: Invalid Google Gemini API Key configured in server." ) # 2. Handle Rate Limiting / Quota Issues elif "429" in error_msg or "Quota" in error_msg: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Service Busy: Google AI Quota exceeded. Please try again in a few moments." ) # 3. Handle Context/Prompt Issues (Optional but good for LangChain) elif "validation" in error_msg.lower() or "prompt" in error_msg.lower(): raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Configuration Error: Failed to compile the AI Prompt template." ) # 4. Generic Fallback else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal Server Error: Could not initialize the Conversation Chain. Please contact the administrator." ) @app.get("/") def home(): # Root endpoint: Provides server status and detailed API usage documentation return { "status": "✅ Online", "service": "Vio's AI Portfolio Assistant API", "version": "1.0.0", "live_urls": { "base_url": "https://silvio0-silvio-portfolio-api.hf.space", "documentation": "https://silvio0-silvio-portfolio-api.hf.space/docs", "chat_endpoint": "https://silvio0-silvio-portfolio-api.hf.space/assistant" }, "usage_guide": { "endpoint": "/assistant", "method": "POST", "description": "Generate AI responses based on Silvio's portfolio context.", # Explicitly defining data types, options, and constraints "payload_structure": { "text": "string (Required) - The user's input question.", "language": "string (Optional) - Options: 'English' | 'Indonesian'. Default: 'English'.", "history": "list (Optional) - Used for conversation context/memory. Format: A list of pairs, where each pair must contain exactly [User_Question, AI_Answer]." }, # A concrete example showing the history format "payload_example": { "text": "Where does he live?", "language": "English", "history": [ ["Who is Silvio?", "Silvio is a Data Scientist."], ["What is his nickname?", "His nickname is Vio."] ] } }, "author": "Silvio Christian Joe" } @app.post("/assistant") def chat(data: ChatRequest): # 1. Retrieve the Vector Store (Knowledge Base) vs = get_vectorstore() # Safety Check: Ensure the database is actually loaded if vs is None: raise HTTPException(status_code=500, detail="Server Error: Knowledge Base failed to load.") try: # 2. Initialize Chain: Create the conversation logic with the user's language chain = get_qa_chain(vs, data.language) # 3. Generate Response: Invoke the chain with current input and past history response = chain.invoke({ "question": data.text, "chat_history": data.history }) # 4. Extract Data: Separate the text answer from the reference documents answer = response["answer"] source_docs = response["source_documents"] # 5. Format Sources: Prepare a clean list of references for the frontend source_list = [] if source_docs: docs_and_scores = vs.similarity_search_with_score(data.text) for (doc, score) in docs_and_scores: source_list.append({ "doc_id": getattr(doc, "id", None), "content": doc.page_content, "source": doc.metadata.get("source", "Unknown").split("/")[-1], "score": float(score) }) return { "answer": answer, "source_list": source_list, } except Exception as e: error_msg = str(e) print(f"❌ Error during Chat Generation: {error_msg}") # 1. Handle Safety Filters (Gemini Specific) # If the input/output violates Google's safety policy (Hate speech, harassment, etc.) if "safety" in error_msg.lower() or "blocked" in error_msg.lower(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Content Policy Violation: The AI refused to answer due to safety restrictions. Please rephrase your question." ) # 2. Handle API Quota Limits (Rate Limiting) elif "429" in error_msg or "Quota" in error_msg: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Server Busy: Google AI Quota exceeded. Please try again in a minute." ) # 3. Handle Timeouts (Network Issues) elif "timed out" in error_msg.lower() or "deadline" in error_msg.lower(): raise HTTPException( status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail="Request Timeout: The AI took too long to respond. Please try again." ) # 4. Generic Server Error else: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Generation Error: An unexpected error occurred. Please try again later." )