Update app.py

app.py

@@ -103,6 +103,9 @@ if __name__ == "__main__":
     demo.launch()
 '''
 
+
+'''
+working
 import os
 import gradio as gr
 import requests
@@ -349,4 +352,402 @@ if __name__ == "__main__":
         })
     
     # Start Gradio app
-    demo.launch()
+    demo.launch()
+
+'''
+
+
+import os
+import gradio as gr
+import requests
+import mlflow
+import dagshub
+from pinecone import Pinecone
+from langchain.prompts import PromptTemplate
+from langchain.chains.llm import LLMChain
+from langchain.llms.base import LLM
+from typing import Optional, List, Mapping, Any
+import time
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from dotenv import load_dotenv
+from datetime import datetime
+
+# DeepEval imports
+try:
+    from deepeval.test_case import LLMTestCase, LLMTestCaseParams
+    from deepeval.metrics import AnswerRelevancyMetric, HallucinationMetric
+    from deepeval.metrics import BaseMetric
+except Exception:
+    raise
+
+# Optional LangChain Google generative integration (Gemini)
+try:
+    import google.generativeai as genai
+    from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
+except Exception:
+    ChatGoogleGenerativeAI = None
+    genai = None
+
+# Load environment variables
+load_dotenv()
+
+PINECONE_API_KEY = os.environ.get("PINECONE_API_KEY", "")
+MLFLOW_TRACKING_URI = os.environ.get("MLFLOW_TRACKING_URI", "http://localhost:5000")
+GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "")
+LITSERVE_ENDPOINT = os.environ.get("LITSERVE_ENDPOINT", "https://8001-01k2h9d9mervcmgfn66ybkpwvq.cloudspaces.litng.ai/predict")
+
+# DagsHub & MLflow Setup (guarded)
+try:
+    dagshub.init(
+        repo_owner='prathamesh.khade20',
+        repo_name='Maintenance_AI_website',
+        mlflow=True
+    )
+except Exception:
+    pass
+
+mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
+mlflow.set_experiment("Maintenance-RAG-Chatbot")
+
+# ----------- App configuration logging -----------
+with mlflow.start_run(run_name=f"App-Config-{datetime.now().strftime('%Y%m%d-%H%M%S')}") as setup_run:
+    mlflow.log_params({
+        "pinecone_index": "rag-granite-index",
+        "embedding_model": "all-MiniLM-L6-v2",
+        "namespace": "rag-ns",
+        "top_k": 3,
+        "llm_endpoint": LITSERVE_ENDPOINT
+    })
+    mlflow.log_text("""
+You are a smart assistant. Based on the provided context, answer the question in 1–2 lines only.
+If the context has more details, summarize it concisely.
+Context:
+{context}
+Question: {question}
+Answer:
+""", "artifacts/prompt_template.txt")
+
+# ----------- 1. Custom LLM for LitServe endpoint -----------
+class LitServeLLM(LLM):
+    endpoint_url: str
+
+    @mlflow.trace
+    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
+        payload = {"prompt": prompt}
+        with mlflow.start_span("lit_serve_request"):
+            start_time = time.time()
+            response = requests.post(self.endpoint_url, json=payload)
+            latency = time.time() - start_time
+            mlflow.log_metric("lit_serve_latency", latency)
+            if response.status_code == 200:
+                data = response.json()
+                mlflow.log_metric("response_tokens", len(data.get("response", "").split()))
+                return data.get("response", "").strip()
+            else:
+                mlflow.log_metric("request_errors", 1)
+                error_info = {
+                    "status_code": response.status_code,
+                    "error": response.text,
+                    "timestamp": datetime.now().isoformat()
+                }
+                mlflow.log_dict(error_info, "artifacts/error_log.json")
+                raise ValueError(f"Request failed: {response.status_code}")
+
+    @property
+    def _identifying_params(self) -> Mapping[str, Any]:
+        return {"endpoint_url": self.endpoint_url}
+
+    @property
+    def _llm_type(self) -> str:
+        return "litserve_llm"
+
+# ----------- 2. Pinecone Connection -----------
+@mlflow.trace
+def init_pinecone():
+    PINECONE_API_KEY = os.environ.get("PINECONE_API_KEY")
+    pc = Pinecone(api_key=PINECONE_API_KEY)
+    return pc.Index("rag-granite-index")
+
+try:
+    index = init_pinecone()
+except Exception:
+    index = None
+
+# ----------- 3. Embedding Model -----------
+embeddings_model = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
+
+# ----------- 4. Context Retrieval with Tracing -----------
+@mlflow.trace
+def get_retrieved_context(query: str, top_k=3):
+    with mlflow.start_span("embedding_generation"):
+        start_time = time.time()
+        query_embedding = embeddings_model.embed_query(query)
+        mlflow.log_metric("embedding_latency", time.time() - start_time)
+
+    if index is None:
+        return ""
+
+    with mlflow.start_span("pinecone_query"):
+        start_time = time.time()
+        results = index.query(
+            namespace="rag-ns",
+            vector=query_embedding,
+            top_k=top_k,
+            include_metadata=True
+        )
+        mlflow.log_metric("pinecone_latency", time.time() - start_time)
+        mlflow.log_metric("retrieved_chunks", len(results['matches']))
+
+    context_parts = [match['metadata']['text'] for match in results['matches']]
+    return "
+".join(context_parts)
+
+# ----------- 5. LLM Chain Setup -----------
+model = LitServeLLM(endpoint_url=LITSERVE_ENDPOINT)
+
+prompt = PromptTemplate(
+    input_variables=["context", "question"],
+    template="""
+You are a smart assistant. Based on the provided context, answer the question in 1–2 lines only.
+If the context has more details, summarize it concisely.
+Context:
+{context}
+Question: {question}
+Answer:
+"""
+)
+
+llm_chain = LLMChain(llm=model, prompt=prompt)
+
+# ----------- 6. RAG Pipeline with Full Tracing -----------
+@mlflow.trace
+def rag_pipeline(question):
+    try:
+        with mlflow.start_run(run_name=f"Query-{datetime.now().strftime('%H%M%S')}", nested=True):
+            mlflow.log_param("user_question", question)
+            retrieved_context = get_retrieved_context(question)
+            mlflow.log_text(retrieved_context, "artifacts/retrieved_context.txt")
+
+            start_time = time.time()
+            response_obj = llm_chain.invoke({
+                "context": retrieved_context,
+                "question": question
+            })
+            response = response_obj.get("text") if isinstance(response_obj, dict) else getattr(response_obj, "text", str(response_obj))
+            response = response.strip()
+
+            if "Answer:" in response:
+                response = response.split("Answer:", 1)[-1].strip()
+
+            mlflow.log_metric("response_latency", time.time() - start_time)
+            mlflow.log_metric("response_length", len(response))
+            mlflow.log_text(response, "artifacts/response.txt")
+
+            return response
+    except Exception as e:
+        mlflow.log_metric("pipeline_errors", 1)
+        error_info = {
+            "error": str(e),
+            "question": question,
+            "timestamp": datetime.now().isoformat()
+        }
+        mlflow.log_dict(error_info, "artifacts/pipeline_errors.json")
+        return f"Error: {str(e)}"
+
+# ----------- 7. DeepEval Wrapper(s) and Metrics Integration (NO EXPECTED OUTPUT) -----------
+from deepeval.models.base_model import DeepEvalBaseLLM
+
+class GoogleVertexAI(DeepEvalBaseLLM):
+    def __init__(self, model):
+        self.model = model
+
+    def load_model(self):
+        return self.model
+
+    def generate(self, prompt: str) -> str:
+        chat_model = self.load_model()
+        res = chat_model.invoke(prompt)
+        if hasattr(res, 'content'):
+            return res.content
+        if isinstance(res, dict):
+            return res.get('content') or res.get('text') or str(res)
+        return str(res)
+
+    async def a_generate(self, prompt: str) -> str:
+        chat_model = self.load_model()
+        res = await chat_model.ainvoke(prompt)
+        return getattr(res, 'content', str(res))
+
+    def get_model_name(self):
+        return "Vertex AI Model"
+
+class LitServeWrapper(DeepEvalBaseLLM):
+    def __init__(self, lit_llm: LitServeLLM):
+        self.lit_llm = lit_llm
+
+    def load_model(self):
+        return self.lit_llm
+
+    def generate(self, prompt: str) -> str:
+        return self.lit_llm._call(prompt)
+
+    async def a_generate(self, prompt: str) -> str:
+        return self.generate(prompt)
+
+    def get_model_name(self):
+        return "LitServeModel"
+
+# Custom metric that DOES NOT require expected_output: Length-based utility metric
+class LengthMetric(BaseMetric):
+    def __init__(self, min_tokens: int = 1, max_tokens: int = 200):
+        self.min_tokens = min_tokens
+        self.max_tokens = max_tokens
+        self.score = 0.0
+        self.success = False
+
+    def measure(self, test_case: LLMTestCase):
+        text = (test_case.actual_output or "")
+        tokens = len(text.split())
+        # Score between 0 and 1 based on how close tokens are to midpoint
+        mid = (self.min_tokens + self.max_tokens) / 2
+        dist = abs(tokens - mid)
+        max_dist = max(mid - self.min_tokens, self.max_tokens - mid)
+        self.score = max(0.0, 1.0 - (dist / max_dist))
+        # success if within [min_tokens, max_tokens]
+        self.success = (self.min_tokens <= tokens <= self.max_tokens)
+        return self.score
+
+    async def a_measure(self, test_case: LLMTestCase):
+        return self.measure(test_case)
+
+    def is_successful(self):
+        return self.success
+
+    @property
+    def name(self):
+        return "Length Metric"
+
+# Helper to get eval model
+def get_deepeval_model(choice: str):
+    if choice == 'gemini' and ChatGoogleGenerativeAI is not None and GOOGLE_API_KEY:
+        try:
+            genai.configure(api_key=GOOGLE_API_KEY)
+        except Exception:
+            pass
+        chat_model = ChatGoogleGenerativeAI(model="gemini-pro", google_api_key=GOOGLE_API_KEY)
+        return GoogleVertexAI(model=chat_model)
+    else:
+        return LitServeWrapper(lit_llm=model)
+
+# Function to run Deepeval tests and log to mlflow (only metrics that don't need expected_output)
+@mlflow.trace
+def run_deepeval_tests(test_cases: List[LLMTestCase], eval_model_choice: str = 'litserve'):
+    model_wrapper = get_deepeval_model(eval_model_choice)
+
+    # Use only metrics that don't require expected output
+    answer_relevancy_metric = AnswerRelevancyMetric(threshold=0.5, model=model_wrapper)
+    hallucination_metric = HallucinationMetric(threshold=0.5, model=model_wrapper)
+    length_metric = LengthMetric(min_tokens=3, max_tokens=200)
+
+    results = []
+    with mlflow.start_run(run_name=f"DeepEval-{datetime.now().strftime('%H%M%S')}", nested=True):
+        for i, tc in enumerate(test_cases):
+            mlflow.log_param(f"tc_{i}_input", tc.input)
+            mlflow.log_param(f"tc_{i}_actual", tc.actual_output)
+            if tc.context:
+                mlflow.log_text("
+".join(tc.context), f"artifacts/tc_{i}_context.txt")
+
+            # Measure metrics
+            answer_relevancy_metric.measure(tc)
+            hallucination_metric.measure(tc)
+            length_metric.measure(tc)
+
+            entry = {
+                "input": tc.input,
+                "actual_output": tc.actual_output,
+                "context": tc.context,
+                "answer_relevancy_score": answer_relevancy_metric.score,
+                "hallucination_score": hallucination_metric.score,
+                "length_score": length_metric.score
+            }
+
+            # Log metrics to mlflow
+            mlflow.log_metric(f"tc_{i}_answer_relevancy", answer_relevancy_metric.score)
+            mlflow.log_metric(f"tc_{i}_hallucination", hallucination_metric.score)
+            mlflow.log_metric(f"tc_{i}_length", length_metric.score)
+
+            results.append(entry)
+
+    return results
+
+# ----------- 8. Gradio UI with Evaluation Tab (NO EXPECTED OUTPUT) -----------
+with gr.Blocks() as demo:
+    gr.Markdown("# 🛠️ Maintenance AI Assistant + DeepEval (No expected output required)")
+
+    with gr.Tabs():
+        with gr.TabItem("Chat (RAG)"):
+            usage_counter = gr.State(value=0)
+            session_start = gr.State(value=datetime.now().isoformat())
+
+            question_input = gr.Textbox(label="Ask your maintenance question")
+            answer_output = gr.Textbox(label="AI Response")
+            ask_button = gr.Button("Get Answer")
+            feedback = gr.Radio(["Helpful", "Not Helpful"], label="Was this response helpful?")
+
+            def track_usage(question, count, session_start, feedback=None):
+                count += 1
+                with mlflow.start_run(run_name=f"User-Interaction-{count}", nested=True):
+                    mlflow.log_param("question", question)
+                    mlflow.log_param("session_start", session_start)
+                    response = rag_pipeline(question)
+                    if feedback:
+                        mlflow.log_param("user_feedback", feedback)
+                        mlflow.log_metric("helpful_responses", 1 if feedback == "Helpful" else 0)
+                    mlflow.log_metric("total_queries", count)
+                    return response, count, session_start
+
+            ask_button.click(
+                track_usage,
+                inputs=[question_input, usage_counter, session_start],
+                outputs=[answer_output, usage_counter, session_start]
+            )
+
+            feedback.change(
+                track_usage,
+                inputs=[question_input, usage_counter, session_start, feedback],
+                outputs=[answer_output, usage_counter, session_start]
+            )
+
+        with gr.TabItem("DeepEval — Model Tests"):
+            gr.Markdown("### Run DeepEval metrics (no expected output needed). Provide input + actual output (or paste model output), optional context.")
+
+            tc_input = gr.Textbox(label="Test Input (prompt)")
+            tc_actual = gr.Textbox(label="Actual Output (paste model response or custom)")
+            tc_context = gr.Textbox(label="Context (optional)")
+
+            model_choice = gr.Radio(["litserve", "gemini"], value="litserve", label="Evaluation backend")
+            run_button = gr.Button("Run DeepEval")
+            eval_output = gr.JSON(label="Evaluation Results")
+
+            def run_single_eval(inp, actual, context, eval_backend):
+                tc = LLMTestCase(input=inp, actual_output=actual, expected_output=None, context=[context] if context else None)
+                results = run_deepeval_tests([tc], eval_model_choice=eval_backend)
+                return results
+
+            run_button.click(
+                run_single_eval,
+                inputs=[tc_input, tc_actual, tc_context, model_choice],
+                outputs=[eval_output]
+            )
+
+if __name__ == "__main__":
+    with mlflow.start_run(run_name="Deployment-Info"):
+        mlflow.log_params({
+            "app_version": "1.2.0",
+            "deployment_platform": "Lightning AI / HuggingFace Space",
+            "deployment_time": datetime.now().isoformat(),
+            "code_version": os.getenv("GIT_COMMIT", "dev")
+        })
+
+    demo.launch()