mirror of
https://github.com/Alkatrazz24/Funk-lab.git
synced 2026-07-11 21:34:43 +02:00
perf(stt-server): mémoire long-terme hors chemin de réponse + résilience embed (#20)
Cause de la latence 30-45s : l'endpoint d'embeddings (gpu-01:1238) peut se geler ; recall ET remember l'attendaient ~20s chacun (timeout → dégradation silencieuse), s'ajoutant à la réponse. Refactor : - store (ex-remember) en BackgroundTasks → APRÈS la réponse, hors latence perçue ; suppression de `?wait=true` (pas d'attente du flush Qdrant) - recall renvoie aussi le vecteur de la requête → store le réutilise (1 embed/tour au lieu de 2, le 2ᵉ portait sur le même texte) - timeout recall serré (4s, STT_MEMORY_RECALL_TIMEOUT) : un embed lent/mort dégrade vite (souvenirs vides) au lieu de bloquer ; store tolère 20s en arrière-plan - clients httpx persistants (pooling/keep-alive) côté brain + longterm, fermés via lifespan (plus de handshake TCP par appel) - log de timing par requête (recall/gen/total/mem) pour diagnostiquer - bump serveur 0.1.0 → 0.2.0 Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
GitHub
parent
7025d7ae70
commit
914942de73
6 changed files with 165 additions and 78 deletions
|
|
@ -27,21 +27,37 @@ class LongTermMemory:
|
|||
self.embed_url = settings.embed_url
|
||||
self.embed_model = settings.embed_model
|
||||
self.top_k = settings.memory_top_k
|
||||
self.recall_timeout = settings.memory_recall_timeout
|
||||
self.store_timeout = settings.memory_store_timeout
|
||||
self._ready = False
|
||||
# Client persistant : pooling + keep-alive (évite un handshake TCP par appel).
|
||||
self._http: httpx.AsyncClient | None = None
|
||||
|
||||
async def _embed(self, client: httpx.AsyncClient, text: str) -> list[float]:
|
||||
r = await client.post(
|
||||
def _client(self) -> httpx.AsyncClient:
|
||||
if self._http is None or self._http.is_closed:
|
||||
self._http = httpx.AsyncClient()
|
||||
return self._http
|
||||
|
||||
async def aclose(self) -> None:
|
||||
if self._http is not None and not self._http.is_closed:
|
||||
await self._http.aclose()
|
||||
|
||||
async def _embed(self, text: str, timeout: float) -> list[float]:
|
||||
r = await self._client().post(
|
||||
self.embed_url,
|
||||
json={"model": self.embed_model, "input": text},
|
||||
timeout=30,
|
||||
timeout=timeout,
|
||||
)
|
||||
r.raise_for_status()
|
||||
return r.json()["data"][0]["embedding"]
|
||||
|
||||
async def _ensure_collection(self, client: httpx.AsyncClient, dim: int) -> None:
|
||||
async def _ensure_collection(self, dim: int) -> None:
|
||||
if self._ready:
|
||||
return
|
||||
r = await client.get(f"{self.qdrant}/collections/{self.collection}")
|
||||
client = self._client()
|
||||
r = await client.get(
|
||||
f"{self.qdrant}/collections/{self.collection}", timeout=self.store_timeout
|
||||
)
|
||||
if r.status_code == 200:
|
||||
# Collection existante : si la dimension a changé (modèle d'embedding
|
||||
# différent, ex. Qwen3 4096 → nomic 768), on recrée — les anciens vecteurs
|
||||
|
|
@ -57,30 +73,38 @@ class LongTermMemory:
|
|||
await client.put(
|
||||
f"{self.qdrant}/collections/{self.collection}",
|
||||
json={"vectors": {"size": dim, "distance": "Cosine"}},
|
||||
timeout=self.store_timeout,
|
||||
)
|
||||
self._ready = True
|
||||
|
||||
async def recall(self, text: str) -> list[str]:
|
||||
"""Souvenirs pertinents (texte) ou [] si indisponible."""
|
||||
async def recall(self, text: str) -> tuple[list[str], list[float] | None]:
|
||||
"""Souvenirs pertinents + le vecteur de la requête (réutilisable pour `store`).
|
||||
|
||||
Sur le chemin de réponse → timeout serré (`recall_timeout`) : si l'embed ou Qdrant
|
||||
traîne, on dégrade vite en `([], None)` plutôt que de faire patienter le client.
|
||||
Le vecteur renvoyé évite de ré-embedder le même texte au moment du `store`.
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=20) as client:
|
||||
vec = await self._embed(client, text)
|
||||
r = await client.post(
|
||||
f"{self.qdrant}/collections/{self.collection}/points/search",
|
||||
json={"vector": vec, "limit": self.top_k, "with_payload": True},
|
||||
)
|
||||
if r.status_code == 404: # collection pas encore créée
|
||||
return []
|
||||
r.raise_for_status()
|
||||
pts = r.json().get("result", [])
|
||||
return [p["payload"]["text"] for p in pts if p.get("payload", {}).get("text")]
|
||||
client = self._client()
|
||||
vec = await self._embed(text, timeout=self.recall_timeout)
|
||||
r = await client.post(
|
||||
f"{self.qdrant}/collections/{self.collection}/points/search",
|
||||
json={"vector": vec, "limit": self.top_k, "with_payload": True},
|
||||
timeout=self.recall_timeout,
|
||||
)
|
||||
if r.status_code == 404: # collection pas encore créée → rien à rappeler
|
||||
return [], vec
|
||||
r.raise_for_status()
|
||||
pts = r.json().get("result", [])
|
||||
texts = [p["payload"]["text"] for p in pts if p.get("payload", {}).get("text")]
|
||||
return texts, vec
|
||||
except Exception: # noqa: BLE001 — dégrade silencieusement
|
||||
return []
|
||||
return [], None
|
||||
|
||||
async def health(self) -> dict:
|
||||
"""Diagnostic actif : sonde embeddings + Qdrant + collection, sans rien avaler.
|
||||
|
||||
Contrairement à recall/remember (qui dégradent en silence), expose les erreurs
|
||||
Contrairement à recall/store (qui dégradent en silence), expose les erreurs
|
||||
pour pouvoir déboguer la mémoire long-terme sans `kubectl exec`.
|
||||
"""
|
||||
out: dict = {
|
||||
|
|
@ -91,49 +115,60 @@ class LongTermMemory:
|
|||
"embed": {"ok": False},
|
||||
"qdrant": {"ok": False},
|
||||
}
|
||||
async with httpx.AsyncClient(timeout=20) as client:
|
||||
# 1) embeddings
|
||||
try:
|
||||
vec = await self._embed(client, "ping mémoire")
|
||||
out["embed"] = {"ok": True, "dim": len(vec)}
|
||||
except Exception as e: # noqa: BLE001 — on veut l'erreur
|
||||
out["embed"] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
|
||||
# 2) Qdrant + collection
|
||||
try:
|
||||
r = await client.get(f"{self.qdrant}/collections")
|
||||
r.raise_for_status()
|
||||
names = [c["name"] for c in r.json().get("result", {}).get("collections", [])]
|
||||
exists = self.collection in names
|
||||
qdrant: dict = {"ok": True, "collections": names, "collection_exists": exists}
|
||||
if exists:
|
||||
cr = await client.get(f"{self.qdrant}/collections/{self.collection}")
|
||||
if cr.status_code == 200:
|
||||
qdrant["points_count"] = cr.json().get("result", {}).get("points_count")
|
||||
out["qdrant"] = qdrant
|
||||
except Exception as e: # noqa: BLE001
|
||||
out["qdrant"] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
|
||||
client = self._client()
|
||||
# 1) embeddings
|
||||
try:
|
||||
vec = await self._embed("ping mémoire", timeout=self.store_timeout)
|
||||
out["embed"] = {"ok": True, "dim": len(vec)}
|
||||
except Exception as e: # noqa: BLE001 — on veut l'erreur
|
||||
out["embed"] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
|
||||
# 2) Qdrant + collection
|
||||
try:
|
||||
r = await client.get(f"{self.qdrant}/collections", timeout=self.store_timeout)
|
||||
r.raise_for_status()
|
||||
names = [c["name"] for c in r.json().get("result", {}).get("collections", [])]
|
||||
exists = self.collection in names
|
||||
qdrant: dict = {"ok": True, "collections": names, "collection_exists": exists}
|
||||
if exists:
|
||||
cr = await client.get(
|
||||
f"{self.qdrant}/collections/{self.collection}", timeout=self.store_timeout
|
||||
)
|
||||
if cr.status_code == 200:
|
||||
qdrant["points_count"] = cr.json().get("result", {}).get("points_count")
|
||||
out["qdrant"] = qdrant
|
||||
except Exception as e: # noqa: BLE001
|
||||
out["qdrant"] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
|
||||
return out
|
||||
|
||||
async def remember(self, session_id: str, text: str) -> None:
|
||||
async def store(
|
||||
self, session_id: str, text: str, vec: list[float] | None = None
|
||||
) -> None:
|
||||
"""Mémorise un tour. Pensé pour tourner **en tâche de fond** (hors chemin de réponse).
|
||||
|
||||
`vec` : si fourni (réutilisé depuis `recall`), on évite un 2ᵉ embed du même texte.
|
||||
Pas de `?wait=true` : on n'attend pas le flush disque de Qdrant — le client a déjà
|
||||
sa réponse, ce point peut être indexé en arrière-plan.
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=20) as client:
|
||||
vec = await self._embed(client, text)
|
||||
await self._ensure_collection(client, len(vec))
|
||||
await client.put(
|
||||
f"{self.qdrant}/collections/{self.collection}/points?wait=true",
|
||||
json={
|
||||
"points": [
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"vector": vec,
|
||||
"payload": {
|
||||
"text": text,
|
||||
"session_id": session_id,
|
||||
"ts": time.time(),
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
if vec is None:
|
||||
vec = await self._embed(text, timeout=self.store_timeout)
|
||||
await self._ensure_collection(len(vec))
|
||||
await self._client().put(
|
||||
f"{self.qdrant}/collections/{self.collection}/points",
|
||||
json={
|
||||
"points": [
|
||||
{
|
||||
"id": str(uuid.uuid4()),
|
||||
"vector": vec,
|
||||
"payload": {
|
||||
"text": text,
|
||||
"session_id": session_id,
|
||||
"ts": time.time(),
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
timeout=self.store_timeout,
|
||||
)
|
||||
except Exception: # noqa: BLE001 — dégrade silencieusement
|
||||
pass
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue