| from dotenv import load_dotenv
|
| from openai import OpenAI
|
| import json
|
| import os
|
| import requests
|
| from pypdf import PdfReader
|
| import gradio as gr
|
| from pathlib import Path
|
|
|
|
|
| load_dotenv(override=True)
|
|
|
| def push(text):
|
| requests.post(
|
| "https://api.pushover.net/1/messages.json",
|
| data={
|
| "token": os.getenv("PUSHOVER_TOKEN"),
|
| "user": os.getenv("PUSHOVER_USER"),
|
| "message": text,
|
| }
|
| )
|
|
|
|
|
| def record_user_details(email, name="Name not provided", notes="not provided"):
|
| push(f"Recording {name} with email {email} and notes {notes}")
|
| return {"recorded": "ok"}
|
|
|
| def record_unknown_question(question):
|
| push(f"Recording {question}")
|
| return {"recorded": "ok"}
|
|
|
| record_user_details_json = {
|
| "name": "record_user_details",
|
| "description": "Use this tool to 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 they provided it"
|
| }
|
| ,
|
| "notes": {
|
| "type": "string",
|
| "description": "Any additional information about the conversation that's worth recording to give context"
|
| }
|
| },
|
| "required": ["email"],
|
| "additionalProperties": False
|
| }
|
| }
|
|
|
| record_unknown_question_json = {
|
| "name": "record_unknown_question",
|
| "description": "Always use this tool to record any question that couldn't be answered as you didn't know the answer",
|
| "parameters": {
|
| "type": "object",
|
| "properties": {
|
| "question": {
|
| "type": "string",
|
| "description": "The question that couldn't be answered"
|
| },
|
| },
|
| "required": ["question"],
|
| "additionalProperties": False
|
| }
|
| }
|
|
|
| tools = [{"type": "function", "function": record_user_details_json},
|
| {"type": "function", "function": record_unknown_question_json}]
|
|
|
|
|
| class Me:
|
|
|
| def _resolve_cv_path(self):
|
| configured_path = os.getenv("CV_PATH")
|
| if configured_path:
|
| path = Path(configured_path)
|
| if not path.is_absolute():
|
| path = Path(__file__).parent / path
|
| if path.exists():
|
| return path
|
|
|
| me_dir = Path(__file__).parent / "me"
|
| pdf_files = sorted(me_dir.glob("*.pdf"), key=lambda p: p.stat().st_mtime, reverse=True)
|
| if not pdf_files:
|
| raise FileNotFoundError(f"No PDF found in {me_dir}")
|
| return pdf_files[0]
|
|
|
| def __init__(self):
|
| self.openai = OpenAI()
|
| self.name = "Venkata Vikranth Jannatha"
|
| cv_path = self._resolve_cv_path()
|
| print(f"Loading CV from: {cv_path}", flush=True)
|
| reader = PdfReader(str(cv_path))
|
| self.linkedin = ""
|
| for page in reader.pages:
|
| text = page.extract_text()
|
| if text:
|
| self.linkedin += text
|
| self.summary = self.linkedin[:2000]
|
|
|
|
|
| def handle_tool_call(self, tool_calls):
|
| results = []
|
| for tool_call in tool_calls:
|
| tool_name = tool_call.function.name
|
| arguments = json.loads(tool_call.function.arguments)
|
| print(f"Tool Called : {tool_name}", flush=True)
|
|
|
| if tool_name == "record_user_details":
|
| result = record_user_details(**arguments)
|
| elif tool_name == "record_unknown_question":
|
| result = record_unknown_question(**arguments)
|
| else:
|
| result = {}
|
|
|
| results.append({"role": "tool","content": json.dumps(result),"tool_call_id": tool_call.id})
|
| return results
|
|
|
| def system_prompt(self):
|
| system_prompt = f"You are acting as {self.name}. You are answering questions on {self.name}'s website, \
|
| particularly questions related to {self.name}'s career, background, skills and experience. \
|
| Your responsibility is to represent {self.name} for interactions on the website as faithfully as possible. \
|
| You are given a summary of {self.name}'s background and LinkedIn profile which you can use to answer questions. \
|
| Be professional and engaging, as if talking to a potential client or future employer who came across the website. \
|
| If you don't know the answer to any question, use your record_unknown_question tool to record the question that you couldn't answer, even if it's about something trivial or unrelated to career. \
|
| If the user is engaging in discussion, try to steer them towards getting in touch via email; ask for their email and record it using your record_user_details tool. "
|
|
|
| system_prompt += f"\n\n## Summary:\n{self.summary}\n\n## LinkedIn Profile:\n{self.linkedin}\n\n"
|
| system_prompt += f"With this context, please chat with the user, always staying in character as {self.name}."
|
| return system_prompt
|
|
|
| def _normalize_history(self, history):
|
| """Convert Gradio chat history into OpenAI chat messages."""
|
| if not history:
|
| return []
|
|
|
| allowed_roles = {"system", "user", "assistant", "tool", "function", "developer"}
|
|
|
|
|
| if isinstance(history, list) and history and isinstance(history[0], dict):
|
| normalized = []
|
| for item in history:
|
| role = item.get("role")
|
| content = item.get("content")
|
| if role in allowed_roles and isinstance(content, str):
|
| normalized.append({"role": role, "content": content})
|
| return normalized
|
|
|
|
|
| normalized = []
|
| for pair in history:
|
| if not (isinstance(pair, (list, tuple)) and len(pair) == 2):
|
| continue
|
| user_msg, assistant_msg = pair
|
| if user_msg is not None and str(user_msg).strip() != "":
|
| normalized.append({"role": "user", "content": str(user_msg)})
|
| if assistant_msg is not None and str(assistant_msg).strip() != "":
|
| normalized.append({"role": "assistant", "content": str(assistant_msg)})
|
| return normalized
|
|
|
| def chat(self, message, history):
|
| history_messages = self._normalize_history(history)
|
| messages = [{"role": "system", "content": self.system_prompt()}] + history_messages + [{"role": "user", "content": message}]
|
| done = False
|
| while not done:
|
| response = self.openai.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools)
|
| finish_reason = response.choices[0].finish_reason
|
| print(finish_reason)
|
|
|
| if finish_reason == "tool_calls":
|
| message_obj = response.choices[0].message
|
| tool_calls = message_obj.tool_calls
|
| results = self.handle_tool_call(tool_calls)
|
| messages.append(message_obj)
|
| messages.extend(results)
|
| else:
|
| done = True
|
| return response.choices[0].message.content
|
|
|
|
|
| if __name__ == "__main__":
|
| me = Me()
|
| gr.ChatInterface(me.chat).launch(share=True)
|
|
|
|
|
| |
