api/services/llm_service.py

import re
import httpx
from api.config import settings

LLM_URL = settings.llm.groq_url
LLM_API_KEY = settings.llm.groq_token
LLM_MODEL = settings.llm.chat_model
LLM_TEMPERATURE = settings.llm.temperature
LLM_MAX_TOKENS = settings.llm.max_tokens

# Chat
LLM_URL = settings.llm.groq_url
LLM_API_KEY = settings.llm.groq_token
LLM_MODEL = settings.llm.chat_model
LLM_TEMPERATURE = settings.llm.temperature
LLM_MAX_TOKENS = settings.llm.max_tokens

# Chat

def _parse_response(text: str) -> dict:
    """ Parse la reponse de Zephyr pour extraire le texte de la réponse, les quick replies et les symptômes (si présents) """
    quick_replies = []
    symptoms = None
    lang = ["en"]  # Par defaut

    print("=== Réponse brute de Zephyr ===")
    print(text)

    # 1. Extraire la langue: [LANGUAGE: [...]]
    lang_match = re.search(r'\[LANGUAGE:\s*(\[.*?\])\]', text)
    if lang_match:
        import json
        try:
            lang = json.loads(lang_match.group(1))
        except Exception:
            pass
        text = text[:lang_match.start()].strip()

    # 2. Extraire les quick replies: [QUICK_REPLIES: [...]]
    qr_match = re.search(r'\[QUICK_REPLIES:\s*(\[.*?\])\]', text)
    if qr_match:
        import json
        try:
            quick_replies = json.loads(qr_match.group(1))
        except Exception:
            pass
        text = text[:qr_match.start()].strip()

    # 3. Extraire les symptômes: [SYMPTOMS: [...]]
    sym_match = re.search(r'\[SYMPTOMS:\s*(\[.*?\])\]', text)
    if sym_match:
        import json
        try:
            symptoms = json.loads(sym_match.group(1))
        except Exception:
            pass
        text = text[:sym_match.start()].strip()

    parsed_response = {
        "lang": lang[0] if lang else "en",
        "reply": text.strip(),
        "quick_replies": quick_replies,
        "symptoms": symptoms
    }

    print("=== Réponse parsée ===")
    print(parsed_response)

    return parsed_response

def _chat_mock_response() -> dict:
    return _parse_response(
        "Hello, I'm HealthCare, an AI health assistant. What seems to be bothering you or what symptoms are you experiencing today? "
        "[SYMPTOMS: [\"headache\", \"fever\", \"nausea\"]] "
        "[QUICK_REPLIES: [\"I have a headache\", \"I have a fever\", \"I feel nauseous\"]] "
        "[LANGUAGE: [en]]"
    )

def _chat_robust_fallback_response(status_code: int, error_text: str) -> dict:
    return _parse_response(
        f"Le service de chat est temporairement indisponible (erreur {status_code}). "
        "Veuillez contacter votre administrateur ou réessayer plus tard. "
        "[SYMPTOMS: []] "
        "[QUICK_REPLIES: []] "
        "[LANGUAGE: [en]]"
    )

def _chat_debug_response(status_code: int, error_text: str) -> dict:
    return {
        "lang": "en",
        "reply": f"DEBUG: LLM chat failed with status {status_code}. {error_text}",
        "quick_replies": [],
        "symptoms": []
    }


async def chat(history: list[dict], new_message: str) -> dict:
    # return _chat_mock_response()
    messages = [{"role": "system", "content": """You are HealthCare, an AI health assistant.
You help users describe their symptoms and ask clarifying questions.
Keep responses short (1-2 sentences).
Ask one follow-up question at a time.
When you have enough info (at least 3-5 symptoms or details), summarize the symptoms and output them in English Language as a JSON block at the very end like:
[SYMPTOMS: [\"fever\", \"headache\", \"vomiting\"]]
If you ask a yes/no question, suggest quick replies at the end like:
[QUICK_REPLIES: [\"Yes\", \"No\", \"Sometimes\"]]
Always add a disclaimer that this is not medical advice.
Respond in the same language as the user. Mention the actual language at the end like:
[LANGUAGE: [\"fr\"]]"""}]

    messages += history
    messages.append({"role": "user", "content": new_message})

    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json",
    }


    payload = {
        "model": LLM_MODEL,
        "messages": messages,
        "max_tokens": LLM_MAX_TOKENS,
        "temperature": LLM_TEMPERATURE,
        "max_tokens": LLM_MAX_TOKENS,
        "temperature": LLM_TEMPERATURE,
    }

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(LLM_URL, json=payload, headers=headers)
    except httpx.HTTPError as exc:
        error_text = str(exc)
        print("HF ERROR: HTTP exception during chat request:", error_text)
        if settings.debug_mode:
            return _chat_debug_response(-1, error_text)
        if settings.robust_mode:
            return _chat_robust_fallback_response(-1, error_text)
        raise

    if response.status_code == 200:
        try:
            generated = response.json()["choices"][0]["message"]["content"]
            return _parse_response(generated)
        except Exception as exc:
            error_text = str(exc)
            print("HF ERROR: malformed chat response:", error_text)
            if settings.debug_mode:
                return _chat_debug_response(response.status_code, error_text)
            if settings.robust_mode:
                return _chat_robust_fallback_response(response.status_code, error_text)
            raise

    print("HF ERROR:", response.status_code, response.text)
    if settings.debug_mode:
        return _chat_debug_response(response.status_code, response.text)
    if settings.robust_mode:
        if response.status_code == 402:
            return _chat_mock_response()
        return _chat_robust_fallback_response(response.status_code, response.text)
    response.raise_for_status()
    return {}


