CiE .dZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ejeZdZdZd gezZd ed<d Zd efd ZdadZded eefdZd0dededed efdZ d1dededeed eefdZ d2deededeededed ef dZ d3dedeededed ef dZ!ed k(rddl"Z"ddl#Z#e"jHd!"Z%e%jMdd#$e%jMd%edd&'e%jMd(edd)'e%jMd*dd+,e%jOZ(e#jRe!e(jTe(jVe(jXe(jZ-Z.e/ej`e.d.e/yy)4u  Quality Gate — KB Ingestion Pipeline Module 12 =============================================== Auto-generated quiz from KB chunks + RAG accuracy evaluation. Stories: 12.01 — generate_quiz: Create Q&A pairs from random KB chunks via Gemini 12.02 — evaluate_accuracy: Run quiz against RAG pipeline and measure hit rate 12.03 — run_quality_gate: Full pipeline (generate → evaluate → report) Usage: python3 -m core.kb.quality_gate hubspot python3 -m core.kb.quality_gate hubspot --questions 10 --threshold 0.9 Dependencies: from core.kb.qdrant_store import search_platform from core.kb.embedder import embed_text from core.rag_query import rag_query from core.kb.pg_store import get_connection import google.genai as genai N)Optional)search_platform) embed_text) rag_queryzgemini-2.0-flashi gư>a<You are a QA engineer generating a quiz to test a retrieval system. Given the following knowledge-base chunk, generate ONE question whose answer is clearly contained within the chunk text. Requirements: - The question must be specific and answerable from the chunk alone. - The answer must be a short phrase or sentence (not a list) extracted or directly inferable from the chunk. - Output ONLY valid JSON in this exact format (no markdown, no extra text): {{"question": "", "answer": ""}} Chunk title: {title} Chunk text: {text} returncvtjdd}|sd}tjj|rrt |5}|D]W}|j }|j ds%|jdddj j d}nddd|S|S#1swY|SxYw) z9Load GEMINI_API_KEY from environment or secrets.env file.GEMINI_API_KEYz(/mnt/e/genesis-system/config/secrets.envzGEMINI_API_KEY==z'"N)osgetenvpathexistsopenstrip startswithsplit)key secrets_pathfhlines -/mnt/e/genesis-system/core/kb/quality_gate.py _load_api_keyrDs ))$b )C A 77>>, 'l# rD::>% ))E"1%C~~f%!"g ZZ_ R 82#5 NNMsSWTWy Y ?EsB-C64C66 D?DDab min_wordscv|jj}|j}|sytt||z dzD]}dj ||||z}||vsyt d|D}|syt |j}t||zt|z } | dk\S)z Heuristic: return True if at least min_words consecutive words from `a` appear as a sub-sequence in `b` (case-insensitive). Used to decide whether a RAG result text contains the expected answer. Fr  Tc3>K|]}t|dkDs|yw)N)len).0ws r z _text_overlap..s4Q!14s?)lowerrrangerGjoinset) r@rArBa_wordsb_loweriphrasea_uniqueb_words overlap_ratios r _text_overlaprWsggiooGggiG  3w<)+a/ 0'!a)m45 W  4g44H '--/"G7*+c(m;M C r#platform num_questions customer_idc ` t|d}t |||dd}|stj d|gSt|t|}tj||}g}|D]n} t| } | |j| d| d | jd d | jd d | jd d | jdd dptj dt||||S#t$r'}tjd|t}Yd}~d}~wwxYw)aI Generate quiz questions from ingested KB to test RAG accuracy. Steps: 1. Get a broad set of chunks from Qdrant for the platform (up to 200). 2. Randomly sample num_questions chunks (or all if fewer available). 3. For each chunk, call Gemini to generate a question answerable from that chunk. 4. Return list of dicts: {question, expected_answer, source_chunk_id, source_text, source_url} Chunks that fail Gemini generation are silently skipped. Args: platform: The platform to build the quiz for (e.g., "hubspot"). num_questions: Desired number of quiz items (actual may be lower if fewer chunks exist or Gemini fails for some). customer_id: Optional customer scope for multi-tenant isolation. Returns: List of quiz item dicts. z documentation knowledge basezEembed_text failed during quiz generation (%s), using discovery vectorNr.r) query_vectorrYr[top_kscore_thresholdz.generate_quiz: no chunks found for platform=%sr,r-idr r' source_urlr&)r,expected_answersource_chunk_id source_textra source_titlez8generate_quiz: generated %d/%d questions for platform=%s)rr8r6r7_DISCOVERY_VECTORrinfominrGrandomsampler?appendr1) rYrZr[r]r>chunks sample_sizesampledquizr$r=s r generate_quizrps78)!XJ.K"LM ! F  DhO mS[1KmmFK0GD   )% 0 :  :!(|$yyr2 99VR0))L"5!IIgr2     KKB D ; KO )^`cd( )sC== D-D((D-皙?ropass_thresholdr^c 4|s |dddd|gdgdSd}g}|D]}|d}|d} |jdd } t|| } d} d}| D]]}|dk(r|jdd}|jdd }| r |r | |k(rd} n'|jdd }|sNt | |s[d} n| r|dz }|j || | | t|ddt|}|dkDr||z nd}||k\}g}|s`||z }|j d|dd|dd|dd|dkr|j d|dk\r|j d|j d|||t|d||||dS#t$r'} tj d |d d | g} Yd } ~ Pd } ~ wwxYw)u Run quiz against the RAG pipeline and measure retrieval accuracy. For each quiz item: 1. Call rag_query(question, platform, top_k=top_k). 2. Mark as "correct" if any result's source_url matches the quiz source_url OR if any result's text has significant overlap with the expected_answer. Args: quiz: Output from generate_quiz(). platform: Platform to evaluate. customer_id: Optional customer scope (reserved for future use). pass_threshold: Minimum accuracy to consider the gate passed (0.0–1.0). top_k: Number of RAG results to check per question. Returns: { "platform": str, "total_questions": int, "correct": int, "accuracy": float, "passed": bool, "threshold": float, "details": list[dict], "recommendations": list[str], } rrFu5No quiz items provided — run generate_quiz() first.)rYtotal_questionscorrectaccuracypassed thresholddetailsrecommendationsr,rbrar )r^z$rag_query failed for question=%r: %sNPscoreTr'r r+)r,rbrafound_in_top_ktop_result_scorez Accuracy z.1%z is z below the z threshold.rKzLow accuracy (<50%) suggests the platform KB may have very few indexed chunks. Re-run the ingestion pipeline to populate Qdrant.zUConsider increasing chunk overlap or reducing chunk size to improve retrieval recall.zQReview failed items in 'details' to identify poorly chunked or ambiguous content.) r1rr8r6r7rWrkroundrG)rorYr[rrr^ruryitemr,rbraresultsr>foundr~result result_url result_texttotalrvrwrzgaps revaluate_accuracyrsDD   ' WX  GG+  #01XXlB/  6G "% F3&#)::gs#;  L"5JjZ:-E!**VR0K}_kJ   qLG .$# %&6 :   K+ Z IE"'!)wH  'F"$O x'~T#c+nS=QQ\ ]  c>  " "D  s?  " "g   _  (A&#*  w  NNA8CR=RU VG s E'' F0FFcKtjd|||t|||}|s tjd||dd|ddSt ||||}||d<|drd nd |d <tjd ||d dz|d |Sw)u Full quality gate pipeline: generate quiz → evaluate → return combined report. This is an async function so it can be awaited from async contexts (e.g., an orchestrator or FastAPI endpoint), but internally all work is synchronous — no I/O concurrency is introduced here. Args: platform: Platform KB to evaluate. customer_id: Optional customer scope. num_questions: Number of quiz questions to generate. pass_threshold: Accuracy fraction required to pass (default 0.80). Returns: Combined report dict. If no chunks found, returns a minimal NO_DATA report. zHrun_quality_gate: starting for platform=%s, questions=%d, threshold=%.2fuGrun_quality_gate: no chunks found for platform=%s — returning NO_DATANO_DATAzNo chunks found for platform 'z<'. Ensure KB ingestion has been run before the quality gate.)rYstatusmessagerorwPASSEDFAILEDrz7run_quality_gate: platform=%s accuracy=%.1f%% status=%srvd)r6rgrpr7r)rYr[rZrrrors rrun_quality_gateros. KKR- =+ >D `bjk 7zBSS  tX{N KFF6N#)(#3xF8 KKA&$s*F8,< MsB B __main__u7KB Quality Gate — auto quiz + RAG accuracy evaluation) descriptionz$Platform to evaluate (e.g., hubspot))helpz --questionsz2Number of quiz questions to generate (default: 20))typedefaultrz --thresholdu1Accuracy pass threshold 0.0–1.0 (default: 0.80)z --customer-idz5Optional customer_id scope for multi-tenant isolation)rr)rYr[rZrr)indentr)r+)rXN)NrqrF)NrXrq)1__doc__r*loggingrritypingrcore.kb.qdrant_storercore.kb.embedderrcore.rag_queryr getLogger__name__r6r4_DISCOVERY_VECTOR_DIMrfr/strrrr"dictr?intboolrWlistrpfloatrrargparseasyncioArgumentParserparser add_argument parse_argsargsrunrYr[ questionsrxreportprintdumpsr#rrs,  0'$   8 $ " E11! & s  $@ S S S  >!%EEE#E $Z E^"&   t*#     P"& //#// /  /l z $X $ $MF  )OP C A E4 @  D    D W[[]]((..>>  F *$**VAs 34=r#