main.py

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()