Diffuser la sortie des agents CLI
Affichez les tokens diffusés caractère par caractère dans les interfaces de terminal.
Diffuser la sortie des agents CLI est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.
Pourquoi la diffusion en continu est importante pour les agents CLI
Sans diffusion en continu, votre agent CLI n’affiche rien tant que la réponse complète du LLM n’est pas prête, ce qui peut prendre de 5 à 30 secondes. Les utilisateurs fixent un terminal vide en se demandant si le programme a planté.
Avec la diffusion en continu, les jetons apparaissent au fur et à mesure de leur génération, ce qui fournit un retour immédiat et une expérience bien meilleure.
Activer la diffusion en continu dans le SDK OpenAI
Transmettez stream=True à chat.completions.create(). L’appel renvoie un générateur au lieu d’un objet de réponse complet. Parcourez-le pour traiter les fragments à mesure de leur arrivée.
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': 'Explain Python generators in 3 sentences.'}],
stream=True # <-- enable streaming
)
# Each chunk arrives as it is generated
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end='', flush=True)
print() # newline after the response is completeprint() ou sys.stdout.write()
Lors d’une diffusion en continu, utilisez print(text, end='', flush=True) ou sys.stdout.write(text), suivi de sys.stdout.flush(). Sans flush=True, Python peut mettre la sortie en mémoire tampon et tout l’afficher d’un seul coup, ce qui annule l’intérêt de la diffusion en continu.
import sys
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
def stream_to_terminal(messages: list):
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True
)
full_response = ''
for chunk in stream:
token = chunk.choices[0].delta.content or ''
full_response += token
# Option 1: print with flush
print(token, end='', flush=True)
# Option 2: sys.stdout.write + flush
# sys.stdout.write(token)
# sys.stdout.flush()
print() # final newline
return full_responseCollecter la réponse complète pendant la diffusion en continu
Vous avez souvent besoin du texte complet de la réponse une fois la diffusion terminée, pour le stocker, le traiter davantage ou l’afficher. Accumulez les jetons dans une chaîne au fur et à mesure que vous les affichez.
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
def stream_and_collect(messages: list) -> str:
full_text = ''
print('Agent: ', end='', flush=True)
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True
)
for chunk in stream:
token = chunk.choices[0].delta.content or ''
full_text += token
print(token, end='', flush=True)
print() # newline
return full_text
# The return value contains the complete response for storage
# response_text = stream_and_collect(history)
# history.append({'role': 'assistant', 'content': response_text})Diffusion asynchrone avec AsyncOpenAI
Pour les architectures d’agents asynchrones, utilisez AsyncOpenAI et async for pour parcourir les fragments diffusés sans bloquer la boucle d’événements.
import asyncio
import openai
async def async_stream_agent(query: str) -> str:
client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')
stream = await client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': query}],
stream=True
)
full_text = ''
print('Agent: ', end='', flush=True)
async for chunk in stream:
token = chunk.choices[0].delta.content or ''
full_text += token
print(token, end='', flush=True)
print()
return full_text
# asyncio.run(async_stream_agent('What is asyncio?'))Détecter la fin de la diffusion avec finish_reason
Le dernier fragment d’une diffusion possède un finish_reason différent de null. Vérifiez-le pour savoir pourquoi la diffusion s’est terminée : 'stop' = fin normale, 'length' = réponse tronquée, 'tool_calls' = appel de fonction nécessaire.
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
def stream_with_finish_detection(messages: list):
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True
)
finish_reason = None
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end='', flush=True)
if chunk.choices[0].finish_reason:
finish_reason = chunk.choices[0].finish_reason
print()
if finish_reason == 'length':
print('[WARNING: Response was truncated. Try increasing max_tokens.]')
elif finish_reason == 'stop':
pass # normal completion
return finish_reasonCouleurs ANSI dans la sortie du terminal
Les codes d’échappement ANSI ajoutent de la couleur à la sortie du terminal. Utilisez-les pour distinguer visuellement le préfixe de l’agent, l’invite de saisie de l’utilisateur et les avertissements. La bibliothèque colorama assure la compatibilité multiplateforme, notamment avec Windows.
# pip install colorama
from colorama import Fore, Style, init
init(autoreset=True) # reset color after each print
def print_colored_stream(messages: list, client):
# Print agent prefix in cyan
print(Fore.CYAN + 'Agent: ' + Style.RESET_ALL, end='', flush=True)
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True
)
for chunk in stream:
token = chunk.choices[0].delta.content or ''
print(token, end='', flush=True)
print()
# Also useful:
# print(Fore.GREEN + 'Success!') — green
# print(Fore.RED + 'Error!') — red
# print(Fore.YELLOW + 'Warning') — yellowLa bibliothèque Rich pour une sortie de terminal plus riche
La bibliothèque rich permet d’afficher du Markdown, des blocs de code avec coloration syntaxique, des tableaux et des indicateurs tournants dans le terminal. Elle s’associe bien à la sortie diffusée de l’agent.
# pip install rich
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown
console = Console()
def stream_with_rich(messages: list, client):
full_text = ''
with Live(console=console, refresh_per_second=10) as live:
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True
)
for chunk in stream:
token = chunk.choices[0].delta.content or ''
full_text += token
# Render accumulated text as Markdown in real time
live.update(Markdown(full_text))
return full_textDiffusion en continu avec appels d’outils
Lorsque la diffusion en continu est combinée à l’appel de fonctions, le champ tool_calls est lui aussi diffusé par fragments. Accumulez la chaîne JSON sur plusieurs fragments avant de l’analyser.
import json
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
def stream_with_tools(messages: list, tools: list) -> dict:
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=tools,
stream=True
)
tool_call_chunks = {}
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in tool_call_chunks:
tool_call_chunks[idx] = {'name': '', 'args': ''}
if tc.function.name:
tool_call_chunks[idx]['name'] += tc.function.name
if tc.function.arguments:
tool_call_chunks[idx]['args'] += tc.function.arguments
# Parse accumulated tool calls
return {v['name']: json.loads(v['args']) for v in tool_call_chunks.values()}Afficher le nombre de jetons
Affichez en direct le nombre de jetons pendant la diffusion pour aider les utilisateurs à suivre leur utilisation et à comprendre les coûts. La diffusion OpenAI inclut les données d’utilisation dans le dernier fragment lorsque stream_options={'include_usage': True} est défini.
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
def stream_with_token_count(messages: list):
stream = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True,
stream_options={'include_usage': True}
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end='', flush=True)
# Last chunk includes usage
if chunk.usage:
print(f'\n[Tokens: prompt={chunk.usage.prompt_tokens}, '
f'completion={chunk.usage.completion_tokens}, '
f'total={chunk.usage.total_tokens}]')Bonnes pratiques pour la diffusion en continu
Résumé des bonnes pratiques pour la sortie en diffusion continue des agents CLI :
- Utiliser toujours
flush=Trueousys.stdout.flush()pour éviter la mise en mémoire tampon - Accumuler les jetons dans une chaîne pour les stocker après la diffusion
- Vérifier
finish_reasonpour détecter les troncatures - Utiliser les couleurs ANSI ou
richpour améliorer la lisibilité visuelle - Gérer la diffusion des appels d’outils en accumulant les fragments d’arguments JSON
Vérification des connaissances : sortie en diffusion continue
Vérifiez votre compréhension de la sortie en diffusion continue dans les agents CLI.
Bilan : sortie en continu dans les agents CLI
Vous pouvez désormais créer des agents CLI avec une sortie en continu, réactifs et modernes :
- Transmettez
stream=Truepour activer la sortie en continu avec le SDK OpenAI - Utilisez
print(token, end='', flush=True)pour afficher immédiatement les jetons - Accumulez les jetons dans une chaîne pour les traiter après la sortie en continu
- Vérifiez
finish_reasondans le dernier fragment pour détecter une troncature - Utilisez
async foravecAsyncOpenAIpour les agents asynchrones - Ajoutez des couleurs ANSI ou
richpour une expérience du terminal soignée
Questions Fréquemment Posées
La leçon « Diffuser la sortie des agents CLI » est-elle gratuite ?
Oui — le texte complet de « Diffuser la sortie des agents CLI » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Diffuser la sortie des agents CLI » ?
Affichez les tokens diffusés caractère par caractère dans les interfaces de terminal. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Agents ?
Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Diffuser la sortie des agents CLI » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?
Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Créer des interfaces d’agents en ligne de commande
- Agents interactifs de type REPL
- Analyser les arguments et rédiger l’aide
- Diffuser la sortie des agents CLI