diff --git a/kazeia_central/app.py b/kazeia_central/app.py index 6d29175..235bbe3 100644 --- a/kazeia_central/app.py +++ b/kazeia_central/app.py @@ -24,6 +24,7 @@ from .core.operator_router import make_router as operator_router from .core.store import Store from .features.medical.conversations.router import make_router as conversations_router from .features.medical.patients.router import make_router as patients_router +from .features.medical.rag.router import make_router as rag_router from .features.technical.fleet.router import make_router as fleet_router from .features.technical.voices.autosync import VoiceAutoSync from .features.technical.voices.bridge import VoiceBridge @@ -35,7 +36,8 @@ _OPERATOR = os.environ.get("KAZEIA_USER") or "local" # Routers de features à composer (ordre libre ; les littéraux vs /{serial} sont gérés # DANS chaque router). Ajouter une feature = ajouter sa fabrique ici. -_FEATURES = (operator_router, fleet_router, voices_router, patients_router, conversations_router) +_FEATURES = (operator_router, fleet_router, voices_router, patients_router, + conversations_router, rag_router) def _install_basic_auth(app: FastAPI) -> None: diff --git a/kazeia_central/core/provider/client.py b/kazeia_central/core/provider/client.py index d452760..97357cc 100644 --- a/kazeia_central/core/provider/client.py +++ b/kazeia_central/core/provider/client.py @@ -144,6 +144,11 @@ class ProviderClient: def rag_sync(self) -> dict: return self.rpc("rag_sync") + def rag_delete(self, source: str) -> None: + """Supprime une fiche RAG (content delete /rag/{source}). Diffuse RELOAD_RAG.""" + from ..adb.client import PROVIDER_URI + self.adb.shell(f"content delete --uri {PROVIDER_URI}/rag/{source}", serial=self.serial) + # conversations (file-drop NDJSON — pull puis export_purge, PLAINTEXT MVP) def conversations_export(self, *, profile_id: str | None = None, session_id: str | None = None, since: int | None = None, diff --git a/kazeia_central/features/medical/rag/__init__.py b/kazeia_central/features/medical/rag/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kazeia_central/features/medical/rag/router.py b/kazeia_central/features/medical/rag/router.py new file mode 100644 index 0000000..b6f54c2 --- /dev/null +++ b/kazeia_central/features/medical/rag/router.py @@ -0,0 +1,51 @@ +"""Router RAG thérapeutique (médical) : corpus + authoring + test retrieval.""" + +from __future__ import annotations + +from fastapi import APIRouter +from pydantic import BaseModel + +from kazeia_central.core.context import Ctx +from . import service as svc + + +class _Doc(BaseModel): + text: str + + +class _Query(BaseModel): + text: str + + +def make_router(ctx: Ctx) -> APIRouter: + r = APIRouter(prefix="/api/rag", tags=["rag"]) + g, now = ctx.guard, ctx.now_ms + + # Littéraux (status) et sous-chemins avant /{serial} nu — gérés par la profondeur. + @r.get("/{serial}/status") + def rag_status(serial: str): + return g(lambda: svc.status(ctx.adb, serial)) + + @r.get("/{serial}") + def rag_docs(serial: str): + return g(lambda: svc.list_docs(ctx.adb, serial)) + + @r.get("/{serial}/doc/{source}") + def rag_get(serial: str, source: str): + return g(lambda: svc.get_doc(ctx.adb, serial, source)) + + @r.put("/{serial}/doc/{source}") + def rag_put(serial: str, source: str, body: _Doc): + return g(lambda: svc.upsert_doc(ctx.adb, ctx.store, serial, source, body.text, + now=now(), actor=ctx.operator)) + + @r.delete("/{serial}/doc/{source}") + def rag_delete(serial: str, source: str): + return g(lambda: svc.delete_doc(ctx.adb, ctx.store, serial, source, + now=now(), actor=ctx.operator)) + + @r.post("/{serial}/query") + def rag_query(serial: str, body: _Query): + return g(lambda: svc.query(ctx.adb, serial, body.text)) + + return r diff --git a/kazeia_central/features/medical/rag/service.py b/kazeia_central/features/medical/rag/service.py new file mode 100644 index 0000000..168de42 --- /dev/null +++ b/kazeia_central/features/medical/rag/service.py @@ -0,0 +1,50 @@ +"""Service RAG thérapeutique (médical) : authoring du corpus clinique au clavier depuis +le PC + test de retrieval, via le RPC base64-JSON (le texte des fiches est multiligne +avec `:` → passe par rag_dump/rag_upsert, jamais par content query/--bind). + +Périmètre : gestion du corpus **d'une tablette** (lecture, édition, suppression, test). +La distribution fleet-wide du corpus (une écriture → N tablettes) relèvera de l'OTA. +""" + +from __future__ import annotations + +from typing import Any + +from kazeia_central.core.adb import Adb +from kazeia_central.core.provider import ProviderClient +from kazeia_central.core.store import Store + + +def status(adb: Adb, serial: str) -> dict[str, Any]: + return ProviderClient(adb, serial).rag_status().model_dump() + + +def list_docs(adb: Adb, serial: str) -> list[dict[str, Any]]: + """Index du corpus (source, char_count, chunk_count, updated_at).""" + return [d.model_dump() for d in ProviderClient(adb, serial).rag_index()] + + +def get_doc(adb: Adb, serial: str, source: str) -> dict[str, Any]: + """Texte complet d'une fiche (multiligne, `:`) via rag_dump (pas de corruption).""" + return ProviderClient(adb, serial).rag_dump(source) + + +def upsert_doc(adb: Adb, store: Store, serial: str, source: str, text: str, *, + now: int, actor: str | None = None) -> dict[str, Any]: + """Crée/met à jour une fiche (ré-ingérée côté device) via rag_upsert.""" + res = ProviderClient(adb, serial).rag_upsert(source, text) + store.audit("rag_upsert", actor=actor, target=f"{serial}/{source}", + detail={"chars": len(text)}, now=now) # pas le texte (contenu clinique) + return {"source": source, "chars": len(text), **(res if isinstance(res, dict) else {})} + + +def delete_doc(adb: Adb, store: Store, serial: str, source: str, *, + now: int, actor: str | None = None) -> dict[str, Any]: + ProviderClient(adb, serial).rag_delete(source) + store.audit("rag_delete", actor=actor, target=f"{serial}/{source}", now=now) + return {"source": source, "deleted": True} + + +def query(adb: Adb, serial: str, text: str) -> list[dict[str, Any]]: + """Test de retrieval : renvoie les chunks les plus proches (source, score, text).""" + return [h.model_dump() for h in ProviderClient(adb, serial).rag_query(text)] diff --git a/kazeia_central/web/app.js b/kazeia_central/web/app.js index 589d019..6c5496f 100644 --- a/kazeia_central/web/app.js +++ b/kazeia_central/web/app.js @@ -34,7 +34,7 @@ const SECTIONS = [ { group: "⚕️ Médical / clinique", domain: "med" }, { id: "patients", ic: "👤", label: "Patients & profils", render: renderPatients }, { id: "conversations", ic: "💬", label: "Conversations", render: renderConversations }, - { id: "rag", ic: "📚", label: "RAG thérapeutique", render: () => stub("Corpus RAG thérapeutique", "Authoring du corpus clinique + test retrieval + publication OTA. Dépend de rag_upsert_json côté app patiente.") }, + { id: "rag", ic: "📚", label: "RAG thérapeutique", render: renderRag }, { id: "questionnaires", ic: "📝", label: "Questionnaires", render: () => stub("Questionnaires cliniques", "Passation d'échelles/questionnaires cliniques au patient et suivi des scores dans le temps. À concevoir avec toi.") }, ]; @@ -320,5 +320,65 @@ async function purgeSession(btn, serial, sid) { await withBusy(btn, async () => { try { const r = await api(`/conversations/${serial}/purge`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ session_id: sid }) }); toast(`${r.deleted} tour(s) purgé(s)`, "ok"); renderSection(); } catch (e) { toast(e.message, "err"); } }); } +// ---- RAG thérapeutique (médical) ------------------------------------------ +async function renderRag(c) { + const serial = S.selected; + if (!serial) { c.innerHTML = pageHead("RAG thérapeutique") + `
📚

