""" app_dashboard.py - Streamlit-Dashboard für Second Brain. Seiten: Übersicht, Engramme, Suche, Graph, Heal-Log, Neural Scorer. """ import json import sys import os from pathlib import Path import streamlit as st _root = Path(__file__).resolve().parent.parent sys.path.insert(0, str(_root)) from src.engram import Engram from src.store import EngramStore from src.chroma_store import ChromaStore from src.retriever import Retriever from src.neural_scorer import NeuralScorer from src.graph_view import generate_graph_html from src.loop_detector import LoopDetector from src.error_healer import ErrorHealer _DEFAULT_DB = _root / "data" / "brain.sqlite" @st.cache_resource def _store(): return EngramStore(str(_DEFAULT_DB)) @st.cache_resource def _chroma(): p = Path(str(_DEFAULT_DB)).parent / "chroma" return ChromaStore(str(p)) _retriever_cache = None def _retriever(): global _retriever_cache if _retriever_cache is None: _retriever_cache = Retriever(_store(), _chroma()) return _retriever_cache @st.cache_resource def _scorer(): return NeuralScorer() @st.cache_resource def _healer(): return ErrorHealer(_store()) st.set_page_config(page_title="Second Brain Dashboard", layout="wide") st.title("🧠 2.Brain v0.3.1") page = st.sidebar.radio("Seite", ["Übersicht", "Engramme", "Suche", "Graph", "Heal-Log", "Neural Scorer"]) if page == "Übersicht": store = _store() engrams = store.get_all(limit=10000) confirmed = sum(1 for e in engrams if e.correctness.confirmed) unconfirmed = len(engrams) - confirmed avg_conf = sum(e.compute_confidence() for e in engrams) / max(1, len(engrams)) errors = [e for e in engrams if "error" in e.metadata.get("tags", [])] c1, c2, c3, c4, c5 = st.columns(5) c1.metric("Total", len(engrams)) c2.metric("Confirmed", confirmed) c3.metric("Pending", unconfirmed) c4.metric("Avg Confidence", f"{avg_conf:.2f}") c5.metric("Errors", len(errors)) st.subheader("Recent Engramme") for eg in sorted(engrams, key=lambda e: e.metadata.get("modified", ""), reverse=True)[:5]: valid = eg.validate_grounding() marker = "✅" if valid["valid"] else "⚠️" with st.expander(f"{marker} {eg.content[:80]}..."): st.write(f"ID: `{eg.id}`") st.write(f"Source: {eg.metadata.get('source')}") st.write(f"Confidence: {eg.compute_confidence():.2f}") st.write(f"Confirmed: {'✅' if eg.correctness.confirmed else '❓'}") st.write("Tags:", ", ".join(eg.metadata.get("tags", []))) if not valid["valid"]: st.warning(f"Grounding: {valid['issue']}") if st.button("Auto-Fix", key=f"af_{eg.id}"): eg.auto_fix_grounding() store.save(eg) st.experimental_rerun() elif page == "Engramme": store = _store() st.subheader("Alle Engramme (max 1000)") tag_filter = st.text_input("Filter tags") source_filter = st.selectbox("Source", ["alle", "user", "agent", "web", "file", "system"]) for eg in store.get_all(limit=1000): tags = eg.metadata.get("tags", []) src = eg.metadata.get("source", "") if tag_filter and tag_filter not in tags: continue if source_filter != "alle" and source_filter != src: continue col1, col2 = st.columns([4, 1]) with col1: conf = eg.compute_confidence() marker = "✅" if conf > 0.7 else "⚠️" st.markdown(f"{marker} **{eg.content[:100]}**") st.caption(f"Conf: {conf:.2f} | Tags: {', '.join(tags)} | Source: {src}") with col2: if st.button("✅ Confirm", key=f"conf_{eg.id}"): eg.correctness.confirm("user") store.save(eg) st.success("Confirmed") if st.button("❌ Reject", key=f"rej_{eg.id}"): eg.correctness.reject("user") store.save(eg) st.warning("Rejected") st.divider() elif page == "Suche": st.subheader("Hybrid Search (Semantic + Keyword)") query = st.text_input("Query", placeholder="Suchbegriff eingeben...") mode = st.radio("Modus", ["Hybrid", "Keyword", "Semantic"], horizontal=True) if st.button("Suchen") and query: ret = _retriever() results = ret.hybrid_retrieve(query, limit=10) if mode == "Hybrid" else \ ret.semantic_retrieve(query, limit=10) if mode == "Semantic" else \ ret.retrieve(query, limit=10) if not results: st.info("Keine Ergebnisse gefunden.") for r in results: eg = r["engram"] with st.container(): st.markdown(f"**{eg.content[:200]}...**") st.write(f"Score: `{r['score']:.3f}` | Match: `{r['match_type']}` | Conf: `{eg.compute_confidence():.2f}`") c1, c2 = st.columns(2) if c1.button("✅ Confirm", key=f"sc_{eg.id}"): eg.correctness.confirm("user") _store().save(eg) st.success("Confirmed") if c2.button("❌ Reject", key=f"sr_{eg.id}"): eg.correctness.reject("user") _store().save(eg) st.warning("Rejected") elif page == "Graph": st.subheader("Graph-Visualisierung") graph_html_path = Path(str(_DEFAULT_DB)).parent / "graph_view.html" if st.button("Graph neu generieren"): with st.spinner("Generiere Graph..."): path = generate_graph_html(_store(), str(graph_html_path)) st.success(f"Graph generiert: {path}") if graph_html_path.exists(): with open(graph_html_path, "r", encoding="utf-8") as f: html = f.read() st.components.v1.html(html, height=800) else: st.info("Graph noch nicht generiert. Klicke oben.") elif page == "Heal-Log": st.subheader("Error Healing & Loop Detection") healer = _healer() stats = healer.get_error_stats() c1, c2, c3 = st.columns(3) c1.metric("Total Errors", stats["total_errors"]) c2.metric("Repeated", stats["repeated_errors"]) c3.metric("Error Types", len(stats.get("error_types", {}))) st.subheader("Error Types") for etype, count in stats.get("error_types", {}).items(): st.write(f"- **{etype}**: {count}") st.subheader("Loop-Checker") q = st.text_input("Query") r = st.text_input("Response") if st.button("Check Loop") and q and r: detector = LoopDetector() result = detector.check(q, r) st.json(result) if result["loop_detected"]: st.error(result["suggestion"]) elif page == "Neural Scorer": st.subheader("Neural Scorer Training") scorer = _scorer() store = _store() engrams = store.get_all(limit=10000) labeled = [e for e in engrams if e.correctness.confirmed or e.correctness.rejections > 0] st.write(f"Labelled Engramme: **{len(labeled)}**") if st.button("Train Neural Scorer"): if len(labeled) < 2: st.error("Mindestens 2 labelierte Engramme nötig (confirm + reject).") else: with st.spinner("Training läuft..."): result = scorer.train(labeled, epochs=30) st.json(result) st.success("Training abgeschlossen!") if st.button("Predict All"): for eg in engrams[:20]: pred = scorer.predict(eg) st.write(f"{eg.content[:50]}... → **{pred:.3f}**")