app.py

import os
import gradio as gr
from huggingface_hub import InferenceClient

# STEP 1 — semantic search imports
from sentence_transformers import SentenceTransformer
import torch

# ---------------------------------------------------------------------------
# AUTH — uses the HF_TOKEN secret set in Space Settings (no visitor login).
# ---------------------------------------------------------------------------
HF_TOKEN = os.environ["HF_TOKEN"]
client = InferenceClient(token=HF_TOKEN, model="openai/gpt-oss-20b")

# STEP 2 — read each knowledge file
with open("programs.txt", "r", encoding="utf-8") as file:
    programs_text = file.read()
with open("faqs.txt", "r", encoding="utf-8") as file:
    faqs_text = file.read()
with open("resources.txt", "r", encoding="utf-8") as file:
    resources_text = file.read()


# STEP 3 — split each file into clean chunks on the *** separator
def preprocess_text(text):
    cleaned_text = text.strip()
    chunks = cleaned_text.split("***")
    cleaned_chunks = []
    for chunk in chunks:
        chunk = chunk.strip()
        if chunk != "":
            cleaned_chunks.append(chunk)
    return cleaned_chunks


cleaned_chunks_programs = preprocess_text(programs_text)
cleaned_chunks_faqs = preprocess_text(faqs_text)
cleaned_chunks_resources = preprocess_text(resources_text)

# STEP 4 — load the embedding model and embed every chunk
model = SentenceTransformer("all-MiniLM-L6-v2")


def create_embeddings(text_chunks):
    chunk_embeddings = model.encode(text_chunks, convert_to_tensor=True)
    return chunk_embeddings


chunk_embeddings_programs = create_embeddings(cleaned_chunks_programs)
chunk_embeddings_faqs = create_embeddings(cleaned_chunks_faqs)
chunk_embeddings_resources = create_embeddings(cleaned_chunks_resources)


# STEP 5 — find the most relevant chunks for a query
def get_top_chunks(query, chunk_embeddings, text_chunks, k=3):
    query_embedding = model.encode(query, convert_to_tensor=True)
    query_embedding_normalized = query_embedding / query_embedding.norm()
    chunk_embeddings_normalized = chunk_embeddings / chunk_embeddings.norm(
        dim=1, keepdim=True
    )
    similarities = torch.matmul(chunk_embeddings_normalized, query_embedding_normalized)
    top_indices = torch.topk(similarities, k=min(k, len(text_chunks))).indices
    top_chunks = []
    for top_index in top_indices:
        top_chunks.append(text_chunks[top_index])
    return top_chunks


# ---------------------------------------------------------------------------
# SYSTEM PROMPT — warm, friendly tone with grounding, routing, crisis safety.
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = """You are Eli, Eliada Homes' friendly virtual helper. Eliada is a caring nonprofit in Asheville, NC that supports children, youth, and families. You're often the first friendly voice someone reaches when they're looking for help, so warmth matters as much as accuracy. If someone asks your name, you're Eli.

Your goal is to make people feel welcomed and pointed in the right direction. Here's how:

1. BE WARM AND HUMAN. Greet people kindly and acknowledge what they're going through ("That sounds really hard - let's get you to the right place."). Use a gentle, encouraging, conversational tone. Reassure people that asking for help is okay. Never sound robotic, clinical, or dismissive.

2. ANSWER ONLY FROM THE PROVIDED INFO. Base every fact - phone numbers, addresses, hours, eligibility - strictly on the "Relevant information" section given with each message. Never invent or guess details. If you don't have the answer, warmly say so and offer Eliada's main line: 828-254-5356.

3. HELP PEOPLE FIND THE RIGHT PLACE. Many people reach out needing help Eliada doesn't provide directly - food, shelter, housing, healthcare, recovery, IDs, and more. When that happens, kindly connect them to the right local organization from the Relevant information, including the phone number or address. This is one of the most important and caring things you do.

4. EMERGENCIES AND CRISIS - STAY CALM AND CLEAR. If someone describes a life-threatening emergency, gently but clearly tell them to call 911 right away. If someone seems to be in a mental health crisis or mentions thoughts of self-harm, kindly share 988 (call or text, available 24/7) and let them know they don't have to go through it alone - but do NOT try to counsel or treat them yourself. Keep these responses calm, caring, and brief.

5. KEEP IT SIMPLE AND KIND. Use plain, friendly language. Lead with the single most helpful next step (usually one phone number or place). Don't overwhelm someone with a long list - offer the best one or two options, and invite them to ask for more.

6. KNOW YOUR ROLE. You're a friendly guide for information and referrals - not a counselor, doctor, or lawyer. Don't give clinical, legal, or medical advice. When something needs a professional, warmly point them to the right one."""