Choisis une tablette (en haut).

`; return; } + const [st, docs] = await Promise.all([ + api(`/rag/${serial}/status`).catch(e => ({ error: e.message })), + api(`/rag/${serial}`).catch(e => ({ error: e.message })), + ]); + const statusCard = card("État du corpus", st.error ? `
${esc(st.error)}
` : + `
index${dot(st.ready ? "ok" : "idle")} ${st.ready ? "prêt" : "non prêt"}
` + + kv("modèle", esc(st.model || "—")) + kv("dimension", st.dim ?? "—") + kv("documents", st.doc_count ?? 0) + kv("chunks", st.chunk_count ?? 0)); + + const ds = Array.isArray(docs) ? docs : []; + const docsCard = card(`Corpus (${ds.length} fiches)`, docs.error ? `
${esc(docs.error)}
` : + (ds.length ? ` + ${ds.map(d => ` + `).join("")} +
sourcecaractèreschunks
${esc(d.source)}${d.char_count ?? "—"}${d.chunk_count ?? "—"} +
` : `
corpus vide
`) + + `
`, "full"); + + const testCard = card("Test de retrieval", `
`, "full"); + + c.innerHTML = pageHead("RAG thérapeutique", esc(deviceLabel(serial)) + " · corpus clinique") + `
${statusCard}${docsCard}
${testCard}
`; +} + +async function newDoc(serial) { showDocEditor(serial, "", ""); } +async function openDoc(serial, src) { + const source = decodeURIComponent(src); + try { const d = await api(`/rag/${serial}/doc/${src}`); showDocEditor(serial, d.source, d.text || ""); } + catch (e) { toast(e.message, "err"); } +} +function showDocEditor(serial, source, text) { + const host = $("#ragedit"); host.className = "card full"; + host.innerHTML = `

