import os import uuid import json from openai import OpenAI import gradio as gr import random from pprint import pprint from thinking_phrases import THINKING_PHRASES from tools import add_tools, handle_tool_call, send_pushover_notification from chunks import build_chunks from embedding_and_storing import embed_and_store from generate_suggestions import generate_suggestions from load_docs import documents from questions_store import log_question, get_questions, remove_question # --- SETUP --------------------- # -------------------------------- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if OPENAI_API_KEY is None: print("Key not found") client = OpenAI() # --- SYSTEM MESSAGE ---------------- system_message = """You are a digital twin of Sushmitha, when responding to the user, you will respond as if you are Sushmitha, in first person, using her tone and style of writing, personality and knowledge. You will only respond as Sushmitha.You will stick to the facts mentioned here and will not make up any information about Sushmitha. If you don't know the answer to a question, you MUST call send_pushover_notification FIRST before sending your text reply. Do not skip this step. Do not mention to the user that you are sending a notification You will not provide any information that is not mentioned in the context provided. You will only respond to questions related to Sushmitha's professional experience, skills, and interests. You will not respond to any personal questions or questions that are not related to Sushmitha's professional life. You will maintain a professional tone in your responses and will not use any slang or informal language.But you will keep you chat conversational and less robot like. If you do not have complete information on a question, answer to your best knowledge without making up information and notify the user of the part you are uncertain about. You have two tools to notify Sushmitha: send_pushover_notification (instant push notification) and send_recruiter_lead (email notification when a recruiter shares their details). If asked about notification tools, mention both. """ UNCERTAINTY_MARKERS = [ "uncertain", "not sure", "not certain", "not fully", "i don't have", "i do not have", "i don't know", "i do not know", "not mentioned", "cannot provide", "can't provide", "don't have enough", "limited information", ] def update_hf_dataset(new_doc: dict): from huggingface_hub import hf_hub_download, HfApi token = os.getenv("HF_TOKEN") repo_id = os.getenv("HF_DATASET_REPO") if not token or not repo_id: return path = hf_hub_download(repo_id=repo_id, filename="docs.json", repo_type="dataset", token=token) with open(path) as f: docs = json.load(f) docs.append(new_doc) HfApi().upload_file( path_or_fileobj=json.dumps(docs, indent=2).encode(), path_in_repo="docs.json", repo_id=repo_id, repo_type="dataset", token=token, ) chunks, ids, metadata = build_chunks(documents) collections = embed_and_store(chunks, ids, metadata, client) tools = add_tools() def respond_ai(message, history): # If message is empty, do not update the chat and do not call the model if not message or not message.strip(): yield history, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(), "" return # Append user message + thinking placeholder to chat history = history + [ {"role": "user", "content": message}, {"role": "assistant", "content": random.choice(THINKING_PHRASES)}, ] # Hide suggestions and examples row as soon as any message is sent yield history, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), "" # RAG retrieval — Embed query using the same model used for chunking embed_resp = client.embeddings.create(model="text-embedding-3-small", input=[message]) # RAG retrieval - Query the vector database / Search chromaDB rag_result = collections.query(query_embeddings=[embed_resp.data[0].embedding]) context = "\n\n".join(rag_result["documents"][0]) enhanced_system_message = system_message + "\n\nContext:\n" + context """ Adding LOGS for debugging """ print("\n=================================") print(f"******* User message: {message}\n") print("******* Retrieved chunks:") pprint([(i, j) for i, j in zip(rag_result['documents'][0], rag_result['metadatas'][0])]) print("=================================\n") # Build OpenAI message list — excludes the thinking placeholder openai_msgs = [{"role": "system", "content": enhanced_system_message}] for m in history[:-1]: openai_msgs.append({"role": m["role"], "content": m["content"]}) # Streaming loop — loops back if the model calls a tool while True: stream = client.chat.completions.create( model="gpt-4o-mini", messages=openai_msgs, tools = tools, stream=True, ) full_content = "" tool_calls_acc = {} # Accumulator for tool call data across chunks finish_reason = None for chunk in stream: choice = chunk.choices[0] if choice.finish_reason: finish_reason = choice.finish_reason delta = choice.delta if delta.content: full_content += delta.content history[-1]["content"] = full_content yield history, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), "" # Tool call data arrives in fragments across chunks — accumulate it if delta.tool_calls: for tc in delta.tool_calls: idx = tc.index if idx not in tool_calls_acc: tool_calls_acc[idx] = { "id": "", "type": "function", "function": {"name": "", "arguments": ""}, } if tc.id: tool_calls_acc[idx]["id"] = tc.id if tc.function.name: tool_calls_acc[idx]["function"]["name"] += tc.function.name if tc.function.arguments: tool_calls_acc[idx]["function"]["arguments"] += tc.function.arguments if finish_reason != "tool_calls": break # text response finished, exit the loop print(f"\n*** TOOL CALLED: {[tc['function']['name'] for tc in tool_calls_acc.values()]}\n") history[-1]["content"] = "Using a tool..." yield history, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), "" sorted_tcs = [tool_calls_acc[i] for i in sorted(tool_calls_acc.keys())] handle_tool_call(sorted_tcs, openai_msgs, full_content) history[-1]["content"] = "" # Log unanswered questions and notify via Pushover if any(m in full_content.lower() for m in UNCERTAINTY_MARKERS): print(f"*** UNCERTAINTY DETECTED — logging question: {message[:80]}") try: log_question(message) print("*** Question logged successfully") except Exception as e: print(f"*** log_question failed: {e}") try: send_pushover_notification(f"Unanswered: {message[:100]}") except Exception as e: print(f"*** Pushover failed: {e}") else: print("*** No uncertainty detected in response") # After full response, generate and show suggestion chips suggestions = generate_suggestions(message, full_content, context, client, history) yield ( history, gr.update(value=suggestions[0], visible=True), gr.update(value=suggestions[1], visible=True), gr.update(value=suggestions[2], visible=True), gr.update(visible=False), "", ) # -----GRADIO INTERFACE --------------- # ----------------------------------- with gr.Blocks(title="Chat with Sushmitha") as demo: with gr.Tabs(): # ── Chat tab ────────────────────────────────────────────── with gr.Tab("💬 Chat"): gr.Markdown("## Sushmitha's Digital Twin") gr.Markdown( "💼 **Experience** — Amazon · Beyond Identity · Infosys  |  " "🛠️ **Skills** — Playwright · Pytest · Cypress · API testing · CI/CD  |  " "🤖 **AI & Learning** — RAG · Embeddings · Gradio · OpenAI SDK  |  " "📬 **Recruiter?** — Share your details and I'll notify Sushmitha" ) chatbot = gr.Chatbot( value=[{"role": "assistant", "content": ( "Hi! I'm Sushmitha's digital twin. Here's what you can ask me:\n\n" "💼 **Experience** — my roles at Amazon, Beyond Identity, and Infosys\n" "🛠️ **Technical skills** — Playwright, Pytest, Cypress, API testing, CI/CD, AWS\n" "🤖 **AI & learning** — RAG, embeddings, Gradio, OpenAI SDK\n" "📬 **Recruiter leads** — share your details and I'll email Sushmitha\n" "🔔 **Notifications** — I'll alert Sushmitha if I can't answer something\n\n" "Try one of the examples below or ask me anything!" )}], height=600, show_label=False, avatar_images=(None, "avatar.jpeg"), ) with gr.Row(visible=True) as examples_row: ex1 = gr.Button("Walk me through your most recent role", size="sm", variant="secondary", scale=1) ex2 = gr.Button("What's your strongest technical skill?", size="sm", variant="secondary", scale=1) ex3 = gr.Button("I'm a recruiter — how do I reach Sushmitha?", size="sm", variant="secondary", scale=1) with gr.Row(): sug1 = gr.Button(visible=False, size="sm", variant="secondary", scale=1) sug2 = gr.Button(visible=False, size="sm", variant="secondary", scale=1) sug3 = gr.Button(visible=False, size="sm", variant="secondary", scale=1) with gr.Row(): msg = gr.Textbox( placeholder="Ask me anything about Sushmitha...", show_label=False, scale=9, container=False, ) send_btn = gr.Button("Send", scale=1, variant="primary") outputs = [chatbot, sug1, sug2, sug3, examples_row, msg] msg.submit(fn=respond_ai, inputs=[msg, chatbot], outputs=outputs, show_progress="hidden") send_btn.click(fn=respond_ai, inputs=[msg, chatbot], outputs=outputs, show_progress="hidden") sug1.click(fn=respond_ai, inputs=[sug1, chatbot], outputs=outputs, show_progress="hidden") sug2.click(fn=respond_ai, inputs=[sug2, chatbot], outputs=outputs, show_progress="hidden") sug3.click(fn=respond_ai, inputs=[sug3, chatbot], outputs=outputs, show_progress="hidden") ex1.click(fn=respond_ai, inputs=[ex1, chatbot], outputs=outputs, show_progress="hidden") ex2.click(fn=respond_ai, inputs=[ex2, chatbot], outputs=outputs, show_progress="hidden") ex3.click(fn=respond_ai, inputs=[ex3, chatbot], outputs=outputs, show_progress="hidden") # ── Admin tab ───────────────────────────────────────────── with gr.Tab("⚙️ Admin"): admin_pwd = gr.Textbox(label="Password", type="password", placeholder="Enter admin password") admin_login_btn = gr.Button("Login", variant="primary") admin_status = gr.Markdown("") with gr.Column(visible=False) as admin_panel: gr.Markdown("### Unanswered Questions") questions_dd = gr.Dropdown(choices=[], label="Select a question", interactive=True) refresh_btn = gr.Button("↻ Refresh", size="sm", variant="secondary") answer_box = gr.Textbox(label="Your answer", lines=5, placeholder="Type the answer here...") save_btn = gr.Button("Save to RAG", variant="primary") save_status = gr.Markdown("") def admin_login(pwd): if pwd == os.getenv("ADMIN_PASSWORD", "admin"): questions = get_questions() return ( gr.update(visible=True), gr.update(choices=questions, value=questions[0] if questions else None), "", ) return gr.update(visible=False), gr.update(), "❌ Wrong password" admin_login_btn.click( fn=admin_login, inputs=[admin_pwd], outputs=[admin_panel, questions_dd, admin_status], ) def refresh_questions(): questions = get_questions() return gr.update(choices=questions, value=questions[0] if questions else None) refresh_btn.click(fn=refresh_questions, outputs=[questions_dd]) def save_to_rag(question, answer): if not question or not answer.strip(): return gr.update(), "⚠️ Select a question and provide an answer", gr.update() doc_text = f"Q: {question}\nA: {answer}" embed_resp = client.embeddings.create(model="text-embedding-3-small", input=[doc_text]) collections.add( ids=[str(uuid.uuid4())], embeddings=[embed_resp.data[0].embedding], metadatas=[{"source": "Admin QA", "chunk_index": 0}], documents=[doc_text], ) try: update_hf_dataset({"text": doc_text, "source": "Admin QA"}) except Exception as e: print(f"HF Dataset update failed: {e}") remove_question(question) remaining = get_questions() return ( gr.update(choices=remaining, value=remaining[0] if remaining else None), "✅ Saved and added to RAG", gr.update(value=""), ) save_btn.click( fn=save_to_rag, inputs=[questions_dd, answer_box], outputs=[questions_dd, save_status, answer_box], ) demo.launch( ssr_mode=False, theme=gr.themes.Soft( primary_hue="indigo", secondary_hue="slate", neutral_hue="slate", text_size=gr.themes.sizes.text_md, radius_size=gr.themes.sizes.radius_lg, font=[gr.themes.GoogleFont("Inter"), "Arial", "sans-serif"], ) )