app.py

import os
from openai import OpenAI
import gradio as gr
import uuid
import chromadb
from pprint import pprint
import json
import requests
import random


#--------------------------------------------------------
#setup
#-------------------------------------------------------
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if OPENAI_API_KEY is None:
    raise Exception("OPENAI_API_KEY environment variable is not set. Please set it in your .env file.")
client = OpenAI()

#--------------------------------------------------------
#Document
#-------------------------------------------------------
#document_overview
document_overview = """
==============================
WHO I AM
==============================

My name is Michael Ma.

I live in Vancouver, British Columbia, Canada.

I work in cardiovascular healthcare with a focus on cardiac catheterization laboratories, structural heart procedures, electrophysiology, and health system operations.

My background combines:

• Registered Nurse
• Clinical Support Coordinator
• Healthcare administration
• Data analytics
• Process improvement
• Digital transformation
• Health policy
• Financial analysis

I am currently completing a Master of Health Administration (MHA).

I enjoy working at the intersection of healthcare, technology, AI, automation, finance, and leadership.

==============================
MY MISSION
==============================

I enjoy solving difficult operational problems.

My goal is to improve healthcare systems so they become:

• safer
• more efficient
• more sustainable
• more patient-centered

I believe small workflow improvements, when multiplied across an entire health system, create enormous value.

I care deeply about responsible stewardship of taxpayer dollars while improving patient outcomes.

==============================
AREAS OF EXPERTISE
==============================

I have extensive knowledge in:

• Cardiac Catheterization Labs
• Structural Heart
• Coronary Intervention
• Pacemakers
• ICDs
• Cardiac Rhythm Devices
• TAVI
• TEER
• Left Atrial Appendage Closure
• Cath Lab equipment
• Medical devices
• Clinical workflows
• Hospital operations
• Healthcare policy
• Quality improvement
• Supply chain
• Business cases
• Microsoft 365
• Excel
• Power Automate
• Power Apps
• AI in healthcare
• Data visualization
• Workflow automation

I enjoy connecting clinical practice with operational improvement.

==============================
HOW I THINK
==============================

When solving problems I usually ask:

• What is the root cause?
• Can this process be simplified?
• Can technology remove manual work?
• Does this improve patient care?
• Does this improve staff experience?
• Is this financially sustainable?
• Can this scale across multiple hospitals?

I prefer systems thinking instead of isolated fixes.

I naturally look for automation opportunities before hiring additional people.

==============================
INVESTING PHILOSOPHY
==============================

I am a long-term investor.

I prefer companies that have:

• durable competitive advantages
• strong cash flow
• visionary leadership
• long growth runways
• AI exposure
• software or healthcare innovation

I am willing to tolerate short-term volatility if long-term fundamentals remain intact.

I enjoy studying macroeconomic trends, disruptive technologies, and secular growth.

==============================
TECHNOLOGY INTERESTS
==============================

I enjoy learning about:

• Artificial Intelligence
• Large Language Models
• Robotics
• Healthcare AI
• Automation
• Local AI
• Microsoft ecosystem
• Productivity systems
• Emerging technologies

I like understanding not only how technology works but how it can create practical value.

==============================
LEARNING STYLE
==============================

I learn by asking lots of questions.

I prefer:

• diagrams
• algorithms
• flowcharts
• practical examples
• first principles
• decision trees

I like turning complex ideas into clear frameworks.

==============================
LEADERSHIP STYLE
==============================

I believe good leaders:

• remove obstacles
• empower people
• improve systems
• use evidence
• make data-informed decisions
• remain humble
• continuously learn

I try to balance operational efficiency with compassion.

==============================
PERSONAL VALUES
==============================

My Christian faith influences how I approach leadership and life.

I value:

• integrity
• humility
• stewardship
• lifelong learning
• service
• excellence
• generosity

I believe knowledge should ultimately be used to help others.

==============================
COMMUNICATION STYLE
==============================

Be:

• practical
• analytical
• curious
• encouraging
• respectful
• evidence-based

Avoid unnecessary jargon.

When explaining something:

1. Start with the big picture.
2. Explain the reasoning.
3. Discuss trade-offs.
4. Give practical recommendations.
5. Mention risks or limitations.

Never exaggerate confidence.

Clearly distinguish facts from opinions.

==============================
WHEN GIVING ADVICE
==============================

When making recommendations:

• Explain why.
• Compare alternatives.
• Discuss pros and cons.
• Consider cost-effectiveness.
• Consider long-term impact.
• Consider operational feasibility.
• Consider implementation challenges.

Whenever possible, think like both a clinician and an administrator.

==============================
OVERALL PERSONALITY
==============================

I am naturally curious.

I enjoy connecting ideas across different disciplines.

I like solving real-world problems more than debating theory.

I believe continuous improvement never stops.

My goal is to leave systems better than I found them.

"""
#--------------------------------------------------------
#document_professional_experience
document_professional_experience = """
Michael Shek Foon Ma

Clinical Systems Support Coordinator RN | Cardiac Cath Lab | Healthcare Operations, Clinical Informatics & Process Improvement | MHA Candidate


Experience


Vancouver Coastal Health logo
Vancouver Coastal Health

Permanent Full-time · 9 yrs 11 mos

Cardiac Cath Lab Clinicals Support Systems Coordinator

Jun 2023 - Present · 3 yrs 2 mos

Vancouver general hospital · On-site

Skills: Medication Administration, Supply Chain Management, +13 skills

Cardiovascular Triage Coordinator

Apr 2022 - Jun 2023 · 1 yr 3 mos

Vancouver, British Columbia, Canada

Skills: Medication Administration, Operations Management, +17 skills

Cardiac Catheterization Laboratory Nurse

Jun 2017 - Apr 2022 · 4 yrs 11 mos

Vancouver, British Columbia, Canada

Skills: Medication Administration, Skilled Multi-tasker, +11 skills

Cardiac Care Unit Nurse

Sep 2016 - Jun 2017 · 10 mos

Vancouver, British Columbia, Canada

Skills: Skilled Multi-tasker, Microsoft Excel, +9 skills

Footcare Nurse

FootCare Mike · Self-employed

Aug 2018 - May 2026 · 7 yrs 10 mos

Vancouver, British Columbia, Canada

Member Directory / Search

MIchael Ma (Vancouver BC, CA). Footcare Service in Vancouver! Fluent in English and Cantonese

 Skilled Multi-tasker, Microsoft Excel and +9 skills

St. Michael's Hospital logo
St. Michael's Hospital

Permanent Full-time · 5 yrs 9 mos

Toronto, Ontario, Canada

Cardiac Care Unit Nurse

May 2014 - Aug 2016 · 2 yrs 4 mos

Skills: Skilled Multi-tasker, Microsoft Excel, +9 skills

Registered Nurse

Dec 2010 - May 2014 · 3 yrs 6 mos

Skills: Skilled Multi-tasker, Microsoft Excel, +8 skills

Profile language

English

正體中文

Who your viewers also viewed
Private to you


Nurse at Southlake Health


View

Nurse at Trillium Health Centre


View

Nurse at Mackenzie Health


View

Nurse at Michael Garron Hospital


View
About

Accessibility

Talent Solutions

Community Guidelines

Careers

Marketing Solutions 

Privacy & Terms

Ad Choices

Advertising

Sales Solutions

Mobile

Small Business

Safety Center

LinkedIn Corporation © 2026

Questions?

Visit our Help Center.

Manage your account and privacy

Go to your Settings.

Recommendation transparency

Learn more about Recommended Content.

Select language


English (English)
Michael Shek Foon MaStatus is online
MessagingYou are on the messaging overlay. Press enter to open the list of conversations.

Compose message
You are on the messaging overlay. Press enter to open the list of conversations.
"""
#--------------------------------------------------------

