app.py

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"<<Document {b['source']} --Chunk{b['chunk_index']} content):\n{a}\n")
    
    #Update system message context (for this conversation turn)
    system_message_enhanced = system_message + "\n\n Context:\n" + context

    #Build messages for this turn
    messages= [{"role": "system", "content": system_message_enhanced}]+history+ [{"role": "user", "content": message}]
    
    #Call LLM
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=messages,
        tools=tools
    )

    message = response.choices[0].message
 # the information if we want to use a tool is stored in message
    while message.tool_calls:
        from pprint import pprint # for debugging
        pprint (message.tool_calls) # for debugging
        tool_result = handle_tool_call(message.tool_calls) # whole list of tool calls
        messages.append(message)
        messages.extend(tool_result) #python error- need to know when to append and when to extend, changed this for multiple tool calls

        response = client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=messages,
            tools=tools   # will add in the future
        )
        message= response.choices[0].message
        #maybe add additional protection against consecutive tool calling
 
    return(message.content)   #here changed print to return


#-------------------------------------------------
# Launch Gradio
#-------------------------------------------------

gr.ChatInterface(
    fn=response_ai,
    title="Aleksandr`s Digital Twin",
    chatbot=gr.Chatbot(avatar_images=(None,"Profile_Aleksandr.jpg")),
    description="Chat with AI Version of Aleksandr Kuternin. Ask about his experience, projects or just say hi.",
    examples= ["What`s your background?", "AI Engineering Projects","Automation Experience"]
).launch()