import streamlit as st import json import torch import numpy as np from sentence_transformers import SentenceTransformer from flashrank import Ranker, RerankRequest import lancedb import os import requests from groq import Groq from dotenv import load_dotenv # Load environment variables load_dotenv() # =============================== # PAGE CONFIG & CONSTITUTION THEME # =============================== st.set_page_config(page_title="Constitution of India - Knowledge System", page_icon="⚖️", layout="wide") st.markdown("""
""", unsafe_allow_html=True) # =============================== # LOAD RESOURCES # =============================== @st.cache_resource def load_resources(): model = SentenceTransformer("BAAI/bge-m3", device='cpu') ranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2", cache_dir="./opt") db = lancedb.connect("./.lancedb") tbl = db.open_table("constitution_knowledge") return model, ranker, tbl model, ranker, tbl = load_resources() # =============================== # LLM LOGIC # =============================== def get_answer_from_llm(query, context): system_prompt = """You are a constitutional law expert and scholar of the Indian Constitution. Answer precisely using the provided context from the Constitution of India. CRITICAL RULES: 1. Cite specific Articles, Parts, and Chapters when answering. 2. Maintain legal accuracy and constitutional interpretation. 3. If the context doesn't contain the answer, say so clearly.""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"Context from Constitution:\n{context}\n\nQuestion: {query}"} ] # Token estimation input_text = system_prompt + f"Context from Constitution:\n{context}\n\nQuestion: {query}" in_tokens = len(input_text) // 4 errors = [] # 1. Try SambaNova First (Ramayana style model name) s_key = os.getenv("SAMBANOVA_API_KEY") if s_key and s_key != "your_key_here": try: url = "https://api.sambanova.ai/v1/chat/completions" headers = {"Authorization": f"Bearer {s_key}", "Content-Type": "application/json"} # Switched to 70B model without 'Meta-' prefix data = {"model": "Llama-3.1-70B-Instruct", "messages": messages, "temperature": 0.1, "stream": False} res = requests.post(url, headers=headers, json=data, timeout=10) if res.status_code == 200: ans = res.json()['choices'][0]['message']['content'] out_tokens = len(ans) // 4 return ans, "SambaNova (70B)", in_tokens, out_tokens else: errors.append(f"SambaNova Error: {res.status_code} - {res.text}") except Exception as e: errors.append(f"SambaNova Exception: {str(e)}") # 2. Fallback to Groq g_key = os.getenv("GROQ_API_KEY") if g_key: try: client = Groq(api_key=g_key) for m in ["llama-3.3-70b-versatile", "mixtral-8x7b-32768", "llama-3.1-8b-instant"]: try: res = client.chat.completions.create(model=m, messages=messages, temperature=0.2) ans = res.choices[0].message.content out_tokens = len(ans) // 4 return ans, m, in_tokens, out_tokens except Exception as e: errors.append(f"Groq {m} Error: {str(e)}") if "429" in str(e) or "400" in str(e): continue else: break except Exception as e: errors.append(f"Groq Setup Error: {str(e)}") # 3. Fallback to Local Ollama try: url = "http://localhost:11434/api/chat" data = { "model": "llama3.1:8b-instruct-q4_K_M", "messages": messages, "stream": False, "options": {"temperature": 0.2} } res = requests.post(url, json=data, timeout=30) if res.status_code == 200: ans = res.json()['message']['content'] out_tokens = len(ans) // 4 return ans, "Ollama (llama3.1)", in_tokens, out_tokens else: errors.append(f"Ollama Error: {res.status_code}") except Exception as e: errors.append(f"Ollama Exception: {str(e)}") return f"[ERROR] All API calls failed. Errors: {errors}", "None", 0, 0 # =============================== # MAIN UI # =============================== st.markdown(""" """, unsafe_allow_html=True) st.markdown('

Constitution of India

', unsafe_allow_html=True) st.markdown('

Legal Knowledge System

', unsafe_allow_html=True) # TRICOLOR DIVIDER st.markdown('
', unsafe_allow_html=True) # Session State for History if "messages" not in st.session_state: st.session_state.messages = [] # Layout: Content First content_placeholder = st.empty() # Render History or Landing Page with content_placeholder.container(): if not st.session_state.messages: st.markdown("""

Welcome to the Constitutional Knowledge System

Access the structured knowledge of the Indian Constitution through semantic search and legal reasoning.

