MCP_Res / app.py
mgbam's picture
Update app.py
994feb6 verified
raw
history blame
6.69 kB
# ──────────────────────────── app.py ─────────────────────────────────
"""Streamlit UI – MedGenesis v2 with gene + variant + trial integration."""
import os, pathlib, asyncio, re
from pathlib import Path
import streamlit as st
import pandas as pd
import plotly.express as px
from fpdf import FPDF
from streamlit_agraph import agraph
from mcp.orchestrator import orchestrate_search, answer_ai_question
from mcp.workspace import get_workspace, save_query
from mcp.knowledge_graph import build_agraph
from mcp.graph_utils import build_nx, get_top_hubs, get_density
from mcp.alerts import check_alerts
# ---- Streamlit telemetry patch -------------------------------------
os.environ.update({
"STREAMLIT_DATA_DIR": "/tmp/.streamlit",
"XDG_STATE_HOME": "/tmp",
"STREAMLIT_BROWSER_GATHERUSAGESTATS": "false",
})
pathlib.Path("/tmp/.streamlit").mkdir(parents=True, exist_ok=True)
ROOT = Path(__file__).parent
LOGO = ROOT / "assets" / "logo.png"
# ---------------- helpers -------------------------------------------
def _latin1_safe(t: str) -> str:
return t.encode("latin-1", "replace").decode("latin-1")
def _export_pdf(papers):
pdf = FPDF()
pdf.set_auto_page_break(auto=True, margin=15)
pdf.add_page()
pdf.set_font("Helvetica", size=11)
pdf.cell(200, 8, _latin1_safe("MedGenesis AI – Results"), ln=True, align="C")
pdf.ln(3)
for i, p in enumerate(papers, 1):
pdf.set_font("Helvetica", "B", 11)
pdf.multi_cell(0, 7, _latin1_safe(f"{i}. {p['title']}"))
pdf.set_font("Helvetica", size=9)
body = f"{p['authors']}\n{p['summary']}\n{p['link']}\n"
pdf.multi_cell(0, 6, _latin1_safe(body))
pdf.ln(1)
return pdf.output(dest="S").encode("latin-1", "replace")
# ---------------- sidebar -------------------------------------------
def _workspace_sidebar():
with st.sidebar:
st.header("πŸ—‚οΈ Workspace")
ws = get_workspace()
if not ws:
st.info("Run a search then press **Save** to populate this list.")
return
for i, item in enumerate(ws, 1):
with st.expander(f"{i}. {item['query']}"):
st.write(item["result"]["ai_summary"])
# ---------------- main ----------------------------------------------
def render_ui():
st.set_page_config("MedGenesis AI", layout="wide")
_workspace_sidebar()
# header ---------------------------------------------------------
c1, c2 = st.columns([0.15, 0.85])
if LOGO.exists():
with c1: st.image(str(LOGO), width=105)
with c2:
st.markdown("## 🧬 **MedGenesis AI**")
st.caption("Multi‑source biomedical assistant Β· OpenAI / Gemini")
llm = st.radio("LLM engine", ["openai", "gemini"], horizontal=True)
query = st.text_input("Enter biomedical question", "CRISPR glioblastoma therapy")
if st.button("Run Search πŸš€") and query:
with st.spinner("Collecting literature & biomedical data …"):
res = asyncio.run(orchestrate_search(query, llm=llm))
st.success(f"Completed with **{res['llm_used'].title()}**")
st.session_state.result = res
st.session_state.last_query = query
st.session_state.last_llm = llm
res = st.session_state.get("result")
if not res:
st.info("Enter a question and press **Run Search πŸš€**")
return
tabs = st.tabs(["Results", "Genes", "Trials", "Graph", "Metrics", "Visuals"])
# results --------------------------------------------------------
with tabs[0]:
for i, p in enumerate(res["papers"], 1):
st.markdown(f"**{i}. [{p['title']}]({p['link']})** *{p['authors']}*")
st.write(p["summary"])
c1, c2 = st.columns(2)
with c1:
st.download_button("CSV", pd.DataFrame(res["papers"]).to_csv(index=False), "papers.csv")
with c2:
st.download_button("PDF", _export_pdf(res["papers"]), "papers.pdf", mime="application/pdf")
if st.button("πŸ’Ύ Save"):
save_query(query, res)
st.success("Saved to workspace")
st.subheader("AI summary")
st.info(res["ai_summary"])
# gene tab -------------------------------------------------------
with tabs[1]:
if not res["genes"]:
st.info("No gene hits (rate‑limited or none found).")
for g in res["genes"]:
st.json(g)
if res["variants"]:
st.markdown("### Tumour variants (cBioPortal)")
for k, v in res["variants"].items():
st.write(f"**{k}** – {len(v)} variants")
# trials tab -----------------------------------------------------
with tabs[2]:
st.header("Clinical trials")
if not res["clinical_trials"]:
st.info("No trials (rate‑limited or none found).")
for t in res["clinical_trials"]:
st.markdown(f"**{t['nctId']}** – {t['briefTitle']}")
st.write(f"Phase {t.get('phase')} | Status {t.get('status')}")
# graph tab ------------------------------------------------------
with tabs[3]:
nodes, edges, cfg = build_agraph(res["papers"], res["umls"], res["drug_safety"])
hl = st.text_input("Highlight node:")
if hl:
pat = re.compile(re.escape(hl), re.I)
for n in nodes:
n.color = "#f1c40f" if pat.search(n.label) else "#d3d3d3"
agraph(nodes, edges, cfg)
# metrics tab ----------------------------------------------------
with tabs[4]:
G = build_nx([n.__dict__ for n in nodes], [e.__dict__ for e in edges])
st.metric("Density", f"{get_density(G):.3f}")
for nid, sc in get_top_hubs(G):
lab = next((n.label for n in nodes if n.id == nid), nid)
st.write(f"- {lab} {sc:.3f}")
# visuals --------------------------------------------------------
with tabs[5]:
years = [p.get("published", "")[:4] for p in res["papers"] if p.get("published")]
if years:
fig = px.histogram(years, nbins=12, title="Publication Year")
st.plotly_chart(fig)
# follow‑up QA ---------------------------------------------------
st.markdown("---")
q = st.text_input("Ask follow‑up question:")
if st.button("Ask AI"):
with st.spinner("Querying LLM …"):
ans = asyncio.run(answer_ai_question(q, context=st.session_state.last_query, llm=st.session_state.last_llm))
st.write(ans["answer"])
if __name__ == "__main__":
render_ui()