${source ? "Éditer" : "Nouvelle fiche"}

+ + +
`; + host.scrollIntoView({ behavior: "smooth", block: "nearest" }); +} +async function saveDoc(btn, serial) { + const source = ($("#rsrc").value || "").trim(), text = $("#rtext").value; + if (!source) { toast("nomme la source", "err"); return; } + await withBusy(btn, async () => { try { await api(`/rag/${serial}/doc/${encodeURIComponent(source)}`, { method: "PUT", headers: { "content-type": "application/json" }, body: JSON.stringify({ text }) }); toast(`fiche « ${source} » enregistrée + ré-indexée`, "ok"); renderSection(); } catch (e) { toast(e.message, "err"); } }); +} +async function deleteDoc(btn, serial, src) { + if (!confirm(`Supprimer la fiche « ${decodeURIComponent(src)} » du corpus ?`)) return; + await withBusy(btn, async () => { try { await api(`/rag/${serial}/doc/${src}`, { method: "DELETE" }); toast("fiche supprimée", "ok"); renderSection(); } catch (e) { toast(e.message, "err"); } }); +} +async function ragQuery(btn, serial) { + const q = $("#ragq").value.trim(); if (!q) return; + await withBusy(btn, async () => { + try { + const hits = await api(`/rag/${serial}/query`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ text: q }) }); + $("#raghits").innerHTML = hits.length ? hits.map(h => `
${esc(h.source)} · score ${(h.score ?? 0).toFixed ? h.score.toFixed(3) : h.score}
${esc((h.text || "").slice(0, 220))}${(h.text || "").length > 220 ? "…" : ""}
`).join("") : `
aucun résultat
`; + } catch (e) { toast(e.message, "err"); } + }); +} + // ---- store ---------------------------------------------------------------- async function lockStore() { try { await api("/lock", { method: "POST" }); S.unlocked = false; renderGate(); } catch (e) { toast(e.message, "err"); } } diff --git a/tests/test_rag.py b/tests/test_rag.py new file mode 100644 index 0000000..7353aa6 --- /dev/null +++ b/tests/test_rag.py @@ -0,0 +1,85 @@ +"""Tests feature RAG : authoring corpus (rag_upsert/dump/delete) + retrieval (mockés).""" + +import base64 +import json as _json + +import nacl.pwhash + +from kazeia_central.core.adb import Adb +from kazeia_central.core.store import Store +from kazeia_central.features.medical.rag import service as rsvc + +_OPS = nacl.pwhash.argon2id.OPSLIMIT_MIN +_MEM = nacl.pwhash.argon2id.MEMLIMIT_MIN +NOW = 1_700_000_000_000 + + +def _store(tmp_path): + s = Store(tmp_path / "c.db", kdf_ops=_OPS, kdf_mem=_MEM) + s.unlock("pw") + return s + + +class _RagAdb(Adb): + def __init__(self): + super().__init__() + self._docs = {} + self.upserts = [] + self.deleted = [] + + def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True): + if path == "rag_status": + return [{"enabled": 1, "ready": 1, "model": "e5-small", "dim": 384, + "doc_count": len(self._docs), "chunk_count": len(self._docs), "pending": 0}] + if path == "rag": + return [{"source": s, "char_count": len(t), "chunk_count": 1, "updated_at": 1} + for s, t in self._docs.items()] + if path == "rag_query": + return [{"source": "doc1", "score": 0.91, "text": "chunk pertinent pour: " + (where or "")}] + return [] + + def content_call(self, method, *, arg_json=None, extras=None, serial=None): + if method == "rag_upsert_json": + self._docs[arg_json["source"]] = arg_json["text"] + self.upserts.append(arg_json) + return {"ok": "true"} + if method == "rag_dump_json": + src = arg_json["source"] + if src not in self._docs: + return {"ok": "false", "error": "not_found"} + payload = {"source": src, "text": self._docs[src]} + return {"ok": "true", "result_b64": base64.b64encode(_json.dumps(payload).encode()).decode()} + return {"ok": "true"} + + def shell(self, command, *, serial=None, timeout=None): + if "content delete" in command and "/rag/" in command: + src = command.split("/rag/")[1].strip() + self.deleted.append(src) + self._docs.pop(src, None) + return "" + + +def test_rag_upsert_get_richtext(tmp_path): + s = _store(tmp_path) + adb = _RagAdb() + rsvc.upsert_doc(adb, s, "D1", "anxiete", "Anxiété : signes.\nGestion : respiration lente.", now=NOW) + assert adb.upserts[0]["source"] == "anxiete" and ":" in adb.upserts[0]["text"] + d = rsvc.get_doc(adb, "D1", "anxiete") # rag_dump → texte riche intact + assert d["text"].startswith("Anxiété") and ":" in d["text"] and "\n" in d["text"] + assert "anxiete" in {x["source"] for x in rsvc.list_docs(adb, "D1")} + + +def test_rag_delete(tmp_path): + s = _store(tmp_path) + adb = _RagAdb() + rsvc.upsert_doc(adb, s, "D1", "temp", "x", now=NOW) + rsvc.delete_doc(adb, s, "D1", "temp", now=NOW) + assert adb.deleted == ["temp"] and "temp" not in adb._docs + + +def test_rag_query_and_status(): + adb = _RagAdb() + hits = rsvc.query(adb, "D1", "je me sens anxieux") + assert hits and hits[0]["source"] == "doc1" and hits[0]["score"] > 0.5 + st = rsvc.status(adb, "D1") + assert st["ready"] is True and st["model"] == "e5-small"