Documentation/Réponses en streaming

Réponses en streaming

Le streaming (Streaming) vous permet de recevoir la sortie en temps réel pendant la génération du modèle, améliorant l'expérience utilisateur et la vitesse perçue.

Principe de fonctionnement#

诺玛AI utilise le protocole Server-Sent Events (SSE) pour les réponses en streaming :

Le client définit `stream: true` lors de l'envoi de la requête
Le serveur renvoie progressivement les fragments de contenu générés (chunk)
Chaque chunk est envoyé via SSE avec le préfixe `data: `
À la fin de la génération, `data: [DONE]` est envoyé

Streaming au protocole OpenAI#

cURL

Code

curl https://as.apinoma.com/v1/chat/completions \
  -H "Authorization: Bearer $APINOMA_API_KEY" \
  -H "Content-Type: application/json" \
  -N \
  -d '{
    "model": "openai/gpt-4o",
    "messages": [{"role": "user", "content": "Écris un poème sur la programmation"}],
    "stream": true
  }'

Python

TypeScript

Streaming au protocole Anthropic#

Python

Code

import anthropic
 
client = anthropic.Anthropic(
    base_url="https://as.apinoma.com/anthropic",
    api_key="<votre APINOMA_API_KEY>"
)
 
with client.messages.stream(
    model="anthropic/claude-sonnet-4.6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Écris un poème sur la programmation"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

TypeScript

Streaming + Function Calling#

Le streaming prend aussi en charge les scénarios d'appel de fonctions. Le modèle diffuse d'abord la requête d'appel d'outil, puis vous poursuivez la conversation une fois le traitement terminé :

Code

stream = client.chat.completions.create(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "Quel temps fait-il à Pékin aujourd'hui ?"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Obtenir la météo d'une ville donnée",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "Nom de la ville"}
                },
                "required": ["city"]
            }
        }
    }],
    stream=True
)
 
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.tool_calls:
        # Traiter l'appel d'outil
        print(f"Appel d'outil : {delta.tool_calls[0].function}")
    elif delta.content:
        print(delta.content, end="", flush=True)

Gestion des erreurs et reconnexion#

La connexion en streaming peut être interrompue en raison de problèmes réseau. Il est recommandé d'implémenter une logique de reconnexion.

Code

import time
 
def stream_with_retry(client, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            stream = client.chat.completions.create(stream=True, **kwargs)
            for chunk in stream:
                yield chunk
            return  # Terminé avec succès
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # Repli exponentiel
                print(f"\nConnexion interrompue, nouvelle tentative dans {wait}s...")
                time.sleep(wait)
            else:
                raise e

Bonnes pratiques#

Toujours définir un délai d'expiration — pour éviter une attente infinie
Gérer les chunks incomplets — certains chunks peuvent ne pas avoir de content
Implémenter un mécanisme de reconnexion — utiliser une stratégie de repli exponentiel
Utiliser `flush` côté frontend — pour garantir l'affichage immédiat du contenu

Dernière mise à jour le 23 juin 2026