Spaces:
Sleeping
Sleeping
| 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() | |