File size: 4,831 Bytes
0d72980
36495b8
0d72980
36495b8
4784b5f
0d72980
7590ff0
 
4784b5f
0d72980
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4784b5f
36495b8
0d72980
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4784b5f
0d72980
 
 
 
 
 
 
 
 
 
 
36495b8
0d72980
 
 
 
36495b8
0d72980
 
36495b8
0d72980
 
 
 
 
36495b8
0d72980
 
36495b8
0d72980
 
 
 
 
 
4784b5f
0d72980
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

import os
import re
import gradio as gr
from huggingface_hub import InferenceClient

# Retrieve the token securely from the Space's environment variables/secrets
HF_TOKEN = os.getenv("HF_TOKEN")
MODEL = os.getenv("HF_MODEL_ID", "meta-llama/Llama-3.1-8B-Instruct")

REFERENCE = {'metformin': 'Contraindicated when eGFR is below 30. Between 30 and 45, do not start it; if the patient is already taking it, halve the maximum to 1000 mg/day and recheck eGFR every 3 months.', 'a1c target': 'General adult target is 7.0 percent. For an older or frail patient, 8.0 percent is reasonable: a tighter target buys little and costs hypoglycaemia risk.', 'ckd stage': 'G1-G2 is eGFR 60 or above. G3a is 45-59. G3b is 30-44. G4 is 15-29. G5 is below 15, which is kidney failure.', 'alert fatigue': 'Degraded human response to a high volume of alarms, caused by the alarms themselves. Around 90 percent of drug-interaction alerts are overridden.', 'lisinopril': 'An ACE inhibitor. Monitor potassium and renal function. Hold it if the patient becomes acutely unwell or dehydrated.'}
CHARTS = {'alvarez': 'Alvarez, F, 68. Type 2 diabetes. eGFR 28. A1c 9.1 percent. Current: metformin 1000 mg twice daily, lisinopril 10 mg daily.', 'okafor': 'Okafor, M, 54. Hypertension. eGFR 71. A1c 5.8 percent. Current: lisinopril 20 mg daily.', 'bianchi': 'Bianchi, F, 81. Type 2 diabetes, frail. eGFR 39. A1c 8.4 percent. Current: metformin 500 mg twice daily.'}
SYSTEM_PROMPT = """You answer clinical questions by looking things up. You have one tool:

lookup: look up one topic in the clinic reference. Input: a short topic, e.g. "metformin".
get_patient: look up a patient chart by surname. Input: a surname, e.g. "Alvarez".

ALWAYS answer in this exact format, ONE step at a time:

Thought: what you know so far and what you need next
Action: lookup
Action Input: the topic to look up

Then STOP and wait. You will be given an Observation with the result.
Never write the Observation yourself.

When you have enough to answer, write instead:

Thought: I now know the final answer
Final Answer: your answer, in one or two sentences

/no_think"""

client = InferenceClient(token=HF_TOKEN)

def lookup(topic: str) -> str:
    hit = REFERENCE.get(topic.strip().lower())
    return hit if hit else f"'{topic}' is not in the reference."

def get_patient(name: str) -> str:
    hit = CHARTS.get(name.strip().lower())
    return hit if hit else f"There is no patient called '{name}' on this list."

TOOLS = {
    "lookup": lookup,
    "get_patient": get_patient
}

def parse_action(text: str):
    tool = re.search(r"Action:\s*([a-zA-Z_]+)", text)
    arg  = re.search(r"Action Input:\s*(.+)", text)
    if not tool:
        return None, None
    return tool.group(1).strip(), (arg.group(1).strip() if arg else "")

def run_tool(name, arg):
    fn = TOOLS.get(name)
    if fn is None:
        return f"There is no tool called '{name}'. Available: {', '.join(TOOLS)}.", False
    try:
        return fn(arg), True
    except Exception as e:
        return f"The tool failed: {type(e).__name__}: {e}", False

def ask_model(messages):
    try:
        r = client.chat.completions.create(
            model=MODEL, messages=messages,
            temperature=0, max_tokens=1000
        )
        text = r.choices[0].message.content or ""
        text = re.sub(r"<think>.*?</think>", "", text, flags=re.S)
        text = re.sub(r"<think>.*$", "", text, flags=re.S).strip()
        return text
    except Exception as e:
        return f"[model call failed: {type(e).__name__}: {e}]"

def chat_interface(question, history):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}]
    trace_output = []

    for step in range(1, 7):
        reply = ask_model(messages)
        trace_output.append(f"**Step {step} (Model)**:\n{reply}")
        yield "\n\n".join(trace_output)

        if "Final Answer:" in reply:
            break

        tool, arg = parse_action(reply)
        if tool is None:
            obs = "I could not find an Action in that."
        else:
            obs, _ = run_tool(tool, arg)

        trace_output.append(f"**Step {step} (Observation)**:\n{obs}")
        yield "\n\n".join(trace_output)

        messages.append({"role": "assistant", "content": reply})
        messages.append({"role": "user", "content": "Observation: " + obs})

demo = gr.ChatInterface(
    fn=chat_interface,
    title="🏥 ReAct Clinical Decision Support Agent",
    description="Ask clinical workflow queries powered by Hugging Face Inference API.",
    examples=[
        "Is Alvarez still OK on her metformin? Check her chart first.",
        "What is the a1c target?",
        "What does our hospital night shift override policy say about 240 alerts in one night?"
    ]
)

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