import streamlit as st
import json
import torch
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer
import lancedb
import os
import requests
from groq import Groq
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# ===============================
# PAGE CONFIG & PREMIUM TECH UI
# ===============================
st.set_page_config(page_title="Git Intelligence System", page_icon="đģ", layout="wide")
st.markdown("""
""", unsafe_allow_html=True)
# ===============================
# RESOURCE LOADING
# ===============================
@st.cache_resource
def load_resources():
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = SentenceTransformer("BAAI/bge-m3", device=device)
# Add FlashRank reranker as in Ramayana app
from flashrank import Ranker, RerankRequest
ranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2")
db = lancedb.connect("./.lancedb")
tbl = db.open_table("git_knowledge")
with open('alias_map.json', 'r', encoding='utf-8') as f:
alias_map = json.load(f)
return model, ranker, tbl, alias_map
try:
model, ranker, tbl, alias_map = load_resources()
except Exception as e:
st.error(f"Failed to load resources: {e}")
st.stop()
# ===============================
# LLM LOGIC
# ===============================
def get_answer_from_llm(query, context):
system_prompt = """You are a STERN and HIGHLY PRECISE Technical Document Analyst.
Your objective is to answer questions using ONLY the provided Context Chunks.
STRICT ANTI-HALLUCINATION PROTOCOLS:
1. ABSOLUTE GROUNDING: Every claim you make MUST be directly supported by the provided text.
2. NO EXTERNAL KNOWLEDGE: Do NOT use your internal training data to explain commands, syntax, or concepts. If the information is not in the context, you MUST say: "The provided documentation does not contain information regarding [Topic]."
3. LIST vs. DESCRIPTION: If the context merely lists a command (e.g., in a 'git --help' style table) but does not provide details or syntax, you MUST state: "The documentation mentions [Command] but provides no instructions or explanations."
4. CITE EVERYTHING: When providing an answer, explicitly mention which 'Source' provided the data.
5. NEGATIVE CONSTRAINT: Do NOT be 'helpful' by providing external tutorials. Your helpfulness is measured by how accurately you report ONLY what is in the document."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Context Chunks:\n{context}\n\nQuestion: {query}"}
]
input_tokens = len(str(messages)) // 4
g_key = os.getenv("GROQ_API_KEY")
if g_key:
try:
client = Groq(api_key=g_key)
# Try different models
for m in ["llama-3.3-70b-versatile", "mixtral-8x7b-32768"]:
try:
res = client.chat.completions.create(model=m, messages=messages, temperature=0.1)
ans = res.choices[0].message.content
return ans, m, input_tokens, len(ans) // 4
except Exception:
continue
except Exception:
pass
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"}
data = {
"model": "Meta-Llama-3.3-70B-Instruct",
"messages": messages,
"temperature": 0.1,
"stream": False
}
res = requests.post(url, headers=headers, json=data, timeout=12)
if res.status_code == 200:
ans = res.json()['choices'][0]['message']['content']
return ans, "SambaNova (Llama-3.3)", input_tokens, len(ans) // 4
except Exception:
pass
return "LLM integration failed. Please check API keys.", "None", 0, 0
# ===============================
# MAIN UI
# ===============================
st.markdown('Git Documentation AI
', unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
st.markdown('> git query --intelligent-search --ultra-premium
', unsafe_allow_html=True)
if "messages" not in st.session_state:
st.session_state.messages = []
# ===============================
# SIDEBAR
# ===============================
with st.sidebar:
st.markdown('', unsafe_allow_html=True)
st.markdown("### đ ī¸ Git Archive Specs")
st.write(f"**Storage:** LanceDB Vector")
st.write(f"**Embedding:** BGE-M3")
st.write(f"**Reranker:** FlashRank")
st.markdown("---")
st.markdown("### đĻ Knowledge Base")
try:
count = len(tbl.to_pandas())
st.write(f"**Total Chunks:** {count}")
st.write(f"**Source:** Git Documentation")
except:
st.write("**Chunks:** 16")
with st.expander("đ System Internals"):
st.markdown("""
- **Pipeline**: Pypdf Extraction
- **Search**: Hybrid (Vector + Rerank)
- **LLM**: Groq Professional
""")
st.markdown('
', unsafe_allow_html=True)
# Display Messages
for msg in st.session_state.messages:
if msg["role"] == "user":
st.markdown(f'$ {msg["content"]}
', unsafe_allow_html=True)
else:
st.markdown(msg["content"], unsafe_allow_html=True)
# Landing Page Content
if not st.session_state.messages:
st.markdown("""
System Initialized
Expert RAG system for official Git Documentation with FlashRank Optimization.
- Command Execution: "How do I rebase my feature branch?"
- Conceptual Clarity: "Explain the Git object model."
- Workflow Optimization: "Best practices for merge conflicts."
v2.0.4-stable
RAG-ACTIVE
""", unsafe_allow_html=True)
# Chat Input
if user_query := st.chat_input("Enter your Git question..."):
# Store user message
st.session_state.messages.append({"role": "user", "content": user_query})
st.rerun()
# Processing the latest user message
if st.session_state.messages and st.session_state.messages[-1]["role"] == "user":
query = st.session_state.messages[-1]["content"]
with st.spinner("Searching documentation..."):
# 1. Expand aliases
processed_query = query.lower()
mapping = alias_map.get("alias_map", {})
for k, v in mapping.items():
processed_query = processed_query.replace(k, v)
# 2. Vector Search (Initial Fetch)
q_vec = model.encode(processed_query)
initial_results = tbl.search(q_vec).limit(15).to_list()
# 3. Rerank with FlashRank
from flashrank import RerankRequest
passages = [{"id": i, "text": r['text'], "meta": r} for i, r in enumerate(initial_results)]
rerank_request = RerankRequest(query=processed_query, passages=passages)
reranked_results = ranker.rerank(rerank_request)[:5]
# 4. Build Context
context_str = "\n---\n".join([f"Source: {r['meta']['title']}\nContent: {r['meta']['text']}" for r in reranked_results])
# 5. Get Answer
answer, model_name, in_t, out_t = get_answer_from_llm(query, context_str)
# 6. Format Output
sources_html = "".join([f'{r["meta"]["title"]}' for r in reranked_results])
formatted_answer = f"""
Response
{answer}
CITED SOURCES:
{sources_html}
in:{in_t}
out:{out_t}
{model_name}
"""
st.session_state.messages.append({"role": "assistant", "content": formatted_answer})
st.rerun()