app.py

import gradio as gr
import smolagents
import json
import pandas as pd
import numpy as np
from huggingface_hub import login, HfApi
from datasets import Dataset, DatasetDict, load_dataset
import difflib
import openai
from langchain_community.utilities.wikipedia import WikipediaAPIWrapper
import os

# Re-define all necessary components that the agent relies on
# This includes data loading, utility functions, and the agent itself

# Setup (copied from qRq0g01h3ZvP)
hf_token_public = os.getenv("token_public")
# login(hf_token_public) # Login is not needed in app.py if HF_TOKEN is set as secret

REPO_ID_TECHSPARK_STAFF = "aslan-ng/CMU_TechSpark_Staff"
REPO_ID_TECHSPARK_COURSES = "aslan-ng/CMU_TechSpark_Courses"
REPO_ID_TECHSPARK_TOOLS = "aslan-ng/CMU_TechSpark_Tools"

SHEET_ID_TECHSPARK = "1cdL_jDglKa-NxZF3j5s2z9ncSFbJSMGC2d-GsKubV-I"

# OPENAI_API = "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" # Use environment variable for API key

OPENAI_API = os.getenv("OPENAI_API")

# Data (copied from rGAiTp0PYvEk, adjusted to load from HF directly)
NUMERIC_PROFILE = ["Laser Cutting",	"Wood Working",	"Wood CNC",	"Metal Machining",	"Metal CNC",	"3D Printer",	"Welding",	"Electronics"]

def load_data_from_huggingface():
    """
    Loads data from HuggingFace.
    """
    # Staff (People)
    ds_staff = load_dataset(REPO_ID_TECHSPARK_STAFF)
    staff_df = ds_staff["train"].to_pandas()

    # Courses
    ds_courses = load_dataset(REPO_ID_TECHSPARK_COURSES)
    courses_df = ds_courses["train"].to_pandas()

    # Tools
    ds_tools = load_dataset(REPO_ID_TECHSPARK_TOOLS)
    tools_df = ds_tools["train"].to_pandas()
    return staff_df, courses_df, tools_df

staff_df, courses_df, tools_df = load_data_from_huggingface()

# LLM (copied from NPPbWry0qUIE)
model = smolagents.OpenAIServerModel(
    model_id="gpt-4.1-mini",
    api_key=OPENAI_API,
)

# General Functions (copied from BwfI-EsvtvVx)
def vector_1st_distance(x: list, y: list):
    """
    Calculate the average 1st distance between two vectors.
    """
    if len(x) != len(y):
      raise ValueError
    return sum(np.array(x) - np.array(y)) / len(x)

def skill_score(
    skill_profile: dict,  # The skill profile that we want to analyze
    laser_cutting: float = None,
    wood_working: float = None,
    wood_cnc: float = None,
    metal_machining: float = None,
    metal_cnc: float = None,
    three_d_printer: float = None,
    welding: float = None,
    electronics: float = None,
):
    """
    Calculate the skill score for a given skill profile. Useful for both staff and courses skill profiles.
    """
    x = []
    y = []
    if laser_cutting is not None:
        x.append(skill_profile['Laser Cutting'])
        y.append(laser_cutting)
    if wood_working is not None:
        x.append(skill_profile['Wood Working'])
        y.append(wood_working)
    if wood_cnc is not None:
        x.append(skill_profile['Wood CNC'])
        y.append(wood_cnc)
    if metal_machining is not None:
        x.append(skill_profile['Metal Machining'])
        y.append(metal_machining)
    if metal_cnc is not None:
        x.append(skill_profile['Metal CNC'])
        y.append(metal_cnc)
    if three_d_printer is not None:
        x.append(skill_profile['3D Printer'])
        y.append(three_d_printer)
    if welding is not None:
        x.append(skill_profile['Welding'])
        y.append(welding)
    if electronics is not None:
        x.append(skill_profile['Electronics'])
        y.append(electronics)
    return vector_1st_distance(x, y)

