app.py

import os
from openai import OpenAI
import gradio as gr
import chromadb
import uuid
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 is missing from environment variables")

client = OpenAI(api_key=OPENAI_API_KEY)

#----------
# Document Overview (full knowledge about Dave)
#----------
document_overview = """Here's information about Dave to help you embody him:
Dave runs Digital6 Technologies, an AI consulting and cloud solutions firm based in Boise, Idaho. He's an AWS Certified Solutions Architect and CMMC Registered Practitioner with 15+ years of experience helping businesses navigate cloud architecture, AI implementation, and compliance requirements.
He has an MBA from Northwest Nazarene University and a BS in Mechanical Engineering from Kansas State. His technical expertise spans AWS, Azure, AI technologies (Bedrock, LangChain, voice AI), and compliance frameworks (HIPAA, CMMC, CJIS, SOC 2, PCI DSS).
Career highlights:

2009-Present: Cloud Security and Solutions Architect at Digital6 Technologies
Built HIPAA-compliant AI solutions using AWS Textract, Comprehend Medical, and Rekognition
Developed voice AI agents for healthcare appointment scheduling and customer service
Led cloud migrations achieving up to 25% cost savings
Created chatbots using LangChain and Python integrated with business databases

What drives him: Dave believes technology should solve real business problems, not create complexity. He's passionate about making advanced tech like AI and cloud computing accessible and practical for regulated industries. He helps businesses balance innovation with compliance and cost control - the three critical priorities that often seem at odds.
His approach: Practical and results-focused. Dave translates technical complexity into business value. He's not interested in theoretical solutions - he builds systems that work in the real world, meet regulatory requirements, and deliver measurable ROI. He reduces AWS AI infrastructure costs by up to 37% while strengthening compliance controls.
Communication style: Direct, consultative, and personable with a dry sense of humor. Dave explains complex technical concepts in business terms without the typical consultant fluff. He's experienced enough to be confident but humble enough to stay current in a fast-moving field. He'll make the occasional wry observation about cloud bills that mysteriously grow overnight or the creative interpretations of "HIPAA-compliant" he's encountered. His humor is understated and often delivered deadpan - he might joke about compliance frameworks being "as exciting as they sound" or mention that yes, he has read all 274 pages of NIST 800-53, and no, he doesn't recommend it as bedtime reading. He asks good questions to understand client needs before proposing solutions, occasionally noting the gap between what vendors promise and what actually happens in production."""

document_education = """
Dave holds an MBA from Northwest Nazarene University and a BS in Mechanical Engineering from Kansas State University. He's also an AWS Certified Solutions Architect and CMMC Registered Practitioner, demonstrating his commitment to staying current with cloud and security technologies. His educational background provides a unique blend of technical depth and business acumen that informs his consulting approach."""

document_professional_experience = """Dave's professional journey began in 2009 as a Cloud Security and Solutions Architect at Digital6 Technologies, where he's spent over 15 years building cloud-native solutions for regulated industries. His career has focused on helping businesses leverage AWS and AI technologies while maintaining compliance with HIPAA, CMMC, CJIS, SOC 2, and PCI DSS standards.

Key accomplishments include:
- Building HIPAA-compliant AI solutions using AWS Textract, Comprehend Medical, and Rekognition
- Developing voice AI agents for healthcare appointment scheduling and customer service
- Leading cloud migrations achieving up to 25% cost savings
- Creating chatbots using LangChain and Python integrated with business databases
His approach combines deep technical expertise with practical business insight, translating complex technical concepts into actionable strategies that drive measurable ROI while meeting regulatory requirements.
 """


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

    def find_natural_boundary(start: int, end: int) -> int:
        for boundary in BOUNDARIES:
            pos = text.rfind(boundary, start, end)
            if pos != -1:
                return pos + 1
        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)
        chunk = text[start:end].strip()
        if chunk:
            chunks.append(chunk)
        if end >= len(text):
            break
        start = max(start + 1, end - overlap)

    return chunks


#----------
# RAG: Chunk, Embed and Store in ChromaDB
#----------
documents = [
    {"text": document_overview, "source": "Overview"},
    {"text": document_education, "source": "Education"},
    {"text": document_professional_experience, "source": "Professional Experience"}
]

chunks = []
ids = []
metadatas = []

for doc in documents:
    chunks_ = split_text_into_chunks(doc["text"], chunk_size=250, overlap=25)
    ids_ = [str(uuid.uuid4()) for _ in range(len(chunks_))]
    metadatas_ = [{"source": doc["source"], "chunk_index": i} for i in range(len(chunks_))]

    chunks.extend(chunks_)
    ids.extend(ids_)
    metadatas.extend(metadatas_)

# Print for logs
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']}, Chunk Index: {metadatas[i]['chunk_index']}:")
    print(chunk)
    print()

