Spaces:
Sleeping
Sleeping
| 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() | |