Spaces:
Sleeping
Sleeping
| 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"], | |
| ) | |
| ) | |