# Formattage du diagnostic final pour affichage a l'utilisateur

def _parse_formatted_diagnosis(text: str) -> dict:
    """ Parse la reponse de Zephyr pour extraire les infos sur les maladies et les recommandations """
    diseases_info = []
    recommendation = ""

    # print("=== Réponse brute de Zephyr pour le formatage du diagnostic ===")
    # print(text)

    # 1. Extraire la recommandation: [RECOMMENDATION: [...]]
    rec_match = re.search(r'\[RECOMMENDATION:\s*(\[.*?\])\]', text, re.DOTALL)
    if rec_match:
        import json
        try:
            recommendation = json.loads(rec_match.group(1))
        except Exception:
            pass
        text = text[:rec_match.start()].strip()

    # 2. Extraire les infos sur les maladies: [DISEASES_INFO: [{\"disease1\": \"description\"}, {\"disease2\": \"description\"}, ...]] -> disease_info = [{"disease1": "description"}, {"disease2": "description"}, ...]
    di_match = re.search(r'\[DISEASES_INFO:\s*(\[.*?\])\]', text, re.DOTALL)
    if di_match:
        import json
        try:
            diseases_info = json.loads(di_match.group(1))
        except Exception:
            pass
        text = text[:di_match.start()].strip()

    parsed_response = {
        "reply": text.strip(),
        "diseases_info": diseases_info,
        "recommendation": recommendation
    }

    # print("=== Réponse parsée ===")
    # print(parsed_response)

    return parsed_response


def _format_diagnosis_mock_response() -> dict:
    return _parse_formatted_diagnosis("""Based on your diagnosis results, here is a clear and concise summary:

**Most Probable Conditions:**

1. **Malaria** (47% probability) - This condition has a low severity level and is characterized by symptoms such as headache, fever, and nausea.
2. **Pneumonia** (29% probability) - This condition has a medium severity level and is also marked by symptoms including headache, fever, and nausea.
3. **Gastroenteritis** (24% probability) - Similar to pneumonia, gastroenteritis has a medium severity level with symptoms like headache, fever, and nausea.

**Global Recommendation:**
Given the potential severity of these conditions, it is crucial to take immediate action. Rest and hydration are key. However, if symptoms persist, it is highly recommended to consult a doctor without delay.

[DISEASES_INFO: [{"Malaria": "A disease characterized by headache, fever, and nausea, with a low severity level"}, {"Pneumonia": "A condition marked by headache, fever, and nausea, with a medium severity level"}, {"Gastroenteritis": "Characterized by headache, fever, and nausea, with a medium severity level"}]]
[RECOMMENDATION: ["Visit a doctor immediately\nGet a blood test to confirm the diagnosis\nStart antimalarial treatment if confirmed"]]
""")

def _format_diagnosis_robust_fallback_response(status_code: int, error_text: str) -> dict:
    return {
        "reply": f"ROBUST FALLBACK: format diagnosis failed with status {status_code}. {error_text}",
        "diseases_info": [],
        "recommendation": "ROBUST FALLBACK: see logs."
    }

def _format_diagnosis_debug_response(status_code: int, error_text: str) -> dict:
    return {
        "reply": f"DEBUG: format diagnosis failed with status {status_code}. {error_text}",
        "diseases_info": [],
        "recommendation": "DEBUG: see logs."
    }