#--------------------------------------------------------
#Chunking Function
#-------------------------------------------------------
def split_text_into_chunks(
    text: str,
    chunk_size: int = 500,
    overlap: int = 50
) -> list[str]:
    boundaries = ["\n\n", "\n", ". ", "? ", "! ", ", ", " "]

    if chunk_size <= 0:
        raise ValueError("chunk_size must be greater than 0")

    if overlap < 0 or overlap >= chunk_size:
        raise ValueError("overlap must be between 0 and chunk_size - 1")

    def find_natural_boundary(start: int, end: int) -> int:
        midpoint = start + (chunk_size // 2)

        for boundary in boundaries:
            position = text.rfind(boundary, midpoint, end)

            if position != -1:
                return position + len(boundary)

        return end
    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 = end - overlap
    return chunks

#--------------------------------------------------------
#RAG: Chunk, Embed & Store in ChromaDB
#-------------------------------------------------------
documents = [
    {"text": document_overview, "source": "document_overview"},
    {"text": document_professional_experience, "source": "document_professional_experience"},
    
]
chunks = []
ids = []
metadatas = []

for doc in documents:
    #Prepare the lists 
    chunks_ = split_text_into_chunks(doc["text"], chunk_size = 300, overlap = 30)
    ids_ = [str(uuid.uuid4()) for _ in range(len(chunks_))]
    metadatas_ = [{"source": doc["source"], "chunk_index": i} for i in range(len(chunks_))]

    #Add to main lists
    chunks.extend(chunks_)
    ids.extend(ids_)
    metadatas.extend(metadatas_)

print(f"Created {len(chunks)} chunks:\n")

for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1} (ID: {ids[i]}, Source: {metadatas[i]['source']}, Index: {metadatas[i]['chunk_index']}, Length: {len(chunk)}:")
    print(chunk)
    print()

#Generate Embeddings for all chunks
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")

#import chromaDB

#initialize  ChrommabDB client (persistent sorage)
#Alternative: initialize ChromaDB client (in memory storage)
#chroma_client = chromadb.Client()
chroma_client = chromadb.PersistentClient(path="./chroma_db_twin")

collection = chroma_client.get_or_create_collection(name="digital_twin")

#Create + empty the collection before adding new data 
if collection.get()["ids"]:
    collection.delete(collection.get()["ids"])
pprint(collection.get())

#Adding data to ChromaDB
collection.add(
    ids=ids,
    embeddings=embeddings,
    documents=chunks,
    metadatas=metadatas
)
pprint(collection.get())
#--------------------------------------------------------
#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 or missing credentials
        return "Notification failed:  Pushover credentials 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 = {
    "name": "send_notification",
    "description": "Sends a push notification to the real Michael Ma.  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 Michasel with the name and contact deatils.  \
        2)  You don't know the answer to a question about Michael - send automatically without asking, include the question so he can add this info later.",
    "parameters": {
        "type": "object",
        "properties": {
            "message" : {"type": "string","description": "The notification message to be sent to the user's phone."}
        },
        "required": ["message"]
    }
}

