""" app.py – Streamlit PoC: AI Matching Assistant pro ČSOB Diplomová práce – Filip Husein Perspektiva ZAMĚSTNANCE (3-polní formulář): 1. Životopis (PDF/DOCX/TXT) – volitelné 2. Současná pozice + důvod pro změnu 3. Cokoliv dodat (koníčky, preference, silné stránky…) – volitelné Systém: - Vyloučí pozice odpovídající AKTUÁLNÍ roli zaměstnance (multilinguálně: CS/SK/EN – díky fine-tuned JobBERT-v3) - Doporučí top 5 pozic v ČSOB - Ollama vygeneruje přátelskou HR odpověď Spuštění: streamlit run app.py Prerekvizity: - Fine-tuned model v models/jobbert-v3-czsk-final/ - FAISS index v index/ - Ollama běží (ollama serve) s modelem llama3.2 """ import json import os import re import sys import time import unicodedata import numpy as np import streamlit as st # ============================================================================= # Konfigurace – musí být DEFINOVÁNA před prvním použitím BASE_DIR # ============================================================================= BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # Přidej aktuální adresář do PYTHONPATH sys.path.insert(0, BASE_DIR) # Model – fallback chain: HF snapshot (stažený v build_index_startup.py) → base MODEL_PATHS = [ os.path.join(BASE_DIR, "model"), # HF snapshot os.path.join(BASE_DIR, "models", "jobbert-v3-czsk-hn-final"), # lokální (volitelné) os.path.join(BASE_DIR, "models", "jobbert-v3-czsk-final"), # fallback "TechWolf/JobBERT-v3", # base ] INDEX_DIR = os.path.join(BASE_DIR, "index") FAISS_PATH = os.path.join(INDEX_DIR, "positions.faiss") META_PATH = os.path.join(INDEX_DIR, "positions_metadata.json") # Kolik kandidátů vytáhnout z FAISS (velká rezerva po filtru) TOP_K_SEARCH = 40 # Kolik finálně zobrazit uživateli TOP_K_SHOW = 5 # ── Exclusion filter ── # 0.88 = jen opravdu podobné role (varianty "Junior/Senior X"), ne příbuzné obory. EXCLUSION_SIM_THRESHOLD = 0.88 EXCLUSION_MAX_COUNT = 30 # hard cap – nikdy nevyhodíme víc pozic # ── Rescale raw cosine similarity pro user-friendly zobrazení ── # Fine-tuned model má gap cca pos=0.83, neg=0.08. # Relevantní "sousední" role padnou do range 0.25–0.70. # Mapujeme: 0.15 → 50%, 0.85 → 99% (lineárně), clamp na [20%, 99%]. DISPLAY_MIN_RAW = 0.15 DISPLAY_MAX_RAW = 0.85 DISPLAY_MIN_PCT = 50.0 DISPLAY_MAX_PCT = 99.0 def display_score(raw: float) -> float: """Přemapuj raw cosine similarity na user-facing procento (0–100).""" if raw is None: return 0.0 t = (raw - DISPLAY_MIN_RAW) / (DISPLAY_MAX_RAW - DISPLAY_MIN_RAW) t = max(0.0, min(1.0, t)) pct = DISPLAY_MIN_PCT + t * (DISPLAY_MAX_PCT - DISPLAY_MIN_PCT) return max(20.0, min(99.0, pct)) # ============================================================================= # Auto-build index + model při prvním startu (HF Spaces) # ============================================================================= if not os.path.exists(FAISS_PATH): import subprocess subprocess.run(["python", "build_index_startup.py"], check=True) from cv_parser import extract_cv_text, summarize_cv from rag_engine import ( synthesize_profile, generate_response, check_llm_available as check_ollama_available, # alias kvůli rest kódu llm_generate as ollama_generate, _fallback_response, ) # ============================================================================= # Lazy loading (cache pro Streamlit) # ============================================================================= @st.cache_resource(show_spinner="Načítám model JobBERT-v3...") def load_model(): """Načti SentenceTransformer model (cached).""" from sentence_transformers import SentenceTransformer for path in MODEL_PATHS: if os.path.isdir(path) or not path.startswith("/"): try: model = SentenceTransformer(path) return model, path except Exception as e: st.warning(f"Model {path} nelze načíst: {e}") continue st.error("Žádný model nebyl nalezen! Spusť nejdříve trénink (12_train_jobbert.py).") st.stop() @st.cache_resource(show_spinner="Načítám FAISS index...") def load_index(): """Načti FAISS index a metadata (cached).""" import faiss if not os.path.exists(FAISS_PATH): st.error(f"FAISS index nenalezen: {FAISS_PATH}\nSpusť: python 14_build_index.py") st.stop() index = faiss.read_index(FAISS_PATH) with open(META_PATH, "r", encoding="utf-8") as f: metadata = json.load(f) return index, metadata # ============================================================================= # Text normalizace (case + diakritika insensitive) # ============================================================================= def _normalize(s: str) -> str: """Lower + NFKD + odstraň diakritiku + squeeze whitespace.""" if not s: return "" s = s.lower() s = unicodedata.normalize("NFKD", s) s = "".join(c for c in s if not unicodedata.combining(c)) s = re.sub(r"\s+", " ", s).strip() return s def _title_contains_any(title: str, phrases: list[str]) -> bool: """Vrátí True, pokud titul obsahuje kteroukoli z frází (normalizovaně).""" t = _normalize(title) for p in phrases: p_norm = _normalize(p) if p_norm and p_norm in t: return True return False # ============================================================================= # Multilinguální varianty role (LLM translate) # ============================================================================= ROLE_TRANSLATE_PROMPT = """Přelož následující název pracovní pozice do češtiny, slovenštiny a angličtiny. Vrať POUZE validní JSON ve formátu {{"cs": "...", "sk": "...", "en": "..."}}. Žádný komentář, žádný markdown, pouze JSON. Pozice: {role} JSON:""" def get_role_variants(role_text: str, use_ollama: bool = True) -> list[str]: """ Vrátí seznam variant role ve 3 jazycích (CS/SK/EN) + původní text. Fallback: pouze původní text, pokud Ollama nedostupná. """ variants = {role_text.strip()} if not use_ollama or not check_ollama_available(): return [v for v in variants if v] try: prompt = ROLE_TRANSLATE_PROMPT.format(role=role_text) raw = ollama_generate(prompt, temperature=0.0, max_tokens=200) # Zkus najít JSON v odpovědi match = re.search(r"\{[^}]*\}", raw, re.DOTALL) if match: data = json.loads(match.group(0)) for key in ("cs", "sk", "en"): val = data.get(key) if isinstance(val, str) and val.strip(): variants.add(val.strip()) except Exception: pass # při chybě pokračujeme jen s původním textem return [v for v in variants if v] # ============================================================================= # Exclusion filter (multilinguální, embedding-based cluster) # ============================================================================= def compute_exclusion_ids( current_role_text: str, model, index, metadata: list[dict], role_variants: list[str] | None = None, similarity_threshold: float = EXCLUSION_SIM_THRESHOLD, max_exclusions: int = EXCLUSION_MAX_COUNT, ) -> tuple[set[int], list[dict]]: """ Vypočti ID pozic, které budou vyloučeny z doporučení. Strategie (multilinguální): 1. Zakóduj current_role (a případné přeložené varianty) fine-tuned modelem. Díky multilinguálnímu alignmentu to funguje napříč CS/SK/EN. 2. Najdi nejbližší pozice (≥ similarity_threshold) = "same role cluster". 3. Doplň substring match na titul (normalizovaně, všechny varianty). Returns: (set indexů k vyloučení, list dictů s debug informacemi) """ if not current_role_text or not current_role_text.strip(): return set(), [] # Texty k zakódování: hlavní + všechny překlady texts = [current_role_text] if role_variants: for v in role_variants: if v and v.lower() != current_role_text.lower(): texts.append(v) # Embed všechny varianty, použij maximum cosine similarity embeddings = model.encode( texts, normalize_embeddings=True, show_progress_bar=False, ) embeddings = np.array(embeddings, dtype=np.float32) k = min(max_exclusions, index.ntotal) # Pro každou variantu vytáhni top-k, sjednoť je exclusion_ids: set[int] = set() debug_rows: list[dict] = [] for emb in embeddings: scores, indices = index.search(emb.reshape(1, -1), k) for idx, score in zip(indices[0], scores[0]): if idx < 0: continue if score >= similarity_threshold: idx_int = int(idx) if idx_int not in exclusion_ids: exclusion_ids.add(idx_int) debug_rows.append({ "index_id": idx_int, "title": metadata[idx_int].get("title", ""), "score": float(score), "reason": "semantic", }) # Substring guard: projdi VŠECHNA metadata a zachyť i ta, co embedding # nechytil (jistota pro přesná shoda titulu) all_variants = list(set(texts)) for i, meta in enumerate(metadata): if i in exclusion_ids: continue if _title_contains_any(meta.get("title", ""), all_variants): exclusion_ids.add(i) debug_rows.append({ "index_id": i, "title": meta.get("title", ""), "score": None, "reason": "substring", }) return exclusion_ids, debug_rows # ============================================================================= # Matching logika # ============================================================================= def search_positions( query_text: str, model, index, metadata, top_k: int = TOP_K_SEARCH, exclude_ids: set[int] | None = None, ): """Zakóduj dotaz a najdi nejbližší pozice ve FAISS, s filtrací exclude_ids.""" embedding = model.encode( [query_text], normalize_embeddings=True, show_progress_bar=False ) embedding = np.array(embedding, dtype=np.float32) # Vytáhni rezervu (2× top_k + velikost exclude setu) extra = len(exclude_ids) if exclude_ids else 0 k = min(top_k + extra, index.ntotal) scores, indices = index.search(embedding, k) results = [] rank_counter = 1 for idx, score in zip(indices[0], scores[0]): if idx < 0: continue idx_int = int(idx) if exclude_ids and idx_int in exclude_ids: continue meta = metadata[idx_int].copy() meta["rank"] = rank_counter meta["score"] = float(score) meta["index_id"] = idx_int results.append(meta) rank_counter += 1 if len(results) >= top_k: break return results # ============================================================================= # Build vyhledávací text # ============================================================================= def build_search_query( current_role: str, reason: str, extras: str, cv_text: str, ) -> str: """ Sestaví text pro embedding / profile synthesis. DŮLEŽITÉ: current_role se sem NEZAPOJUJE přímo (jinak by model vytahoval podobné pozice zpět). Použijeme jen: důvod změny + extras + CV. """ parts = [] if reason and reason.strip(): parts.append(f"Hledám novou pozici, protože: {reason.strip()}") if extras and extras.strip(): parts.append(f"Další informace o mně: {extras.strip()}") if cv_text and cv_text.strip(): parts.append(f"Z životopisu: {cv_text.strip()[:1500]}") # Pokud není nic, aspoň roli jako fallback (aby se něco našlo) if not parts and current_role: parts.append(current_role) return "\n\n".join(parts) # ============================================================================= # Streamlit UI # ============================================================================= def main(): # Page config st.set_page_config( page_title="ČSOB – AI Matching Assistant", page_icon="🏦", layout="wide", initial_sidebar_state="collapsed", ) # Custom CSS st.markdown(""" """, unsafe_allow_html=True) # Header st.markdown("""