# ---------------------------------------------------------------------------
# TRANSCRIBE — turn recorded audio into text via HF Inference (Whisper).
# ---------------------------------------------------------------------------
def transcribe(audio_path):
    if not audio_path:
        return ""
    try:
        with open(audio_path, "rb") as f:
            audio_bytes = f.read()
        result = client.automatic_speech_recognition(
            audio_bytes, model="openai/whisper-large-v3"
        )
        # result may be an object with .text or a plain dict
        text = getattr(result, "text", None)
        if text is None and isinstance(result, dict):
            text = result.get("text", "")
        return (text or "").strip()
    except Exception as e:
        return f"[Could not transcribe audio: {e}]"


# ---------------------------------------------------------------------------
# SPEAK — turn Eli's latest reply into audio via HF Inference (text-to-speech).
# ---------------------------------------------------------------------------
def speak_last_reply(history):
    if not history:
        return None
    # Find the most recent assistant message.
    last_reply = ""
    for turn in reversed(history):
        if turn.get("role") == "assistant" and turn.get("content"):
            last_reply = to_text(turn["content"])
            break
    if not last_reply:
        return None

    # Try a few TTS models in case one isn't currently served by the API.
    tts_models = [
        "hexgrad/Kokoro-82M",
        "facebook/mms-tts-eng",
        "espnet/kan-bayashi_ljspeech_vits",
    ]
    for tts_model in tts_models:
        try:
            audio_bytes = client.text_to_speech(last_reply, model=tts_model)
            out_path = "eli_reply.flac"
            with open(out_path, "wb") as f:
                f.write(audio_bytes)
            print(f"[TTS] success with {tts_model}")
            return out_path
        except Exception as e:
            print(f"[TTS] failed with {tts_model}: {e}")
            continue
    print("[TTS] all models failed")
    return None


# ---------------------------------------------------------------------------
# LOGGING TOOL — record anonymous request categories to a private HF Dataset.
# Degrades gracefully: if anything fails, Eli keeps working and just skips it.
#
# ONE-TIME SETUP:
#   1. On huggingface.co, create a new Dataset (e.g. "your-username/eli-logs"),
#      set it to Private.
#   2. Add a Space secret named LOG_DATASET with the value "your-username/eli-logs".
#   3. Your existing HF_TOKEN (Write role) is reused to append rows.
# If LOG_DATASET isn't set, logging is silently disabled and Eli runs normally.
# ---------------------------------------------------------------------------
import csv
import datetime
from huggingface_hub import HfApi

LOG_DATASET = os.environ.get("LOG_DATASET", "")  # e.g. "username/eli-logs"
LOG_FILE = "logs.csv"
_hf_api = HfApi(token=HF_TOKEN) if LOG_DATASET else None

VALID_CATEGORIES = {
    "food", "housing", "shelter", "healthcare", "mental_health",
    "crisis", "recovery", "transportation", "documents", "veterans",
    "clothing", "childcare", "eliada_programs", "escalated", "other",
}