# Staff Tools (copied from Q47nRn9_Zz1P)
def all_staff():
    """
    Return a list of all staff.
    """
    return staff_df["Name"].dropna().tolist()

def match_staff_name(name: str):
    """
    Match the staff name to the closest match in the staff list.
    """
    matches = difflib.get_close_matches(name, all_staff(), n=1, cutoff=0.2)
    return matches[0] if matches else None

def all_available_staff(exclude: list):
    """
    Return a list of all staff with exclusion.
    """
    try:
      exclude = list(exclude)
    except:
      pass
    if exclude is None or len(exclude) == 0:
        return all_staff()
    excluded_names = []
    for raw_name in exclude:
        excluded_name = match_staff_name(raw_name)
        if excluded_name:
            excluded_names.append(excluded_name)
    return [name for name in all_staff() if name not in excluded_names]

def get_staff_full_profile(name: str):
    """
    Get the staff full profile given its name (including description and skill).
    """
    name = match_staff_name(name)
    if name:
        full_profile = staff_df[staff_df["Name"] == name].iloc[0].to_dict()
        return full_profile
    return None

def get_staff_skills_profile(name: str):
    """
    Get the staff skills profile given its name.
    """
    full_profile = get_staff_full_profile(name)
    return {k: full_profile[k] for k in NUMERIC_PROFILE}

def get_staff_profile(name: str):
    """
    Get the staff profile without skill part.
    """
    full_profile = get_staff_full_profile(name)
    return {k: v for k, v in full_profile.items() if k not in NUMERIC_PROFILE}

def search_staff_by_skills(
    laser_cutting: float = None,
    wood_working: float = None,
    wood_cnc: float = None,
    metal_machining: float = None,
    metal_cnc: float = None,
    three_d_printer: float = None,
    welding: float = None,
    electronics: float = None,
    exclude: list = None,
):
    names = all_available_staff(exclude)
    best_name = None
    best_score = float("inf")
    for name in names:
        skills_profile = get_staff_skills_profile(name)
        score = skill_score(
            skill_profile = skills_profile,
            laser_cutting = laser_cutting,
            wood_working = wood_working,
            wood_cnc = wood_cnc,
            metal_machining = metal_machining,
            metal_cnc = metal_cnc,
            three_d_printer = three_d_printer,
            welding = welding,
            electronics = electronics,
        )
        # keep only positive scores
        if score is not None and score > 0 and score < best_score:
            best_score = score
            best_name = name
    return best_name

class SearchStaffInformation(smolagents.tools.Tool):
    name = "search_staff_information"
    description = (
        "Search the staff information by its name."
    )
    inputs = {
        "name": {"type": "string", "description": "Name of the staff member."},
    }
    output_type = "object"

    def forward(self, name: str) -> str:
        return json.dumps(get_staff_profile(name))

