import os import json import time import uvicorn import itertools from fastapi import FastAPI, HTTPException, Security, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.security import APIKeyHeader from pydantic import BaseModel from langchain_google_genai import GoogleGenerativeAIEmbeddings from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_groq import ChatGroq from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_core.documents import Document from langchain_classic.chains import ConversationalRetrievalChain from langchain_core.prompts import PromptTemplate # ── Env ─────────────────────────────────────────────────────────────────────── GROQ_API_KEY = os.environ.get("GROQAPIKEY", "") API_KEY = os.environ.get("APIKEY", "") os.environ["GROQ_API_KEY"] = GROQ_API_KEY KNOWLEDGE_BASE_DIR = "knowledge_base" # ── Auth ────────────────────────────────────────────────────────────────────── api_key_header = APIKeyHeader(name="X-API-Key", auto_error=True) def verify_key(key: str = Security(api_key_header)): if key != API_KEY: raise HTTPException(status_code=403, detail="Geçersiz API key.") # ── App ─────────────────────────────────────────────────────────────────────── app = FastAPI( docs_url=None, redoc_url=None, openapi_url=None, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["POST", "GET"], allow_headers=["X-API-Key", "Content-Type"], ) # ── Global state ────────────────────────────────────────────────────────────── qa_chain = None chat_histories: dict = {} # ── Google API Key Rotation ─────────────────────────────────────────────────── def get_google_api_keys() -> list: keys = [] i = 1 while True: suffix = "" if i == 1 else str(i) key = os.environ.get(f"GOOGLEAPIKEY{suffix}", "") if key: keys.append(key) i += 1 else: break return keys def get_hf_embeddings(): print("🔄 Loading HuggingFace embeddings (fallback)...") emb = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) print("✅ HuggingFace embeddings loaded.") return emb def embed_with_rotation(docs: list, google_keys: list): """Embed documents using Google keys in rotation, fall back to HuggingFace.""" if not google_keys: print("⚠️ No Google keys found, using HuggingFace for all chunks.") return Chroma.from_documents(docs, get_hf_embeddings()) key_cycle = itertools.cycle(google_keys) batch_size = 80 db = None for i in range(0, len(docs), batch_size): batch = docs[i:i + batch_size] batch_num = i // batch_size + 1 print(f"📤 Embedding batch {batch_num} ({len(batch)} chunks)...") tried_keys = set() success = False while not success: key = next(key_cycle) if key in tried_keys: print(f"⚠️ All Google keys failed for batch {batch_num}, switching to HuggingFace.") hf = get_hf_embeddings() if db is None: db = Chroma.from_documents(batch, hf) else: db.add_documents(batch) success = True break tried_keys.add(key) try: os.environ["GOOGLE_API_KEY"] = key embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-001") if db is None: db = Chroma.from_documents(batch, embeddings) else: db.add_documents(batch) success = True print(f"✅ Batch {batch_num} embedded with Google key.") except Exception as e: err = str(e) if "429" in err or "RESOURCE_EXHAUSTED" in err: print(f"⚠️ Key rate limited, trying next Google key...") else: print(f"⚠️ Key error: {err[:120]}, trying next Google key...") return db # ── JSON → Readable Text ────────────────────────────────────────────────────── def json_to_text(data: dict, source_type: str, filename: str = "") -> str: if source_type == "personal": langs = ", ".join([f"{l['language']} ({l['level']})" for l in data.get("languages", [])]) return f"""PERSONAL INFORMATION Name: {data.get('name')} | Full name: {data.get('full_name')} Title: {data.get('title')} Summary: {data.get('summary')} Email: {data.get('email')} Phone: {data.get('phone')} Location: {data.get('location')} Website: {data.get('website')} LinkedIn: {data.get('linkedin')} GitHub: {data.get('github')} Instagram: {data.get('instagram')} YouTube: {data.get('youtube')} Languages: {langs} Open to: {data.get('open_to')}""" elif source_type == "education": lines = ["EDUCATION"] for edu in data.get("education", []): lines.append(f""" Degree: {edu.get('degree')} Institution: {edu.get('institution')} | Location: {edu.get('location')} Period: {edu.get('period')} GPA: {edu.get('gpa')} Achievement: {edu.get('achievement')} Link: {edu.get('link', 'N/A')}""") return "\n".join(lines) elif source_type == "skills": s = data.get("skills", {}) return f"""SKILLS Programming Languages: {", ".join(s.get("languages", []))} Frameworks: {", ".join(s.get("frameworks", []))} AI & ML: {", ".join(s.get("ai_ml", []))} Databases: {", ".join(s.get("databases", []))} Tools: {", ".join(s.get("tools", []))}""" elif source_type == "experience": lines = ["WORK EXPERIENCE"] for exp in data.get("experience", []): highlights = "\n".join([f"- {h}" for h in exp.get("highlights", [])]) links = ", ".join(exp.get("links", [])) lines.append(f""" Title: {exp.get('title')} | Type: {exp.get('type')} Company: {exp.get('company', 'Freelance')} Period: {exp.get('period')} Technologies: {", ".join(exp.get('technologies', []))} Links: {links if links else 'N/A'} Highlights: {highlights}""") lines.append("\nINTERNSHIPS") for intern in data.get("internships", []): highlights = "\n".join([f"- {h}" for h in intern.get("highlights", [])]) lines.append(f""" Title: {intern.get('title')} | Company: {intern.get('company')} Location: {intern.get('location', '')} Period: {intern.get('period')} Technologies: {", ".join(intern.get('technologies', []))} Highlights: {highlights}""") return "\n".join(lines) elif source_type == "volunteering": lines = ["VOLUNTEERING"] for v in data.get("volunteering", []): lines.append(f""" Title: {v.get('title')} Organization: {v.get('organization')} Period: {v.get('period')} Description: {v.get('description')} Document: {v.get('document_image', 'N/A')}""") return "\n".join(lines) elif source_type == "certificates": lines = ["CERTIFICATES"] for c in data.get("certificates", []): lines.append(f"- {c.get('title')} | {c.get('issuer')} | {c.get('year')} | Category: {c.get('category')}") return "\n".join(lines) elif source_type == "references": lines = ["REFERENCES"] for r in data.get("references", []): lines.append(f""" Name: {r.get('name')} Title: {r.get('title')} Institution: {r.get('institution')} Email: {r.get('email')} Profile: {r.get('profile_url', 'N/A')}""") return "\n".join(lines) elif source_type == "project": techs = ", ".join([t.get("name", "") for t in data.get("technologies", [])]) tags = ", ".join(data.get("tags", [])) challenges = "\n".join([f"- {c}" for c in data.get("challenges", [])]) solutions = "\n".join([f"- {s}" for s in data.get("solutions", [])]) results = "\n".join([ f"- {r.get('metric')}: {r.get('value')} — {r.get('description')}" for r in data.get("results", []) ]) content_text = "" for block in data.get("contentBlocks", []): if block.get("type") == 0: content_text += f"\n### {block.get('heading', '')}\n{block.get('content', '')}\n" return f"""PROJECT: {data.get('title')} Subtitle: {data.get('subtitle')} Description: {data.get('description')} Category: {", ".join(data.get("category", []))} Tags: {tags} Technologies: {techs} Date: {data.get('date')} Duration: {data.get('duration', 'N/A')} Team Size: {data.get('teamSize')} Role: {data.get('role')} Demo Link: {data.get('demoLink', 'N/A')} GitHub: {data.get('githubLink', 'N/A')} Featured: {data.get('isFeatured')} Long Description: {data.get('longDescription', '')} {content_text} Challenges: {challenges} Solutions: {solutions} Results: {results}""" elif source_type == "article": content_text = "" for block in data.get("contentBlocks", []): if block.get("type") == 0: content_text += f"\n### {block.get('heading', '')}\n{block.get('content', '')}\n" return f"""ARTICLE: {data.get('title')} Subtitle: {data.get('subtitle', '')} Description: {data.get('description', '')} Category: {data.get('category', '')} Tags: {", ".join(data.get("tags", []))} Date: {data.get('date', '')} Link: {data.get('link', data.get('demoLink', 'N/A'))} {content_text}""" return json.dumps(data, ensure_ascii=False, indent=2) # ── Load all JSON files ─────────────────────────────────────────────────────── def load_knowledge_base() -> list: documents = [] root_files = { "personal.json": "personal", "education.json": "education", "skills.json": "skills", "experience.json": "experience", "volunteering.json": "volunteering", "certificates.json": "certificates", "references.json": "references", } for filename, source_type in root_files.items(): filepath = os.path.join(KNOWLEDGE_BASE_DIR, filename) if os.path.exists(filepath): try: with open(filepath, "r", encoding="utf-8") as f: data = json.load(f) text = json_to_text(data, source_type, filename) documents.append(Document( page_content=text, metadata={"source": filename, "type": source_type} )) print(f"✅ Loaded: {filename}") except Exception as e: print(f"❌ Error loading {filename}: {e}") else: print(f"⚠️ Not found: {filepath}") projects_dir = os.path.join(KNOWLEDGE_BASE_DIR, "projects") if os.path.exists(projects_dir): for fname in sorted(os.listdir(projects_dir)): if fname.endswith(".json"): filepath = os.path.join(projects_dir, fname) try: with open(filepath, "r", encoding="utf-8") as f: data = json.load(f) text = json_to_text(data, "project", fname) documents.append(Document( page_content=text, metadata={ "source": f"projects/{fname}", "type": "project", "title": data.get("title", fname), "isFeatured": data.get("isFeatured", False), "demoLink": data.get("demoLink", ""), "githubLink": data.get("githubLink", ""), "category": ", ".join(data.get("category", [])), } )) print(f"✅ Loaded project: {fname}") except Exception as e: print(f"❌ Error loading project {fname}: {e}") articles_dir = os.path.join(KNOWLEDGE_BASE_DIR, "articles") if os.path.exists(articles_dir): for fname in sorted(os.listdir(articles_dir)): if fname.endswith(".json"): filepath = os.path.join(articles_dir, fname) try: with open(filepath, "r", encoding="utf-8") as f: data = json.load(f) text = json_to_text(data, "article", fname) documents.append(Document( page_content=text, metadata={ "source": f"articles/{fname}", "type": "article", "title": data.get("title", fname), "link": data.get("link", data.get("demoLink", "")), } )) print(f"✅ Loaded article: {fname}") except Exception as e: print(f"❌ Error loading article {fname}: {e}") return documents # ── Build RAG chain ─────────────────────────────────────────────────────────── def build_chain(): documents = load_knowledge_base() if not documents: raise ValueError("Knowledge base is empty.") splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=200) docs = splitter.split_documents(documents) print(f"📦 Total chunks: {len(docs)}") google_keys = get_google_api_keys() print(f"🔑 Found {len(google_keys)} Google API key(s).") db = embed_with_rotation(docs, google_keys) retriever = db.as_retriever(search_type="similarity", search_kwargs={"k": 6}) qa_prompt = PromptTemplate( input_variables=["context", "question"], template="""You are an AI assistant for Yasir Alrawi's personal portfolio website. Use ONLY the information provided in the context below. NEVER invent, guess, or generate links, URLs, names, or facts that are not explicitly in the context. If a link exists in the context, use it exactly as written. If no link exists, do not provide one. LANGUAGE RULE: Always detect and match the language of the CURRENT question only. If the current question is in English, answer in English. If Turkish, answer in Turkish. If Arabic, answer in Arabic. Do NOT follow the language of previous messages. STYLE RULES: - In Turkish: use pronouns naturally. Use "Yasir" by name only when needed for clarity. - In English: use "he" naturally. - Do NOT start sentences with just "O," alone. - Do NOT repeat the full name "Yasir Alrawi" in every sentence. - Be natural, friendly and professional. --- PROJECT LISTING RULES — Apply when the user asks "what projects do you have", "projelerin neler", "show me your projects" or similar broad project listing questions: - Group projects by their Category field. - For each project show ONLY the name and its link. Use demoLink if available, otherwise githubLink. If neither exists, show just the name. - Do NOT write descriptions, technologies, or any other details. - Keep it clean and brief. Example format: Yasir'in çeşitli alanlarda projeleri var: **Yapay Zeka** - Proje Adı [link] **Web** - Proje Adı [link] --- FEATURED / LATEST PROJECTS RULE — Apply when the user asks "latest projects", "son projeler", "featured projects", "öne çıkan projeler" or similar: - Show ONLY projects where isFeatured = true. - List each with its name and link (demoLink first, otherwise githubLink). - Do NOT include non-featured projects or extra details. --- ARTICLE RULES — When asked about articles or blog posts: - Always include the article's link if one exists in the context. --- REFERENCE DETAIL RULES — When asked about a specific reference person: - Always include their profile_url link if it exists in the context. --- EDUCATION DETAIL RULES — When asked about a specific school or university: - Always include the institution link if it exists in the context. --- PROJECT DETAIL RULES — When asked for details about a specific project: - Give a brief summary: what the project is, Yasir's role, key technologies used, and notable results. - Do NOT paste every field verbatim. Keep it conversational and concise. - Always end your answer with (in the detected language): - Turkish: "Proje hakkında merak ettiğin spesifik bir şey varsa sorabilirsin!" - English: "Feel free to ask if you'd like to know more about a specific aspect of the project!" - Arabic: "لا تتردد في السؤال إذا أردت معرفة المزيد عن جانب معين من المشروع!" --- Context: {context} Question: {question} Answer:""" ) chain = ConversationalRetrievalChain.from_llm( llm=ChatGroq( model="llama-3.1-8b-instant", temperature=0, api_key=GROQ_API_KEY, ), chain_type="stuff", retriever=retriever, combine_docs_chain_kwargs={"prompt": qa_prompt}, return_source_documents=True, return_generated_question=True, ) return chain # ── Startup ─────────────────────────────────────────────────────────────────── @app.on_event("startup") def startup(): global qa_chain try: qa_chain = build_chain() print("✅ Knowledge base loaded successfully.") except Exception as e: print(f"❌ Failed to load knowledge base: {e}") # ── Schemas ─────────────────────────────────────────────────────────────────── class AskRequest(BaseModel): question: str session_id: str = "default" class AskResponse(BaseModel): answer: str sources: list[dict] # ── Endpoints ───────────────────────────────────────────────────────────────── @app.get("/health") def health(): return {"status": "ok"} @app.post("/ask", response_model=AskResponse, dependencies=[Depends(verify_key)]) def ask(body: AskRequest): global chat_histories if qa_chain is None: raise HTTPException(status_code=503, detail="Model henüz hazır değil.") question = body.question.strip() session_id = body.session_id if not question: raise HTTPException(status_code=422, detail="Soru boş olamaz.") if session_id not in chat_histories: chat_histories[session_id] = [] session_history = chat_histories[session_id] if session_history: history_str = "\n".join( f"User: {q}\nAssistant: {a}" for q, a in session_history[-4:] ) full_question = f"Previous conversation:\n{history_str}\n\nNew question: {question}" else: full_question = question result = None last_error = None for attempt in range(3): try: result = qa_chain.invoke({"question": full_question, "chat_history": []}) break except Exception as e: last_error = e if attempt < 2: time.sleep(3) if result is None: raise HTTPException(status_code=503, detail=f"Model hatası: {str(last_error)[:200]}") answer = result["answer"] src_docs = result.get("source_documents", []) sources, seen = [], set() for doc in src_docs: src = doc.metadata.get("source", "Unknown") if src not in seen: seen.add(src) sources.append({ "file": src, "type": doc.metadata.get("type", ""), "title": doc.metadata.get("title", os.path.basename(src)), "excerpt": doc.page_content[:300], }) session_history.append((question, answer)) if len(session_history) > 10: session_history.pop(0) chat_histories[session_id] = session_history if len(chat_histories) > 100: oldest = list(chat_histories.keys())[0] del chat_histories[oldest] return AskResponse(answer=answer, sources=sources) # ── Run ─────────────────────────────────────────────────────────────────────── if __name__ == "__main__": uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)