keepalive pendant le preremplissage : le flux ne peut plus rester muet

anthropic_proxy.py

@@ -11,6 +11,7 @@ UPSTREAM.
 
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 import re
@@ -473,7 +474,41 @@ async def stream_anthropic(payload: dict, req_model: str):
             # comme je le faisais, empechait donc cette reprise.
             yield _sse("error", {"type": "error", "error": _err_of(body, r.status_code)})
             return
-        async for line in r.aiter_lines():
+        # Le `ping` periodique ne suffisait pas : il ne se declenchait qu'a
+        # l'interieur de cette boucle, donc uniquement quand vLLM emettait deja
+        # quelque chose. Or pendant le PREREMPLISSAGE vLLM n'envoie rien -- et
+        # un prompt de plusieurs centaines de milliers de jetons met plusieurs
+        # minutes a etre calcule. Le flux restait muet, et le proxy Runpod
+        # coupait vers 125 s.
+        #
+        # MESURE : une aiguille a ~700 k jetons echouait a 126,5 s avec zero
+        # jeton, alors que la meme a ~358 k passait en 94,4 s. Le plafond
+        # n'etait pas le modele mais le silence : aucune requete au-dela de
+        # ~500 k ne pouvait aboutir, quel que soit le contexte annonce.
+        #
+        # On decouple donc la lecture amont de l'emission aval : une tache
+        # pompe les lignes dans une file, et l'absence de ligne pendant
+        # KEEPALIVE_S produit un `ping` au lieu d'un silence.
+        file: asyncio.Queue = asyncio.Queue()
+
+        async def _pompe() -> None:
+            try:
+                async for _l in r.aiter_lines():
+                    await file.put(_l)
+            finally:
+                await file.put(None)
+
+        _tache = asyncio.create_task(_pompe())
+        try:
+          while True:
+            try:
+                line = await asyncio.wait_for(file.get(), timeout=KEEPALIVE_S)
+            except asyncio.TimeoutError:
+                yield _sse("ping", {"type": "ping"})
+                last_out = time.monotonic()
+                continue
+            if line is None:
+                break
             if not line.startswith("data: "):
                 continue
             chunk = line[6:].strip()
@@ -547,6 +582,8 @@ async def stream_anthropic(payload: dict, req_model: str):
 
             if ch.get("finish_reason"):
                 stop_reason = _STOP.get(ch["finish_reason"], "end_turn")
+        finally:
+            _tache.cancel()
 
     if think_open:
         yield _sse("content_block_stop", {"type": "content_block_stop", "index": idx})