Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| from urllib.parse import urlparse | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from groq import Groq | |
| from langchain_core.documents import Document | |
| from langchain_core.vectorstores import InMemoryVectorStore | |
| from langchain_huggingface import HuggingFaceEmbeddings | |
| from langchain_text_splitters import RecursiveCharacterTextSplitter | |
| # Set this in environment (recommended for Hugging Face Spaces secrets). | |
| # Fallback to empty string for local runs. | |
| GROQ_API_KEY = os.getenv("GROQ_API_KEY", "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").strip() | |
| BRANCH_LOCATOR_URL = "https://www.hdfcbank.com/personal/locators/branch-locator" | |
| ALLOWED_HOSTS = {"www.hdfcbank.com", "hdfcbank.com"} | |
| DEFAULT_APPROVED_URLS = [ | |
| "https://www.hdfcbank.com/personal/locators/branch-locator", | |
| "https://www.hdfcbank.com/personal/locators/atm-locator", | |
| "https://www.hdfcbank.com/personal/pay/cards/debit-cards", | |
| "https://www.hdfcbank.com/personal/pay/cards/credit-cards", | |
| "https://www.hdfcbank.com/personal/borrow/popular-loans/personal-loan", | |
| "https://www.hdfcbank.com/personal/save/accounts/savings-accounts", | |
| "https://www.hdfcbank.com/personal/pay/money-transfer", | |
| "https://www.hdfcbank.com/personal/help", | |
| ] | |
| TEXT_SPLITTER = RecursiveCharacterTextSplitter( | |
| chunk_size=900, | |
| chunk_overlap=150, | |
| separators=["\n\n", "\n", ". ", " ", ""], | |
| ) | |
| VECTOR_STORE = None | |
| def normalize_text(value): | |
| return re.sub(r"\s+", " ", str(value or "").strip().lower()) | |
| EMBEDDINGS = HuggingFaceEmbeddings( | |
| model_name="sentence-transformers/all-MiniLM-L6-v2", | |
| model_kwargs={"device": "cpu"}, | |
| encode_kwargs={"normalize_embeddings": True}, | |
| ) | |
| def clean_location_text(value): | |
| """Remove timing/bank filler words from extracted location phrases.""" | |
| text = normalize_text(value) | |
| if not text: | |
| return "" | |
| removable_words = { | |
| "hdfc", | |
| "bank", | |
| "branch", | |
| "branches", | |
| "time", | |
| "timing", | |
| "timings", | |
| "hours", | |
| "hour", | |
| "open", | |
| "close", | |
| "working", | |
| "in", | |
| "at", | |
| "near", | |
| "for", | |
| } | |
| tokens = [token for token in re.sub(r"[^a-z\s-]", " ", text).split() if token not in removable_words] | |
| return " ".join(tokens).strip() | |
| def get_approved_public_urls(): | |
| env_value = os.getenv("HDFC_APPROVED_URLS", "").strip() | |
| if not env_value: | |
| return DEFAULT_APPROVED_URLS | |
| urls = [value.strip() for value in env_value.split(",") if value.strip()] | |
| return urls if urls else DEFAULT_APPROVED_URLS | |
| def is_allowed_hdfc_url(url): | |
| parsed = urlparse(url) | |
| if parsed.scheme not in {"http", "https"}: | |
| return False | |
| return parsed.netloc.lower() in ALLOWED_HOSTS | |
| def html_to_clean_text(html_content): | |
| soup = BeautifulSoup(html_content, "html.parser") | |
| for tag in soup(["script", "style", "noscript"]): | |
| tag.decompose() | |
| return re.sub(r"\s+", " ", soup.get_text(separator=" ")).strip() | |
| def ingest_public_hdfc_documents(urls, request_timeout=12): | |
| documents = [] | |
| session = requests.Session() | |
| session.headers.update( | |
| { | |
| "User-Agent": "Mozilla/5.0 (compatible; HDFC-RAG-Bot/1.0; +https://www.hdfcbank.com)", | |
| "Accept": "text/html,application/xhtml+xml", | |
| } | |
| ) | |
| for url in urls: | |
| if not is_allowed_hdfc_url(url): | |
| continue | |
| try: | |
| response = session.get(url, timeout=request_timeout) | |
| response.raise_for_status() | |
| text = html_to_clean_text(response.text) | |
| title = urlparse(url).path.strip("/") or "hdfc-home" | |
| title = title.replace("-", " ").replace("/", " > ").strip() or "HDFC public page" | |
| base_document = Document( | |
| page_content=text, | |
| metadata={"title": title, "source_url": url}, | |
| ) | |
| for idx, chunk in enumerate(TEXT_SPLITTER.split_documents([base_document]), start=1): | |
| documents.append( | |
| { | |
| "title": f"{title} (chunk {idx})", | |
| "content": chunk.page_content, | |
| "source_url": url, | |
| } | |
| ) | |
| except Exception: | |
| continue | |
| return documents | |
| def initialize_knowledge_base(): | |
| global VECTOR_STORE | |
| retriever_documents = [] | |
| seed_documents = ingest_public_hdfc_documents(get_approved_public_urls()) | |
| for doc in seed_documents: | |
| content = str(doc.get("content", "")).strip() | |
| title = str(doc.get("title", "")).strip() or "HDFC public info" | |
| source_url = str(doc.get("source_url", "")).strip() | |
| if not content: | |
| continue | |
| retriever_documents.append( | |
| Document( | |
| page_content=content, | |
| metadata={ | |
| "title": title, | |
| "source_url": source_url, | |
| }, | |
| ) | |
| ) | |
| if not retriever_documents: | |
| VECTOR_STORE = None | |
| return | |
| VECTOR_STORE = InMemoryVectorStore(EMBEDDINGS) | |
| VECTOR_STORE.add_documents(retriever_documents) | |
| def retrieve_relevant_context(query, top_k=4): | |
| if not VECTOR_STORE: | |
| return "", [] | |
| top_docs = VECTOR_STORE.similarity_search(query, k=top_k) | |
| context_parts = [] | |
| citations = [] | |
| for doc in top_docs: | |
| context_parts.append( | |
| f"Title: {doc.metadata.get('title', 'HDFC public info')}\n" | |
| f"Source URL: {doc.metadata.get('source_url', 'N/A')}\n" | |
| f"Snippet: {doc.page_content}" | |
| ) | |
| citations.append( | |
| { | |
| "title": doc.metadata.get("title", "HDFC public info"), | |
| "source_url": doc.metadata.get("source_url", "N/A"), | |
| } | |
| ) | |
| return "\n\n".join(context_parts), citations | |
| def extract_branch_location(query): | |
| normalized = re.sub(r"\s+", " ", (query or "").strip()).lower() | |
| if not normalized: | |
| return None | |
| if "branch" not in normalized and "hdfc" not in normalized: | |
| return None | |
| patterns = [ | |
| r"\b(?:branch\s+)?(?:in|at|near)\s+([a-z][a-z\s\-]{2,40})", | |
| r"\bhdfc\s+(?:bank\s+)?(?:branch\s+)?([a-z][a-z\s\-]{2,40})", | |
| ] | |
| for pattern in patterns: | |
| match = re.search(pattern, normalized) | |
| if match: | |
| location = clean_location_text(match.group(1)) | |
| if location: | |
| return location.title() | |
| tokens = re.sub(r"[^a-z\s-]", " ", normalized).split() | |
| stop_words = { | |
| "hdfc", | |
| "bank", | |
| "branch", | |
| "branches", | |
| "timing", | |
| "timings", | |
| "open", | |
| "close", | |
| "hours", | |
| "hour", | |
| "for", | |
| "in", | |
| "at", | |
| "near", | |
| "what", | |
| "is", | |
| "the", | |
| "me", | |
| } | |
| candidates = [token for token in tokens if token not in stop_words] | |
| if candidates: | |
| location = clean_location_text(" ".join(candidates[:3])) | |
| if location: | |
| return location.title() | |
| return None | |
| def banking_chat_handler(message, history, model_name): | |
| if not GROQ_API_KEY: | |
| return "Configuration Error: Please set GROQ_API_KEY as an environment variable." | |
| branch_location = extract_branch_location(message) | |
| retrieved_context, citations = retrieve_relevant_context(message, top_k=4) | |
| def call_direct_llm_fallback(): | |
| fallback_prompt = ( | |
| "You are a helpful HDFC Bank support assistant. " | |
| "Answer user questions directly and clearly. " | |
| "Do not invent exact branch addresses, phone numbers, fees, or timings when uncertain. " | |
| "If exact details are unavailable, say EXACT_ANSWER_NOT_FOUND. " | |
| "Never provide personal account data or ask for OTP/PIN/CVV." | |
| ) | |
| fallback_messages = [{"role": "system", "content": fallback_prompt}] | |
| for user_msg, assistant_msg in history[-6:]: | |
| if user_msg: | |
| fallback_messages.append({"role": "user", "content": user_msg}) | |
| if assistant_msg: | |
| fallback_messages.append({"role": "assistant", "content": assistant_msg}) | |
| fallback_messages.append({"role": "user", "content": message}) | |
| fallback_completion = client.chat.completions.create( | |
| model=model_name, | |
| messages=fallback_messages, | |
| temperature=0.0, | |
| max_tokens=700, | |
| ) | |
| return fallback_completion.choices[0].message.content | |
| system_prompt = ( | |
| "You are a precise HDFC Bank support assistant using RAG context. " | |
| "Answer only from the retrieved context below. " | |
| "Do not invent branch addresses, phone numbers, fees, timings, or policies not present in context. " | |
| "If exact details are unavailable in context, respond with EXACT_ANSWER_NOT_FOUND. " | |
| "Avoid generic filler language, apologies, and disclaimers. " | |
| "For branch/location queries, mention the requested location explicitly if it is in context. " | |
| "Never provide personal account data or ask for OTP/PIN/CVV.\n\n" | |
| "Response format:\n" | |
| "1) Exact Answer\n" | |
| "2) Evidence: short bullets grounded in context\n\n" | |
| f"--- RETRIEVED CONTEXT ---\n{retrieved_context}\n" | |
| "-------------------------" | |
| ) | |
| if branch_location: | |
| system_prompt += ( | |
| f" The user likely needs branch details for {branch_location}. " | |
| "Prioritize this location in your response." | |
| ) | |
| try: | |
| client = Groq(api_key=GROQ_API_KEY) | |
| if not retrieved_context: | |
| return call_direct_llm_fallback() | |
| messages = [{"role": "system", "content": system_prompt}] | |
| # Keep recent turns so follow-up questions get complete context. | |
| for user_msg, assistant_msg in history[-6:]: | |
| if user_msg: | |
| messages.append({"role": "user", "content": user_msg}) | |
| if assistant_msg: | |
| messages.append({"role": "assistant", "content": assistant_msg}) | |
| messages.append({"role": "user", "content": message}) | |
| completion = client.chat.completions.create( | |
| model=model_name, | |
| messages=messages, | |
| temperature=0.0, | |
| max_tokens=700, | |
| ) | |
| answer_text = completion.choices[0].message.content | |
| if "EXACT_ANSWER_NOT_FOUND" in (answer_text or ""): | |
| return call_direct_llm_fallback() | |
| sources_block = "\n\nSources:\n" + "\n".join( | |
| f"- {item['title']}: {item['source_url']}" for item in citations | |
| ) | |
| return answer_text + sources_block | |
| except Exception as e: | |
| return f"Groq API Call Failed: {str(e)}" | |
| initialize_knowledge_base() | |
| if __name__ == "__main__": | |
| model_name = "llama-3.1-8b-instant" | |
| history = [] | |
| print("HDFC Bank Customer Support AI Agent (CLI)") | |
| print("Type your question and press Enter. Type 'exit' to quit.") | |
| while True: | |
| user_message = input("You: ").strip() | |
| if user_message.lower() in {"exit", "quit"}: | |
| print("Session ended.") | |
| break | |
| answer = banking_chat_handler(user_message, history, model_name) | |
| print(f"Agent: {answer}\n") | |
| history.append((user_message, answer)) | |