import os # Configure thread pool limits BEFORE PyTorch/numpy are imported to conserve memory in Codespace os.environ["OMP_NUM_THREADS"] = "1" os.environ["MKL_NUM_THREADS"] = "1" os.environ["OPENBLAS_NUM_THREADS"] = "1" os.environ["VECLIB_MAXIMUM_THREADS"] = "1" os.environ["NUMEXPR_NUM_THREADS"] = "1" import gc import hashlib import re from datetime import datetime from typing import Optional, Any, List, Dict, cast, AsyncGenerator import numpy as np import json import logging from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field, field_validator from supabase import create_client, Client from groq import Groq from fastapi.middleware.cors import CORSMiddleware from sentence_transformers import SentenceTransformer, CrossEncoder from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from bhashini import translate_text, LANGUAGE_CODES from eligibility.engine import check_eligibility from neuro_symbolic import run_neuro_symbolic_pipeline from graph_rag import run_graph_rag from whatsapp.webhook import router as whatsapp_router from config import settings from services.rag_coordinator import parse_interleaved_stream # ── PROJECT INDRA: Poincaré Re-Ranking Engine (Sprint 29) ───── # Instantiated globally at startup to claim memory once. # batch_size=400 matches the get_indra_candidates() oversample_limit. from indra_engine import IndraProjectionEngine indra_engine = IndraProjectionEngine(batch_size=400, dimensions=768) # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") logger = logging.getLogger("govbridge.api") # Variables for AI models embedding_model = None cross_encoder_model = None def get_embedding_model(): global embedding_model if embedding_model is None: logger.info("⏳ Lazily loading Nomic Embedding Model (768-dim)...") embedding_model = SentenceTransformer("nomic-ai/nomic-embed-text-v1", trust_remote_code=True) return embedding_model def get_reranker_model(): global cross_encoder_model if cross_encoder_model is None: logger.info("⏳ Lazily loading Cross-Encoder Reranker Model...") cross_encoder_model = CrossEncoder("cross-encoder/ettin-reranker-68m-v1", max_length=512, device='cpu', trust_remote_code=True) return cross_encoder_model # --- SECURE KEYS --- SUPABASE_URL = settings.SUPABASE_URL SUPABASE_KEY = settings.SUPABASE_KEY GROQ_API_KEY = settings.GROQ_API_KEY ADMIN_SECRET = settings.ADMIN_SECRET supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY) groq_client = Groq(api_key=GROQ_API_KEY) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("🚀 Lifespan started (lazy model loading enabled).") yield app = FastAPI(lifespan=lifespan) app.include_router(whatsapp_router) # --- RATE LIMITER CONFIG --- limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, cast(Any, _rate_limit_exceeded_handler)) # --- CORS MIDDLEWARE --- app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], expose_headers=["X-Sources", "X-Translation-Provider"], ) from middleware.memory_guard import MemoryGuardMiddleware app.add_middleware(MemoryGuardMiddleware) # --- MODELS --- class SearchQuery(BaseModel): question: str = Field(..., min_length=3, max_length=500) language: str = Field(default="english") @field_validator('language') @classmethod def validate_language(cls, v): valid = ["english","hindi","tamil","bengali","telugu", "marathi","gujarati","kannada","malayalam","punjabi"] if v.lower() not in valid: return "english" return v.lower() @field_validator('question') @classmethod def clean_input(cls, v): v = re.sub(r'<[^>]+>', '', v) v = re.sub(r'\s+', ' ', v).strip() if not v: raise ValueError('Question is empty after cleaning') return v class IngestRequest(BaseModel): title: str text: str ministry: Optional[str] = None state: Optional[str] = None source_url: Optional[str] = None doc_type: Optional[str] = "scheme" class EligibilityRequest(BaseModel): annual_income: float = Field(default=0, ge=0) age: int = Field(default=0, ge=0, le=120) is_farmer: bool = False state: str = "" caste_category: str = "General" gender: str = "Unknown" is_bpl: bool = False land_size_hectares: float = Field(default=0.0, ge=0) is_disabled: bool = False occupation: str = "Unknown" class GazetteSearchQuery(BaseModel): """Search query for the Gazette Vault hybrid search pipeline.""" query: str = Field(..., min_length=2, max_length=500) gazette_type: Optional[str] = Field(default=None, description="Filter: central, state, extraordinary") state: Optional[str] = Field(default=None, description="Filter by state applicability") limit: int = Field(default=10, ge=1, le=50) @field_validator('query') @classmethod def clean_gazette_query(cls, v: str) -> str: v = re.sub(r'<[^>]+>', '', v) v = re.sub(r'\s+', ' ', v).strip() if not v: raise ValueError('Query is empty after cleaning') return v @field_validator('gazette_type') @classmethod def validate_gazette_type(cls, v: Optional[str]) -> Optional[str]: if v is None: return None valid = ["central", "state", "extraordinary"] if v.lower() not in valid: return None return v.lower() class GraphNode(BaseModel): """A node in the knowledge graph visualization.""" id: str name: str group: str # 'anchor' | 'hop1' | 'hop2' class GraphLink(BaseModel): """A directional edge in the knowledge graph visualization.""" source: str target: str label: str hop: int class GraphResponse(BaseModel): """Force-graph-compatible response payload.""" anchor_id: str nodes: List[GraphNode] links: List[GraphLink] total_nodes: int total_links: int def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]: paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()] chunks = [] current = "" for para in paragraphs: if len(current) + len(para) < chunk_size: current += " " + para else: if current.strip(): chunks.append(current.strip()) current = para if current.strip(): chunks.append(current.strip()) if len(chunks) <= 1: return chunks overlapped = [chunks[0]] for i in range(1, len(chunks)): tail = chunks[i-1][-overlap:] if len(chunks[i-1]) > overlap else chunks[i-1] overlapped.append(tail + " " + chunks[i]) return overlapped def rerank_chunks(query: str, chunks: list) -> list: if not chunks: return [] # Selective Reranking: Pre-filter to top 20 candidates to reduce CPU load candidates = chunks[:20] reranker = get_reranker_model() pairs = [[query, c.get('chunk_text', '')] for c in candidates] # Optimization: batch_size=32 and explicit Device handling (CPU) scores = reranker.predict(pairs, batch_size=32, show_progress_bar=False) for i, score in enumerate(scores): candidates[i]['rerank_score'] = float(score) candidates.sort(key=lambda x: x['rerank_score'], reverse=True) return candidates[:5] def rerank_gazette_chunks(query: str, chunks: list) -> list: """ Cross-encoder re-ranking for gazette search results. Takes top 30 RRF candidates (gazette chunks are denser than scheme chunks), runs them through Ettin reranker, applies a hard quality gate, and returns ONLY genuinely relevant results. QUALITY GATE (5.0 raw logit): ───────────────────────────────────────────────────── The Ettin reranker outputs raw logits roughly in [-10, +15]. Empirical calibration from GovBridge test corpus: Score ≥ 8.0 → Highly relevant (exact topic match) Score 6-8 → Relevant (same domain, related content) Score 5-6 → Borderline (tangential relevance) Score 3-5 → NOISE. Garbage queries, random names, and unrelated terms consistently score here. Score < 3 → Definitively irrelevant. FLOOR = 5.0 eliminates ALL noise while preserving every genuinely relevant result. A World No. 1 system NEVER shows irrelevant results — it shows "No results found" instead. ───────────────────────────────────────────────────── """ if not chunks: return [] RELEVANCE_FLOOR = 5.0 # Hard quality gate restored for production # Gazette-specific: wider pool (30) because legislative text # has higher semantic density than scheme descriptions candidates = chunks[:30] reranker = get_reranker_model() pairs = [[query, c.get('chunk_text', '')] for c in candidates] scores = reranker.predict(pairs, batch_size=32, show_progress_bar=False) for i, score in enumerate(scores): candidates[i]['cross_encoder_score'] = float(score) candidates.sort(key=lambda x: x['cross_encoder_score'], reverse=True) # Hard quality gate: Remove all results below the relevance floor filtered = [c for c in candidates if c['cross_encoder_score'] >= RELEVANCE_FLOOR] dropped = len(candidates) - len(filtered) if dropped > 0: print(f" 🔽 Quality gate: {dropped}/{len(candidates)} results below floor ({RELEVANCE_FLOOR})") if not filtered: print(f" ⛔ ALL results below quality gate — returning empty (best was {candidates[0]['cross_encoder_score']:.2f})") return filtered[:10] # --- ENDPOINTS --- @app.get("/health") async def health_check(): checks = {} try: model = get_embedding_model() test_vec = model.encode("search_query: test", normalize_embeddings=True) checks["embedding_model"] = {"status": "ok", "dims": len(test_vec)} except Exception as e: checks["embedding_model"] = {"status": "error", "detail": str(e)} try: supabase.table("document_chunks").select("id").limit(1).execute() checks["supabase"] = {"status": "ok"} except Exception as e: checks["supabase"] = {"status": "error", "detail": str(e)} checks["groq"] = { "status": "ok" if settings.GROQ_API_KEY else "missing" } all_ok = all(v["status"] == "ok" for v in checks.values()) return { "status": "healthy" if all_ok else "degraded", "timestamp": datetime.utcnow().isoformat(), "checks": checks } @app.get("/debug/memory") async def debug_memory(): """Live memory inspection endpoint (Sprint 18).""" from middleware.memory_guard import get_memory_mb return { "rss_mb": round(get_memory_mb(), 1), "gc_counts": gc.get_count(), "gc_threshold": gc.get_threshold(), "translation_provider": "groq-llm", "embedding_loaded": embedding_model is not None, "reranker_loaded": reranker_model is not None, } @app.post("/api/eligibility/check") async def check_scheme_eligibility(request: EligibilityRequest): profile = request.model_dump() results = check_eligibility(profile) eligible_schemes = [k.replace("eligible_", "").replace("_", " ").title() for k, v in results.items() if v] return { "eligible_schemes": eligible_schemes, "full_results": results, "profile_checked": profile } @app.post("/api/admin/ingest") async def ingest_document(request: IngestRequest, admin_key: str = ""): if admin_key != ADMIN_SECRET: async def denied(): yield "Unauthorized" return StreamingResponse(denied(), status_code=401, media_type="text/plain") doc_hash = hashlib.sha256(request.text.encode()).hexdigest()[:16] chunks = chunk_text(request.text) model = get_embedding_model() # Nomic requires search_document prefix for schemes/documents prefixed_chunks = [f"search_document: {c}" for c in chunks] embeddings = model.encode(prefixed_chunks, normalize_embeddings=True, batch_size=32, show_progress_bar=False).tolist() rows = [] dual_write = settings.DUAL_WRITE_EMBEDDINGS.lower() == "true" for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): row = { "chunk_index": i, "chunk_text": chunk, "scheme_title": request.title, "ministry": request.ministry, "state": request.state, "source_url": request.source_url, "doc_type": request.doc_type, "embedding": embedding, "content_hash": doc_hash } # During expand-contract migration: write to BOTH columns if dual_write: row["embedding_v2"] = embedding rows.append(row) supabase.table("document_chunks").upsert(rows, on_conflict="content_hash,chunk_index").execute() return {"status": "success", "title": request.title, "chunks_created": len(chunks), "doc_hash": doc_hash} @app.post("/api/rag/query") @limiter.limit("10/minute") async def rag_query(request: Request, query: SearchQuery): print(f"Citizen asked: {query.question} | Target Language: {query.language}") try: # --- STEP 1: TRANSLATE QUERY TO ENGLISH --- english_query = query.question user_language = query.language.lower() if user_language != 'english': english_query = await translate_text(query.question, user_language, 'english') print(f"Translated query: '{query.question}' → '{english_query}'") # --- STEP 2.5: NEURO-SYMBOLIC ELIGIBILITY DETECTION (Sprint 19) --- # If the query is about eligibility, route through OpenFisca # instead of the probabilistic RAG pipeline. ns_result = run_neuro_symbolic_pipeline(english_query, groq_client) if ns_result["type"] in ("eligibility_result", "followup_question"): ns_response = ns_result["response"] # Translate the response back to user's language if needed if user_language != 'english': ns_response = await translate_text(ns_response, 'english', user_language) async def ns_stream(): yield ns_response headers = { "X-Accel-Buffering": "no", "X-Translation-Provider": "OpenFisca-Deterministic", "X-Eligibility-Type": ns_result["type"], } if ns_result.get("eligible_schemes"): headers["X-Sources"] = "|".join(ns_result["eligible_schemes"]) return StreamingResponse( ns_stream(), media_type="text/plain", headers=headers ) # If intent is "informational", continue to the standard RAG pipeline below # --- STEP 2: EMBED THE ENGLISH QUERY --- model = get_embedding_model() # Nomic requires search_query prefix for citizens' questions query_numbers = model.encode(f"search_query: {english_query}", normalize_embeddings=True).tolist() # --- STEP 3: SEMANTIC CACHE LOOKUP (ZERO-COLLISION MODE) --- # We only use the cache for 100% identical questions to ensure precision. # Semantic 'guessing' is disabled to prevent different topics from returning the same answer. exact_match = supabase.table("query_cache")\ .select("response_text, sources")\ .eq("question_lower", english_query.lower().strip())\ .limit(1)\ .execute() if exact_match.data: print("⚡ EXACT CACHE HIT: Returning instant response") hit = cast(Dict[str, Any], exact_match.data[0]) final_cached_response = hit.get("response_text", "") if user_language != 'english': final_cached_response = await translate_text(final_cached_response, 'english', user_language) async def cached_stream() -> AsyncGenerator[str, None]: safe_data = final_cached_response.replace("\n", "\\n") yield f"event: message\ndata: {safe_data}\n\n" yield f"event: stream_end\ndata: {{}}\n\n" return StreamingResponse( cached_stream(), media_type="text/event-stream", headers={ "X-Accel-Buffering": "no", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Sources": "|".join(cast(List[str], hit.get("sources", []))), "X-Translation-Provider": "GovBridge-Exact-Cache" } ) # SEMANTIC RPC DISABLED (Bypassing to Step 4 for guaranteed accuracy) # --- STEP 4: HYBRID SEARCH WITH ENGLISH QUERY --- result = supabase.rpc("hybrid_search", { "query_text": english_query, "query_embedding": query_numbers, "match_count": 5 # Sprint 40: Compressed to 5 to prevent LLM context blowout for streaming }).execute() # After hybrid search returns results, before reranking raw_data = cast(List[Dict[str, Any]], result.data or []) if raw_data: first_hit = raw_data[0] top_score = first_hit.get("rrf_score", 0) top_title = first_hit.get("scheme_title", "Unknown") # Log low confidence queries for coverage gap analysis if float(top_score) < 0.02: supabase.table("low_confidence_queries").insert({ "query_text": english_query, "rrf_score": float(top_score), "language": user_language, "top_result_title": top_title }).execute() print(f"⚠️ Low confidence query logged: '{english_query}' (score: {float(top_score):.3f})") chunks = raw_data # --- STEP 5: CROSS-ENCODER RERANKING --- if chunks: chunks = cast(List[Dict[str, Any]], rerank_chunks(english_query, chunks)) if not chunks: context = "This information is not available in the GovBridge database for this query." source_titles: List[str] = [] else: context = "\n\n---\n\n".join([ f"[Document: {str(c.get('scheme_title', 'Unknown'))}]\n{str(c.get('chunk_text', ''))}" for c in chunks ]) source_titles = list(set([ str(c.get('scheme_title', 'Unknown')) for c in chunks if c.get('scheme_title') ])) # --- STEP 6: TRUE SSE STREAMING WITH RAG COORDINATION MATRIX (Sprint 40) --- # The LLM generates prose interleaved with structured JSON parameter # blocks enclosed in <|param_start|> / <|param_end|> sentinels. # parse_interleaved_stream routes prose → UI, JSON → OpenFisca. # Step 6a: Attempt Graph-RAG for legal relationship context graph_rag_result = run_graph_rag( query=english_query, context_chunks=chunks if chunks else [], groq_client=groq_client, supabase_client=supabase, embedding_model=get_embedding_model(), user_language=user_language, ) # Log graph traversal metrics if graph_rag_result.get("graph_traversals", 0) > 0: logger.info( f"🔗 Graph-RAG: {graph_rag_result['graph_traversals']} traversal(s), " f"TOON context: {len(graph_rag_result.get('graph_context', ''))} chars" ) # Merge graph sources if any graph_sources = graph_rag_result.get("sources", []) if graph_sources: source_titles = list(set(source_titles + graph_sources)) # Step 6b: Build the streaming system prompt sse_system_prompt = ( "You are GovBridge AI, India's sovereign civic intelligence assistant. " "Answer using ONLY the provided context. Respond exclusively in English. " "Keep your answer under 200 words. Be precise, not exhaustive. " "Cite the scheme name. Format benefits as bullet points where applicable.\n\n" "PARAMETER EXTRACTION PROTOCOL:\n" "If the user's question contains demographic information (income, age, state, " "caste, gender, occupation, farming status, disability, BPL status, land size), " "extract those parameters into a JSON object and wrap it EXACTLY like this:\n" "<|param_start|>{\"annual_income\": 250000, \"age\": 35, \"is_farmer\": true}" "<|param_end|>\n" "Embed this parameter block naturally within your conversational response. " "Only include fields that are EXPLICITLY stated by the user. " "Continue your prose after the parameter block." ) # If Graph-RAG already produced a full response, use it directly (no re-streaming) graph_rag_response = graph_rag_result.get("response", "") if graph_rag_response: # Graph-RAG succeeded — serve its response via SSE with no re-generation async def graph_rag_sse_stream() -> AsyncGenerator[str, None]: # Cache the response try: if len(graph_rag_response) > 50: supabase.table("query_cache").upsert({ "query_embedding": query_numbers, "query_text": english_query, "question": query.question, "response_text": graph_rag_response, "sources": source_titles, "language": user_language, "created_at": datetime.utcnow().isoformat(), "hit_count": 1 }).execute() logger.info("✅ Graph-RAG response cached") except Exception as cache_err: logger.warning(f"⚠️ Cache write failed: {cache_err}") final_text = graph_rag_response if user_language != 'english': final_text = await translate_text(graph_rag_response, 'english', user_language) yield f"event: message\ndata: {final_text}\n\n" yield f"event: stream_end\ndata: {{}}\n\n" return StreamingResponse( graph_rag_sse_stream(), media_type="text/event-stream", headers={ "X-Accel-Buffering": "no", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Sources": "|".join(source_titles), "X-Translation-Provider": "Graph-RAG-SSE", } ) # Step 6c: Graph-RAG returned empty — fall through to TRUE streaming groq_stream = groq_client.chat.completions.create( model="llama-3.3-70b-versatile", messages=[ {"role": "system", "content": sse_system_prompt}, {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {english_query}"} ], temperature=0.1, max_tokens=512, stream=True # Sprint 40: TRUE STREAMING ENABLED ) # Step 6d: Build async token iterator from Groq sync stream async def groq_token_iterator() -> AsyncGenerator[str, None]: """Wraps the synchronous Groq stream into an async iterator.""" for chunk in groq_stream: delta = chunk.choices[0].delta if delta and delta.content: yield delta.content # Step 6e: SSE Event Generator with Dual-Pipeline Routing async def sse_event_generator() -> AsyncGenerator[str, None]: full_response_parts: list[str] = [] async for msg_type, content in parse_interleaved_stream(groq_token_iterator()): if msg_type == "text": full_response_parts.append(content) # Translate chunk if needed display_content = content if user_language != 'english': # For streaming, we send English first for low latency # Translation happens per-chunk only for short segments pass # Keep English for real-time; translate in final cache # SSE: prose event # Escape newlines in data field for SSE compliance safe_data = content.replace("\n", "\\n") yield f"event: message\ndata: {safe_data}\n\n" elif msg_type == "json": # SSE: parameter update event — route through OpenFisca try: params = json.loads(content) # Run deterministic eligibility check eligibility_results = check_eligibility(params) eligible_schemes = [ k.replace("eligible_", "").replace("_", " ").title() for k, v in eligibility_results.items() if v ] payload = json.dumps({ "extracted_params": params, "eligible_schemes": eligible_schemes, "full_results": eligibility_results, }) yield f"event: parameter_update\ndata: {payload}\n\n" logger.info(f"🎯 SSE parameter_update: {len(eligible_schemes)} schemes matched") except json.JSONDecodeError as jde: logger.warning(f"⚠️ Invalid JSON in parameter block: {jde}") # Yield the malformed content as prose instead of dropping it safe_data = content.replace("\n", "\\n") yield f"event: message\ndata: {safe_data}\n\n" full_response_parts.append(content) # Cache the full assembled response full_response = "".join(full_response_parts) try: if len(full_response) > 50: supabase.table("query_cache").upsert({ "query_embedding": query_numbers, "query_text": english_query, "question": query.question, "response_text": full_response, "sources": source_titles, "language": user_language, "created_at": datetime.utcnow().isoformat(), "hit_count": 1 }).execute() logger.info("✅ Streaming response cached") except Exception as cache_err: logger.warning(f"⚠️ Cache write failed: {cache_err}") # Terminal SSE event yield f"event: stream_end\ndata: {{}}\n\n" return StreamingResponse( sse_event_generator(), media_type="text/event-stream", headers={ "X-Accel-Buffering": "no", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Sources": "|".join(source_titles), "X-Translation-Provider": "Groq-SSE-Stream", } ) except Exception as e: error_detail = str(e) print(f"System Error: {error_detail}") async def crash_msg(): yield f"The AI Brain encountered an internal logic error. Please try again. ({error_detail})" return StreamingResponse(crash_msg(), media_type="text/plain") @app.get("/api/admin/coverage-gaps") async def get_coverage_gaps(request: Request, limit: int = 20): # Verify admin secret auth = request.headers.get("X-Admin-Secret") if auth != os.environ.get("ADMIN_SECRET"): from fastapi import HTTPException raise HTTPException(status_code=403, detail="Unauthorized") result = supabase.table("low_confidence_queries")\ .select("query_text, rrf_score, language, created_at")\ .lt("rrf_score", 0.3)\ .order("created_at", desc=True)\ .limit(limit)\ .execute() return { "total_gaps": len(result.data), "queries": result.data } @app.post("/api/gazette/search") @limiter.limit("15/minute") async def gazette_search(request: Request, query: GazetteSearchQuery): """ Gazette Vault Search — PROJECT INDRA Pipeline (Sprint 29). Pipeline: 1. Embed query with Nomic (768-dim) using search_query: prefix 2. Call get_indra_candidates RPC (oversampled HNSW, 400 candidates + embeddings) 3. Poincaré ball projection → hyperbolic distance re-ranking 4. Cross-encoder re-rank top 30 with Ettin 5. Return page-pinned results to frontend GazetteViewer """ logger.info(f"📜 Gazette search [INDRA]: '{query.query}' | Type: {query.gazette_type} | State: {query.state}") try: # Step 1: Embed query model = get_embedding_model() query_embedding_raw = model.encode( f"search_query: {query.query}", normalize_embeddings=True ) query_embedding = np.array(query_embedding_raw).flatten().tolist() # Step 2: Call INDRA oversampling RPC rpc_params: Dict[str, Any] = { "query_embedding": query_embedding, "oversample_limit": 400, } # Add optional filters if query.gazette_type: rpc_params["filter_gazette_type"] = query.gazette_type if query.state: rpc_params["filter_state"] = query.state result = supabase.rpc("get_indra_candidates", rpc_params).execute() raw_results = cast(List[Dict[str, Any]], result.data or []) if not raw_results: return { "query": query.query, "results": [], "total": 0, "pipeline": "indra_poincare + ettin_reranker" } n_candidates = len(raw_results) logger.info(f" 📊 INDRA: {n_candidates} Euclidean candidates retrieved") # Step 3: Poincaré projection & hyperbolic re-ranking # Extract embedding vectors from RPC results query_vec = np.array(query_embedding, dtype=np.float32) candidate_embeddings = np.array( [json.loads(r["embedding"]) if isinstance(r["embedding"], str) else r["embedding"] for r in raw_results], dtype=np.float32 ) # Project and rank by hyperbolic distance sorted_indices = indra_engine.project_and_rank( query_vec, candidate_embeddings, n_candidates ) # Compute distances for logging hyp_distances = indra_engine.compute_poincare_distances( query_vec, candidate_embeddings, n_candidates ) # Re-order results by hyperbolic distance (ascending = most relevant first) hyperbolic_ranked = [] for idx in sorted_indices[:30]: # Take top 30 for cross-encoder stage entry = raw_results[int(idx)].copy() entry["hyperbolic_distance"] = float(hyp_distances[int(idx)]) # Remove embedding from the result (not needed downstream, saves memory) entry.pop("embedding", None) hyperbolic_ranked.append(entry) logger.info(f" 🔮 Poincaré: top-1 distance={hyperbolic_ranked[0]['hyperbolic_distance']:.4f}, " f"top-30 distance={hyperbolic_ranked[-1]['hyperbolic_distance']:.4f}") # Step 4: Cross-encoder re-rank (Ettin verifies semantic coherence) reranked = rerank_gazette_chunks(query.query, hyperbolic_ranked) # Step 5: Trim to requested limit reranked = reranked[:query.limit] # Step 6: Format response with page-pinned results results = [] for chunk in reranked: results.append({ "id": chunk.get("id"), "document_title": chunk.get("document_title", "Unknown Gazette"), "source_url": chunk.get("source_url"), "page_number": chunk.get("page_number", 1), "snippet_text": chunk.get("chunk_text", ""), "gazette_type": chunk.get("gazette_type", "central"), "issuing_authority": chunk.get("issuing_authority"), "notification_date": str(chunk.get("notification_date", "")), "cross_encoder_score": round(float(chunk.get("cross_encoder_score", 0.0)), 4), "hyperbolic_distance": round(float(chunk.get("hyperbolic_distance", 0.0)), 6), "rrf_score": round(float(chunk.get("rrf_score", chunk.get("l2_distance", 0.0))), 6), }) if results: logger.info(f" ✅ Returning {len(results)} gazette results " f"(best CE score: {results[0]['cross_encoder_score']:.4f})") else: logger.info(" ⛔ No results survived quality gate") return { "query": query.query, "results": results, "total": len(results), "pipeline": "indra_poincare + ettin_reranker" } except Exception as e: logger.error(f"❌ Gazette search [INDRA] error: {e}", exc_info=True) return { "query": query.query, "results": [], "total": 0, "error": "An internal error occurred during the search. Please try again later.", "pipeline": "indra_poincare + ettin_reranker" } # ═══════════════════════════════════════════════════════════════ # PROJECT INDRA Phase 3.1 — Knowledge Graph Neighborhood API # Sprint 34 # ═══════════════════════════════════════════════════════════════ @app.get("/api/gazette/graph/{chunk_id}") @limiter.limit("30/minute") async def get_graph_neighborhood(request: Request, chunk_id: str, depth: int = 1, limit: int = 100): """ Knowledge Graph Neighborhood — PROJECT INDRA Phase 3.1 (Sprint 34). Returns the 1-hop or 2-hop graph neighborhood of a gazette document, formatted as a node-link structure for force-directed graph rendering. Args: chunk_id: UUID of the anchor gazette_chunks document. depth: Traversal depth (1 or 2). Default 1. limit: Max edges to return. Default 100, max 500. """ logger.info(f"🔗 Graph neighborhood: chunk_id={chunk_id} | depth={depth} | limit={limit}") try: # Validate UUID format import uuid as uuid_mod try: uuid_mod.UUID(chunk_id, version=4) except ValueError: return { "anchor_id": chunk_id, "nodes": [], "links": [], "total_nodes": 0, "total_links": 0, "error": "Invalid UUID format" } # Clamp parameters depth = max(1, min(2, depth)) limit = max(1, min(500, limit)) # Call the graph neighborhood RPC result = supabase.rpc("get_graph_neighborhood", { "anchor_id": chunk_id, "max_depth": depth, "edge_limit": limit, }).execute() raw_edges = cast(List[Dict[str, Any]], result.data or []) if not raw_edges: logger.info(f" ⛔ No graph edges found for chunk_id={chunk_id}") return { "anchor_id": chunk_id, "nodes": [], "links": [], "total_nodes": 0, "total_links": 0 } # ── Transform edge-centric data to node-link format ── nodes_map: Dict[str, GraphNode] = {} links: List[GraphLink] = [] for edge in raw_edges: source_id = edge.get("source_id", "") target_id = edge.get("target_id", "") source_title = edge.get("source_title", "Unknown Document") target_title = edge.get("target_title", "Unresolved Entity") edge_type = edge.get("edge_type", "cross_references") hop_depth = edge.get("hop_depth", 1) # Skip edges with null target (unresolved entities from Sprint 30) if not target_id: continue # Deduplicate nodes — assign group based on relationship to anchor if source_id and source_id not in nodes_map: group = "anchor" if source_id == chunk_id else f"hop{hop_depth}" nodes_map[source_id] = GraphNode( id=source_id, name=source_title[:80], # Truncate long titles for Canvas rendering group=group, ) if target_id and target_id not in nodes_map: group = "anchor" if target_id == chunk_id else f"hop{hop_depth}" nodes_map[target_id] = GraphNode( id=target_id, name=target_title[:80], group=group, ) # Create link if source_id and target_id: links.append(GraphLink( source=source_id, target=target_id, label=edge_type, hop=hop_depth, )) nodes = list(nodes_map.values()) logger.info(f" ✅ Graph: {len(nodes)} nodes, {len(links)} links (depth={depth})") return { "anchor_id": chunk_id, "nodes": [n.model_dump() for n in nodes], "links": [l.model_dump() for l in links], "total_nodes": len(nodes), "total_links": len(links), } except Exception as e: logger.error(f"❌ Graph neighborhood error: {e}", exc_info=True) return { "anchor_id": chunk_id, "nodes": [], "links": [], "total_nodes": 0, "total_links": 0, "error": "An internal error occurred retrieving the graph.", }