Update app.py

app.py

@@ -1,44 +1,36 @@
+# app.py — NVIDIA NIM + Tool Calling + Gradio Chatbot
+# Model: meta/llama3-8b-instruct  (supports OpenAI-style tools)
+# Works on HuggingFace Spaces with your nvapi-... key.
 
-
-from openai import OpenAI
 import os
 import json
 import requests
 from pypdf import PdfReader
+from openai import OpenAI
 import gradio as gr
-import traceback
 
-# ---------------------------
-# Configuration (environment)
-# ---------------------------
-OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")  # nvapi-... (NVIDIA NIM)
-# NVIDIA NIM compatible base URL:
-OPENAI_BASE_URL = "https://integrate.api.nvidia.com/v1"
+# ===============================
+# ENVIRONMENT CONFIG
+# ===============================
+OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")   # Your nvapi-XXXX key
+BASE_URL = "https://integrate.api.nvidia.com/v1"
+MODEL = "meta/llama3-8b-instruct"
 
-# Pushover (optional)
 PUSHOVER_TOKEN = os.environ.get("PUSHOVER_TOKEN")
 PUSHOVER_USER = os.environ.get("PUSHOVER_USER")
 
-# Model to call on NVIDIA NIM
-MODEL = "qwen/qwen2-7b-instruct"
-
-# Initialize OpenAI client (NIM via OpenAI-compatible client)
-# The OpenAI Python client accepts api_key and base_url in constructor.
-oa_client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
-
-# ---------------------------
-# Utility: push notifications
-# ---------------------------
-def push(text: str):
-    """
-    Send a Pushover notification if credentials are available.
-    If not available, just print to stdout (no failure).
-    """
+client = OpenAI(api_key=OPENAI_API_KEY, base_url=BASE_URL)
+
+
+# ===============================
+# OPTIONAL: Pushover notification
+# ===============================
+def push(text):
     try:
         if not PUSHOVER_TOKEN or not PUSHOVER_USER:
-            print("Pushover not configured - message would be:", text)
+            print("Pushover not configured:", text)
             return
-        resp = requests.post(
+        requests.post(
             "https://api.pushover.net/1/messages.json",
             data={
                 "token": PUSHOVER_TOKEN,
@@ -47,275 +39,166 @@ def push(text: str):
             },
             timeout=10
         )
-        if resp.status_code != 200:
-            print("Pushover returned", resp.status_code, resp.text)
     except Exception as e:
-        print("Failed sending pushover:", e)
-
-# ---------------------------
-# Tools definitions (JSON schemas)
-# ---------------------------
-record_user_details_json = {
-    "name": "record_user_details",
-    "description": "Record that a user is interested in being in touch and provided an email address",
-    "parameters": {
-        "type": "object",
-        "properties": {
-            "email": {"type": "string", "description": "The email address of this user"},
-            "name": {"type": "string", "description": "The user's name, if provided"},
-            "notes": {"type": "string", "description": "Any additional info about the conversation"}
-        },
-        "required": ["email"],
-        "additionalProperties": False
-    }
-}
-
-record_unknown_question_json = {
-    "name": "record_unknown_question",
-    "description": "Record any question that couldn't be answered",
-    "parameters": {
-        "type": "object",
-        "properties": {
-            "question": {"type": "string", "description": "The question that couldn't be answered"}
-        },
-        "required": ["question"],
-        "additionalProperties": False
-    }
-}
+        print("Pushover failed:", e)
 
-tools = [
-    {"type": "function", "function": record_user_details_json},
-    {"type": "function", "function": record_unknown_question_json}
-]
 
-# ---------------------------
-# Local tools (callable functions)
-# ---------------------------
+# ===============================
+# TOOL IMPLEMENTATIONS
+# ===============================
 def record_user_details(email, name="Name not provided", notes="not provided"):
-    push(f"Recording {name} with email {email} and notes {notes}")
-    print(f"[TOOL] record_user_details: email={email} name={name} notes={notes}", flush=True)
-    # Here you might append to a DB / google sheet / file. For Space demo we just return ok.
-    return {"recorded": "ok", "email": email, "name": name, "notes": notes}
+    push(f"New lead → {name} | {email} | Notes: {notes}")
+    return {"status": "ok", "email": email, "name": name, "notes": notes}
 
 def record_unknown_question(question):
-    push(f"Recording unknown question: {question}")
-    print(f"[TOOL] record_unknown_question: {question}", flush=True)
-    return {"recorded": "ok", "question": question}
+    push(f"Unknown question recorded: {question}")
+    return {"status": "ok", "question": question}
+
+
+# ===============================
+# TOOL JSON DEFINITIONS
+# ===============================
+tools = [
+    {
+        "type": "function",
+        "function": {
+            "name": "record_user_details",
+            "description": "Record user interest and their email.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "email": {"type": "string"},
+                    "name": {"type": "string"},
+                    "notes": {"type": "string"}
+                },
+                "required": ["email"]
+            }
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "record_unknown_question",
+            "description": "Record any question the assistant could not answer.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "question": {"type": "string"}
+                },
+                "required": ["question"]
+            }
+        }
+    }
+]
 