def _append_local_row(category):
    new_file = not os.path.exists(LOG_FILE)
    with open(LOG_FILE, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        if new_file:
            writer.writerow(["timestamp", "category"])
        writer.writerow(
            [datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), category]
        )


def log_request(category):
    """Tool: record an anonymous request category. No personal data, ever."""
    if not LOG_DATASET or _hf_api is None:
        return  # logging disabled — Eli runs normally
    category = (category or "other").strip().lower()
    if category not in VALID_CATEGORIES:
        category = "other"
    try:
        _append_local_row(category)
        _hf_api.upload_file(
            path_or_fileobj=LOG_FILE,
            path_in_repo=LOG_FILE,
            repo_id=LOG_DATASET,
            repo_type="dataset",
        )
        print(f"[LOG] recorded category: {category}")
    except Exception as e:
        print(f"[LOG] failed (continuing anyway): {e}")


def classify_request(message):
    """Ask the model to label the request with one category (the 'agent' decision)."""
    try:
        result = client.chat_completion(
            [
                {
                    "role": "system",
                    "content": (
                        "You are a classifier. Reply with EXACTLY ONE word from this "
                        "list and nothing else - no punctuation, no explanation:\n"
                        + ", ".join(sorted(VALID_CATEGORIES)) + "\n"
                        "Guidance: 'crisis' for emergencies or self-harm. "
                        "'eliada_programs' for questions about Eliada's own services "
                        "(child care, foster care, residential treatment). "
                        "'food' for meals, pantries, groceries. "
                        "'housing' or 'shelter' for places to live or stay. "
                        "'other' only if truly nothing fits."
                    ),
                },
                {"role": "user", "content": message},
            ],
            max_tokens=512,
            temperature=0,
        )
        raw = (result.choices[0].message.content or "").strip().lower()
        # Normalize: keep letters and underscores only.
        cleaned = "".join(c if (c.isalpha() or c == "_") else " " for c in raw)
        tokens = cleaned.split()

        # 1) exact single-word match
        if raw in VALID_CATEGORIES:
            return raw
        # 2) any token is a valid category
        for t in tokens:
            if t in VALID_CATEGORIES:
                return t
        # 3) any valid category appears as a substring of the reply
        for cat in VALID_CATEGORIES:
            if cat in raw:
                return cat
        print(f"[LOG] classify unmatched, raw output: {raw!r}")
        return "other"
    except Exception as e:
        print(f"[LOG] classify failed: {e}")
        return "other"


# ---------------------------------------------------------------------------
# RESPOND — retrieve from all three files, build the prompt, stream the answer.
# ---------------------------------------------------------------------------
def to_text(value):
    """Coerce a Gradio message (str, dict, or list) into plain text."""
    if value is None:
        return ""
    if isinstance(value, str):
        return value
    if isinstance(value, dict):
        # Newer Gradio may wrap content as {"type": "text", "text": "..."}
        if "text" in value and isinstance(value["text"], str):
            return value["text"]
        if "content" in value:
            return to_text(value["content"])
        return ""
    if isinstance(value, (list, tuple)):
        return " ".join(to_text(v) for v in value).strip()
    return str(value)


def respond(message, history, max_tokens=512, temperature=0.3, top_p=0.95):
    message = to_text(message)

    top_programs = get_top_chunks(
        message, chunk_embeddings_programs, cleaned_chunks_programs, k=2
    )
    top_faqs = get_top_chunks(
        message, chunk_embeddings_faqs, cleaned_chunks_faqs, k=2
    )
    top_resources = get_top_chunks(
        message, chunk_embeddings_resources, cleaned_chunks_resources, k=3
    )

    context = (
        "Eliada programs:\n" + "\n".join(top_programs) + "\n\n"
        "Eliada FAQ:\n" + "\n".join(top_faqs) + "\n\n"
        "Community resources:\n" + "\n".join(top_resources)
    )

    system_with_context = (
        f"{SYSTEM_PROMPT}\n\n"
        f"--- Relevant information ---\n{context}\n--- End of information ---"
    )

    messages = [{"role": "system", "content": system_with_context}]
    for turn in history:
        messages.append(
            {"role": turn.get("role", "user"), "content": to_text(turn.get("content"))}
        )
    messages.append({"role": "user", "content": message})

    response = ""
    for chunk in client.chat_completion(
        messages,
        max_tokens=max_tokens,
        stream=True,
        temperature=temperature,
        top_p=top_p,
    ):
        choices = chunk.choices
        if len(choices) and choices[0].delta.content:
            response += choices[0].delta.content
            yield response


