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 shivam@gmail.com 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 shivam@gmail.com 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 shivam@gmail.com 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 shivam@gmail.com 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()