-# Register tool functions in globals so they can be invoked by name
 globals()["record_user_details"] = record_user_details
 globals()["record_unknown_question"] = record_unknown_question
 
-# ---------------------------
-# The assistant wrapper
-# ---------------------------
+
+# ===============================
+# MAIN ASSISTANT CLASS
+# ===============================
 class Me:
     def __init__(self):
         self.name = "Ayush Tyagi"
-        # load LinkedIn pdf and summary (if present)
-        self.linkedin = ""
-        self.summary = "Summary not provided."
-        try:
-            pdf_path = "me/Ayush_linkdin.pdf"
-            if os.path.exists(pdf_path):
-                reader = PdfReader(pdf_path)
-                text = []
-                for page in reader.pages:
-                    page_text = page.extract_text()
-                    if page_text:
-                        text.append(page_text)
-                self.linkedin = "\n\n".join(text).strip()
-            else:
-                print(f"{pdf_path} not found in repo; skipping PDF loading.")
-        except Exception as e:
-            print("Error loading PDF:", e)
-            traceback.print_exc()
-
-        try:
-            summary_path = "me/summary.txt"
-            if os.path.exists(summary_path):
-                with open(summary_path, "r", encoding="utf-8") as f:
-                    self.summary = f.read()
-            else:
-                print(f"{summary_path} not found in repo; using fallback summary.")
-        except Exception as e:
-            print("Error reading summary.txt:", e)
-            traceback.print_exc()
-
-        # Prebuild the system prompt (keeps it simple & safe)
-        self._system_prompt = self._build_system_prompt()
-
-    def _build_system_prompt(self):
-        sp = f"""You are acting as {self.name}. Your role is to answer questions on {self.name}'s personal website,
-specifically those related to {self.name}'s career, background, skills, and professional experience.
-Represent {self.name} accurately, professionally and engagingly.
-
-If you don't know the answer to any question, say you don't know and use the record_unknown_question tool
-to record the question. If the user wants to stay in touch, ask for their email and use the record_user_details tool.
-
-## Summary:
+        self.summary = ""
+        self.linkedin_text = ""
+
+        # Load summary text
+        if os.path.exists("me/summary.txt"):
+            self.summary = open("me/summary.txt", "r", encoding="utf-8").read()
+
+        # Load LinkedIn PDF (optional)
+        pdf_path = "me/Ayush_linkdin.pdf"
+        if os.path.exists(pdf_path):
+            text = []
+            reader = PdfReader(pdf_path)
+            for page in reader.pages:
+                page_text = page.extract_text()
+                if page_text:
+                    text.append(page_text)
+            self.linkedin_text = "\n\n".join(text)
+
+    def system_prompt(self):
+        return f"""
+You are acting as {self.name}. Answer questions professionally about Ayush's skills,
+career, background, and experience.
+
+If you DON'T know something → call the tool:
+- record_unknown_question
+
+If the user shows interest → ask for an email and call:
+- record_user_details
+
+Be polite, confident, friendly and helpful.
+
+### Summary:
 {self.summary}
 
