| python |
| import os, io, re |
| import streamlit as st |
| from pdfminer.high_level import extract_text |
| from sentence_transformers import SentenceTransformer |
| from qdrant_client import QdrantClient |
| from qdrant_client.http import models as qmodels |
| import requests |
|
|
| st.set_page_config(page_title="EDM & Fanuc – Hybrid RAG", layout="wide") |
| st.title("🔧 EDM & Fanuc – Hybrid RAG (PDF + Web)") |
|
|
| |
| with st.sidebar: |
| st.header("🔐 Secrets") |
| QDRANT_URL = st.text_input("QDRANT_URL", value=os.getenv("QDRANT_URL", "")) |
| QDRANT_API_KEY = st.text_input("QDRANT_API_KEY", value=os.getenv("QDRANT_API_KEY", ""), type="password") |
| TAVILY_API_KEY = st.text_input("TAVILY_API_KEY (optional)", value=os.getenv("TAVILY_API_KEY", ""), type="password") |
|
|
| st.header("⚙️ Settings") |
| collection = st.text_input("Qdrant collection", value="docs") |
| topic = st.selectbox("Topic tag", ["EDM","FANUC","general"], index=0) |
| chunk_chars = st.slider("Chunk size (chars)", 800, 2500, 1500, 100) |
| overlap = st.slider("Overlap (chars)", 0, 400, 200, 20) |
| topk = st.slider("Top-K passages", 1, 8, 5, 1) |
|
|
| |
| @st.cache_resource(show_spinner=False) |
| def get_embedder(): |
| return SentenceTransformer("intfloat/multilingual-e5-small") |
|
|
| @st.cache_resource(show_spinner=False) |
| def get_qdrant(url, key): |
| return QdrantClient(url=url, api_key=key) if url else None |
|
|
| |
| def chunk_text(text: str, size: int, ov: int): |
| text = re.sub(r"\s+", " ", text) |
| chunks, i = [], 0 |
| step = max(1, size - ov) |
| while i < len(text): |
| chunk = text[i:i+size] |
| if chunk.strip(): |
| chunks.append(chunk) |
| i += step |
| return chunks |
|
|
| def ensure_collection(client: QdrantClient, coll: str, dim: int = 384): |
| try: |
| client.get_collection(coll) |
| except Exception: |
| client.recreate_collection( |
| collection_name=coll, |
| vectors_config=qmodels.VectorParams(size=dim, distance=qmodels.Distance.COSINE) |
| ) |
|
|
| def embed_passages(texts): |
| emb = get_embedder() |
| return emb.encode([f"passage: {t}" for t in texts], normalize_embeddings=True).tolist() |
|
|
| def embed_query(text: str): |
| emb = get_embedder() |
| return emb.encode([f"query: {text}"], normalize_embeddings=True)[0].tolist() |
|
|
| def upsert_chunks(client, coll, chunks, topic): |
| vectors = embed_passages(chunks) |
| points = [] |
| for c, v in zip(chunks, vectors): |
| points.append(qmodels.PointStruct(vector=v, payload={"text": c, "topic": topic})) |
| client.upsert(collection_name=coll, points=points) |
| return len(points) |
|
|
| def search_local(client, coll, query, topk, filt): |
| qvec = embed_query(query) |
| qfilter = None |
| if filt: |
| must = [qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=v)) for k,v in filt.items()] |
| qfilter = qmodels.Filter(must=must) |
| res = client.search(collection_name=coll, query_vector=qvec, limit=topk, query_filter=qfilter) |
| hits = [{"score": r.score, "text": r.payload.get("text",""), "payload": r.payload} for r in res] |
| return hits |
|
|
| def web_search(query: str, n: int = 3): |
| if not TAVILY_API_KEY: |
| return [] |
| r = requests.post("https://api.tavily.com/search", json={ |
| "api_key": TAVILY_API_KEY, |
| "query": query, |
| "search_depth": "basic", |
| "max_results": n |
| }, timeout=30) |
| data = r.json() |
| out = [] |
| for item in data.get("results", [])[:n]: |
| text = (item.get("title","") + "\n" + item.get("content",""))[:1500] |
| out.append({"url": item.get("url"), "text": text}) |
| return out |
|
|
| |
| ingest_tab, chat_tab = st.tabs(["📥 Ingest (PDF)", "💬 Chat"]) |
|
|
| with ingest_tab: |
| st.subheader("Nahraj PDF (EDM / Fanuc)") |
| uploaded = st.file_uploader("Vyber PDF súbory", type=["pdf"], accept_multiple_files=True) |
| if st.button("Indexovať do Qdrant"): |
| if not (QDRANT_URL and QDRANT_API_KEY and uploaded): |
| st.error("Vyplň Qdrant URL/API key a nahraj aspoň 1 PDF.") |
| else: |
| client = get_qdrant(QDRANT_URL, QDRANT_API_KEY) |
| ensure_collection(client, collection) |
| total_chunks = 0 |
| with st.spinner("Spracúvam PDF…"): |
| for f in uploaded: |
| text = extract_text(io.BytesIO(f.getvalue())) |
| chunks = chunk_text(text, chunk_chars, overlap) |
| total_chunks += upsert_chunks(client, collection, chunks, topic) |
| st.success(f"Hotovo. Uložených chunkov: {total_chunks}") |
|
|
| with chat_tab: |
| st.subheader("Otázka") |
| q = st.text_area("Zadaj otázku (EDM/Fanuc)", height=100) |
| do_web = st.checkbox("Povoliť web doplnenie (Tavily)", value=True) |
| if st.button("Spýtať sa"): |
| if not (QDRANT_URL and QDRANT_API_KEY and q): |
| st.error("Vyplň Qdrant URL/API key a otázku.") |
| else: |
| client = get_qdrant(QDRANT_URL, QDRANT_API_KEY) |
| filt = {"topic": topic} |
| hits = search_local(client, collection, q, topk, filt) |
| have_good_local = any(h.get("score",0) >= 0.35 for h in hits) |
| web_snips = web_search(q, n=3) if (do_web and not have_good_local) else [] |
|
|
| st.markdown("### Výsledky z PDF") |
| for i,h in enumerate(hits): |
| st.write(f"[{i+1}] (score {h['score']:.2f}) {h['text'][:500]}…") |
|
|
| if web_snips: |
| st.markdown("### Výsledky z webu") |
| for j,w in enumerate(web_snips): |
| st.write(f"[W{j+1}] {w['url']} — {w['text'][:300]}…") |
|
|
| if not hits and not web_snips: |
| st.warning("Nenašli sa žiadne relevantné zdroje.") |
