import os import chromadb from openai import OpenAI import gradio as gr import spaces# import uuid import json import requests import chromadb import random from pprint import pprint #------------------------------------------------- # Setup #------------------------------------------------- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if OPENAI_API_KEY is None: raise Exception("API Key is missing") client=OpenAI() #------------------------------------------------- # Import Spaces #------------------------------------------------- @spaces.GPU def _dummy(): pass #------------------------------------------------- # Document #------------------------------------------------- document_overview =""" Who is Aleksandr? Aleksandr is an IT professional in the field of Information Systems and Data with deep understanding of the subject. He enjoys working on complex data problems and finding innovative solutions to them. He likes to streamline and improve processes, find insights and deliver value to the stakeholders. Communication Style: You want to respond short and concise. Include emoji for emphasis and use a professional tone, also include bullet points for clarity. and keep the responses structured consistently. Direct, friendly and encouraging. Happy to share what he had learned, and his experience but also include current projects. Keep the format consistent especially for dates, roles and formatting and font. Emojis should be used sparingly and appropriately. If person says Alek, Alex, Aleks, Aleksander, Alexander - assume it refers to Aleksandr. Additionl info: -Aleksandr was a a professional distance runner and enjoys running outdoors. -Aleksandr lives in Los Angeles, California """ document_education =""" My education listed below: (from most recent to oldest) Provider: Udacity Course Name: AI Product Manager When: May 2023 – Aug 2023 Content: Project 1: Created Medical Image Annotation Data Set with Appen - Developed annotation instructions using best practices - Used Figure Eight platform Project 2: Build classification system to flag serious cases of pneumonia using Google AutoML - Developed and evaluated the model according to metrics like accuracy, precision, and recall Project 3: Measuring Business Impact & Mitigating Bias - Suggested further improvements to the model 2nd: Provider:Udacity Course Name: Data Analyst Nanodegree, Data Analysis When: 2019 – 2020 Content: Data Wrangling, Exploration, Visualization using Python: Pandas, NumPy, Matplotlib 3rd: Provider:York University Degree:Bachelor of Commerce and Information Technology, Information Technology and Business Systems Analysis When:2012 – 2016 - Conducted Independent Research Project for Professor Luiz Cysneiros at York University by helping build a knowledge graph. The project required to analyze and conclude the type of relationships that exists between two non-functional requirements (NRF) Transparency and Trust. At least 40 academic sources were used to determine well over 100 different NRFs that either support or hurt these two requirements. The final report was presented in both paper form and as a knowledge graph (modeled diagram), which was used in further research projects. 4th: Provider:Udacity Course Name:Digital Marketing Nanodegree, Marketing When: 2017 – 2018 Content: Market Research: Moz, AdWords Web Platforms and Analytics: Facebook Ads, Google AdSense, Hootsuite, MailChimp Projects: Facebook and Google Analytics ads campaign and A/B testing """ document_Professional_experience =""" Below is professional experience and related work experience: Title: Assistant Director - Data Analyst Dates:Sep 2019 - Present (July 16,2026 and after) Location: New York, New York, United States Projects and experience: Worked in progressive data role as : Data Analyst, progressing through complexity of different projects. ● Partnered with Data Ops and engineering teams to troubleshoot data issues and optimize data ingestion pipelines, improving data quality and availability. ● Developed test scripts and executed UAT for CRM data migrations and financial reporting enhancements; triaged defects and captured stakeholder sign-off. ● Built dashboards in Power BI and Tableau to track NPS responses, to view qualification and engagement trends over time ● Optimized customer insights delivery by reducing NPS Survey processing from 5 weeks to 2 days, improving data accuracy from 55% to 92% through data pipeline enhancements leveraging AWS, SQL, and Gainsight. ● Created process documentation, data mapping specs, KPI frameworks, and runbooks in Confluence to support system go-lives and post-production reporting. ● Led internal Contact Platform backlog, aligning with Product and Engineering to ensure readiness for go-live, user training, and issue escalation ● Served as Data SME, leading cross-functional teams of 2-3 members to enhance forecasting, revenue reporting, and data accessibility, significantly reducing availability time from months to weeks. ● Implemented Lakeflow automation solutions in Databricks to streamline recurring processes ● Used genie on top of older data extracts to aid client support team with information discovery and improve data retrieval efficiency 2nd: Title: IT Consultant: Data Management Organization: Randstad · Contract Dates: Nov 2018 - Sep 2019 Location: New York City Metropolitan Area Projects and experience: Data Governance Specialist working at Moody`s Analytics, focusing on: - Building key Process Lineage - CDE Identification and ownership - Documentation using Collibra - Identification and remediation of Data Quality issues 3rd: Title: Business Analyst / Scrum Master Organization: Royal Bank of Canada, Financial Crimes, Trading Compliance, Toronto, Ontario Dates: September 2016 - November 2018 Location: Toronto, Ontario From April 2015- to November 2018 he worked at RBC as Business Analyst and Scrum Master. ● Gathered and translated requirements into actionable user stories; worked with QA and Dev teams to ensure sprint success and business alignment. ● Led sprint ceremonies for a 12-member team, prioritizing delivery of data governance improvements for Wealth Management and Capital Markets. ● Reviewed technical artifacts and ensured stakeholder feedback was integrated into releases; supported integration of internal tools with enterprise compliance systems. ● Designed dashboards and reporting metrics that improved compliance monitoring and reduced reporting turnaround time. 4th: Title: Business Systems Analyst Organization: Royal Bank of Canada, Enterprise & International Applications, Trading Compliance, Toronto, Ontario Dates: September 2015- April 2016 Location: Toronto, Ontario ● Conducted over 100 user interviews and documented more than 250 business processes, providing foundational insights for large-scale regulatory compliance projects. ● Developed UI mockups and wireframes, enhancing user experience and aligning technical solutions with business goals """ #------------------------------------------------- # Chunking Function #------------------------------------------------- def split_text_into_chunks(text: str, chunk_size: int = 300, overlap: int = 30): BOUNDARIES = ["\n\n", "\n", ". ", "? ", "! ", " "] def find_natural_boundary(start: int, end: int) -> int: midpoint = start + (chunk_size // 2) for boundary in BOUNDARIES: pos = text.rfind(boundary, midpoint, end) if pos != -1: return pos + len(boundary) return end def find_overlap_start(end: int) -> int: window_start = max(0, end - overlap) for boundary in BOUNDARIES: pos = text.find(boundary, window_start, end) if pos != -1 and pos + len(boundary) < end: return pos + len(boundary) return window_start chunks = [] start = 0 while start < len(text): end = min(start + chunk_size, len(text)) if end < len(text): end = find_natural_boundary(start, end) chunks.append(text[start:end]) if end >= len(text): break start = max(start + 1, find_overlap_start(end)) return chunks #------------------------------------------------- # RAG: Chunk, EMBED & Store in ChromoDB #------------------------------------------------- #Prep multiple documents: now the data is coming from 3 sources, how do we chunk- use metadata! #1 Step add all documents to list documents =[ {"text" : document_overview, "source": "Overview"}, {"text" : document_education, "source": "Education"}, {"text" : document_Professional_experience, "source": "Professional Experience"} ] # 2Prepare variables chunks =[] ids =[] metadatas =[] for doc in documents: #Prepare the lists chunks_ = split_text_into_chunks(doc["text"], 300, 30) ids_ = [str(uuid.uuid4()) for _ in range(len(chunks_))] # have to be unique and instead of assigning - lets bring it throuhg uuid. _ is through away variable metadatas_ = [{"source": doc["source"], "chunk_index": i} for i in range(len(chunks_))] # metadas_ creating a range from 0 to eg. 100 based on number of chunks above, then # have a loop that iterates over the chunks to assign the correct chunk_index for each chunk in the metadata list. #Add to main lists and extending because they are already lists chunks.extend(chunks_) ids.extend(ids_) metadatas.extend(metadatas_) #Print for logs print(f"Created {len(chunks)} chunks") for i, chunk in enumerate(chunks): print(f"Chunk {i+1} (ID: {ids[i]}, source: {metadatas[i]['source']}, chunk_index: {metadatas[i]['chunk_index']},length: {len(chunk)})") print(chunk) print() #Generate embeddings for all chunks using openAI - there are many other solutions response = client.embeddings.create( model = "text-embedding-3-small", input = chunks ) embeddings = [item.embedding for item in response.data] #Verify embeddings for logs print(f"Generated {len(embeddings)} embeddings") print(f"Each embedding has {len(embeddings[0])} dimensions") #Initialize ChromaDB client (persistent for local storage) chroma_client=chromadb.PersistentClient(path="./chroma_db_twin") #Persistent client will keep refreshing every run from scratch #Alternative initialize ChromaDB client (in-memory storage) #chroma_client=chromadb.Client() collection=chroma_client.get_or_create_collection(name="digital_twin") #Get or Create + Empty collection before adding new data (for testing purposes) if collection.get()["ids"]: collection.delete(collection.get()["ids"]) #Adding data to ChromaDB collection.add( ids=ids, embeddings=embeddings, #data comes from above - values we created documents=chunks, #data comes from above - values we created metadatas=metadatas ) pprint(collection.get()) #commas in text indicate where chunk ends #------------------------------------------------- # System message #------------------------------------------------- #System message from ChromaDB system_message =""" = Your are a digital twin of Aleksandr Kuternin. When people talk to you, you respond AS Aleksandr - in first person, using his voice, personality and knowledge. Important: Don`t make things up. If you don`t know the answer, say you don`t know. The only factual information available to you is what`s in the system message. You cannot get any more facts about Aleksandr from the internet or make them up. IMPORTANT: Whenever you don`t know something about Aleksandr, ALWAYS use the send_notification tool to alert real Aleksandr - do this automatically without asking the user. """ #------------------------------------------------- # Tools #------------------------------------------------- tools = [] pushover_user = os.getenv("PUSHOVER_USER") pushover_token = os.getenv("PUSHOVER_TOKEN") pushover_url = "https://api.pushover.net/1/messages.json" #Create send_notification function def send_notification(message:str): if pushover_user is None or pushover_token is None: # handling of potential error missing credentials return "Notification failed: pushover not configured." payload = {"user": pushover_user, "token": pushover_token, "message": message} requests.post(pushover_url, data=payload) return f"notification sent: {message}" #Describe Pushover as an LLM tool send_notification_function = { # create a function that describes LLM - and tells what it is. Creating dictionary "name": "send_notification", "description": "Send a push notification to real Aleksandr. Use this when:\ 1) Someone wants to get in touch, hire, or collaborate,\ -ask for their name and contact details first, then send notification to Aleksandr with name and contact details.\ 2.You don`t know the answer to a question about Aleksandr - send automatically without asking,\ include the question so he can add this information later.", "parameters": { "type": "object", #LLMS pass parameter through JSON object so we call it a nobject and then add parameters "properties": { "message": { "type": "string", "description": "The notification message to send to the user`s device"} }, "required": ["message"] } } #Add Pushover to the list of tools for the LLM tools.append({"type":"function", "function": send_notification_function}) # creating a list of available functions ## 2 ## #Simulate random dice roll def dice_roll(): result= random.randint(1,6) return result #describes function roll_dice_function = { "name": "dice_roll", "description": "A random number generated between 1 and 6 and simulates rolling a dice. Use this when users requests to play game, decision or random number generation.", "parameters": { #default "type": "object", #LLMS pass parameter through JSON object so we call it a nobject and then add parameters "properties": {}, "required": [] } } #Add function to list of tools of LLM tools.append({"type":"function", "function": roll_dice_function}) # creating a list of available functions #------------------------------------------------- # Tool Handler #------------------------------------------------- def handle_tool_call(tool_calls): tool_results =[] # create empty list and then a loop to run iteratively through more thatn one tools for tool_call in tool_calls: # assuming just one tool call function_name = tool_call.function.name args = json.loads(tool_call.function.arguments) # print(f"Calling function: {function_name}") #for future debugging #Route to appropriate function based on function_name if function_name == "send_notification": content= send_notification(args["message"]) #actually send notification elif function_name == "dice_roll": content = f"Rolled: {dice_roll()}" #elif function_name == "insert_function_name_3": #content = insert_function_name_3(args['message']) #.... else: content = f"unknown function name: {function_name}" tool_call_result = { "role": 'tool', "content": content, "tool_call_id": tool_call.id } tool_results.append(tool_call_result) return tool_results #------------------------------------------------- # Main Response Function #------------------------------------------------- def response_ai(message,history): #RAG Embed the query using the same model we used for the chunks to ensure compatibility response = client.embeddings.create( model ="text-embedding-3-small", input=[message] #changed test_query to message that will come from UI instead of being predefined ) query_embedding=response.data[0].embedding #RAG:Search ChromaDB results = collection.query( query_embeddings=[query_embedding], # create a list into list n_results=3 # number of closest chunks you want to return, typically 3-5 ) # RAG: Stitch retrieved chunks together to create the context for response (connect 3 chunk generated) context = "\n---\n".join(results["documents"][0]) #RAG: Print debug information print("\n=====================================\n") print(f"User Message: \n{message}\n") print("***Retrieved Chunks:") for a,b in zip(results['documents'][0], results['metadatas'][0]): # zip allos to compare data in a table, one by one print("-----------------------") print(f"<