-## LinkedIn (extracted text, if available):
-{self.linkedin}
+### LinkedIn Data:
+{self.linkedin_text}
 """
-        return sp
-
-    def system_prompt(self):
-        return self._system_prompt
-
-    def handle_tool_call(self, tool_calls):
-        """
-        Accepts a list of tool call objects returned by the model (tool_calls).
-        Each tool_call is expected to have:
-          - function.name (string)
-          - function.arguments (JSON string)
-          - id (optional)
-        """
-        results = []
-        for tool_call in tool_calls:
-            try:
-                # Different client shapes exist; be defensive
-                func_name = None
-                func_args_json = None
-                call_id = None
-
-                # attempt several shapes
-                if hasattr(tool_call, "function"):
-                    # tool_call.function may be a simple namespace with .name and .arguments
-                    func = tool_call.function
-                    func_name = getattr(func, "name", None) or func.get("name") if isinstance(func, dict) else func_name
-                    func_args_json = getattr(func, "arguments", None) or (func.get("arguments") if isinstance(func, dict) else None)
-                # fallback for dict-like
-                if not func_name and isinstance(tool_call, dict):
-                    func_name = tool_call.get("function", {}).get("name") or tool_call.get("name")
-                    func_args_json = tool_call.get("function", {}).get("arguments") or tool_call.get("arguments")
-                    call_id = tool_call.get("id")
-
-                # also check top-level
-                if not func_args_json:
-                    func_args_json = getattr(tool_call, "arguments", None) or tool_call.get("arguments") if isinstance(tool_call, dict) else func_args_json
-
-                if not func_name:
-                    print("Could not determine tool name for tool_call:", tool_call)
-                    continue
-
-                # parse JSON args (models often return JSON-string)
-                args = {}
-                if func_args_json:
-                    if isinstance(func_args_json, str):
-                        try:
-                            args = json.loads(func_args_json)
-                        except Exception:
-                            # sometimes arguments come as dict already
-                            try:
-                                args = eval(func_args_json)
-                            except Exception:
-                                args = {}
-                    elif isinstance(func_args_json, dict):
-                        args = func_args_json
-
-                print(f"Tool called: {func_name} with args: {args}", flush=True)
-                tool = globals().get(func_name)
-                if callable(tool):
-                    result = tool(**args)
-                else:
-                    print(f"No tool function found for {func_name}")
-                    result = {}
-                # build tool result entry in a shape compatible with continuing the chat
-                result_content = json.dumps(result)
-                results.append({"role": "tool", "content": result_content, "tool_call_id": call_id})
-            except Exception as e:
-                print("Error during tool handling:", e)
-                traceback.print_exc()
-                results.append({"role": "tool", "content": json.dumps({"error": str(e)}), "tool_call_id": None})
-        return results
 
+    # ===============================
+    # CHAT LOOP WITH TOOL CALLING
+    # ===============================
     def chat(self, message, history):
-        """
-        Gradio ChatInterface-compatible function: (message, history) -> str
-        history is a list of tuples (user, assistant) or a list of message dicts depending on Gradio version.
-        We'll convert to a message list compatible with the model.
-        """
-        # Build messages array (OpenAI chat format)
-        # Start with system prompt
         messages = [{"role": "system", "content": self.system_prompt()}]
 
-        # Convert gradio history into assistant/user messages robustly:
-        # history can be list of pairs [[user, bot], ...] or list of dict messages
-        try:
-            if history:
-                # If history items are tuples/lists of length 2
-                if isinstance(history, list) and len(history) and isinstance(history[0], (list, tuple)) and len(history[0]) == 2:
-                    for user_msg, assistant_msg in history:
-                        if user_msg:
-                            messages.append({"role": "user", "content": user_msg})
-                        if assistant_msg:
-                            messages.append({"role": "assistant", "content": assistant_msg})
-                else:
-                    # Fallback: if history is list of dicts with role/content
-                    for item in history:
-                        if isinstance(item, dict) and "role" in item and "content" in item:
-                            messages.append({"role": item["role"], "content": item["content"]})
-        except Exception as e:
-            print("Failed to normalize history:", e)
-            traceback.print_exc()
-
-        # Add the latest user message
+        # Convert history from Gradio into chat API format
+        for user_msg, bot_msg in history:
+            if user_msg:
+                messages.append({"role": "user", "content": user_msg})
+            if bot_msg:
+                messages.append({"role": "assistant", "content": bot_msg})
+
         messages.append({"role": "user", "content": message})
 
-        # Loop to support tool-calls (the client may return tool_calls finish_reason)
-        done = False
-        last_response_text = "Sorry — something went wrong."
-        try:
-            while not done:
-                # Call the model
-                response = oa_client.chat.completions.create(
-                    model=MODEL,
-                    messages=messages,
-                    tools=tools,
-                    max_tokens=512
-                )
-                # defensive extraction
-                choice0 = response.choices[0]
-                finish_reason = getattr(choice0, "finish_reason", None) or (choice0.get("finish_reason") if isinstance(choice0, dict) else None)
-                # If model asks to call tools:
-                if finish_reason == "tool_calls" or getattr(choice0, "message", None) and getattr(choice0.message, "tool_calls", None):
-                    # extract the message object
-                    model_message = getattr(choice0, "message", None) or choice0.get("message", {})
-                    tool_calls = getattr(model_message, "tool_calls", None) or model_message.get("tool_calls", [])
-                    # handle tool calls
-                    results = self.handle_tool_call(tool_calls)
-                    # append the model message (which requested tool calls) and the tool result messages
-                    messages.append(model_message if isinstance(model_message, dict) else {"role": "assistant", "content": getattr(model_message, "content", "")})
-                    messages.extend(results)
-                else:
-                    # final content
-                    msg_obj = getattr(choice0, "message", None) or choice0.get("message", {})
-                    content = getattr(msg_obj, "content", None) or (msg_obj.get("content") if isinstance(msg_obj, dict) else None)
-                    if not content:
-                        # Some clients put the text at choice0.text or similar
-                        content = getattr(choice0, "text", None) or (choice0.get("text") if isinstance(choice0, dict) else "")
-                    last_response_text = content or " (no content returned by model) "
-                    done = True
-        except Exception as e:
-            # Log and return a helpful message
-            print("Error calling model:", e)
-            traceback.print_exc()
-            last_response_text = "Sorry, the model call failed. Check logs in Space build/runtime for details."
-
-        return last_response_text
-
-# ---------------------------
-# Instantiate and run Gradio
-# ---------------------------
+        while True:
+            response = client.chat.completions.create(
+                model=MODEL,
+                messages=messages,
+                tools=tools,
+                tool_choice="auto",
+                max_tokens=600
+            )
+
+            choice = response.choices[0]
+            finish = choice.finish_reason
+            msg = choice.message
+
+            # ---- TOOL CALL ----
+            if finish == "tool_calls":
+                for tool_call in msg.tool_calls:
+                    func = tool_call.function
+                    name = func.name
+                    args = json.loads(func.arguments)
+
+                    tool_fn = globals().get(name)
+                    result = tool_fn(**args)
+
+                    messages.append(msg.dict())  # append tool call request
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "content": json.dumps(result)
+                    })
+                continue  # loop again and let model respond
+
+            # ---- NORMAL RESPONSE ----
+            return msg.content
+
+
+# ===============================
+# GRADIO APP
+# ===============================
 me = Me()
 
-# Gradio ChatInterface (simpler)
-iface = gr.ChatInterface(fn=me.chat, type="messages", title="Ayush Tyagi — Personal Assistant")
+ui = gr.ChatInterface(
+    fn=me.chat,
+    title="Ayush Tyagi — Personal Assistant",
+    type="messages"
+)
 
 if __name__ == "__main__":
-    # In Spaces, it's recommended to bind to 0.0.0.0 and use the PORT envvar if provided.
-    server_name = "0.0.0.0"
-    server_port = int(os.environ.get("PORT", 7860))
-    iface.launch(server_name=server_name, server_port=server_port)
+    ui.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))