# ---------------------------------------------------------------------------
# CHAT HANDLERS — used by both the text box and the mic.
# ---------------------------------------------------------------------------
def add_user_message(message, history):
    # Show the user's message immediately, clear the textbox.
    history = history + [{"role": "user", "content": to_text(message)}]
    return "", history


def stream_bot_reply(history):
    # The last item is the user's message; generate the reply to it.
    user_message = to_text(history[-1]["content"])
    prior = history[:-1]
    history = history + [{"role": "assistant", "content": ""}]
    for partial in respond(user_message, prior):
        history[-1]["content"] = partial
        yield history

    # After replying, classify and log the request category (anonymous, non-blocking).
    try:
        category = classify_request(user_message)
        log_request(category)
    except Exception as e:
        print(f"[LOG] post-reply logging skipped: {e}")


def voice_to_text(audio_path):
    # Transcribe and place the text in the textbox for the user to review/send.
    return transcribe(audio_path)


# ---------------------------------------------------------------------------
# UI — text box + microphone, built with Blocks for full control.
# ---------------------------------------------------------------------------
with gr.Blocks(title="Ask Eli - Eliada Homes") as demo:
    gr.Markdown(
        "## Ask Eli 💛\n"
        "Hi! I'm Eli, Eliada Homes' friendly helper. Tell me what you're looking "
        "for - by typing or using the microphone - and I'll do my best to point "
        "you to the right place. You're always welcome here.\n\n"
        "**In an emergency, please call 911. For a mental health crisis, call or text 988.**"
    )

    chat_box = gr.Chatbot(height=460, label="Conversation")

    with gr.Row():
        msg = gr.Textbox(
            placeholder="Type your message, or use the microphone below...",
            label="Your message",
            scale=4,
        )
        send_btn = gr.Button("Send", variant="primary", scale=1)

    mic = gr.Audio(
        sources=["microphone"],
        type="filepath",
        label="Or speak your message (it will appear in the box above to review before sending)",
    )

    with gr.Row():
        speak_btn = gr.Button("🔊 Read Eli's last reply aloud", scale=1)
    reply_audio = gr.Audio(
        label="Eli's voice", autoplay=True, visible=True, interactive=False
    )

    gr.Examples(
        examples=[
            "What programs does Eliada offer?",
            "I need help finding food.",
            "I'm looking for a place to stay tonight.",
            "How do I enroll my child in child care?",
        ],
        inputs=msg,
    )

    # Read Eli's last reply aloud when the button is tapped.
    speak_btn.click(fn=speak_last_reply, inputs=chat_box, outputs=reply_audio)

    # Mic -> transcribe -> fill the textbox (user reviews, then sends).
    mic.stop_recording(fn=voice_to_text, inputs=mic, outputs=msg)

    # Send via button.
    send_btn.click(
        fn=add_user_message, inputs=[msg, chat_box], outputs=[msg, chat_box]
    ).then(fn=stream_bot_reply, inputs=chat_box, outputs=chat_box)

    # Send via Enter key.
    msg.submit(
        fn=add_user_message, inputs=[msg, chat_box], outputs=[msg, chat_box]
    ).then(fn=stream_bot_reply, inputs=chat_box, outputs=chat_box)


if __name__ == "__main__":
    demo.launch()