class FindSuitableStaff(smolagents.tools.Tool):
    name = "find_suitable_staff"
    description = (
        "Find the most suitable staff member for the task based on required skills."
    )
    inputs = {
        "laser_cutting": {"type": "number", "nullable": True, "description": "Laser cutting skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "wood_working": {"type": "number", "nullable": True, "description": "Wood working skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "wood_cnc": {"type": "number", "nullable": True, "description": "Wood CNC skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "metal_machining": {"type": "number", "nullable": True, "description": "Metal machining skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "metal_cnc": {"type": "number", "nullable": True, "description": "Metal CNC skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "three_d_printer": {"type": "number", "nullable": True, "description": "3D printer skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "welding": {"type": "number", "nullable": True, "description": "Welding skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "electronics": {"type": "number", "nullable": True, "description": "Electronics skill required for the task. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "exclude": {"type": "number", "nullable": True, "description": "A list of names that we want to exclude from searching. Default is None or an empty list."}
    }
    output_type = "object"

    def forward(self,
        laser_cutting: float = None,
        wood_working: float = None,
        wood_cnc: float = None,
        metal_machining: float = None,
        metal_cnc: float = None,
        three_d_printer: float = None,
        welding: float = None,
        electronics: float = None,
        exclude: list = None,
    ) -> str:
        name = search_staff_by_skills(
            laser_cutting = laser_cutting,
            wood_working = wood_working,
            wood_cnc = wood_cnc,
            metal_machining = metal_machining,
            metal_cnc = metal_cnc,
            three_d_printer = three_d_printer,
            welding = welding,
            electronics = electronics,
            exclude = exclude,
        )
        return json.dumps(get_staff_profile(name))

# Course Functions (copied from _P8TTwcOaUkN)
def all_courses_code():
    """
    Return a list of all course codes.
    """
    return courses_df["Code"].dropna().astype(str).tolist()

def all_courses_name():
    """
    Return a list of all course names.
    """
    return courses_df["Name"].dropna().tolist()

def course_name_to_code(course_name):
    """
    Convert the course name to course code.
    """
    return str(courses_df[courses_df["Name"] == course_name]["Code"].iloc[0])

def course_code_to_name(course_code):
    """
    Convert the course code to course name.
    """
    return str(courses_df[courses_df["Code"].astype(str) == str(course_code)]["Name"].iloc[0])

def match_course_name_code(input):
    """
    Match the course to the closest match in the course list and return their codes.
    """
    input = str(input)
    matches = None
    code_matches = difflib.get_close_matches(input, all_courses_code(), n=3, cutoff=0.2)
    name_matches_code = difflib.get_close_matches(input, all_courses_name(), n=2, cutoff=0.3)
    if name_matches_code:
        name_matches = [course_name_to_code(name) for name in name_matches_code]
    else:
        name_matches = None
    if code_matches and name_matches:
        matches = code_matches + name_matches
    elif code_matches and not name_matches:
        matches = code_matches
    elif name_matches and not code_matches:
        matches = name_matches
    return matches

def get_course_full_profile(course):
    """
    Get the course full profile given its code (including description and skill).
    """
    # Ensure the input code is a string for comparison
    matches = match_course_name_code(course)
    code = matches[0] if matches else None
    if code:
        full_profile = courses_df[courses_df["Code"].astype(str) == code].iloc[0].to_dict()
        return full_profile
    return None

def get_course_skills_profile(course_code):
    """
    Get the course skills profile given its code.
    """
    full_profile = get_course_full_profile(course_code)
    return {k: full_profile[k] for k in NUMERIC_PROFILE}

def get_course_profile(course_code):
    """
    Get the course profile without skill part.
    """
    full_profile = get_course_full_profile(course_code)
    return {k: v for k, v in full_profile.items() if k not in NUMERIC_PROFILE}

def search_course_by_skills(
    laser_cutting: float = None,
    wood_working: float = None,
    wood_cnc: float = None,
    metal_machining: float = None,
    metal_cnc: float = None,
    three_d_printer: float = None,
    welding: float = None,
    electronics: float = None,
    n_results: int = 1,
):
    names = all_courses_code()
    scored_courses = []

    for name in names:
        skills_profile = get_course_skills_profile(name)

        score = skill_score(
            skill_profile=skills_profile,
            laser_cutting=laser_cutting,
            wood_working=wood_working,
            wood_cnc=wood_cnc,
            metal_machining=metal_machining,
            metal_cnc=metal_cnc,
            three_d_printer=three_d_printer,
            welding=welding,
            electronics=electronics,
        )

        if score is not None:
            scored_courses.append((abs(score), name))
            # store (absolute_score, course_name)

    # Sort by closeness to zero
    scored_courses.sort(key=lambda x: x[0])

    # Return only the names of top N matches
    return [name for _, name in scored_courses[:n_results]]

class SearchCourseInformation(smolagents.tools.Tool):
    name = "search_course_information"
    description = (
        "Search the course information by the course name or course number (code)."
    )
    inputs = {
        "name": {"type": "string", "description": "Course name or course number (code)."},
    }
    output_type = "object"

    def forward(self, name: str) -> str:
        return json.dumps(get_course_profile(name))

class FindSuitableCourses(smolagents.tools.Tool):
    name = "find_suitable_courses"
    description = (
        "Find the top 3 most suitable courses for the task based on required skills. The first element is the best match."
    )
    inputs = {
        "laser_cutting": {"type": "number", "nullable": True, "description": "Laser cutting skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "wood_working": {"type": "number", "nullable": True, "description": "Wood working skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "wood_cnc": {"type": "number", "nullable": True, "description": "Wood CNC skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "metal_machining": {"type": "number", "nullable": True, "description": "Metal machining skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "metal_cnc": {"type": "number", "nullable": True, "description": "Metal CNC skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "three_d_printer": {"type": "number", "nullable": True, "description": "3D printer skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "welding": {"type": "number", "nullable": True, "description": "Welding skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
        "electronics": {"type": "number", "nullable": True, "description": "Electronics skill being taught during the course. It is a number between 0 (no expertise required) to 3 (high expertise expertise). Default is None. If left None, it will be ignored. (Optional)"},
    }
    output_type = "object"

    def forward(self,
        laser_cutting: float = None,
        wood_working: float = None,
        wood_cnc: float = None,
        metal_machining: float = None,
        metal_cnc: float = None,
        three_d_printer: float = None,
        welding: float = None,
        electronics: float = None,
    ) -> str:
        matches = search_course_by_skills(
            laser_cutting = laser_cutting,
            wood_working = wood_working,
            wood_cnc = wood_cnc,
            metal_machining = metal_machining,
            metal_cnc = metal_cnc,
            three_d_printer = three_d_printer,
            welding = welding,
            electronics = electronics,
            n_results = 3,
        )
        options = [get_course_profile(course) for course in matches]
        return json.dumps(options)

# Machine Functions (copied from OKKlHB88tt1r)
def all_tools():
    """
    Return a list of all tools and machines.
    """
    return tools_df["Name"].dropna().astype(str).tolist()

def match_tool_name(input):
    """
    Match the course to the closest match in the course list and return their codes.
    """
    input = str(input)
    matches = difflib.get_close_matches(input, all_tools(), n=1, cutoff=0.2)
    return matches[0] if matches else None

def get_tool_location(name: str):
    """
    Get the tool location given its name.
    """
    tool_name = match_tool_name(name)
    if tool_name is not None:
      return tools_df[tools_df["Name"] == tool_name].iloc[0]["Location"]
    else:
      raise ValueError("Not found.")

def is_tool_accessible(name):
    """
    Check if the machine is accessible to students, and if they require taking mandatory courses.
    """
    result = None
    tool_name = match_tool_name(name)
    if tool_name is not None:
        accessible = tools_df[tools_df["Name"] == tool_name].iloc[0]["Accessible by Students"]
        accessible = bool(accessible)
        course_code = tools_df[tools_df["Name"] == tool_name].iloc[0]["Required Course"]
    else:
        raise ValueError("Not found.")

    if accessible:
        if course_code:
            # Accessible but conditional (only by passing the course)
            result_short = "Conditional"
            result_description = f"Student can access it only if they take the {course_code}: {course_code_to_name(course_code)}."
        else:
            # Accessible
            result_short = "Yes"
            result_description = "Student can access it."
    else:
        # Not accessible by students. Need staff members!
        result_short = "No"
        result_description = "Student cannot access it. Only available to staff memebers. Ask them to do your task for you."
    result = {
        "short answer": result_short,
        "description": result_description
    }
    return json.dumps(result)

class SearchMachineLocation(smolagents.tools.Tool):
    name = "search_machine_location"
    description = (
        "Search the machine or tool location in the TechSpark."
    )
    inputs = {
        "name": {"type": "string", "description": "Tool or machine name."},
    }
    output_type = "object"

    def forward(self, name: str) -> str:
        return json.dumps(get_tool_location(name))

class CheckMachineAccessibility(smolagents.tools.Tool):
    name = "check_machine_accessibility"
    description = (
        "Check whether machine or tool is accessible to students. Some are accessible, some need to take a course to become accessible, and some are only available to staff members."
    )
    inputs = {
        "name": {"type": "string", "description": "Tool or machine name."},
    }
    output_type = "object"

    def forward(self, name: str) -> str:
        return json.dumps(is_tool_accessible(name))

# Wikipedia Search (copied from 6AHceBzBXISE)
class WikipediaSearch(smolagents.Tool):
    """
    Create tool for searching Wikipedia
    """
    name = "wikipedia_search"
    description = "Search Wikipedia, the free encyclopedia."
    inputs = {
        "query": {"type": "string", "nullable": True, "description": "The search terms"},
    }
    output_type = "string"

    def forward(self, query: str | None = None) -> str:
        if not query:
            return "Error: 'query' is required."
        wikipedia_api = WikipediaAPIWrapper(top_k_results=1)
        answer = wikipedia_api.run(query)
        return answer

# Agent (copied from 9iwR_e424jfJ)
techspark_definition = """
TechSpark is the largest makerspace at CMU (Carnegie Mellon University), located in the College of Engineering.  
Its mission is to promote a vibrant, student-centric making culture to enhance educational, extracurricular, and research activities across the entire campus community.
"""

instruction = """
You are a helpful assistant for the CMU TechSpark facility. Your purpose is to assist users with inquiries related to staff, courses, and tools.
Use the available tools to find information about staff members, suggest suitable staff based on skills, or provide training information for machines.
Respond concisely and directly with the information requested by the user, utilizing the output from the tools.
Which machines to use for a task, and where to find them.
When you were in doubt, try searching wikipedia to gain more knowledge.

Safety is important. So:
- When talking about any machines, check whether it is accessbile to students or not.
- Try to match them to correct staff member specially when you are not sure about your answer or the student work might be dangerous.
"""

system_prompt = f"""
{techspark_definition}
{instruction}
"""

agent = smolagents.CodeAgent(
    tools=[
        smolagents.FinalAnswerTool(),
        SearchStaffInformation(),
        FindSuitableStaff(),
        SearchCourseInformation(),
        FindSuitableCourses(),
        SearchMachineLocation(),
        CheckMachineAccessibility(),
        WikipediaSearch(),
    ],
    instructions=system_prompt,
    model=model,
    add_base_tools=False,
    max_steps=10,
    verbosity_level=0, # Changed to 0 for deployment
)


# UI (copied from w0g2EzpD7fUy, adjusted for app.py)
# Minimal Gradio chat


with gr.Blocks(theme=gr.themes.Ocean()) as demo:
    
    # Centered title and description using HTML
    gr.HTML("""
            <div style="text-align: center; font-family: 'Arial', sans-serif;">
                <h1 style="color:#1f77b4; margin-bottom: 20px; font-weight: 300;">
                    🤖 TechSpark AI Assistant
                </h1>
                
                <p style="margin-top: 0; font-weight: 300; font-size: 16px; color:#555;">
                    Welcome to the TechSpark AI Assistant!<br>
                    Ask anything about TechSpark staff, tools, courses or location of tools.<br>
                    This assistant is powered by OpenAI's GPT model via smolagents, 
                    accessing accurate information from our curated dataset verified by TechSpark staff!
                </p>
            </div>
            """)

    chat = gr.Chatbot(height=420)
    inp = gr.Textbox(placeholder="Ask your question in natural language.", label="Your question")

    # No gr.State for agent — just close over `agent`
    def respond(message, history):
        try:
            # 1. Use agent.chat() to maintain internal history
            out = str(agent.run(message))
        except Exception as e:
            out = f"[Error] {e}"

        # This just updates the Gradio UI history
        history = (history or []) + [(message, out)]
        return "", history

    gr.Examples(
        examples=[
            "Who is Ed?",
            "Who to talk to to create a wooden table?",
            "how to access laser cutter"
        ],
        inputs=[inp],
        outputs=[inp, chat],
        fn=respond,
        cache_examples=False, # Set to False for dynamic content or to avoid caching issues
    )

    inp.submit(respond, [inp, chat], [inp, chat])

demo.launch(share=True)