from dotenv import load_dotenv import os import chromadb from chromadb.config import Settings from openai import OpenAI from sentence_transformers import SentenceTransformer from pypdf import PdfReader import requests from pydantic import BaseModel import gradio as gr import json load_dotenv(override=True) def handle_tool_call(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) tool = globals().get(tool_name) result = tool(**arguments) if tool else {} results.append({"role": "tool","content": json.dumps(result),"tool_call_id": tool_call.id}) return results class Evaluation(BaseModel): is_acceptable: bool feedback: str model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') 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 __init__(self, name, cv_path): self.name = name self.chroma_client = chromadb.HttpClient(host=os.environ["CHROMA_DB_CLIEN"], port=8000, settings=Settings( chroma_client_auth_provider="chromadb.auth.token_authn.TokenAuthClientProvider", chroma_client_auth_credentials=os.environ["CHROMA_TOKEN"] )) self.collection = self.chroma_client.get_collection(name="all-my-projects") self.model = model self.openai = OpenAI() self.cv = "" reader = PdfReader(cv_path) for page in reader.pages: text = page.extract_text() if text: self.cv += text self.system_prompt = f"You are acting as {self.name}. You are answering questions on {self.name}'s website, \ particularly questions related to {name}'s career, background, skills and experience. \ Your responsibility is to represent {name} for interactions on the website as faithfully as possible. \ You are given a summary of {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. \ " self.system_prompt += f"\n\## CV:\n{self.cv}\n\n" self.system_prompt += f"With this context, please chat with the user, always staying in character as {self.name}." self.gemini = OpenAI( api_key=os.getenv("GOOGLE_API_KEY"), base_url="https://generativelanguage.googleapis.com/v1beta/openai/") self.evaluator_system_prompt = f"You are an evaluator that decides whether a response to a question is acceptable. \ You are provided with a conversation between a User and an Agent. Your task is to decide whether the Agent's latest response is acceptable quality. \ The Agent is playing the role of {name} and is representing {self.name} on their website. \ The Agent has been instructed to be professional and engaging, as if talking to a potential client or future employer who came across the website. \ The Agent has been provided with context on {self.name} in the form of their summary and LinkedIn details. Here's the information:" self.evaluator_system_prompt += f"## CV:\n{self.cv}\n\n" self.evaluator_system_prompt += f"With this context, please evaluate the latest response, replying with whether the response is acceptable and your feedback." def evaluator_user_prompt(self, reply, message, history): user_prompt = f"Here's the conversation between the User and the Agent: \n\n{history}\n\n" user_prompt += f"Here's the latest message from the User: \n\n{message}\n\n" user_prompt += f"Here's the latest response from the Agent: \n\n{reply}\n\n" user_prompt += f"Please evaluate the response, replying with whether it is acceptable and your feedback." return user_prompt def evaluate(self, reply, message, history) -> Evaluation: messages = [{"role": "system", "content": self.evaluator_system_prompt}] + [{"role": "user", "content": self.evaluator_user_prompt(reply, message, history)}] response = self.gemini.beta.chat.completions.parse(model="gemini-2.0-flash", messages=messages, response_format=Evaluation) return response.choices[0].message.parsed def embed(self, text): return self.model.encode(text) def find_similars(self, question): results = self.collection.query(query_embeddings=self.embed(question).astype(float).tolist(), n_results=5,include=['documents',"distances"]) documents = results['documents'][0][:] distances=results['distances'][0][:] filtered_documents = [ doc for doc, dist in zip(documents, distances) if dist < 1.7 ] return filtered_documents def rerun(self, reply, message, history, feedback): updated_system_prompt = self.system_prompt + f"\n\n## Previous answer rejected\nYou just tried to reply, but the quality control rejected your reply\n" updated_system_prompt += f"## Your attempted answer:\n{reply}\n\n" updated_system_prompt += f"## Reason for rejection:\n{feedback}\n\n" messages = [{"role": "system", "content": updated_system_prompt}] + history + [{"role": "user", "content": message}] response = self.openai.chat.completions.create(model="gpt-4o-mini", messages=messages) return response.choices[0].message.content def make_context(self,similars): if len(similars)==0: return "" message = f"To provide some context, here are some projects done by {self.name} that might be related to the question that you need to answer.\n\n" for similar in similars: message += f"Potentially related projects:\n{similar}\n\n" return message def chat(self,message,history): similars=self.find_similars(message) message+=self.make_context(similars) messages = [{"role": "system", "content": self.system_prompt}]+history + [{"role": "user", "content": message}] done = False while not done: response = self.openai.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools) if response.choices[0].finish_reason=="tool_calls": message = response.choices[0].message tool_calls = message.tool_calls results = handle_tool_call(tool_calls) messages.append(message) messages.extend(results) else: done = True reply=response.choices[0].message.content evaluation = self.evaluate(reply, message, history) if evaluation.is_acceptable: return reply else: return self.rerun(reply, message, history, evaluation.feedback) if __name__ == "__main__": me = Me(name="Djallel BRAHMIA", cv_path="documents/CV/CV.pdf") gr.ChatInterface(me.chat, type="messages").launch()