🏦 AI Matching Assistant

Najdi svou další kariérní příležitost v ČSOB · Diplomová práce

""", unsafe_allow_html=True) # ─── Sidebar: Status ─── with st.sidebar: st.header("⚙️ Status systému") model, model_path = load_model() model_name = os.path.basename(model_path) if "/" in model_path else model_path is_finetuned = "final" in model_path or "czsk" in model_path if is_finetuned: st.success(f"Model: {model_name}\n(fine-tuned)") else: st.info(f"Model: {model_name}\n(base)") index, metadata = load_index() st.info(f"Pozice v indexu: {index.ntotal}") ollama_ok = check_ollama_available() if ollama_ok: st.success("LLM: Groq (llama-3.3-70b)") else: st.warning("LLM: nedostupný\n(fallback režim)") st.divider() st.markdown("**Parametry**") st.caption(f"Vyhledání: top-{TOP_K_SEARCH}") st.caption(f"Zobrazení: top-{TOP_K_SHOW}") st.caption(f"Exclusion práh: {EXCLUSION_SIM_THRESHOLD:.2f}") st.caption(f"Exclusion cap: max {EXCLUSION_MAX_COUNT}") st.divider() show_debug = st.checkbox("🔧 Debug info", value=False, help="Zobrazí raw cosine similarity a interní metriky (pro autora/vedoucího)") st.divider() st.caption("Diplomová práce – Filip Husein") st.caption("VŠE Praha · 2026") # ─── Hlavní formulář ─── st.markdown("### 📝 Vstupní údaje") st.markdown( "Vyplň pole níže. Systém pak doporučí top 5 ČSOB pozic šitých na míru – " "**s vyloučením tvé aktuální pozice** (a podobných)." ) with st.form("matching_form", clear_on_submit=False): # 1. CV upload uploaded_cv = st.file_uploader( "**1. Životopis** (volitelné – PDF / DOCX / TXT)", type=["pdf", "docx", "txt"], help="CV pomůže lépe porozumět tvým dovednostem. PII (email, telefon) se automaticky odstraní.", ) col1, col2 = st.columns([1, 2]) with col1: # 2a. Aktuální pozice current_role = st.text_input( "**2. Tvoje současná pozice** *", placeholder="např. Finanční analytik", help="Pozice, kterou aktuálně zastáváš. Vyloučíme ji (a její varianty CS/SK/EN) z doporučení.", ) with col2: # 2b. Důvod pro změnu reason = st.text_area( "**Proč hledáš změnu?** *", placeholder=( "např. Chtěl bych se posunout víc k analytice dat, " "baví mě Python a chci práci s většími datasety…" ), height=100, ) # 3. Extras extras = st.text_area( "**3. Cokoliv dodat?** (volitelné)", placeholder=( "Koníčky, silné stránky, preference (remote/hybrid), " "zájmové oblasti (AI, ESG, fintech…), jazykové schopnosti…" ), height=80, ) submit = st.form_submit_button("🔍 Najít pozice", type="primary", use_container_width=True) # ─── Zpracování ─── if submit: # Validace povinných polí if not current_role.strip(): st.error("Vyplň prosím svoji současnou pozici.") st.stop() if not reason.strip() and not uploaded_cv and not extras.strip(): st.error( "Vyplň alespoň jedno z: důvod pro změnu, životopis, nebo dodatečné informace." ) st.stop() # Extract CV cv_text = "" if uploaded_cv: with st.spinner("Zpracovávám životopis..."): raw = extract_cv_text( uploaded_cv.getvalue(), uploaded_cv.name, remove_personal=True, ) cv_text = summarize_cv(raw, max_chars=1500) if cv_text: st.success(f"✅ CV načteno ({len(cv_text)} znaků)") else: st.warning("⚠️ Nepodařilo se extrahovat text z CV.") start = time.time() # 1) Získej multilinguální varianty current_role with st.spinner("Zjišťuji varianty aktuální pozice (CS/SK/EN)..."): role_variants = get_role_variants(current_role, use_ollama=ollama_ok) # 2) Vypočti exclusion IDs with st.spinner("Filtruju aktuální pozici z kandidátů..."): exclusion_ids, exclusion_debug = compute_exclusion_ids( current_role_text=current_role, model=model, index=index, metadata=metadata, role_variants=role_variants, similarity_threshold=EXCLUSION_SIM_THRESHOLD, ) # 3) Sestav dotaz (BEZ current_role) a proveď search search_query = build_search_query( current_role=current_role, reason=reason, extras=extras, cv_text=cv_text, ) # 4) Syntéza profilu (Ollama) with st.spinner("Syntetizuji tvůj profil..."): profile_text = synthesize_profile( query=search_query, cv_text=cv_text, use_ollama=ollama_ok, ) # 5) FAISS search s exclusion filter with st.spinner(f"Hledám pozice (vyloučeno {len(exclusion_ids)})..."): results = search_positions( profile_text, model, index, metadata, top_k=TOP_K_SEARCH, exclude_ids=exclusion_ids, ) # 5b) Rescale skóre pro zobrazení (raw → user-friendly %) for r in results: r["raw_score"] = r["score"] # uchovat pro debug r["score"] = display_score(r["raw_score"]) / 100.0 # nahradit displayem # 6) Generace odpovědi with st.spinner("Generuji doporučení..."): if ollama_ok: response = generate_response( profile=profile_text, positions=results[:TOP_K_SHOW], use_ollama=True, ) else: response = _fallback_response(profile_text, results[:TOP_K_SHOW]) elapsed = time.time() - start # ─── Zobrazit výsledky ─── st.divider() # Info: co bylo vyloučeno with st.expander(f"🚫 Vyloučeno z doporučení ({len(exclusion_ids)} pozic)", expanded=False): st.markdown( f"Aktuální role **„{current_role}“** – detekovány varianty:" ) st.code(" · ".join(role_variants), language=None) st.markdown( f"Níže je seznam pozic, které byly vyloučeny, " f"protože sémanticky odpovídají tvé aktuální pozici " f"(podobnost ≥ {EXCLUSION_SIM_THRESHOLD:.2f}) nebo titul obsahuje variantu role." ) if exclusion_debug: # Seřaď podle score (None na konec) exclusion_debug.sort( key=lambda r: (r["score"] is None, -(r["score"] or 0)) ) for row in exclusion_debug[:50]: if row["score"] is not None: st.text(f" [{row['reason']:9s}] {row['score']:.3f} – {row['title']}") else: st.text(f" [{row['reason']:9s}] — – {row['title']}") if len(exclusion_debug) > 50: st.caption(f"… a dalších {len(exclusion_debug) - 50}") else: st.caption("Nic nebylo vyloučeno (aktuální role se v indexu nevyskytuje).") # CV preview if cv_text: with st.expander("📄 Extrakt z CV"): st.text(cv_text[:1000] + ("..." if len(cv_text) > 1000 else "")) # Profil (co bylo zakódováno) with st.expander("🧠 Syntetizovaný profil (vstup do matching)", expanded=False): st.info(profile_text) # Hlavní HR odpověď st.markdown("### 💬 Doporučení od AI") st.markdown(response) # Karty pozic if results: st.divider() st.markdown(f"### 📊 Top {min(TOP_K_SHOW, len(results))} pozic ({elapsed:.1f}s)") for pos in results[:TOP_K_SHOW]: score = pos["score"] # už rescaled (0–1) raw = pos.get("raw_score") # původní cosine similarity # Badge podle rescaled skóre (≥80 % výborná, 60–80 % dobrá, <60 % inspirace) if score >= 0.80: badge_emoji = "🟢" badge_label = "Výborná shoda" elif score >= 0.60: badge_emoji = "🟡" badge_label = "Dobrá shoda" else: badge_emoji = "🔵" badge_label = "Zajímavá inspirace" header = f"{badge_emoji} #{pos['rank']} – {pos['title']} ({score:.0%})" if show_debug and raw is not None: header += f" [raw={raw:.3f}]" with st.expander(header, expanded=(pos['rank'] <= 2)): cols = st.columns([3, 1]) with cols[0]: st.markdown(f"**{pos['title']}** · {badge_label}") desc = pos.get("description", "") if desc: st.write(desc[:400] + ("..." if len(desc) > 400 else "")) with cols[1]: st.metric("Shoda", f"{score:.0%}") if show_debug and raw is not None: st.caption(f"raw cos: {raw:.3f}") # Metadata meta_parts = [] if pos.get("professions"): meta_parts.append(f"**Profese:** {', '.join(pos['professions'])}") if pos.get("fields"): meta_parts.append(f"**Obor:** {', '.join(pos['fields'])}") if pos.get("location"): meta_parts.append(f"**Lokace:** {pos['location']}") if pos.get("employment_types"): meta_parts.append(f"**Úvazek:** {', '.join(pos['employment_types'])}") if pos.get("suitable_for_graduate"): meta_parts.append("✅ Vhodné pro absolventy") if pos.get("url"): meta_parts.append(f"[🔗 Odkaz na pozici]({pos['url']})") if meta_parts: st.markdown(" · ".join(meta_parts)) # Debug tabulka s raw skóre if show_debug: st.divider() st.markdown("#### 🔧 Debug – raw cosine similarity") debug_rows = [ { "rank": r["rank"], "title": r["title"][:60], "raw_cos": round(r.get("raw_score", 0), 4), "display_%": f"{r['score']*100:.1f}%", } for r in results[:TOP_K_SHOW] ] st.table(debug_rows) st.caption( f"Mapování: raw {DISPLAY_MIN_RAW:.2f} → {DISPLAY_MIN_PCT:.0f} % | " f"raw {DISPLAY_MAX_RAW:.2f} → {DISPLAY_MAX_PCT:.0f} % (lineárně, clamp 20–99 %)" ) else: st.warning( "Po odfiltrování aktuální pozice nezbyly žádné kandidátní pozice. " "Zkus upřesnit dotaz, rozšířit popis, nebo snížit práh exclusion filtru." ) if __name__ == "__main__": main()