51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
"""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)]
|