Blame for agent.py

import os
import json
import requests
import gradio as gr
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

import warnings
warnings.filterwarnings("ignore")

# 1. Environment & Config
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

openai = OpenAI()
NAME = 'RISHIVEL S'
PDF_PATH = "me/Rishivel_Personal_KB.pdf"
FAISS_INDEX_PATH = "faiss_index"

# Pushover Setup
PUSHOVER_USER = os.getenv("PUSHOVER_USER")
PUSHOVER_TOKEN = os.getenv("PUSHOVER_TOKEN")
PUSHOVER_URL = "https://api.pushover.net/1/messages.json"

if PUSHOVER_USER and PUSHOVER_TOKEN:
    print(f"Pushover configured for {PUSHOVER_USER[0]}***")
else:
    print("Pushover credentials not found or incomplete.")

def push(message):
    print(f"Internal Note: {message}")
    if PUSHOVER_USER and PUSHOVER_TOKEN:
        payload = {'user': PUSHOVER_USER, 'token': PUSHOVER_TOKEN, 'message': message}
        try:
            requests.post(PUSHOVER_URL, data=payload)
        except Exception as e:
            print(f"Failed to send Pushover notification: {e}")

# 2. Vector DB Storage & Retrieval Logic
print("Initializing Vector Database...")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

if os.path.exists(FAISS_INDEX_PATH):
    print("Loading existing FAISS index from disk...")
    vectorstore = FAISS.load_local(FAISS_INDEX_PATH, embeddings, allow_dangerous_deserialization=True)
else:
    print("Creating new FAISS index...")
    loader = PyPDFLoader(PDF_PATH)
    documents = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=50)
    texts = text_splitter.split_documents(documents)
    
    vectorstore = FAISS.from_documents(texts, embeddings)
    vectorstore.save_local(FAISS_INDEX_PATH)
    print(f"FAISS index saved to {FAISS_INDEX_PATH}")


def retrive_context(query):
    results = vectorstore.similarity_search(query)
    context = ""
    for result in results:
        context += result.page_content + "\n\n"
    return context

# 3. Tools Definitions
def record_user_details_func(email, name='name not provided', notes="not provided"):
    push(f"New user details recorded: {email}, name : {name}, conversation : {notes}")
    return {'recorded': 'ok'}

def record_unknown_question_func(question):
    push(f"Recording '{question}' asked that I couldn't answer")
    return {"recorded": "ok"}

record_user_details = {
    "name": "record_user_details",
    "description": "Record user details like email, name and notes",
    "parameters": {
        "type": "object",
        "properties": {
            "email": {"type": "string", "description": "email of the user"},
            "name": {"type": "string", "description": "name of the user"},
            "notes": {"type": "string", "description": "any additional notes about the user"}
        },
        "required": ["email"],
        "additionalProperties": False    
    }
}

record_unknown_question = {
    "name": 'record_unknown_question',
    "description": "Record any question that I couldn't answer to improve my knowledge base",
    "parameters": {
        "type": "object",
        "properties": {
            "question": {
                "type": "string",
                "description": "the question that I couldn't answer"
            }
        },  
        "required": ["question"],
        "additionalProperties": False
    }
}

tools = [
    {'type': 'function', 'function': record_user_details}, 
    {'type': 'function', 'function': record_unknown_question}
]

# 4. Agent Configuration
SYSTEM_PROMPT = f"""You are acting as {NAME}. You are answering questions on {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 context of the given query 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, name and record it using your record_user_details tool. \
You will be given the context for the user's query; try to frame your sentence using that."""

class Evaluator(BaseModel):
    valid_output: bool
    feedback: str

def evaluator(user_message, output, history) -> Evaluator:
    eval_system_prompt = "You are an evaluator for the personal agent. Your task is to evaluate the output of the agent based on the user message and the conversation history. You will be given the user message, the agent's output and the conversation history. You need to evaluate if the agent's output is valid, if not provide feedback on how to improve it. A valid output is one that answers the user's question accurately and is relevant to the conversation history. The feedback should be constructive and specific, pointing out what was good about the output and what can be improved."
    messages = [{"role": "system", "content": eval_system_prompt}] + history + [{"role": "user", "content": f"user message : {user_message} \n agent output : {output} \n "}]
    response = openai.chat.completions.parse(
        model="gpt-4o-mini",
        messages=messages,
        response_format=Evaluator
    )
    return response.choices[0].message.parsed   

def rerun(message, context, history):
    rerun_prompt = f"The provided answer is not valid based on the user message and conversation history. Please provide a new answer based on the following context that is relevant to the user message: {context} and the conversation history: {history}\nMake sure to answer the user's question accurately and be relevant to the conversation history. Don't deviate from the context and conversation history while answering the question. Don't answer questions that are not relevant to personal career, background, skills and experience. If the question is not relevant, say you don't know the answer but would love to share more about my career, background, skills, and experience instead."
    
    messages = [{"role": "system", "content": rerun_prompt}] + history + [{"role": "user", "content": message}]
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools
    )
    return response.choices[0].message.content

def chat_logic(user_message, history_gradio):
    # Convert Gradio history format to OpenAI format
    history = []
    for user_msg, sys_msg in history_gradio:
        history.append({"role": "user", "content": user_msg})
        history.append({"role": "assistant", "content": sys_msg})

    context = retrive_context(user_message)
    full_prompt = SYSTEM_PROMPT + f"\nContext:\n{context}"

    messages = [{'role': 'system', 'content': full_prompt}] + history + [{'role': 'user', 'content': user_message}]

    done = False
    msg = None
    
    while not done:
        response = openai.chat.completions.create(
            model="gpt-4o-mini",
            tools=tools,
            messages=messages
        )

        msg = response.choices[0].message
        finish_reason = response.choices[0].finish_reason

        messages.append(msg)

        if finish_reason == "tool_calls":
            tool_calls = msg.tool_calls
            for tool_call in tool_calls:
                name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)

                if name == "record_user_details":
                    result = record_user_details_func(**args)
                elif name == "record_unknown_question":
                    result = record_unknown_question_func(**args)
                else:
                    result = "none"

                messages.append({
                    "role": "tool",
                    "tool_name": name,
                    "content": str(result),
                    "tool_call_id": tool_call.id
                })
        else:
            done = True
            
    if msg and msg.content:
        # evaluate the output
        evaluator_decision = evaluator(user_message, msg.content, history)
        if evaluator_decision.valid_output:
            return msg.content
        else:
            print(f"Feedback for improvement: {evaluator_decision.feedback}")
            return rerun(user_message, context, history)
            
    return "I couldn't quite put an answer together for that. Could you ask me something else about my background?"

# 5. Gradio UI
demo = gr.ChatInterface(
    fn=chat_logic,
    title=f"{NAME}'s Personal AI Agent",
    description="Welcome! I'm an AI representative for Rishivel S. Ask me anything about my background, projects, skills, or career goals!",
    examples=[
        "What are your top projects?",
        "Tell me about your hackathon experience",
        "What technologies do you use for backend?",
        "Can I get your contact details?"
    ],
    theme=gr.themes.Soft(primary_hue="blue", neutral_hue="slate")
)

if __name__ == "__main__":
    demo.launch()