Delete langchain

langchain/langgraph_main.py

@@ -1,613 +0,0 @@
-from pydantic import BaseModel, Field, validator
-from typing import List, Optional, Dict, Any, TypedDict,Generic, TypeVar
-import uuid
-import io
-import os
-import PyPDF2
-import re
-import logging
-import time
-from docx import Document as dx
-from langchain_text_splitters import RecursiveCharacterTextSplitter
-import tempfile
-import faiss
-from langchain_community.docstore.in_memory import InMemoryDocstore
-from langchain_community.vectorstores import FAISS
-from langchain_core.prompts import PromptTemplate
-from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
-from langchain_core.documents import Document
-from langchain_huggingface import HuggingFaceEmbeddings
-from langgraph.checkpoint.memory import MemorySaver
-from langgraph.graph import StateGraph, END
-from sqlalchemy import create_engine, Column, String, Integer, DateTime, ForeignKey, Text
-from sqlalchemy.dialects.sqlite import JSON as SQLiteJSON
-# from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker, relationship
-from sentence_transformers import SentenceTransformer
-from huggingface_hub import login
-from langchain_google_genai import ChatGoogleGenerativeAI
-import datetime
-from enum import Enum as PyEnum
-from sqlalchemy.orm import DeclarativeBase
-from config import Config
-from functools import lru_cache
-from dotenv import load_dotenv
-
-# load_dotenv()
-# # hf_token = os.environ.get("hf_user_token") or os.getenv("hf_user_token")
-# def login_hf():
-#     hf_token = os.environ.get("hf_user_token") or os.getenv("hf_user_token")
-#     if hf_token:
-#         login(token=hf_token,add_to_git_credential=True)
-#     else:
-#         raise ValueError("HF_TOKEN environment variable is not set.")
-
-T = TypeVar("T")
-# --- 1. Database Setup ---
-DATABASE_URL = "sqlite:///src/database_telemetry.db"
-if os.path.exists(DATABASE_URL):
-    engine = create_engine(DATABASE_URL)
-    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-else:
-    DATABASE_URL = "sqlite:///database_telemetry.db"
-    engine = create_engine(DATABASE_URL)
-    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-class Base(DeclarativeBase):
-    pass
-
-class FeedbackScore(PyEnum):
-    POSITIVE = 1
-    NEGATIVE = -1
-
-class Telemetry(Base):
-    __tablename__ = "telemetry_table"
-    transaction_id = Column(String, primary_key=True)
-    session_id = Column(String)
-    user_question = Column(Text)
-    response = Column(Text)
-    context = Column(Text)
-    model_name = Column(String)
-    input_tokens = Column(Integer)
-    output_tokens = Column(Integer)
-    total_tokens = Column(Integer)
-    latency = Column(Integer)
-    dtcreatedon = Column(DateTime)
-    
-    feedback = relationship("Feedback", back_populates="telemetry_entry", uselist=False)
-
-class Feedback(Base):
-    __tablename__ = "feedback_table"
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    telemetry_entry_id = Column(String, ForeignKey("telemetry_table.transaction_id"), nullable=False, unique=True)
-    feedback_score = Column(Integer, nullable=False)
-    feedback_text = Column(Text, nullable=True)
-    user_query = Column(Text, nullable=False)
-    llm_response = Column(Text, nullable=False)
-    timestamp = Column(DateTime, default=datetime.datetime.now)
-    
-    telemetry_entry = relationship("Telemetry", back_populates="feedback")
-
-class ConversationHistory(Base):
-    __tablename__ = "conversation_history"
-    session_id = Column(String, primary_key=True)
-    messages = Column(SQLiteJSON, nullable=False)
-    last_updated = Column(DateTime, default=datetime.datetime.now)
-
-# --- 2. Initialize LLM and Embeddings ---
-
-gak = os.environ.get("Gapi_key")
-llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite",google_api_key=gak)
-
-def init_embed():
-    embedding_model = HuggingFaceEmbeddings(
-        model_name="ibm-granite/granite-embedding-english-r2",
-        model_kwargs={'device': 'cpu'},
-        encode_kwargs={'normalize_embeddings': False}
-    )
-    # embedding_model = SentenceTransformer("ibm-granite/granite-embedding-english-r2")
-    return embedding_model
-# embedding_model = SentenceTransformer("ibm-granite/granite-embedding-english-r2")
-
-# my_model_name = "gemma3:1b-it-qat"
-# llm = ChatOllama(model=my_model_name)
-
-
-# --- 3. LangGraph State and Workflow ---
-class GraphState(TypedDict):
-    chat_history: List[Dict[str, Any]]
-    retrieved_documents: List[str]
-    user_question: str
-    session_id: str
-    telemetry_id: Optional[str] = None
-
-vectorstore_retriever = None
-compiled_app = None
-memory = MemorySaver()
-
-# --- 4. LangGraph Nodes ---
-def retrieve_documents(state: GraphState):
-    global vectorstore_retriever
-    user_question = state["user_question"]
-    if vectorstore_retriever is None:
-        raise ValueError("Knowledge base not loaded. Please upload documents first.")
-    retrieved_docs = vectorstore_retriever.as_retriever(search_type="mmr", search_kwargs={"k": 3})
-    top_docs = retrieved_docs.invoke(user_question)
-    print("Top Docs: ", top_docs)
-    retrieved_docs_content = [doc.page_content if doc.page_content else doc for doc in top_docs]
-    print("retrieved_documents List: ", retrieved_docs_content)
-    return {"retrieved_documents": retrieved_docs_content}
-
-def generate_response(state: GraphState):
-    global llm
-    user_question = state["user_question"]
-    retrieved_documents = state["retrieved_documents"]
-    
-    formatted_chat_history = []
-    for msg in state["chat_history"]:
-        if msg['role'] == 'user':
-            formatted_chat_history.append(HumanMessage(content=msg['content']))
-        elif msg['role'] == 'assistant':
-            formatted_chat_history.append(AIMessage(content=msg['content']))
-    
-    if not retrieved_documents:
-        response_content = "I couldn't find any relevant information in the uploaded documents for your question. Can you please rephrase or provide more context?"
-        response_obj = AIMessage(content=response_content)
-    else:
-        context = "\n\n".join(retrieved_documents)
-        template = """
-            You are a helpful AI assistant. Answer the user's question based on the provided context {context} and the conversation history {chat_history}.
-            If the answer is not in the context, state that you don't have enough information.
-            Do not make up answers. Only use the given context and chat_history.
-            Remove unwanted words like 'Response:' or 'Answer:' from answers.
-            \n\nHere is the Question:\n{user_question}
-        """
-        rag_prompt = PromptTemplate(
-            input_variables=["context", "chat_history", "user_question"],
-            template=template
-        )
-        rag_chain = rag_prompt | llm
-        time.sleep(3)
-        response_obj = rag_chain.invoke({
-            "context": [SystemMessage(content=context)],
-            "chat_history": formatted_chat_history,
-            "user_question": [HumanMessage(content=user_question)]
-        })
-    
-    telemetry_data = response_obj.model_dump()
-    input_tokens = telemetry_data.get('usage_metadata', {}).get('input_tokens', 0)
-    output_tokens = telemetry_data.get('usage_metadata', {}).get('output_tokens', 0)
-    total_tokens = telemetry_data.get('usage_metadata', {}).get('total_tokens', 0)
-    model_name = telemetry_data.get('response_metadata', {}).get('model', 'unknown')
-    total_duration = telemetry_data.get('response_metadata', {}).get('total_duration', 0)
-    
-    db = SessionLocal()
-    transaction_id = str(uuid.uuid4())
-    try:
-        telemetry_record = Telemetry(
-            transaction_id=transaction_id,
-            session_id=state.get("session_id"),
-            user_question=user_question,
-            response=response_obj.content,
-            context="\n\n".join(retrieved_documents) if retrieved_documents else "No documents retrieved",
-            model_name=model_name,
-            input_tokens=input_tokens,
-            output_tokens=output_tokens,
-            total_tokens=total_tokens,
-            latency=total_duration,
-            dtcreatedon=datetime.datetime.now()
-        )
-        db.add(telemetry_record)
-        
-        new_messages = state["chat_history"] + [
-            {"role": "user", "content": user_question}, 
-            {"role": "assistant", "content": response_obj.content, "telemetry_id": transaction_id}
-        ]
-        
-        # --- FIX: Refactored Database Save Logic ---
-        print(f"Saving conversation for session_id: {state.get('session_id')}")
-        conversation_entry = db.query(ConversationHistory).filter_by(session_id=state.get("session_id")).first()
-        if conversation_entry:
-            print(f"Updating existing conversation for session_id: {state.get('session_id')}")
-            conversation_entry.messages = new_messages
-            conversation_entry.last_updated = datetime.datetime.now()
-        else:
-            print(f"Creating new conversation for session_id: {state.get('session_id')}")
-            new_conversation_entry = ConversationHistory(
-                session_id=state.get("session_id"),
-                messages=new_messages,
-                last_updated=datetime.datetime.now()
-            )
-            db.add(new_conversation_entry)
-        
-        db.commit()
-        print(f"Successfully saved conversation for session_id: {state.get('session_id')}")
-
-    except Exception as e:
-        db.rollback()
-        print(f"***CRITICAL ERROR***: Failed to save data to database. Error: {e}")
-    finally:
-        db.close()
-    
-    return {
-        "chat_history": new_messages,
-        "telemetry_id": transaction_id
-    }
-
-
-# Build and compile the workflow
-workflow = StateGraph(GraphState)
-workflow.add_node("retrieve", retrieve_documents)
-workflow.add_node("generate", generate_response)
-workflow.set_entry_point("retrieve")
-workflow.add_edge("retrieve", "generate")
-workflow.add_edge("generate", END)
-compiled_app = workflow.compile(checkpointer=memory)
-
-
-# --- 5. API Models ---
-class ChatHistoryEntry(BaseModel):
-    role: str
-    content: str
-    telemetry_id: Optional[str] = None
-
-class ChatRequest(BaseModel):
-    user_question: str
-    session_id: str
-    chat_history: Optional[List[ChatHistoryEntry]] = Field(default_factory=list)
-
-    @validator('user_question')
-    def validate_prompt(cls, v):
-        v = v.strip()
-        if not v:
-            raise ValueError('Question cannot be empty')
-        return v
-
-class ChatResponse(BaseModel):
-    ai_response: str
-    updated_chat_history: List[ChatHistoryEntry]
-    telemetry_entry_id: str
-    is_restricted: bool = False
-    moderation_reason: Optional[str] = None
-
-class FeedbackRequest(BaseModel):
-    session_id: str
-    telemetry_entry_id: str
-    feedback_score: int
-    feedback_text: Optional[str] = None
-
-class ConversationSummary(BaseModel):
-    session_id: str
-    title: str
-
-# Content Moderation Service
-class ContentModerator:
-    def __init__(self):
-        self.blacklist_words = Config.BLACKLIST_WORDS
-        self.suspicious_patterns = [re.compile(pattern, re.IGNORECASE) 
-                                  for pattern in Config.SUSPICIOUS_PATTERNS]
-        self.allowed_topics = Config.ALLOWED_TOPICS
-    
-    def contains_blacklisted_words(self, text: str) -> bool:
-        text_lower = text.lower()
-        return any(word in text_lower for word in self.blacklist_words)
-    
-    def contains_suspicious_patterns(self, text: str) -> bool:
-        return any(pattern.search(text) for pattern in self.suspicious_patterns)
-    
-    def has_encoding_attempts(self, text: str) -> bool:
-        # Check for encoding/obfuscation attempts
-        encoding_patterns = [
-            r"%[0-9A-Fa-f]{2}",  # URL encoding
-            r"\\x[0-9A-Fa-f]{2}",  # Hex encoding
-            r"&#x?[0-9a-f]+;",  # HTML entities
-        ]
-        return any(re.search(pattern, text) for pattern in encoding_patterns)
-    
-    def has_excessive_special_chars(self, text: str) -> bool:
-        # Check for excessive special characters that might indicate obfuscation
-        special_chars = len(re.findall(r'[^\w\s]', text))
-        total_chars = len(text)
-        if total_chars == 0:
-            return False
-        return (special_chars / total_chars) > 0.3  # More than 30% special chars
-    
-    def is_prompt_injection(self, text: str) -> bool:
-        # Check for common prompt injection techniques
-        injection_indicators = [
-            self.contains_suspicious_patterns(text),
-            self.contains_blacklisted_words(text),
-            self.has_encoding_attempts(text),
-            self.has_excessive_special_chars(text)
-        ]
-        return any(injection_indicators)
-    
-    def moderate_content(self, text: str) -> Dict[str, Any]:
-        # Check for prompt injection first
-        if self.is_prompt_injection(text):
-            return {
-                "is_restricted": True,
-                "reason": "Potential prompt injection detected",
-                "response_type": "injection"
-            }
-        
-        # Check for harmful content
-        if self.contains_blacklisted_words(text):
-            harmful_words = [word for word in self.blacklist_words if word in text.lower()]
-            return {
-                "is_restricted": True,
-                "reason": f"Contains restricted content: {', '.join(harmful_words[:3])}",
-                "response_type": "harmful"
-            }
-        
-        return {"is_restricted": False, "reason": None, "response_type": None}
-
-moderator = ContentModerator()
-
-@lru_cache(maxsize=5)
-def process_text(file):
-    string_data = (file.read()).decode("utf-8")
-    return string_data
-    
-@lru_cache(maxsize=5)
-def process_pdf(file):
-    pdf_bytes = io.BytesIO(file.read())
-    reader = PyPDF2.PdfReader(pdf_bytes)
-    pdf_text = "".join([page.extract_text() + "\n" for page in reader.pages])
-    return pdf_text
-    
-@lru_cache(maxsize=5)
-def process_docx(file):
-    docx_bytes = io.BytesIO(file.read())
-    docx_docs = dx(docx_bytes)
-    docx_content = "\n".join([para.text for para in docx_docs.paragraphs])
-    return docx_content
-    
-
-class UploadDocs:
-    def upload_documents(files):
-        global vectorstore_retriever
-    
-        embedding_model = init_embed()
-        
-        all_documents = []
-        for uploaded_file in files:
-            
-            if uploaded_file.type == "text/plain":
-                # string_data = ( uploaded_file.read()).decode("utf-8")
-                string_data = process_text(uploaded_file)
-                all_documents.append(Document(page_content=string_data, metadata={"source": uploaded_file.name}))
-            elif uploaded_file.type == "application/pdf":
-                pdf_text = process_pdf(uploaded_file)
-                
-                # pdf_bytes = io.BytesIO( uploaded_file.read())
-                # reader = PyPDF2.PdfReader(pdf_bytes)
-                # pdf_text = "".join([page.extract_text() + "\n" for page in reader.pages])
-                all_documents.append(Document(page_content=pdf_text, metadata={"source": uploaded_file.name}))
-                
-            elif uploaded_file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
-                docx_content = process_docx(uploaded_file)
-                
-                # docx_bytes = io.BytesIO( uploaded_file.read())
-                # docx_docs = dx(docx_bytes)
-                # docx_content = "\n".join([para.text for para in docx_docs.paragraphs])
-                all_documents.append(Document(page_content=docx_content, metadata={"source": uploaded_file.name}))
-            else:
-                raise Exception(status_code=400, detail=f"Unsupported file type: {uploaded_file.name} ({uploaded_file.type})")
-        
-        if not all_documents:
-            raise Exception(status_code=400, detail="No supported documents uploaded.")
-    
-        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
-        text_chunks = text_splitter.split_documents(all_documents)
-        print("text_chucks: ", text_chunks[:100])
-    
-        processed_chunks_with_ids = []
-        for i, chunk in enumerate(text_chunks):
-            # Generate a unique ID for each chunk
-            # Option 1 (Recommended): Using UUID for global uniqueness
-            # chunk_id = str(uuid.uuid4())
-            
-            # Option 2 (Alternative): Combining source file path with chunk index
-            # This is good if you want IDs to be deterministic based on file/chunk.
-            # You might need to make the file path more robust (e.g., hash it or normalize it).
-            file_source = chunk.metadata.get('source', 'unknown_source')
-            chunk_id = f"{file_source.replace('.','_')}_chunk_{i}"
-    
-            # Add the unique ID to the chunk's metadata
-            # It's good practice to keep original metadata and just add your custom ID.
-            chunk.metadata['doc_id'] = chunk_id
-            
-            
-            processed_chunks_with_ids.append(chunk)
-        # embeddings = [embedding_model.encode(doc_chunks.page_content, convert_to_numpy=True) for doc_chunks in processed_chunks_with_ids]
-        
-        print(f"Split {len(processed_chunks_with_ids)} chunks.")
-        print(f"Assigned unique 'doc_id' to each chunk in metadata.")
-        # dimension = 768
-        # # hnsw_m = 32
-        # # index = faiss.IndexHNSWFlat(dimension, hnsw_m, faiss.METRIC_INNER_PRODUCT)
-        # index = faiss.IndexFlatL2(dimension)
-        # vector_store = FAISS(
-        #     embedding_function=embedding_model.embed_query,
-        #     index=index,
-        #     docstore= InMemoryDocstore(),
-        #     index_to_docstore_id={}
-        # )
-        vectorstore = FAISS.from_documents(documents=processed_chunks_with_ids, embedding=embedding_model)
-        vectorstore.add_documents(processed_chunks_with_ids, ids = [cid.metadata['doc_id'] for cid in processed_chunks_with_ids])
-        # vectorstore_retriever = vectorstore.as_retriever(search_kwargs={'k': 5})
-        vectorstore_retriever = vectorstore
-        msg = f"Successfully processed {len(files)} documents and created knowledge base."
-        return msg
-
-# @app.post("/chat", response_model=ChatResponse)
-def chat_with_rag(chatdata):
-    global compiled_app
-    global vectorstore_retriever
-    if vectorstore_retriever is None:
-        raise Exception(status_code=400, detail="Knowledge base not loaded. Please upload documents first.")
-    print(f"Received request: {chatdata}")
-    # moderation_result = moderator.moderate_content(request.user_question)
-    # if moderation_result["is_restricted"]:
-    #     # Get appropriate response based on restriction type
-    #     response_type = moderation_result.get("response_type", "general")
-    #     response_text = Config.RESTRICTED_RESPONSES.get(
-    #         response_type, 
-    #         Config.RESTRICTED_RESPONSES["general"]
-    #     )
-        
-    #     logger.warning(
-    #         f"Restricted query: {request.prompt[:100]}... "
-    #         f"Reason: {moderation_result['reason']}"
-    #     )
-        
-    #     return ChatResponse(
-    #         ai_response=response_text,
-    #         updated_chat_history=[],
-    #         telemetry_entry_id=request.session_id,
-    #         is_restricted=True,
-    #         moderation_reason=moderation_result["reason"],
-    #     )
-    print("✅ Question passed the RAI check.........")
-    initial_state = {
-        # "chat_history": [msg.model_dump() for msg in chatdata.get('chat_history')],
-        "chat_history": [msg for msg in chatdata.get('chat_history')],
-        "retrieved_documents": [],
-        "user_question": chatdata.get('user_question'),
-        "session_id": chatdata.get('session_id')
-    }
-
-    try:
-        config = {"configurable": {"thread_id": chatdata.get('session_id')}}
-        final_state = compiled_app.invoke(initial_state, config=config)
-        
-        ai_response_message = final_state["chat_history"][-1]["content"]
-        updated_chat_history_dicts = final_state["chat_history"]
-
-        response_chat = ChatResponse(
-            ai_response=ai_response_message,
-            updated_chat_history=updated_chat_history_dicts,
-            telemetry_entry_id=final_state.get("telemetry_id"),
-            is_restricted=False,
-        )
-        return response_chat.dict()
-    except Exception as e:
-        print(f"Internal Server Error: {e}")
-        raise Exception(status_code=500, detail=f"An error occurred during chat processing: {e}")
-
-# @app.post("/feedback")
-# def submit_feedback(feedbackdata):
-#     db = SessionLocal()
-#     try:
-#         telemetry_record = db.query(Telemetry).filter(
-#             Telemetry.transaction_id == feedbackdata.telemetry_entry_id,
-#             Telemetry.session_id == feedbackdata.session_id
-#         ).first()
-
-#         if not telemetry_record:
-#             raise Exception(status_code=404, detail="Telemetry entry not found or session ID mismatch.")
-
-#         existing_feedback = db.query(Feedback).filter(
-#             Feedback.telemetry_entry_id == feedbackdata.telemetry_entry_id
-#         ).first()
-
-#         if existing_feedback:
-#             existing_feedback.feedback_score = feedbackdata.feedback_score
-#             existing_feedback.feedback_text = feedbackdata.feedback_text
-#             existing_feedback.timestamp = datetime.datetime.now()
-#         else:
-#             feedback_record = Feedback(
-#                 telemetry_entry_id=feedbackdata.telemetry_entry_id,
-#                 feedback_score=feedbackdata.feedback_score,
-#                 feedback_text=feedbackdata.feedback_text,
-#                 user_query=telemetry_record.user_question,
-#                 llm_response=telemetry_record.response,
-#                 timestamp=datetime.datetime.now()
-#             )
-#             db.add(feedback_record)
-            
-#         db.commit()
-
-#         return {"message": "Feedback submitted successfully."}
-
-#     except Exception as e:
-#         raise e
-#     except Exception as e:
-#         db.rollback()
-#         raise Exception(status_code=500, detail=f"An error occurred: {str(e)}")
-#     finally:
-#         db.close()
-
-def submit_feedback(feedbackdata):
-    db = SessionLocal()
-    try:
-        telemetry_record = db.query(Telemetry).filter(
-            Telemetry.transaction_id == feedbackdata['telemetry_entry_id'],
-            Telemetry.session_id == feedbackdata['session_id']
-        ).first()
-
-        if not telemetry_record:
-            raise Exception(status_code=404, detail="Telemetry entry not found or session ID mismatch.")
-
-        existing_feedback = db.query(Feedback).filter(
-            Feedback.telemetry_entry_id == feedbackdata['telemetry_entry_id']
-        ).first()
-
-        if existing_feedback:
-            existing_feedback.feedback_score = feedbackdata['feedback_score']
-            existing_feedback.feedback_text = feedbackdata['feedback_text']
-            existing_feedback.timestamp = datetime.datetime.now()
-        else:
-            feedback_record = Feedback(
-                telemetry_entry_id=feedbackdata['telemetry_entry_id'],
-                feedback_score=feedbackdata['feedback_score'],
-                feedback_text=feedbackdata['feedback_text'],
-                user_query=telemetry_record.user_question,
-                llm_response=telemetry_record.response,
-                timestamp=datetime.datetime.now()
-            )
-            db.add(feedback_record)
-            
-        db.commit()
-
-        return {"message": "Feedback submitted successfully."}
-
-    except Exception as e:
-        raise e
-    except Exception as e:
-        db.rollback()
-        raise Exception(status_code=500, detail=f"An error occurred: {str(e)}")
-    finally:
-        db.close()
-
-# @app.get("/conversations", response_model=List[ConversationSummary])
-def get_conversations():
-    db = SessionLocal()
-    try:
-        conversations = db.query(ConversationHistory).order_by(ConversationHistory.last_updated.desc()).all()
-        summaries = []
-        for conv in conversations:
-            for msg in conv.messages:
-                print(msg)
-            first_user_message = next((msg for msg in conv.messages if msg["role"] == "user"), None)
-            title = first_user_message.get("content") if first_user_message else "New Conversation"
-            summaries.append({"session_id":conv.session_id, "title":title[:30] + "..." if len(title) > 30 else title})
-        return summaries
-    finally:
-        db.close()
-
-# @app.get("/conversations/{session_id}", response_model=List[ChatHistoryEntry])
-def get_conversation_history(session_id: str):
-    db = SessionLocal()
-    try:
-        conversation = db.query(ConversationHistory).filter(ConversationHistory.session_id == session_id).first()
-        if not conversation:
-            raise Exception(status_code=404, detail="Conversation not found.")
-        return conversation.messages
-    finally:
-        db.close()
-
-if __name__ == "__main__":
-    pass
-#     uvicorn.run(app, host="0.0.0.0", port=8000)