async def format_diagnosis(diagnosis: dict, lang: str) -> dict:

    # return _format_diagnosis_mock_response()
    messages = [{"role": "system", "content": "You are a helpful assistant that formats medical diagnosis results for user display. Format the following diagnosis results in a clear and concise way for a user, highlighting the most probable conditions and their severity. Propose a global recommendation with respect to the diagnosis you've got. Respond using the language indicated by the language code. At the end of your message, simply return the descriptions of each disease and the symptoms likely associated with them in the specified language (filter the appropriate symptoms for each disease the diagnosis, if there's no final match just choose random symptoms from the list) followed by the recommendation you've given in a fixed format like: [DISEASES_INFO: [{\"disease_1\": \"description\", \"symptoms\": [\"symptom1\", \"symptom2\"]}, {\"disease_2\": \"description\", \"symptoms\": [\"symptom3\", \"symptom4\"]}, ...]]\n[RECOMMENDATION: [\"Visit a doctor immediately\"]]"}]
    messages.append({"role": "user", "content": f"Format this diagnosis for user display: {diagnosis}. Language code {lang}."})

    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json",
    }
    
    payload = {
        "model": LLM_MODEL,
        "messages": messages,
        "max_tokens": LLM_MAX_TOKENS,
        "temperature": LLM_TEMPERATURE
    }

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(LLM_URL, json=payload, headers=headers)
    except httpx.HTTPError as exc:
        error_text = str(exc)
        print("HF ERROR: HTTP exception during format_diagnosis request:", error_text)
        if settings.debug_mode:
            return _format_diagnosis_debug_response(-1, error_text)
        if settings.robust_mode:
            return _format_diagnosis_robust_fallback_response(-1, error_text)
        raise

    if response.status_code == 200:
        try:
            generated = response.json()["choices"][0]["message"]["content"]
            return _parse_formatted_diagnosis(generated.strip())
        except Exception as exc:
            error_text = str(exc)
            print("HF ERROR: malformed format_diagnosis response:", error_text)
            if settings.debug_mode:
                return _format_diagnosis_debug_response(response.status_code, error_text)
            if settings.robust_mode:
                return _format_diagnosis_robust_fallback_response(response.status_code, error_text)
            raise

    print("HF ERROR:", response.status_code, response.text)
    if settings.debug_mode:
        return _format_diagnosis_debug_response(response.status_code, response.text)
    if settings.robust_mode:
        if response.status_code == 402:
            return _format_diagnosis_mock_response()
        return _format_diagnosis_robust_fallback_response(response.status_code, response.text)
    response.raise_for_status()
    return {}

# Resumés médicaux de documents

def _summarize_mock_response() -> str:
    return "Résumé médical du document (mock): Patient présente des symptômes de fièvre, toux et fatigue. Diagnostic possible de grippe. Recommandation: repos, hydratation, et consultation médicale si les symptômes persistent."

def _summarize_robust_fallback_response(status_code: int, error_text: str) -> str:
    return (
        f"Le résumé médical n'est pas disponible pour le moment (erreur {status_code}). "
        "Veuillez contacter votre administrateur ou réessayer plus tard."
    )

def _summarize_debug_response(status_code: int, error_text: str) -> str:
    return f"DEBUG: summarization failed with status {status_code}. {error_text}"

async def summarize_for_medical_context(raw_text: str) -> str:
    messages = [{"role": "system", "content": "You are a medical document summarizer. Extract only medically relevant information: symptoms, diagnoses, medications, test results, dates. Be concise (max 5 sentences). Output in the same language as the document language."}]
    
    messages.append({"role": "user", "content": f"Summarize this medical document: {raw_text}"})

    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": LLM_MODEL,
        "messages": messages,
        "max_tokens": LLM_MAX_TOKENS,
        "temperature": LLM_TEMPERATURE
    }

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(LLM_URL, json=payload, headers=headers)
    except httpx.HTTPError as exc:
        error_text = str(exc)
        print("HF ERROR: HTTP exception during summarize request:", error_text)
        if settings.debug_mode:
            return _summarize_debug_response(-1, error_text)
        if settings.robust_mode:
            return _summarize_robust_fallback_response(-1, error_text)
        raise

    if response.status_code == 200:
        try:
            generated = response.json()["choices"][0]["message"]["content"]
            return generated.strip()
        except Exception as exc:
            error_text = str(exc)
            print("HF ERROR: malformed summarize response:", error_text)
            if settings.debug_mode:
                return _summarize_debug_response(response.status_code, error_text)
            if settings.robust_mode:
                return _summarize_robust_fallback_response(response.status_code, error_text)
            raise

    print("HF ERROR:", response.status_code, response.text)
    if settings.debug_mode:
        return _summarize_debug_response(response.status_code, response.text)
    if settings.robust_mode:
        if response.status_code == 402:
            return _summarize_mock_response()
        return _summarize_robust_fallback_response(response.status_code, response.text)
    response.raise_for_status()
    return ""