#Add Pushover to the list of tools for the LLM
tools.append({"type": "function", "function": send_notification_function})

#simulates rolling a single six-sided die
def dice_roll():
    result = random.randint(1,6)
    return result

#Describe function for the LLM
roll_dice_function = {
        "name": "dice_roll",
    "description": "Simulates rooling a single six-sided die and returns the result.  Use this when the user wants to roll a die for games, decisions, or random number generation.",
    "parameters": {
        "type": "object",
        "properties": {},
        "required": []
                  }       
                    }

#add function to list of tools of LLM
tools.append({"type":"function", "function":roll_dice_function})

#--------------------------------------------------------
#Tool handler
#-------------------------------------------------------
def handle_tool_call(tool_calls):
  tool_results = []

  for tool_call in tool_calls:
      function_name = tool_call.function.name
      args = json.loads(tool_call.function.arguments)
    #print(f"Calling function{function_name}")  #for future debugging
    # actually send the notification ie.e call the tool
    
    #Route to appropriate function
      if function_name == "send_notification":
           #Actually send the notification
          content = send_notification(args["message"])
      elif function_name == "dice_roll":
          content = f"Rolled: {dice_roll()}"
      #elif function_name == "insert_function_name3":
      #   content = insert_function_name_3(args["message"])
      
      else:
         content = f"Unknown function: {function_name}"
         
      tool_call_result = {
      "role":"tool",
      "content": content,
      "tool_call_id": tool_call.id
    }

      tool_results.append(tool_call_result)

  return tool_results

    


#--------------------------------------------------------
#System Message
#-------------------------------------------------------
system_message = """
You are a digital twin of Michael Ma.
When people talk to you, you respond AS Michael—in first person—using my reasoning style, communication style, professional experience, and values.

IMPORTANT:  do not make things up.  If you don't know an answer, sya you don't know.  
The only factual information available to you is what's in this system message.
You cannot get any more factts about Michael from the internet or make them up.

IMPORTANT:  Whenever you don't know something about Kirill,
ALWAYS use the send_notification tool to alert the real Michael - do this automatically without asking the user.
"""
#--------------------------------------------------------
#Main response function
#-------------------------------------------------------
def respond_ai(message, history):
   
 #RAG
    #Embed the query ussing the same model we used for teh chunks to ensure compatiblity
    response = client.embeddings.create(
    model = "text-embedding-3-small",
    input = [message]
    )
    query_embedding = response.data[0].embedding
    
    #RAGSearch ChromaDB
    results = collection.query(
    query_embeddings=query_embedding,
    n_results=3,
    )
    #RAG: #stitch retrived chunks together to create the context for the response.
    context = "\n---\n".join(results["documents"][0])
    #Logs print for debugging
    print("\n===========================\n")
    print(f"User message:\n{message}\n")
    print("***Retrienved Chunks:")
    for a, b in zip(results["documents"][0], results["metadatas"][0]):
        print("---------------------")
        print(f"Document {b['source']} -- Chunk {b['chunk_index']} (Chunk content):\n{a}\n")
    
    #Update the system message with context (for this conversation turn)
    system_message_enhanced = system_message + "\n\nContext:\n" + context
    #Build message 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
#Check if model wants to call a tool
    while message.tool_calls:
        from pprint import pprint
        pprint (message.tool_calls)

        tool_result = handle_tool_call(message.tool_calls) 
        messages.append(message) 
        messages.extend(tool_result)   # change from append to eextend when we swtich to multiple too call handling

        response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=messages,
        tools=tools 
        )
        message = response.choices[0].message
        #Note:  Maybe consider adding protection from infinite consecutive tool calling

    return(message.content)
    
#--------------------------------------------------------
#Lanunch Gradio
#-------------------------------------------------------
gr.ChatInterface(
    fn=respond_ai,
    title= "Michael's Digital Twin",
    chatbot=gr.Chatbot(avatar_images=(None,"Michael.jpg")),
    description= "Chat with an AI version of Michael Ma.  Ask about his experience, project, or just say hi!",
    examples= ["What's your background?", "AI engineering expereince", "Do you like Pineapple on Pizza?"]
    ).launch()