Streaming (SSE)

Le streaming envoie la réponse au fur et à mesure de sa génération, en Server-Sent Events. L'utilisateur voit le texte apparaître immédiatement au lieu d'attendre la fin : c'est ce qui rend une interface de chat agréable.

Activer#

Ajoutez "stream": true. Pour recevoir aussi le décompte de jetons, ajoutez "stream_options": {"include_usage": true}.

from openai import OpenAI

client = OpenAI(api_key="zir_ai_...", base_url="https://api.cloudzir.com/v1")

flux = client.chat.completions.create(
    model="mina-low",
    messages=[{"role": "user", "content": "Explique le DNS en 5 phrases."}],
    stream=True,
    stream_options={"include_usage": True},
)

complet = []
for chunk in flux:
    if chunk.choices and chunk.choices[0].delta.content:
        morceau = chunk.choices[0].delta.content
        complet.append(morceau)
        print(morceau, end="", flush=True)
    if chunk.usage:                      # dernier événement
        print(f"\n\n{chunk.usage.total_tokens} jetons")

const flux = await client.chat.completions.create({
  model: "mina-low",
  messages: [{ role: "user", content: "Explique le DNS en 5 phrases." }],
  stream: true,
  stream_options: { include_usage: true },
});

let complet = "";
for await (const chunk of flux) {
  const morceau = chunk.choices?.[0]?.delta?.content;
  if (morceau) {
    complet += morceau;
    process.stdout.write(morceau);
  }
  if (chunk.usage) console.log(`\n\n${chunk.usage.total_tokens} jetons`);
}

curl -N https://api.cloudzir.com/v1/chat/completions \
  -H "Authorization: Bearer $CLOUDZIR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "mina-low",
    "messages": [{"role": "user", "content": "Explique le DNS en 5 phrases."}],
    "stream": true
  }'

L'option -N de curl est indispensable : sans elle, la sortie est tamponnée et vous ne voyez rien avant la fin.

Format du flux (surface OpenAI)#

data: {"id":"chatcmpl-…","object":"chat.completion.chunk","model":"mina-low",
       "choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

data: {"id":"chatcmpl-…","object":"chat.completion.chunk","model":"mina-low",
       "choices":[{"index":0,"delta":{"content":"Le "},"finish_reason":null}]}

data: {"id":"chatcmpl-…","object":"chat.completion.chunk","model":"mina-low",
       "choices":[{"index":0,"delta":{"content":"DNS"},"finish_reason":null}]}

: keep-alive

data: {"id":"chatcmpl-…","object":"chat.completion.chunk","model":"mina-low",
       "choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: {"id":"chatcmpl-…","object":"chat.completion.chunk","model":"mina-low",
       "choices":[],"usage":{"prompt_tokens":18,"completion_tokens":96,"total_tokens":114}}

data: [DONE]

Points à retenir :

• Les lignes commençant par : sont des commentaires de maintien de connexion (toutes les 15 s pendant l'attente d'un moteur). Tout client SSE conforme les ignore.
• Le chunk portant usage n'apparaît que si vous avez demandé include_usage ; son tableau choices est vide.
• data: [DONE] termine toujours le flux.

Appels d'outils en streaming#

Les arguments arrivent par fragments qu'il faut concaténer :

appels = {}
for chunk in flux:
    for choix in chunk.choices or []:
        for tc in choix.delta.tool_calls or []:
            entree = appels.setdefault(tc.index, {"id": "", "name": "", "args": ""})
            if tc.id:
                entree["id"] = tc.id
            if tc.function and tc.function.name:
                entree["name"] = tc.function.name
            if tc.function and tc.function.arguments:
                entree["args"] += tc.function.arguments   # à concaténer !

# À la fin seulement, les arguments forment un JSON valide
import json
for appel in appels.values():
    arguments = json.loads(appel["args"])

Erreurs pendant le flux#

Le code HTTP est déjà 200 quand le flux commence : une panne survenue ensuite arrive comme un objet error dans le flux, suivi de [DONE]. Vérifiez donc la présence de error dans chaque événement :

data: {"error":{"message":"Erreur interne.","type":"api_error","param":null,"code":"api_error"}}
data: [DONE]

Retransmettre vers un navigateur#

N'exposez jamais votre clé côté client. Faites transiter le flux par votre serveur :

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, json

app = FastAPI()

@app.post("/chat")
async def chat(payload: dict):
    async def relais():
        async with httpx.AsyncClient(timeout=None) as http:
            async with http.stream(
                "POST",
                "https://api.cloudzir.com/v1/chat/completions",
                headers={"Authorization": f"Bearer {CLE}"},
                json={**payload, "stream": True},
            ) as amont:
                async for ligne in amont.aiter_lines():
                    if ligne:
                        yield f"{ligne}\n\n"

    return StreamingResponse(relais(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no"})

app.post("/chat", async (req, res) => {
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("X-Accel-Buffering", "no");

  const amont = await fetch("https://api.cloudzir.com/v1/chat/completions", {
    method: "POST",
    headers: { "Content-Type": "application/json", Authorization: `Bearer ${CLE}` },
    body: JSON.stringify({ ...req.body, stream: true }),
  });

  const reader = amont.body.getReader();
  const decoder = new TextDecoder();
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    res.write(decoder.decode(value));
  }
  res.end();
});

Facturation#

Une requête interrompue en cours de route est facturée pour ce qui a été généré. Si le fournisseur ne renvoie pas d'usage, CloudZIR l'estime et le marque comme tel dans Usage.

Format Anthropic#

Le streaming côté /anthropic utilise les événements nommés (message_start, content_block_delta, …) : voir Utilisation de l'API Anthropic.