Spaces:
Sleeping
Sleeping
| from dotenv import load_dotenv | |
| from openai import OpenAI | |
| import json | |
| import os | |
| import requests | |
| from pypdf import PdfReader | |
| import gradio as gr | |
| from pydantic import BaseModel | |
| import logging | |
| # Load environment variables | |
| load_dotenv(override=True) | |
| TOOL_SIMULATION = os.getenv("TOOL_SIMULATION", "false").lower() == "true" | |
| # Setup logging | |
| logging.basicConfig(filename="tool_logs.log", level=logging.INFO) | |
| def push(text): | |
| if TOOL_SIMULATION: | |
| print(f"[SIMULATED PUSH]: {text}") | |
| else: | |
| 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"): | |
| msg = f"Recording {name} with email {email} and notes {notes}" | |
| push(msg) | |
| logging.info(msg) | |
| return {"recorded": "ok"} | |
| def record_unknown_question(question): | |
| msg = f"Recording unknown question: {question}" | |
| push(msg) | |
| logging.info(msg) | |
| 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", | |
| "format": "email", | |
| "pattern": "^\\S+@\\S+\\.\\S+$" | |
| }, | |
| "name": { | |
| "type": "string", | |
| "description": "The user's name, if they provided it" | |
| }, | |
| "notes": { | |
| "type": "string", | |
| "description": "generate a summary of the conversation. 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 Evaluation(BaseModel): | |
| is_acceptable: bool | |
| feedback: str | |
| class Me: | |
| def __init__(self): | |
| self.openai = OpenAI() | |
| self.name = "Sarthak Pawar" | |
| reader = PdfReader("me/linkedin.pdf") | |
| self.linkedin = "" | |
| for page in reader.pages: | |
| text = page.extract_text() | |
| if text: | |
| self.linkedin += text | |
| with open("me/summary.txt", "r", encoding="utf-8") as f: | |
| self.summary = f.read() | |
| def handle_tool_call(self, tool_calls): | |
| results = [] | |
| valid_tools = { | |
| "record_user_details": record_user_details, | |
| "record_unknown_question": record_unknown_question | |
| } | |
| for tool_call in tool_calls: | |
| try: | |
| tool_name = tool_call.function.name | |
| arguments = json.loads(tool_call.function.arguments) | |
| print(f"Tool called: {tool_name} with args: {arguments}", flush=True) | |
| if tool_name not in valid_tools: | |
| push(f"Invalid tool call attempted: {tool_name}") | |
| results.append({ | |
| "role": "tool", | |
| "content": json.dumps({"error": f"Unknown tool: {tool_name}"}), | |
| "tool_call_id": tool_call.id | |
| }) | |
| else: | |
| result = valid_tools[tool_name](**arguments) | |
| results.append({ | |
| "role": "tool", | |
| "content": json.dumps(result), | |
| "tool_call_id": tool_call.id | |
| }) | |
| except Exception as e: | |
| push(f"Error handling tool call: {str(e)}") | |
| results.append({ | |
| "role": "tool", | |
| "content": json.dumps({"error": f"Error handling tool call: {str(e)}"}), | |
| "tool_call_id": tool_call.id | |
| }) | |
| return results | |
| def system_prompt(self): | |
| 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. \ | |
| IMPORTANT: If you don't know the answer to any question OR if the question is unrelated to {self.name}'s career/background/skills/experience, YOU MUST USE THE `record_unknown_question` tool. \ | |
| If the user is engaging in discussion related to {self.name}'s career/background/skills/experience or wants to work with {self.name}, try to steer them towards getting in touch via email. \ | |
| If the user provides their name and email, you should use the `record_user_details` tool. | |
| ## gaurd rails: | |
| - if the user is asking about {self.name}'s career/background/skills/experience or wants to work with {self.name}, the agent should get the user to provide their name and email and do not call any tools. | |
| - if the user is asking about anything unrelated to {self.name}'s career/background/skills/experience, the agent should use the `record_unknown_question` tool and try to guide the user back to the topic of {self.name}'s career/background/skills/experience. | |
| - if and only if the user provides their name and email, the agent should use the `record_user_details` tool and reply with a message that says "great, I'll get back to you as soon as possible." | |
| - if you think the conversation is going off topic, restric the bot to bring it back on topic: {self.name}'s career/background/skills/experience | |
| """ | |
| prompt += """ | |
| ## Tool Call Evaluation Criteria: | |
| tool information: | |
| - only call record_user_details if the user provides their name or email. | |
| - only call record_unknown_question if the user is asking about anything unrelated to {self.name}'s career/background/skills/experience or you don't know the answer. | |
| if a specific tool was spposed to be called, but wasn't, that's a failure. | |
| if a tool was called, but the response is not related to the tool call, that's a failure. | |
| for example, | |
| user: can you help me with my web development project? | |
| assistant: sure, I can help you with that. | |
| tool call: null | |
| this is a failure because the tool call was null. it should have been record_unknown_question. | |
| user: how can I contact you? | |
| assistant: please provide your name and email and I'll get back to you as soon as possible. | |
| tool call: null | |
| this is a success. the user is asking to be contacted, so the agent should get the user to provide their name and email. | |
| user: my name is shivam and you can contact me at [email protected] | |
| assistant: null | |
| tool call: record_user_details | |
| this is a success because the tool call was record_user_details and the response is related to the tool call. | |
| user: do you watch f1 | |
| assistant: yes, I do. | |
| tool call: null | |
| this is a failure because the tool call was null. it should have been record_unknown_question. any question unrelated to the user's career/background/skills/experience should call record_unknown_question. | |
| user: who is your father | |
| assistant: null | |
| tool call: record_unknown_question | |
| this is a success because the tool call was record_unknown_question and the response is related to the tool call. | |
| user: how many stars are there in the milky way? | |
| assistant: null | |
| tool call: record_unknown_question | |
| this is a success because the tool call was record_unknown_question and the response is related to the tool call. | |
| user: would you be open to working with me? | |
| assistant: sure, I'd be happy to work with you. please provide your name and email and I'll get back to you as soon as possible. | |
| tool call: null | |
| this is a success because the user is asking to work with the agent, so the agent should get the user to provide their name and email. | |
| user: you can contact me at [email protected] | |
| assistant: null | |
| tool call: record_user_details | |
| this is a success because the user is providing their name and email, so the agent should use the `record_user_details` tool. | |
| """ | |
| prompt += f"\n\n## Summary:\n{self.summary}\n\n## LinkedIn Profile:\n{self.linkedin}\n\n" | |
| return prompt | |
| def get_evaluator_prompt(self) -> str: | |
| return self.system_prompt() + "\n\nYou are now evaluating if the assistant is behaving correctly per these guidelines." | |
| def get_tool_response_prompt(self) -> str: | |
| return f"""You are {self.name} responding to a user after a tool has been called. | |
| ## Context: | |
| - A tool was just executed (either recording user details or recording an unknown question) | |
| - You should now provide a natural, conversational response to the user | |
| - Do NOT call any tools - just respond conversationally | |
| - Keep the response professional and engaging | |
| - If the tool was `record_user_details`, acknowledge that you'll get back to them | |
| - If the tool was `record_unknown_question`, guide the conversation back to your career/background/skills/experience | |
| ## Your background: | |
| {self.summary} | |
| ## LinkedIn Profile: | |
| {self.linkedin} | |
| Remember: You are representing {self.name} professionally. Be helpful, engaging, and steer conversations toward your expertise and career opportunities.""" | |
| def evaluator_user_prompt(self, reply: str, message: str, history: str, tool_call) -> str: | |
| return f"""You are evaluating a conversation between a user and an AI assistant impersonating a real person. | |
| --- | |
| ## Conversation History: | |
| {history} | |
| --- | |
| ## Latest User Message: | |
| {message} | |
| --- | |
| ## Assistant's Latest Reply: | |
| {reply} | |
| --- | |
| ## Tool Call: | |
| {tool_call} | |
| ## Tool Call Evaluation Criteria: | |
| tool information: | |
| - only call record_user_details if the user provides their name or email. | |
| - only call record_unknown_question if the user is asking about anything unrelated to {self.name}'s career/background/skills/experience or ##you don't know the answer. | |
| if a specific tool was spposed to be called, but wasn't, that's a failure. | |
| if a tool was called, but the response is not related to the tool call, that's a failure. | |
| for example, | |
| user: can you help me with my web development project? | |
| assistant: sure, I can help you with that. | |
| tool call: null | |
| this is a failure because the tool call was null. it should have been record_unknown_question. | |
| user: how can I contact you? | |
| assistant: please provide your name and email and I'll get back to you as soon as possible. | |
| tool call: null | |
| this is a success. the user is asking to be contacted, so the agent should get the user to provide their name and email. | |
| user: my name is shivam and you can contact me at [email protected] | |
| assistant: null | |
| tool call: record_user_details | |
| this is a success because the tool call was record_user_details and the response is related to the tool call. | |
| user: do you watch f1 | |
| assistant: yes, I do. | |
| tool call: null | |
| this is a failure because the tool call was null. it should have been record_unknown_question. any question unrelated to the user's career/background/skills/experience should call record_unknown_question. | |
| user: who is your father | |
| assistant: null | |
| tool call: record_unknown_question | |
| this is a success because the tool call was record_unknown_question and the response is related to the tool call. | |
| user: how many stars are there in the milky way? | |
| assistant: null | |
| tool call: record_unknown_question | |
| this is a success because the tool call was record_unknown_question and the response is related to the tool call. | |
| user: would you be open to working with me? | |
| assistant: sure, I'd be happy to work with you. please provide your name and email and I'll get back to you as soon as possible. | |
| tool call: null | |
| this is a success because the user is asking to work with the agent, so the agent should get the user to provide their name and email. | |
| user: you can contact me at [email protected] | |
| assistant: null | |
| tool call: record_user_details | |
| this is a success because the user is providing their name and email, so the agent should use the `record_user_details` tool. | |
| ## gaurd rails: | |
| - when a tool call is made, the agent will not provide a reply. that will be handled in the next step. so don't judge the reply when a tool call is made. because it will be null. | |
| - if the user is asking about {self.name}'s career/background/skills/experience or wants to work with {self.name}, the agent should get the user to provide their name and email and do not call any tools. | |
| - if the user is asking about anything unrelated to {self.name}'s career/background/skills/experience, the agent should use the `record_unknown_question` tool and try to guide the user back to the topic of {self.name}'s career/background/skills/experience. | |
| - if and only if the user provides their name and email, the agent should use the `record_user_details` tool and reply with a message that says "great, I'll get back to you as soon as possible." | |
| - if you think the conversation is going off topic, restric the bot to bring it back on topic: {self.name}'s career/background/skills/experience | |
| Please evaluate the assistant's response. | |
| - Is the response acceptable? (True/False) | |
| - Feedback: (Explain what was good or what needs improvement)""" | |
| def evaluate(self, reply, message, history, tool_call) -> Evaluation: | |
| messages = [ | |
| {"role": "system", "content": self.get_evaluator_prompt()}, | |
| {"role": "user", "content": self.evaluator_user_prompt(reply, message, history, tool_call)} | |
| ] | |
| try: | |
| response = self.openai.beta.chat.completions.parse( | |
| model="gpt-4.1-mini", | |
| messages=messages, | |
| response_format=Evaluation | |
| ) | |
| return response.choices[0].message.parsed | |
| except Exception as e: | |
| push(f"Evaluation failed: {str(e)}") | |
| return Evaluation(is_acceptable=False, feedback="Evaluation parsing failed or incomplete.") | |
| def rerun(self, reply, message, history, feedback): | |
| updated_system_prompt = self.system_prompt() + f"\n\n## Previous answer rejected:\n{reply}\n\nReason: {feedback}\n respond appropriately according to the gaurd rails." | |
| messages = [{"role": "system", "content": updated_system_prompt}] + history + [{"role": "user", "content": message}] | |
| return self.openai.chat.completions.create(model="gpt-4.1-mini", messages=messages, tools=tools) | |
| def chat(self, message, history): | |
| history = history[-10:] | |
| messages = [{"role": "system", "content": self.system_prompt()}] + history + [{"role": "user", "content": message}] | |
| satisfied = False | |
| max_retries = 3 | |
| retries = 0 | |
| response = self.openai.chat.completions.create(model="gpt-4.1-mini", messages=messages, tools=tools) | |
| reply = response.choices[0].message.content | |
| finish_reason = response.choices[0].finish_reason | |
| tool_call = response.choices[0].message.tool_calls | |
| feedback = "" | |
| while not satisfied and retries < max_retries: | |
| if(retries > 0): | |
| response = self.rerun(reply, message, history, feedback) | |
| reply = response.choices[0].message.content | |
| finish_reason = response.choices[0].finish_reason | |
| tool_call = response.choices[0].message.tool_calls | |
| print(f"rerun_message: {response.choices[0].message}\n\n", flush=True) | |
| print(f"rerun_reply: {reply}\n\n", flush=True) | |
| print(f"rerun_message: {message}\n\n", flush=True) | |
| print(f"rerun_history: {history}\n\n", flush=True) | |
| print(f"rerun_tool_call: {tool_call}\n\n", flush=True) | |
| evaluation = self.evaluate(reply, message, history, tool_call) | |
| if evaluation.is_acceptable: | |
| retries = 0 | |
| print(f"Evaluation successful: {evaluation.feedback}", flush=True) | |
| if finish_reason == "tool_calls": | |
| print(f"Tool calls: {tool_call}", flush=True) | |
| tool_calls = response.choices[0].message.tool_calls | |
| results = self.handle_tool_call(tool_calls) | |
| messages.append(response.choices[0].message) | |
| messages.extend(results) | |
| response_messages = [{"role": "system", "content": self.get_tool_response_prompt()}] + history + [{"role": "user", "content": message}] | |
| response = self.openai.chat.completions.create(model="gpt-4.1-mini", messages=response_messages) | |
| reply = response.choices[0].message.content | |
| print(f"response: {response.choices[0].message}\n\n", flush=True) | |
| satisfied = True | |
| return reply | |
| print(f"satisfied\n\n", flush=True) | |
| satisfied = True | |
| else: | |
| print(f"reply: {reply}\n\n", flush=True) | |
| print(f"message: {message}\n\n", flush=True) | |
| print(f"history: {history}\n\n", flush=True) | |
| print(f"tool_call: {tool_call}\n\n", flush=True) | |
| print(f"Evaluation failed: {evaluation.feedback}\n\n", flush=True) | |
| feedback = evaluation.feedback | |
| retries += 1 | |
| if(retries >= max_retries): | |
| print(f"Max retries reached\n\n", flush=True) | |
| return "I'm sorry, I'm having trouble answering your question. can we move back to talking about my career/background/skills/experience?" | |
| return reply | |
| if __name__ == "__main__": | |
| me = Me() | |
| gr.ChatInterface(me.chat, type="messages").launch() |