""", unsafe_allow_html=True) # Render Conversation for msg in st.session_state.messages: if msg["role"] == "user": st.markdown(f'
Query: {msg["content"]}
', unsafe_allow_html=True) elif msg["role"] == "assistant": st.markdown(msg["content"], unsafe_allow_html=True) # Input Pinned to Bottom if user_query := st.chat_input("Ask a question about the Constitution..."): # Append User Msg st.session_state.messages.append({"role": "user", "content": user_query}) with content_placeholder.container(): st.markdown(f'
Query: {user_query}
', unsafe_allow_html=True) # Process Query limit_n = 3 is_summary = any(k in user_query.lower() for k in ["summarize", "summary", "overview", "explain all"]) # PART DETECTION target_part = None parts = { "fundamental rights": "PART III", "directive principles": "PART IV", "union": "PART V", "states": "PART VI", "judiciary": "PART V", "elections": "PART XV", "emergency": "PART XVIII", "amendment": "PART XX" } for k, v in parts.items(): if k in user_query.lower(): target_part = v break # Increase search depth for better recall limit_n = 10 if is_summary: limit_n = 15 if target_part: limit_n = 20 q_vec = model.encode(user_query) # Search top 30-60 candidates to ensure we don't miss key Articles like Art 3 search_res = tbl.search(q_vec).limit(limit_n * 3).to_list() # Filter by Part if detected if target_part: filtered_res = [r for r in search_res if target_part in r.get('part', '')] if filtered_res: search_res = filtered_res passages = [{"id": i, "text": r['text'], "meta": r} for i, r in enumerate(search_res)] # Rerank top 8 candidates (increased from 5) reranked = ranker.rerank(RerankRequest(query=user_query, passages=passages))[:10 if is_summary or target_part else 8] # Deduplicate chunks based on Article Number OR Text content (CRITICAL FIX) seen_keys = set() unique_reranked = [] for r in reranked: # Create a unique key: Try Article Number first + Part, else full text art_num = r['meta'].get('article_number', '').strip() part = r['meta'].get('part', '').strip() if art_num and "Article" in art_num: key = f"{art_num}-{part}" else: key = r['meta']['text'][:100] # Use first 100 chars of text as proxy if key not in seen_keys: seen_keys.add(key) unique_reranked.append(r) # Take top 3 UNIQUE results reranked = unique_reranked[:3] ctx = "\n---\n".join([f"{r['meta'].get('article_number', 'Section')}: {r['meta']['text']}" for r in reranked]) if len(ctx) > 6000: ctx = ctx[:6000] + "... [TRUNCATED]" # Generate Answer with st.spinner("Analyzing constitutional provisions..."): errors = [] # Prepare Prompt (Strict Context Only) system_prompt = """You are a strict legal assistant. Answer the question ONLY using the provided Context from the Constitution of India. CRITICAL RULES: 1. Do NOT use your own internal knowledge. Use ONLY the provided context. 2. If the answer is not explicitly mentioned in the context, state: "The provided context does not contain information about [topic]." 3. Do not assume or hallucinate Articles not present in the text below. 4. Cite specific Articles from the context provided.""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"Context from Constitution:\n{ctx}\n\nQuestion: {user_query}"} ] # Calculate Input Tokens (Approx) in_tokens = len(system_prompt + ctx + user_query) // 4 # 1. Try SambaNova First (DISABLED) # s_key = os.getenv("SAMBANOVA_API_KEY") # ans = None # Ensure ans is initialized # if s_key and s_key != "your_key_here": # ... (Logic commented out) ... ans = None # 2. Fallback to Groq (DISABLED) # if not ans: # ... (Logic commented out) ... # 3. Use Local Ollama (EXCLUSIVE) if not ans: try: url = "http://localhost:11434/api/chat" data = { "model": "llama3.1:8b-instruct-q4_K_M", "messages": messages, "stream": False, "options": {"temperature": 0.2, "num_ctx": 8192} } res = requests.post(url, json=data, timeout=120) # Increased timeout for local processing if res.status_code == 200: ans_text = res.json()['message']['content'] out_tokens = len(ans_text) // 4 ans, m_used, in_tok, out_tok = ans_text, "Ollama (local)", in_tokens, out_tokens else: errors.append(f"Ollama Error: {res.status_code} - {res.text}") except Exception as e: errors.append(f"Ollama Exception: {str(e)}") # Final Check if not ans: ans = f"[ERROR] All API calls failed. Errors: {errors}" m_used, in_tok, out_tok = "None", 0, 0 # Build HTML sources_html = "".join([ f'{r["meta"].get("article_number", "Section")} - {r["meta"].get("part", "")}' for r in reranked ]) answer_html = f"""

Legal Analysis

{ans}

Citations:
{sources_html}
In: {in_tok} Out: {out_tok} {m_used}
""" # Append Assistant Msg st.session_state.messages.append({"role": "assistant", "content": answer_html}) st.rerun() # =============================== # SIDEBAR # =============================== with st.sidebar: st.markdown('
', unsafe_allow_html=True) st.markdown("### System Metadata") st.write(f"**Index Type:** LanceDB Vector") st.write(f"**Model:** BGE-M3 (1024D)") st.write(f"**Reranker:** FlashRank") st.write(f"**Total Articles:** {len(tbl)}") with st.expander("Example Queries"): st.markdown(""" - What are fundamental rights?
- Role of the President
- How to amend the Constitution?
- Directive Principles explained
- Emergency provisions """, unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True)