# Generate embeddings for each chunk
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=chunks
)
embeddings = [item.embedding for item in response.data]

# Verify embeddings
print(f"Generated {len(embeddings)} embeddings")
print(f"Each embedding has {len(embeddings[0])} dimensions")

# Initialize ChromaDB collection (in-memory for HF Spaces)
chroma_client = chromadb.Client()

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

# Adding data to ChromaDB
collection.add(
    ids=ids,
    embeddings=embeddings,
    metadatas=metadatas,
    documents=chunks
)

# Show everything (or just what you need)
result = collection.get(include=["documents", "metadatas", "embeddings"])
pprint(result)

#----------
# 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 not pushover_user or not pushover_token:
        return "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 = {
  "name": "send_notification",
  "description": "Sends a push notification to the real world verion of you via Pushover. 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 real dave with this information.\
          2) you dont know the answer - send automatically without asking, include the question so he can add this information later. Tell user good question and I will send that answer to you later.",
  "parameters": {
    "type": "object",
    "properties": {
      "message": {
        "type": "string",
        "description": "The notification message to send to the user's device"
      }
    },
    "required": ["message"]
  }
}

#Add Pushover to tools
tools.append({"type":"function", "function":send_notification_function})


#Simulates DIce rolling tool
def dice_roll():
    result = random.randint(1,6)
    return result

#Describe function for the LLM
roll_dice_function = {
    "name": "dice_roll",
    "description": "Simulates rolling a 6-sided die and returns the result. Use this when the user wants to roll a die or to give the user a random number",
    "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)
               
        #Route to the approperiate function based on function_name
        if function_name == "send_notification":
            content = f"Notification sent: {args['message']}"
        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: {function_name}"
        
        
        tool_call_result = {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": content
        }
        
        tool_results.append(tool_call_result)
        
    return tool_results




#----------
# System Message
#----------
system_message = """You are a digital twin of David Fraas. When people talk to you, you respond AS Dave - in first person, using his voice, personality, and knowledge.
CRITICAL INSTRUCTIONS:
- Only share information you actually know about Dave from the context provided
- If you don't know something about Dave's experience, projects, or opinions, say so honestly
- Never invent projects, clients, certifications, or experience that aren't documented
- When unsure, respond like Dave would in a job interview: acknowledge what you do know, admit what you don't, and offer to find out or clarify
- It's better to say "I don't have that specific information" than to guess
- Important: Whenever you don't know something about Dave, Always use the send_notification tool to alert the real Dave - do this automatically without asking user."""


#----------
# Main Response Function
#----------
def respond_ai(message, history):
    if not message:
        return ""

    # RAG: Embed the query
    query_response = client.embeddings.create(
        input=[message],
        model="text-embedding-3-small"
    )
    query_embedding = query_response.data[0].embedding

    # RAG: Search ChromaDB
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=3,
    )

    # RAG: Stitch context together
    context = "\n---\n".join(results["documents"][0])

    # Print logs for debugging
    print("\n==============================\n")
    print(f"User message:\n{message}\n")
    print("Retrieved Chunks:")
    for a, b in zip(results["documents"][0], results["metadatas"][0]):
        print("-----------------")
        print(f"- Source: {b['source']} -- Chunk {b['chunk_index']}>>\n{a}\n")

    # Update system message with context
    system_message_enhanced = system_message + "\n\nContext:\n" + context

    # Build messages list from history
    messages = [{"role": "system", "content": system_message_enhanced}]
    for h in history:
        messages.append({"role": "user", "content": h["content"] if isinstance(h, dict) else h[0]})
        # Only add assistant message if it exists
        if isinstance(h, dict):
            pass  # Gradio type="messages" handled below
        elif h[1]:
            messages.append({"role": "assistant", "content": h[1]})

    # Handle both Gradio history formats
    if history and isinstance(history[0], dict):
        messages = [{"role": "system", "content": system_message_enhanced}]
        for h in history:
            messages.append({"role": h["role"], "content": h["content"]})

    messages.append({"role": "user", "content": message})

   # Call LLM
    llm_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools
    )

    assistant_message = llm_response.choices[0].message

    # Check if model wants to call a tool
    while assistant_message.tool_calls:
        pprint(assistant_message.tool_calls)
        tool_result = handle_tool_call(assistant_message.tool_calls)
        messages.append(assistant_message)
        messages.extend(tool_result)

        llm_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=tools
        )
        assistant_message = llm_response.choices[0].message

    return assistant_message.content


#----------
# Launch Gradio
#----------
gr.ChatInterface(
    fn=respond_ai,
    title="Dave Fraas - Digital Twin",
    chatbot=gr.Chatbot(avatar_images=(None, "david_fraas-rbk.png")),
    description="Chat with Digital Dave. Ask me anything about my experience, services, or career!",
    examples=["Tell me about your experience", "What services do you offer?", "How can I contact you?"], 
).launch()