1a5cc72067
Recognises products by trade name or substance (Slovak inflection aware), queries LightRAG and shows the answer with the retrieved subgraph and ADC source texts.
368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""Web demo of the pharmacy decision-support agent (Streamlit).
|
||
|
||
The agent answers questions about indications, contraindications and
|
||
interactions using the LightRAG knowledge graph built from ADC leaflets.
|
||
Every answer is shown together with its justification: the part of the
|
||
knowledge graph that was retrieved and the ADC leaflet texts it came from.
|
||
|
||
Run (LightRAG server must be running, see start_servers.py):
|
||
|
||
streamlit run demo/app.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import html
|
||
import sys
|
||
import re
|
||
import unicodedata
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
import streamlit as st
|
||
from pyvis.network import Network
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
from pharmacy_agent import QUERY_SETTINGS, normalize, build_aliases, detect_drugs, describe_match, prepare_query
|
||
PRODUCT_INDEX = ROOT / "outputs" / "demo" / "product_index.json"
|
||
DEFAULT_URL = "http://localhost:9621"
|
||
|
||
MODES = {
|
||
"mix": "mix – graf + textové úryvky (odporúčané)",
|
||
"hybrid": "hybrid – lokálny + globálny graf",
|
||
"local": "local – entity a ich susedia",
|
||
"global": "global – vzťahy",
|
||
"naive": "naive – len textové úryvky (bez grafu)",
|
||
}
|
||
|
||
EXAMPLES = [
|
||
"Môžem užívať Nurofen spolu s warfarínom?",
|
||
"Kedy sa nesmie užívať metformín?",
|
||
"Na čo sa používa Aclexa?",
|
||
"Aké lieky sa nemajú kombinovať s klaritromycínom?",
|
||
]
|
||
|
||
ENTITY_COLORS = {
|
||
"drug": "#4C78A8",
|
||
"activesubstance": "#72B7B2",
|
||
"drugclass": "#B279A2",
|
||
"indication": "#54A24B",
|
||
"condition": "#F58518",
|
||
"contraindication": "#E45756",
|
||
"interaction": "#EECA3B",
|
||
"patientgroup": "#9D755D",
|
||
}
|
||
DEFAULT_ENTITY_COLOR = "#BAB0AC"
|
||
|
||
# Relation keyword -> (label, colour) used to highlight edges in the graph.
|
||
RELATION_KINDS = (
|
||
("kontraindik", "kontraindikácia", "#E45756"),
|
||
("interak", "interakcia", "#F58518"),
|
||
("indik", "indikácia", "#54A24B"),
|
||
("upozorn", "upozornenie", "#EECA3B"),
|
||
)
|
||
|
||
# Words that start product names but do not identify a product.
|
||
|
||
|
||
# ---------------------------------------------------------------- helpers
|
||
|
||
|
||
|
||
def api(base_url: str, method: str, path: str, timeout: int = 300, **kwargs):
|
||
response = requests.request(method, f"{base_url}{path}", timeout=timeout, **kwargs)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
|
||
@st.cache_data(show_spinner="Načítavam index liekov ADC ...")
|
||
def load_product_index() -> list[dict] | None:
|
||
if not PRODUCT_INDEX.exists():
|
||
return None
|
||
return json.loads(PRODUCT_INDEX.read_text(encoding="utf-8"))
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def clean(text: str | None, limit: int = 600) -> str:
|
||
"""LightRAG joins merged descriptions with <SEP>; show them as paragraphs."""
|
||
text = str(text or "").replace("<SEP>", "\n\n")
|
||
return text if len(text) <= limit else text[:limit].rsplit(" ", 1)[0] + " …"
|
||
|
||
|
||
def relation_kind(keywords: str, description: str) -> tuple[str, str]:
|
||
text = normalize(f"{keywords} {description}")
|
||
for needle, label, color in RELATION_KINDS:
|
||
if needle in text:
|
||
return label, color
|
||
return "iný vzťah", "#9E9E9E"
|
||
|
||
|
||
def render_graph(nodes: list[dict], edges: list[dict], highlight: set[str], height: int = 560) -> None:
|
||
net = Network(height=f"{height}px", width="100%", bgcolor="#ffffff", font_color="#222222", cdn_resources="in_line")
|
||
net.barnes_hut(gravity=-800, central_gravity=0.15, spring_length=100)
|
||
for node in nodes:
|
||
entity_type = str(node.get("type") or "").lower()
|
||
is_focus = normalize(node["id"]) in highlight
|
||
net.add_node(
|
||
node["id"],
|
||
label=node["id"][:40],
|
||
title=f"{node['id']} ({entity_type or '?'})\n\n{clean(node.get('description'))}",
|
||
color=ENTITY_COLORS.get(entity_type, DEFAULT_ENTITY_COLOR),
|
||
size=28 if is_focus else 14,
|
||
borderWidth=4 if is_focus else 1,
|
||
font={"size":18,"strokeWidth":3,"strokeColor":"#ffffff"},
|
||
)
|
||
names = {node["id"] for node in nodes}
|
||
for edge in edges:
|
||
if edge["src"] not in names or edge["tgt"] not in names:
|
||
continue
|
||
label, color = relation_kind(edge.get("keywords", ""), edge.get("description", ""))
|
||
net.add_edge(
|
||
edge["src"], edge["tgt"], color=color, width=3 if label != "iný vzťah" else 1,
|
||
title=f"{label}: {edge.get('keywords', '')}\n\n{clean(edge.get('description'))}",
|
||
)
|
||
graph_html = net.generate_html(notebook=False)
|
||
graph_html = graph_html.replace("</body>", """<script>
|
||
network.once('stabilizationIterationsDone', function () {
|
||
network.setOptions({physics: false, nodes: {scaling: {label: {drawThreshold: 0}}}});
|
||
network.fit({animation: false});
|
||
});
|
||
</script></body>""")
|
||
st.iframe(graph_html, height=height + 20)
|
||
|
||
|
||
def graph_legend() -> None:
|
||
entity_html = " ".join(
|
||
f"<span style='background:{c};color:white;padding:2px 6px;border-radius:4px;margin-right:4px'>{t}</span>"
|
||
for t, c in ENTITY_COLORS.items()
|
||
)
|
||
relation_html = " ".join(
|
||
f"<span style='border-bottom:3px solid {c};margin-right:10px'>{label}</span>" for _, label, c in RELATION_KINDS
|
||
)
|
||
st.markdown(f"<small>Uzly: {entity_html}<br>Hrany: {relation_html}</small>", unsafe_allow_html=True)
|
||
|
||
|
||
def product_name(url: str, by_url: dict) -> str:
|
||
product = by_url.get(url)
|
||
return product["name"] if product else url
|
||
|
||
|
||
# ---------------------------------------------------------------- page
|
||
|
||
st.set_page_config(page_title="Lekárnický asistent", page_icon="💊", layout="wide")
|
||
|
||
with st.sidebar:
|
||
st.header("Nastavenia")
|
||
base_url = st.text_input("LightRAG server", DEFAULT_URL)
|
||
mode = st.selectbox("Režim vyhľadávania", list(MODES), format_func=MODES.get)
|
||
top_k = st.slider("Počet entít / vzťahov (top_k)", 10, 80, 40, step=5)
|
||
max_graph_nodes = st.slider("Max. uzlov v zobrazenom grafe", 10, 120, 50, step=10)
|
||
try:
|
||
health = api(base_url, "GET", "/health", timeout=5)
|
||
config = health.get("configuration") or {}
|
||
st.success(f"Server beží · LLM: {config.get('llm_model', '?')}")
|
||
except Exception as exc: # noqa: BLE001
|
||
st.error(f"Server nedostupný: {exc}")
|
||
st.caption(
|
||
"Odpovede sú generované jazykovým modelom na základe príbalových letákov ADC "
|
||
"a nenahrádzajú konzultáciu s lekárom alebo lekárnikom."
|
||
)
|
||
|
||
products = load_product_index()
|
||
if products is None:
|
||
st.warning(
|
||
f"Chýba index liekov `{PRODUCT_INDEX.relative_to(ROOT)}` – rozpoznávanie obchodných názvov je vypnuté. "
|
||
"Vytvorte ho príkazom `python scripts/demo/build_product_index.py`."
|
||
)
|
||
aliases, by_prefix, by_url = {}, {}, {}
|
||
else:
|
||
aliases, by_prefix, by_url = build_aliases(products)
|
||
|
||
st.title("💊 Lekárnický asistent nad znalostným grafom ADC")
|
||
st.caption("Indikácie · kontraindikácie · interakcie — odpoveď s odôvodnením zo znalostného grafu a príbalových letákov")
|
||
|
||
tab_ask, tab_explore = st.tabs(["Otázka", "Prieskum grafu"])
|
||
|
||
# ---------------------------------------------------------------- tab: question
|
||
|
||
with tab_ask:
|
||
if "question" not in st.session_state:
|
||
st.session_state.question = EXAMPLES[0]
|
||
cols = st.columns(len(EXAMPLES))
|
||
for col, example in zip(cols, EXAMPLES):
|
||
if col.button(example, width="stretch"):
|
||
st.session_state.question = example
|
||
question = st.text_area("Otázka", key="question", height=80)
|
||
ask = st.button("Opýtať sa", type="primary")
|
||
|
||
if ask and question.strip():
|
||
prepared = prepare_query(question, products or [], aliases, by_prefix)
|
||
matches = prepared["matches"]
|
||
query = prepared["query"]
|
||
payload = {**QUERY_SETTINGS,"query":query,"mode":mode,"top_k":top_k,
|
||
"user_prompt":prepared["user_prompt"]}
|
||
try:
|
||
with st.spinner("Vyhľadávam v znalostnom grafe ..."):
|
||
data = api(base_url, "POST", "/query/data", json=payload)
|
||
with st.spinner("Generujem odpoveď ..."):
|
||
answer = api(base_url, "POST", "/query", json={**payload, "include_references": True, "include_chunk_content": True})
|
||
except Exception as exc: # noqa: BLE001
|
||
st.error(f"Chyba pri dopyte na LightRAG: {exc}")
|
||
st.stop()
|
||
st.session_state.result = {
|
||
"question": question, "query": query, "user_prompt": payload.get("user_prompt"),
|
||
"matches": matches, "data": data, "answer": answer,
|
||
}
|
||
|
||
result = st.session_state.get("result")
|
||
if result:
|
||
matches, data, answer = result["matches"], result["data"].get("data") or {}, result["answer"]
|
||
|
||
if matches:
|
||
st.subheader("Rozpoznané lieky")
|
||
for col, m in zip(st.columns(min(len(matches), 4)), matches):
|
||
with col.container(border=True):
|
||
st.markdown(f"**{m['label']}** \n{'účinná látka' if m['kind'] == 'substance' else 'obchodný názov'}")
|
||
if m["substances"]:
|
||
st.write("Látka: " + ", ".join(m["substances"]))
|
||
if m.get("ambiguous"):
|
||
st.warning("Značka má viac zložení. Uveďte presný prípravok.")
|
||
ex = m["example"]
|
||
st.write(f"{ex['rx'] or ''} \n{ex['form'] or ''}")
|
||
if m["in_graph"]:
|
||
st.success("Index obsahuje klinického reprezentanta")
|
||
else:
|
||
st.warning("Index neobsahuje klinického reprezentanta; dôvod treba overiť v ADC.")
|
||
st.caption(f"{m['count']} balení v ADC · [ADC]({ex['url']})")
|
||
|
||
answer_tab, graph_tab, source_tab, debug_tab = st.tabs(
|
||
["Odpoveď", "Znalostný graf", "Zdroje z ADC", "Detaily dopytu"]
|
||
)
|
||
|
||
with answer_tab:
|
||
st.markdown(answer.get("response", ""))
|
||
refs = answer.get("references") or []
|
||
if refs:
|
||
st.markdown("**Referencie**")
|
||
for ref in refs:
|
||
path = ref.get("file_path", "")
|
||
st.markdown(f"[{ref.get('reference_id')}] [{product_name(path, by_url)}]({path})")
|
||
|
||
with graph_tab:
|
||
entities = data.get("entities") or []
|
||
relations = data.get("relationships") or []
|
||
focus = {normalize(s) for m in matches for s in m["substances"] + [m["label"]]}
|
||
# Keep the most connected entities when the retrieved subgraph is large.
|
||
degree = Counter()
|
||
for r in relations:
|
||
degree[r["src_id"]] += 1
|
||
degree[r["tgt_id"]] += 1
|
||
# Context limits can omit endpoint entities while retaining their edges.
|
||
# Recover endpoint labels from the actual relations, without inventing descriptions.
|
||
present = {e["entity_name"] for e in entities}
|
||
entities = list(entities) + [
|
||
{"entity_name":name,"entity_type":"unknown","description":""}
|
||
for name in degree if name not in present
|
||
]
|
||
ranked = sorted(entities, key=lambda e: (normalize(e["entity_name"]) not in focus, -degree[e["entity_name"]]))
|
||
shown = ranked[:max_graph_nodes]
|
||
st.caption(
|
||
f"Samostatne vyhľadaný podgraf: {len(entities)} entít a {len(relations)} vzťahov "
|
||
f"(zobrazených {len(shown)} uzlov). Prejdite myšou na uzol alebo hranu pre popis."
|
||
)
|
||
graph_legend()
|
||
render_graph(
|
||
[{"id": e["entity_name"], "type": e.get("entity_type"), "description": e.get("description")} for e in shown],
|
||
[{"src": r["src_id"], "tgt": r["tgt_id"], "keywords": r.get("keywords", ""),
|
||
"description": r.get("description", "")} for r in relations],
|
||
focus,
|
||
)
|
||
with st.expander(f"Tabuľka vzťahov ({len(relations)})"):
|
||
st.dataframe(
|
||
[{
|
||
"typ": relation_kind(r.get("keywords", ""), r.get("description", ""))[0],
|
||
"od": r["src_id"], "k": r["tgt_id"],
|
||
"kľúčové slová": r.get("keywords", ""),
|
||
"popis": r.get("description", ""),
|
||
"zdroj": product_name(str(r.get("file_path", "")).split("<SEP>")[0], by_url),
|
||
} for r in relations],
|
||
width="stretch", hide_index=True,
|
||
)
|
||
|
||
with source_tab:
|
||
chunks = [
|
||
{"file_path":ref.get("file_path", ""),"content":content}
|
||
for ref in answer.get("references") or [] for content in ref.get("content") or []
|
||
]
|
||
st.caption(f"{len(chunks)} zdrojových úryvkov vrátených s touto odpoveďou. "
|
||
"Podgraf bol vyhľadaný samostatným volaním a nemusí presne zodpovedať kontextu generovania.")
|
||
if not chunks:
|
||
st.warning("Odpoveď neobsahuje zdrojové úryvky. Jej tvrdenia nemožno overiť z tohto výstupu.")
|
||
grouped = defaultdict(list)
|
||
for chunk in chunks:
|
||
grouped[chunk.get("file_path", "")].append(chunk)
|
||
for path, items in grouped.items():
|
||
product = by_url.get(path)
|
||
title = product["name"] if product else path
|
||
with st.expander(f"📄 {title} ({len(items)} úryvkov)"):
|
||
if product:
|
||
st.markdown(f"[Detail v ADC]({path}) · [Príbalový leták]({product['pil']})")
|
||
for chunk in items:
|
||
st.markdown(f"<div style='white-space:pre-wrap;font-size:0.9em;border-left:3px solid #ccc;"
|
||
f"padding-left:8px;margin-bottom:12px'>{html.escape(chunk.get('content', ''))}</div>",
|
||
unsafe_allow_html=True)
|
||
|
||
with debug_tab:
|
||
meta = result["data"].get("metadata") or {}
|
||
st.write("**Dopyt odoslaný do LightRAG:**")
|
||
st.code(result["query"], language=None)
|
||
if result.get("user_prompt"):
|
||
st.write("**Doplňujúce údaje z ADC pre model:**")
|
||
st.code(result["user_prompt"], language=None)
|
||
st.write("**Kľúčové slová extrahované LLM:**", meta.get("keywords"))
|
||
st.json(meta.get("processing_info") or {})
|
||
|
||
# ---------------------------------------------------------------- tab: explore
|
||
|
||
with tab_explore:
|
||
st.write("Vyhľadajte entitu (liečivo, ochorenie, skupinu ...) a zobrazte jej okolie v znalostnom grafe.")
|
||
term = st.text_input("Hľadať entitu", "warfarín")
|
||
labels = []
|
||
if term.strip():
|
||
try:
|
||
labels = api(base_url, "GET", "/graph/label/search", params={"q": term, "limit": 30}, timeout=30)
|
||
except Exception as exc: # noqa: BLE001
|
||
st.error(f"Chyba: {exc}")
|
||
if labels:
|
||
col1, col2 = st.columns([3, 1])
|
||
label = col1.selectbox("Entita", labels)
|
||
depth = col2.radio("Hĺbka", [1, 2], horizontal=True)
|
||
graph = api(base_url, "GET", "/graphs", params={"label": label, "max_depth": depth, "max_nodes": max_graph_nodes}, timeout=60)
|
||
nodes = [{
|
||
"id": n["id"],
|
||
"type": (n.get("properties") or {}).get("entity_type"),
|
||
"description": (n.get("properties") or {}).get("description"),
|
||
} for n in graph.get("nodes", [])]
|
||
edges = [{
|
||
"src": e["source"], "tgt": e["target"],
|
||
"keywords": (e.get("properties") or {}).get("keywords", ""),
|
||
"description": (e.get("properties") or {}).get("description", ""),
|
||
} for e in graph.get("edges", [])]
|
||
st.caption(f"{len(nodes)} uzlov, {len(edges)} hrán")
|
||
graph_legend()
|
||
render_graph(nodes, edges, {normalize(label)})
|
||
center = next((n for n in nodes if n["id"] == label), None)
|
||
if center:
|
||
st.markdown(f"**{label}** ({center['type']})")
|
||
st.markdown(clean(center["description"], limit=3000))
|
||
elif term.strip():
|
||
st.info("Nenašla sa žiadna entita.")
|