Keep Ollama chat streams alive

This commit is contained in:
Michael committed 2026-06-30 10:55:15 +02:00
1 parent e9c83e3a0d
commit cdbb5f03fa
2 files changed
+51 -15

No files matched your search

+43 -13
View File
@@ -8,8 +8,10 @@ Flux :
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import json import json
import logging import logging
from contextlib import suppress
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
@@ -58,14 +60,43 @@ async def chat(req: ChatRequest) -> StreamingResponse:
tools_meta: list[dict] = [] tools_meta: list[dict] = []
error_message = "" error_message = ""
queue: asyncio.Queue[dict | None] = asyncio.Queue()
async def produce_events() -> None:
try:
async for event in run_agent(
model,
convo,
options=agent_config.ollama_options(cfg),
enabled_tools=agent_config.enabled_tool_names(cfg),
confirm_shell=cfg.get("confirm_shell", True),
):
await queue.put(event)
except Exception as exc:
logger.exception("Échec inattendu du flux de chat")
await queue.put(
{
"type": "error",
"message": f"Erreur interne du chat : {exc}",
}
)
finally:
await queue.put(None)
producer = asyncio.create_task(produce_events())
try: try:
async for ev in run_agent( while True:
model, try:
convo, ev = await asyncio.wait_for(queue.get(), timeout=15.0)
options=agent_config.ollama_options(cfg), except TimeoutError:
enabled_tools=agent_config.enabled_tool_names(cfg), # Empêche OpenResty/Nginx/Cloudflare de fermer le SSE pendant
confirm_shell=cfg.get("confirm_shell", True), # le chargement parfois long d'un modèle Ollama.
): yield _sse("ping", {"status": "waiting"})
continue
if ev is None:
break
etype = ev.pop("type") etype = ev.pop("type")
if etype in ( if etype in (
"token", "token",
@@ -82,12 +113,11 @@ async def chat(req: ChatRequest) -> StreamingResponse:
elif etype == "final": elif etype == "final":
final_content = ev["content"] final_content = ev["content"]
tools_meta = ev["tools"] tools_meta = ev["tools"]
except Exception as exc: finally:
# Après l'envoi des en-têtes SSE, une exception non gérée coupe le if not producer.done():
# socket sans explication et laisse l'interface bloquée. producer.cancel()
logger.exception("Échec inattendu du flux de chat") with suppress(asyncio.CancelledError):
error_message = f"Erreur interne du chat : {exc}" await producer
yield _sse("error", {"message": error_message})
if final_content or tools_meta: if final_content or tools_meta:
db.add_message( db.add_message(
+8 -2
View File
@@ -190,8 +190,11 @@ export async function streamChat(
body: JSON.stringify(body), body: JSON.stringify(body),
}); });
} catch (err) { } catch (err) {
const raw = err instanceof Error ? err.message : "serveur Loki injoignable";
handlers.onError( handlers.onError(
err instanceof Error ? err.message : "serveur Loki injoignable" /network error|failed to fetch/i.test(raw)
? "Serveur Loki injoignable. Vérifiez le reverse proxy et le conteneur."
: raw
); );
return; return;
} }
@@ -271,8 +274,11 @@ export async function streamChat(
} catch (err) { } catch (err) {
if (!failed) { if (!failed) {
failed = true; failed = true;
const raw = err instanceof Error ? err.message : "connexion interrompue";
handlers.onError( handlers.onError(
err instanceof Error ? err.message : "connexion au chat interrompue" /network error|failed to fetch/i.test(raw)
? "Connexion interrompue pendant le chargement du modèle. Vérifiez OLLAMA_HOST et le délai du reverse proxy."
: raw
); );
} }
} }