From 1a5cc72067c4958baa9045e0d5fa3e5caebbfe17 Mon Sep 17 00:00:00 2001 From: Oleh Poiasnik Date: Mon, 2 Nov 2026 16:14:32 +0100 Subject: [PATCH] Add pharmacy agent and Streamlit web demo Recognises products by trade name or substance (Slovak inflection aware), queries LightRAG and shows the answer with the retrieved subgraph and ADC source texts. --- ask_aclasta.py | 84 +++++++ demo/app.py | 367 ++++++++++++++++++++++++++++ demo/requirements.txt | 3 + pharmacy_agent.py | 138 +++++++++++ scripts/demo/build_product_index.py | 87 +++++++ 5 files changed, 679 insertions(+) create mode 100644 ask_aclasta.py create mode 100644 demo/app.py create mode 100644 demo/requirements.txt create mode 100644 pharmacy_agent.py create mode 100644 scripts/demo/build_product_index.py diff --git a/ask_aclasta.py b/ask_aclasta.py new file mode 100644 index 0000000..f3a3a41 --- /dev/null +++ b/ask_aclasta.py @@ -0,0 +1,84 @@ + + +"""Ask the local LightRAG server a question that exists in the current graph. + +Usage: + python ask_aclasta.py +""" + +from __future__ import annotations + +import argparse +import json +import sys +import urllib.error +import urllib.request +from pathlib import Path + +from pharmacy_agent import QUERY_SETTINGS, prepare_query + + +DEFAULT_URL = "http://localhost:9621" +DEFAULT_MODE = "mix" +DEFAULT_QUESTION = "Ake su kontraindikacie lieku ACARIZAX?" + + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(encoding="utf-8") + + +def post_query(base_url: str, question: str, mode: str, timeout: int) -> dict: + index = Path(__file__).resolve().parent / "outputs/demo/product_index.json" + products = json.loads(index.read_text(encoding="utf-8")) if index.exists() else [] + prepared = prepare_query(question, products) + payload = json.dumps( + {**QUERY_SETTINGS, "query": prepared["query"], "user_prompt": prepared["user_prompt"], + "mode": mode, "include_references": True, "include_chunk_content": True}, + ensure_ascii=False, + ).encode("utf-8") + request = urllib.request.Request( + f"{base_url.rstrip('/')}/query", + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Query LightRAG about ACARIZAX contraindications.") + parser.add_argument("--url", default=DEFAULT_URL, help="LightRAG server URL.") + parser.add_argument("--mode", default=DEFAULT_MODE, choices=("mix", "hybrid", "local", "global", "naive")) + parser.add_argument("--timeout", type=int, default=120) + parser.add_argument("--question", default=DEFAULT_QUESTION) + args = parser.parse_args() + + try: + result = post_query(args.url, args.question, args.mode, args.timeout) + except urllib.error.URLError as exc: + raise SystemExit( + "LightRAG server is not reachable. Start it first with:\n" + " python start_servers.py\n\n" + f"Details: {exc}" + ) from exc + except json.JSONDecodeError as exc: + raise SystemExit(f"LightRAG returned invalid JSON: {exc}") from exc + + response = result.get("response") + if response is None: + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + print(response) + references = result.get("references") or [] + if references: + print("\nZdroje:") + for ref in references: + print(f"[{ref.get('reference_id', '?')}] {ref.get('file_path', '')}") + + +if __name__ == "__main__": + main() diff --git a/demo/app.py b/demo/app.py new file mode 100644 index 0000000..346b90b --- /dev/null +++ b/demo/app.py @@ -0,0 +1,367 @@ +"""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 ; show them as paragraphs.""" + text = str(text or "").replace("", "\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("", """""") + st.iframe(graph_html, height=height + 20) + + +def graph_legend() -> None: + entity_html = " ".join( + f"{t}" + for t, c in ENTITY_COLORS.items() + ) + relation_html = " ".join( + f"{label}" for _, label, c in RELATION_KINDS + ) + st.markdown(f"Uzly: {entity_html}
Hrany: {relation_html}", 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("")[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"
{html.escape(chunk.get('content', ''))}
", + 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.") diff --git a/demo/requirements.txt b/demo/requirements.txt new file mode 100644 index 0000000..56deb51 --- /dev/null +++ b/demo/requirements.txt @@ -0,0 +1,3 @@ +streamlit>=1.40 +pyvis>=0.3.2 +requests>=2.31 diff --git a/pharmacy_agent.py b/pharmacy_agent.py new file mode 100644 index 0000000..33f4c68 --- /dev/null +++ b/pharmacy_agent.py @@ -0,0 +1,138 @@ +"""Shared, offline-testable product recognition for the demo and evaluation.""" +from __future__ import annotations +import re +import unicodedata +from collections import Counter, defaultdict + +FORM_WORDS = { + "tbl", "cps", "sol", "crm", "gel", "ung", "inj", "inf", "plv", "sus", "sir", "gtt", "spr", + "flm", "obd", "eff", "ret", "pro", "por", "ijf", "dis", "lyo", "aer", "emp", "pst", "liq", +} + + +def normalize(text: str) -> str: + """Lowercase, strip diacritics and punctuation (Slovak inflection is handled by prefix matching).""" + text = unicodedata.normalize("NFKD", text) + text = "".join(ch for ch in text if not unicodedata.combining(ch)) + return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip() + + +def build_aliases(_products: list[dict]) -> tuple[dict, dict, dict]: + """Map normalized names to products. + + Returns (aliases, by_prefix, by_url): aliases maps an alias to + {"kind": "substance"|"product", "label": str, "products": [index, ...]}. + """ + aliases: dict[str, dict] = {} + substance_products: dict[str, list[int]] = defaultdict(list) + brand_products: dict[str, list[int]] = defaultdict(list) + + for i, product in enumerate(_products): + for substance in product["subs"]: + key = normalize(substance) + if len(key) >= 4: + substance_products[key].append(i) + tokens = normalize(product["name"]).split() + brand = [] + for token in tokens: + if any(ch.isdigit() for ch in token) or token in FORM_WORDS: + if brand: + break + continue # leading "3M", "0 9 %" ... are not the brand + brand.append(token) + if brand and len(brand[0]) >= 3: + brand_products[brand[0]].append(i) + if len(brand) >= 2: + brand_products[" ".join(brand[:2])].append(i) + + for key, ids in substance_products.items(): + label = Counter(s for i in ids for s in _products[i]["subs"] if normalize(s) == key).most_common(1)[0][0] + aliases[key] = {"kind": "substance", "label": label, "products": ids} + for key, ids in brand_products.items(): + if key in aliases: + continue + # A word shared by many unrelated products (e.g. "vitamin") is not a brand. + if len({tuple(_products[i]["subs"]) for i in ids}) > 6: + continue + aliases[key] = {"kind": "product", "label": key.title(), "products": ids} + + by_prefix: dict[str, list[str]] = defaultdict(list) + for key in aliases: + by_prefix[key[:4]].append(key) + by_url = {p["url"]: p for p in _products if p["url"]} + return aliases, by_prefix, by_url + + +def detect_drugs(question: str, aliases: dict, by_prefix: dict) -> list[dict]: + """Find products and active substances mentioned in the question. + + Slovak nouns are inflected (warfarínom, ibuprofénu), so a question word + matches an alias it starts with, allowing a short ending. + """ + tokens = normalize(question).split() + found, used = {}, set() + for n in (2, 1): + for i in range(len(tokens) - n + 1): + if any(j in used for j in range(i, i + n)): + continue + phrase = " ".join(tokens[i:i + n]) + for alias in by_prefix.get(phrase[:4], []): + if len(alias.split()) != n: + continue + exact = phrase == alias + inflected = len(alias) >= 5 and phrase.startswith(alias) and len(phrase) - len(alias) <= 4 + if exact or inflected: + found[alias] = aliases[alias] + used.update(range(i, i + n)) + break + return list(found.values()) + + +def describe_match(match: dict, products: list[dict]) -> dict: + items = [products[i] for i in match["products"]] + substances = Counter(tuple(p["subs"]) for p in items if p["subs"]).most_common() + covered = [p for p in items if p["rep"]] + example = (covered or items)[0] + return { + "label": match["label"], + "kind": match["kind"], + "substances": list(substances[0][0]) if len(substances) == 1 else [], + "ambiguous": len(substances) > 1, + "substance_variants": [list(s) for s, _ in substances], + "count": len(items), + "in_graph": bool(covered), + "example": example, + } + + +BASE_PROMPT = ( + "Odpovedaj po slovensky a iba v rozsahu dostupných dôkazov. " + "Rozlišuj indikácie, kontraindikácie, upozornenia a interakcie. " + "Neprítomnosť informácie neznamená neprítomnosť interakcie ani bezpečnosť kombinácie. " + "Uvádzaj zdroje k tvrdeniam, ak ich kontext poskytuje. Ak dôkaz chýba, priznaj to. " + "Nehádaj silu, liekovú formu ani zloženie nejednoznačnej značky. " + "Údaje z reprezentanta rovnakej účinnej látky nemusia pokrývať konkrétny prípravok." +) + +# Reserve space for original ADC chunks after graph context and prompt overhead. +QUERY_SETTINGS = {"top_k":40,"chunk_top_k":10,"max_total_tokens":8000, + "max_entity_tokens":1000,"max_relation_tokens":1500,"enable_rerank":False} + + +def prepare_query(question, products, aliases=None, by_prefix=None): + if aliases is None or by_prefix is None: + aliases, by_prefix, _ = build_aliases(products) + matches = [describe_match(m, products) for m in detect_drugs(question, aliases, by_prefix)] + hints, notes = [], [] + for match in matches: + if match["ambiguous"]: + notes.append(f"Názov {match['label']} je nejednoznačný; vyžiadaj presný názov a zloženie.") + elif match["substances"]: + hints.append(f"{match['label']} = účinná látka {', '.join(match['substances'])}") + if not match["in_graph"]: + ex = match["example"] + notes.append(f"Pre {ex['name']} nie je v indexe reprezentant s klinickými údajmi; " + "príčinou môže byť chýbajúci leták, parsovanie alebo výber dokumentov. " + f"Základný údaj ADC: {ex.get('ind','')}. Chýbajúce údaje neznamenajú bezpečnosť.") + query = question + ("\n\n(Lieky v otázke: " + "; ".join(hints) + ")" if hints else "") + return {"query":query,"matches":matches,"user_prompt":BASE_PROMPT + "\n" + "\n".join(notes)} diff --git a/scripts/demo/build_product_index.py b/scripts/demo/build_product_index.py new file mode 100644 index 0000000..5615d70 --- /dev/null +++ b/scripts/demo/build_product_index.py @@ -0,0 +1,87 @@ +"""Build a compact index of all ADC products for the web demo. + +The demo cannot load the 800 MB structured dataset, so this script extracts +the fields needed to recognise a product in a question (name, active +substances, form, prescription status, ...) and links every product to the +LightRAG document that represents its active substance set (`rep`). Products +without a representative have no clinical data in the knowledge graph. + + python scripts/demo/build_product_index.py +""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections import Counter +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "scripts" / "lightrag_ingest")) +from ingest_adc_structured import DEFAULT_INPUT, dedupe_key, load_json, write_json # noqa: E402 + +DEFAULT_SECTION_DOCS = Path("outputs/lightrag_sections/adc_section_docs.json") +DEFAULT_OUTPUT = Path("outputs/demo/product_index.json") + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") + + +def as_list(value) -> list[str]: + if not value: + return [] + if isinstance(value, str): + return [value] + return [str(item) for item in value if str(item).strip()] + + +def compact(record: dict, representative: str | None) -> dict: + product = record.get("product") or {} + hints = record.get("graph_hints") or {} + urls = record.get("urls") or {} + fields = product.get("detail_fields") or {} + classes = [c.get("name") for c in hints.get("classification_codes") or [] if isinstance(c, dict)] + return { + "name": product.get("name") or hints.get("drug") or "", + "url": urls.get("detail") or "", + "pil": urls.get("pil") or "", + "subs": as_list(hints.get("active_substances")), + "form": hints.get("dosage_form") or "", + "rx": fields.get("vydaj") or "", + "type": fields.get("typ_produktu") or "", + "group": classes[-1] if classes else "", + "sukl": hints.get("sukl_code") or "", + "ind": str(hints.get("indications_text") or "")[:500], + "rep": representative, + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--input", type=Path, default=DEFAULT_INPUT) + parser.add_argument("--section-docs", type=Path, default=DEFAULT_SECTION_DOCS) + parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) + args = parser.parse_args() + + # Same grouping as build_section_docs.py: group key -> URL of the uploaded document. + representatives = { + dedupe_key(doc, "substance"): doc["urls"]["detail"] for doc in load_json(args.section_docs) + } + + print(f"Loading {args.input} ...") + records = load_json(args.input) + index = [] + stats: Counter = Counter() + for record in records: + rep = representatives.get(dedupe_key(record, "substance")) + stats["in_graph" if rep else "not_in_graph"] += 1 + index.append(compact(record, rep)) + + write_json(args.output, index) + print(f"Products: {len(index)} | {dict(stats)}") + print(f"Saved: {args.output}") + + +if __name__ == "__main__": + main()