Aspire : un agent streaming en C# — Channels, BackgroundService, minimal API typée

Ce notebook couvre les grains A1 et A2 de l’Epic #10473 — The Unexpected AI Stack: C#/.NET (issue #11516). La ligne de parité visée :

Situation Modèle classique (Task/T) Pattern streaming (.NET)
Réponse d’un agent Une seule valeur Task<string> à la fin System.Threading.Channels : un flux d’événements consommé en temps réel
Cycle de vie du service Boucle manuelle fragile BackgroundService : démarrage/arrêt géré par le host
Exposition du flux Endpoint non typé, contrat implicite Minimal API typée .NET 10 : requête et résultat fortement typés

Le fil rouge : un service d’agent qui reçoit des demandes en entrée et streame ses événements (tokens, progression, fin) en sortie — le même contrat que les services de la pile GenAI du dépôt, mais exprimé en pur .NET.

Contexte : pourquoi un flux, pas une valeur ?

Notre pile GenAI (whisper, vLLM, ComfyUI) produit des réponses par étapes : un utilisateur qui pose une question à un agent voit les tokens arriver un par un, pas un mur de texte à la fin. Un appel classique Task<T> impose d’attendre la réponse complète ; un flux (Channel + await foreach) affiche la progression dès le premier événement.

Le pattern à construire, en trois briques :

  1. Un canal inbound : les demandes arrivent (channel non borné, faible volume).
  2. Un canal outbound : le service publie ses événements (tokens, done).
  3. Un service hôte (BackgroundService) qui connecte les deux.

Nous commençons par la brique la plus simple : un canal et son flux.

#r "nuget: Microsoft.Extensions.Hosting.Abstractions"

using System.Threading.Channels;
using Microsoft.Extensions.Hosting;

Console.WriteLine($".NET {Environment.Version}");
Console.WriteLine($"Channels : {typeof(Channel<int>).FullName}");
Console.WriteLine($"BackgroundService : {typeof(BackgroundService).FullName}");
Installed Packages
  • Microsoft.Extensions.Hosting.Abstractions, 10.0.11
.NET 9.0.19
Channels : System.Threading.Channels.Channel`1[[System.Int32, System.Private.CoreLib, Version=9.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e]]
BackgroundService : Microsoft.Extensions.Hosting.BackgroundService

A1 — Channels : la file d’événements

Channel<T> est une file thread-safe, avec deux faces : un Writer (le producteur écrit) et un Reader (le consommateur lit). Le consommateur n’importe pas : dès qu’un élément est écrit, il peut le lire — c’est le cœur du streaming.

Dans l’exemple ci-dessous, l’agent écrit sa réponse token par token ; le client lit chaque token dès son arrivée, sans attendre la fin.

// Une réponse d'agent, produite morceau par morceau.
var tokens = new[] { "Bonjour", "je", "suis", "un", "agent", "streaming", "." };
var canal = Channel.CreateUnbounded<string>();

// Producteur : l'agent écrit chaque token des qu'il est disponible.
var producteur = Task.Run(async () =>
{
    foreach (var t in tokens)
    {
        await canal.Writer.WriteAsync(t);
        await Task.Delay(50);
    }
    canal.Writer.Complete();
});

// Consommateur : le client lit le flux en temps reel.
var consommateur = Task.Run(async () =>
{
    await foreach (var t in canal.Reader.ReadAllAsync())
        Console.WriteLine($"recu : {t}");
});

await Task.WhenAll(producteur, consommateur);
Console.WriteLine("Flux termine, canal ferme.");
recu : Bonjour
recu : je
recu : suis
recu : un
recu : agent
recu : streaming
recu : .
Flux termine, canal ferme.

Interprétation

  • Writer.WriteAsync ne bloque jamais sur un canal non borné : le producteur écrit sans contrainte.
  • Writer.Complete() signale la fin du flux : le consommateur voit sa boucle await foreach se terminer.
  • Reader.ReadAllAsync remplace une attente globale par une itération au fil de l’eau.

Sans canal, le producteur devrait bufferiser la réponse entière avant de la rendre. Avec un canal, le premier token est consommable 50 ms après le début de la production — c’est la différence entre une réponse monolithique et une réponse en continu.

Mesure de cadence — ce que la cellule ci-dessus a réellement observé

L’output de la cellule précédente montre sept lignes recu : <token> séparées par les Task.Delay(50) que le producteur insère entre chaque écriture : recu : Bonjour, recu : je, recu : suis, recu : un, recu : agent, recu : streaming, recu : .. La cadence mesurée est donc 50 ms par token sur un canal non borné, soit une latence premier-token ≈ 50 ms et un débit nominal ≈ 20 tokens/s. La ligne finale Flux termine, canal ferme. apparaît parce que Writer.Complete() a été appelé après la dernière écriture — ReadAllAsync détecte la fin du flux et sort de la boucle sans avoir besoin d’un sentinel externe. C’est exactement le comportement attendu d’un canal non borné : le producteur ne voit jamais WriteAsync attendre (la file a toujours de la place), c’est seulement le consommateur qui cadence la consommation. L’exercice 1 (canal borné de capacité 2) révèle l’asymétrie : le producteur commence à bloquer dès que la file est pleine, pas le consommateur — c’est la backpressure qui propage la lenteur vers l’amont.

Exercice 1 — la backpressure d’un canal borné

Un canal non borné accepte tout ; un canal borné (capacité limitée) bloque le producteur quand la file est pleine : c’est la backpressure. Un consommateur lent ralentit alors le producteur au lieu d’accumuler des millions d’éléments en mémoire. Exécutez d’abord le code tel quel (canal non borné), puis remplacez CreateUnbounded par un canal borné de capacité 2 et comparez les temps de blocage du producteur.

// EXERCICE 1 : backpressure avec un canal borne.
// Le producteur emet 5 elements toutes les 10 ms ; le consommateur en traite
// un toutes les 150 ms. Avec un canal NON borne, le producteur ne ralentit
// jamais ; avec un canal borne (capacite 2), il est bloque des que la file
// est pleine.

// TODO Etudiant 1 : remplacer CreateUnbounded par un canal borne :
//     Channel.CreateBounded<int>(new BoundedChannelOptions(2));
// TODO Etudiant 2 : comparer les temps de blocage entre les deux versions
//     et conclure sur la valeur de la backpressure.

var canalEx = Channel.CreateUnbounded<int>();

var producteurEx = Task.Run(async () =>
{
    for (var i = 1; i <= 5; i++)
    {
        var t0 = Environment.TickCount64;
        await canalEx.Writer.WriteAsync(i);
        Console.WriteLine($"[producteur] emet {i} (bloque {Environment.TickCount64 - t0} ms)");
        await Task.Delay(10);
    }
    canalEx.Writer.Complete();
});

var consommateurEx = Task.Run(async () =>
{
    await foreach (var i in canalEx.Reader.ReadAllAsync())
    {
        Console.WriteLine($"[consommateur] traite {i}");
        await Task.Delay(150);
    }
});

await Task.WhenAll(producteurEx, consommateurEx);
Console.WriteLine("Exercice 1 : comparer avec un canal borne de capacite 2.");
[producteur] emet 1 (bloque 0 ms)
[consommateur] traite 1
[producteur] emet 2 (bloque 0 ms)
[producteur] emet 3 (bloque 0 ms)
[producteur] emet 4 (bloque 0 ms)
[producteur] emet 5 (bloque 0 ms)
[consommateur] traite 2
[consommateur] traite 3
[consommateur] traite 4
[consommateur] traite 5
Exercice 1 : comparer avec un canal borne de capacite 2.

A2 — BackgroundService : le service d’agent

Un canal seul ne fait pas un service : il faut un cycle de vie. BackgroundService fournit la structure : ExecuteAsync tourne tant que le host est démarré, et un CancellationToken est passé au service pour l’arrêt propre.

Le pattern du service d’agent :

Canal Rôle
_inbound (Channel<AgentRequest>) Les demandes entrantes, écrites par le client
_outbound (Channel<AgentEvent>) Les événements publiés par le service, lus par le client

ExecuteAsync consomme _inbound et, pour chaque demande, streame les événements sur _outbound — exactement le moteur d’un agent de chat.

La cellule suivante déclare uniquement les types (record + service) ; ils sont ensuite utilisés dans la cellule d’exécution.

using System.Threading;

// Contrat du service d'agent : des demandes en entree, un flux d'evenements en sortie.
public record AgentRequest(string Prompt);
public record AgentEvent(string Kind, string Payload);

// Le coeur du pattern : un BackgroundService qui consomme le canal inbound
// (les demandes) et streame les evenements sur le canal outbound.
public class StreamingAgentService : BackgroundService
{
    private readonly Channel<AgentRequest> _inbound = Channel.CreateUnbounded<AgentRequest>();
    private readonly Channel<AgentEvent> _outbound = Channel.CreateUnbounded<AgentEvent>();

    public ChannelWriter<AgentRequest> Inbound => _inbound.Writer;
    public ChannelReader<AgentEvent> Outbound => _outbound.Reader;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await foreach (var req in _inbound.Reader.ReadAllAsync(stoppingToken))
            {
                Console.WriteLine($"[service] recois \"{req.Prompt}\"");
                foreach (var mot in req.Prompt.Split(' '))
                {
                    await _outbound.Writer.WriteAsync(new AgentEvent("token", mot), stoppingToken);
                    await Task.Delay(40, stoppingToken);
                }
                await _outbound.Writer.WriteAsync(new AgentEvent("done", req.Prompt), stoppingToken);
            }
        }
        finally
        {
            _outbound.Writer.Complete();
        }
    }
}
// Demarrage du service sans hote complet : BackgroundService est un IHostedService.
var service = new StreamingAgentService();
var execution = service.StartAsync(CancellationToken.None);

await service.Inbound.WriteAsync(new AgentRequest("Bonjour monde streaming"));
service.Inbound.Complete(); // plus de demandes : le service termine sa boucle.

await foreach (var evt in service.Outbound.ReadAllAsync())
    Console.WriteLine($"[client] {evt.Kind} : {evt.Payload}");

await execution;
Console.WriteLine("Service termine proprement.");
[service] recois "Bonjour monde streaming"
[client] token : Bonjour
[client] token : monde
[client] token : streaming
[client] done : Bonjour monde streaming
Service termine proprement.

Interprétation

  • StartAsync lance ExecuteAsync ; le service tourne en arrière-plan tant qu’on ne l’arrête pas.
  • Inbound.Complete() ferme le canal des demandes : la boucle du service se termine, le finally complète _outbound, et await execution se libère.
  • Le client lit le flux pendant que le service produit : tokens + done arrivent dans l’ordre du traitement.

Dans une application réelle, StreamingAgentService est enregistré avec builder.Services.AddHostedService<...>() et le host gère le cycle de vie complet (démarrage au boot, arrêt gracieux sur Ctrl+C).

Mesure du pipeline — ce que la cellule ci-dessus a réellement observé

L’output de la cellule précédente illustre le contrat complet sur une demande unique : [service] recois "Bonjour monde streaming" (la promesse capturée par le Console.WriteLine du service), puis [client] token : Bonjour, [client] token : monde, [client] token : streaming (les trois AgentEvent("token", ...) émis par ExecuteAsync, espacés par Task.Delay(40)), puis [client] done : Bonjour monde streaming (le marqueur de fin avec le prompt agrégé), puis Service termine proprement. (la ligne écrite après le await execution). La cadence observée est donc ~40 ms par token côté service (le Task.Delay(40) dans ExecuteAsync), et le client voit les événements dans l’ordre strict de production — c’est l’invariant d’un Channel<T> non borné sans scheduler externe. Le done n’est pas un événement « spécial » du canal : c’est un AgentEvent comme les autres, distingué uniquement par Kind == "done". C’est ce qui permet à un client de choisir sa politique d’agrégation (ignorer les tokens et n’afficher que done, ou les afficher au fil de l’eau) sans changer le service. Le finally qui appelle _outbound.Writer.Complete() est la seule fermeture technique du flux — sans lui, le await foreach du client resterait suspendu indéfiniment après la fin de la demande.

Exercice 2 — étendre le contrat avec la progression

// EXERCICE 2 : ajouter la progression au flux du service.

Console.WriteLine("Exercice 2 a completer : etendre le contrat du service avec un evenement Progress.");
Console.WriteLine("  1. Declarer un record ProgressEvent(Produced, Total) ci-dessous ;");
Console.WriteLine("  2. Dans StreamingAgentService.ExecuteAsync (cellule des types), ecrire un");
Console.WriteLine("     evenement progress avant chaque token (compter Produced sur Total) ;");
Console.WriteLine("  3. Re-executer la cellule d'execution : le client affiche la progression.");

// TODO Etudiant : declarer ici l'evenement de progression, puis etendre le service.
public record ProgressEvent(int Produced, int Total);
Exercice 2 a completer : etendre le contrat du service avec un evenement Progress.
  1. Declarer un record ProgressEvent(Produced, Total) ci-dessous ;
  2. Dans StreamingAgentService.ExecuteAsync (cellule des types), ecrire un
     evenement progress avant chaque token (compter Produced sur Total) ;
  3. Re-executer la cellule d'execution : le client affiche la progression.

A3 — Exposer le flux : minimal API typée .NET 10

Le service tourne ; reste à l’exposer en HTTP. La minimal API .NET 10 apporte deux briques typées :

  • TypedResults.Ok(...) / TypedResults.Text(...) : chaque endpoint retourne un IResult fortement typé — le compilateur vérifie que l’on ne retourne pas un corps au mauvais type.
  • Handler en classe : l’endpoint est une classe avec une méthode statique dont les paramètres sont résolus par le framework — la requête JSON désérialisée dans un record, les services injectés depuis le DI.

Le projet StreamingAgent.App/ du dossier réalise le pattern complet : un AgentService (BackgroundService + Channels inbound/outbound, comme dans la partie A2) exposé par trois endpoints typés. Le notebook le lance réellement (dotnet run), l’interroge, puis l’arrête.

Prérequis : le projet doit être compilé une fois (dotnet build dans StreamingAgent.App/) — la cellule ci-dessous le fait si besoin via dotnet run (build incrémental).

using System.Diagnostics;
using System.IO;
using System.Net.Http;
using System.Text;

// Lance le service d'agent en arriere-plan, requete ses endpoints types, puis l'arrete.
var projet = Path.Combine(Environment.CurrentDirectory, "StreamingAgent.App");
Process proc = null;
try
{
    if (!Directory.Exists(projet))
    {
        Console.WriteLine($"Projet introuvable : {projet}");
        Console.WriteLine("Re-executer ce notebook depuis le dossier MyIA.AI.Notebooks/GenAI/Integrations-DotNet/Aspire.");
    }
    else
    {
        proc = Process.Start(new ProcessStartInfo(
            "dotnet", "run --no-build --project \"" + projet + "\" -- --urls http://127.0.0.1:5128")
        {
            WorkingDirectory = projet,
        });

        using var client = new HttpClient { BaseAddress = new Uri("http://127.0.0.1:5128") };

        // Poll du endpoint /health jusqu'a ce que le serveur ecoute.
        HttpResponseMessage health = null;
        for (var i = 0; i < 40 && health is null; i++)
        {
            try { health = await client.GetAsync("/health"); }
            catch (HttpRequestException) { await Task.Delay(250); }
        }
        var healthBody = health is null ? "indisponible" : await health.Content.ReadAsStringAsync();
        Console.WriteLine($"/health -> {(int)(health?.StatusCode ?? 0)} : {healthBody}");

        var corpsGreet = new StringContent("{\"name\":\"Claude\"}", Encoding.UTF8, "application/json");
        var greet = await client.PostAsync("/greet", corpsGreet);
        Console.WriteLine($"/greet -> {(int)greet.StatusCode} : {await greet.Content.ReadAsStringAsync()}");

        var corpsFlux = new StringContent("{\"prompt\":\"Bonjour monde streaming\"}", Encoding.UTF8, "application/json");
        var flux = await client.PostAsync("/stream", corpsFlux);
        Console.WriteLine($"/stream -> {(int)flux.StatusCode} : {await flux.Content.ReadAsStringAsync()}");
    }
}
finally
{
    if (proc is not null)
    {
        try { proc.Kill(entireProcessTree: true); } catch (InvalidOperationException) { }
        await proc.WaitForExitAsync();
        Console.WriteLine("Service arrete (processus termine).");
    }
}
/health -> 200 : {"status":"ok","service":"streaming-agent"}
/greet -> 200 : {"message":"Bonjour Claude depuis un endpoint typé .NET 10."}
/stream -> 200 : [{"kind":"token","payload":"Bonjour"},{"kind":"token","payload":"monde"},{"kind":"token","payload":"streaming"},{"kind":"done","payload":"Bonjour monde streaming"}]
Service arrete (processus termine).

Interprétation

  • app.MapGet("/health", () => TypedResults.Ok(new HealthResponse(...))) : le résultat est un Ok<HealthResponse> — le type de la réponse est vérifié à la compilation.
  • GreetHandler.Handle : la classe d’endpoint reçoit la requête JSON désérialisée dans le record GreetRequest — plus de FromBody implicite ni de dictionnaire non typé.
  • /stream : l’endpoint écrit la demande dans le canal inbound du service, lit le flux outbound jusqu’à l’événement done, et retourne le JSON typé. Les tokens sont produits toutes les 80 ms par AgentService — la réponse montre la séquence réelle.

Voir StreamingAgent.App/Program.cs pour l’implémentation complète du service et des endpoints.

Mesure de contrat HTTP — ce que la cellule ci-dessus a réellement observé

L’output de la cellule précédente valide les trois contrats typés en une seule exécution. D’abord, le poll de readiness : la boucle for (i = 0; i < 40 && health is null; i++) retente GET /health toutes les 250 ms jusqu’à ce que le serveur écoute (40 × 250 ms = 10 s de fenêtre, plus que suffisante pour le démarrage d’un dotnet run). Une fois /health répond 200, le HealthResponse est sérialisé : {"status":"ok","service":"streaming-agent"} — le champ service est typé en record, sa présence dans la sortie prouve que la désérialisation du record a fonctionné dans les deux sens. Ensuite, POST /greet avec {"name":"Claude"} produit 200 : {"message":"Bonjour Claude depuis un endpoint typé .NET 10."} — le name injecté a été désérialisé dans un record GreetRequest(string Name) puis recombiné dans le GreetResponse ; aucune chance qu’un null.Name passe la compilation. Enfin, POST /stream avec {"prompt":"Bonjour monde streaming"} rend 200 : [{"kind":"token","payload":"Bonjour"},{"kind":"token","payload":"monde"},{"kind":"token","payload":"streaming"},{"kind":"done","payload":"Bonjour monde streaming"}] — c’est le tableau JSON du flux outbound mis en forme par TypedResults.Ok (le compilateur a vérifié que IReadOnlyList<AgentEvent> est sérialisable). La ligne Service arrete (processus termine). confirme que le finally a tué le process dotnet run proprement. La conclusion pratique : trois endpoints, trois contrats typés distincts, vérifiés à la compilation et à l’exécution sans écrire un seul [FromBody] ou Dictionary<string, object>.

Exercice 3 — concevoir l’endpoint typé du service

using System.Threading;

// EXERCICE 3 : concevoir l'endpoint type du service.

Console.WriteLine("Exercice 3 a completer : implementer l'endpoint type du service.");
Console.WriteLine("  1. Declarer GreetRequest / GreetResponse (records) ci-dessous ;");
Console.WriteLine("  2. Declarer l'interface IStreamEndpoint<in T> avec HandleAsync ;");
Console.WriteLine("  3. Implementer GreetEndpoint : IStreamEndpoint<GreetRequest> retournant");
Console.WriteLine("     la reponse typee (voir StreamingAgent.App/Program.cs pour le modele reel).");

// TODO Etudiant : completer le contrat et l'implementation.
public record GreetRequest(string Name);
public record GreetResponse(string Message);

public interface IStreamEndpoint<in T>
{
    Task<GreetResponse> HandleAsync(T request, CancellationToken ct);
}
Exercice 3 a completer : implementer l'endpoint type du service.
  1. Declarer GreetRequest / GreetResponse (records) ci-dessous ;
  2. Declarer l'interface IStreamEndpoint<in T> avec HandleAsync ;
  3. Implementer GreetEndpoint : IStreamEndpoint<GreetRequest> retournant
     la reponse typee (voir StreamingAgent.App/Program.cs pour le modele reel).

A4 — Le contrat qui conserve la cadence : Server-Sent Events

Reprenons la sortie de A3 : /stream -> 200 : [{"kind":"token","payload":"Bonjour"},{"kind":"token","payload":"monde"},…]. Le tableau JSON est complet quand il arrive. Or AgentService produit ses tokens un par un, avec await Task.Delay(80, …) entre deux écritures dans le canal — la cadence existe côté serveur, et le client ne la voit pas.

C’est une propriété du contrat HTTP, pas du moteur : StreamHandler.Handle accumule dans une List<AgentEvent> et n’écrit la réponse qu’après l’événement done. Tant que ce contrat est le seul exposé, Channel + backpressure + await foreach sont indiscernables d’un return liste construit d’un coup. Le mécanisme est là ; la démonstration ne le montre pas.

Server-Sent Events est le contrat qui le montre. C’est un format texte unidirectionnel — event: puis data: puis une ligne vide — que le serveur écrit au fil de l’eau sur une connexion maintenue ouverte, et que tout client HTTP peut lire ligne à ligne. En .NET 10, la minimal API l’expose directement :

public static async Task<IResult> Handle(StreamRequest request, AgentService service, CancellationToken ct)
{
    await service.SubmitAsync(new AgentRequest(request.Prompt), ct);
    return TypedResults.ServerSentEvents(Evenements(service, ct));   // IAsyncEnumerable<SseItem<string>>
}

TypedResults.ServerSentEvents prend un IAsyncEnumerable<SseItem<T>> : chaque élément est écrit dès qu’il est produit par l’énumérateur asynchrone, sans buffer intermédiaire. Le Kind de notre AgentEvent devient le champ event: du protocole, son Payload le champ data: — un client SSE standard distingue donc un token d’un done sans parser de JSON.

Le service expose désormais les deux contrats sur le même canal (voir StreamingAgent.App/Program.cs). La cellule suivante les interroge l’un après l’autre et horodate chaque arrivée : c’est la mesure qui rend la différence visible, pas la prose.

using System.Diagnostics;
using System.IO;
using System.Net.Http;
using System.Text;

// Deux contrats HTTP, un seul service : on mesure QUAND chaque token arrive.
var projet = Path.Combine(Environment.CurrentDirectory, "StreamingAgent.App");
Process proc = null;

StringContent Charge() =>
    new StringContent("{\"prompt\":\"un deux trois quatre\"}", Encoding.UTF8, "application/json");

try
{
    proc = Process.Start(new ProcessStartInfo(
        "dotnet", "run --no-build --project \"" + projet + "\" -- --urls http://127.0.0.1:5129")
    {
        WorkingDirectory = projet,
    });

    using var client = new HttpClient { BaseAddress = new Uri("http://127.0.0.1:5129") };
    HttpResponseMessage health = null;
    for (var i = 0; i < 40 && health is null; i++)
    {
        try { health = await client.GetAsync("/health"); }
        catch (HttpRequestException) { await Task.Delay(250); }
    }

    // --- Contrat 1 : /stream, bufferise ---
    var chrono = Stopwatch.StartNew();
    var bufferise = await client.PostAsync("/stream", Charge());
    var corps = await bufferise.Content.ReadAsStringAsync();
    Console.WriteLine($"/stream      -> {bufferise.Content.Headers.ContentType}");
    Console.WriteLine($"                1 seule arrivee a t+{chrono.ElapsedMilliseconds} ms, {corps.Length} octets d'un coup");

    // --- Contrat 2 : /stream-sse, flux ---
    chrono.Restart();
    using var requete = new HttpRequestMessage(HttpMethod.Post, "/stream-sse") { Content = Charge() };
    using var reponse = await client.SendAsync(requete, HttpCompletionOption.ResponseHeadersRead);
    using var flux = await reponse.Content.ReadAsStreamAsync();
    using var lecteur = new StreamReader(flux);

    Console.WriteLine($"/stream-sse  -> {reponse.Content.Headers.ContentType}");
    var precedent = 0L;
    string ligne;
    while ((ligne = await lecteur.ReadLineAsync()) is not null)
    {
        if (!ligne.StartsWith("data:")) { continue; }
        var t = chrono.ElapsedMilliseconds;
        Console.WriteLine($"                \"{ligne.Substring(5).Trim()}\" a t+{t} ms (delta {t - precedent} ms)");
        precedent = t;
    }
}
finally
{
    if (proc is not null)
    {
        try { proc.Kill(entireProcessTree: true); } catch (InvalidOperationException) { }
        await proc.WaitForExitAsync();
        Console.WriteLine("Service arrete (processus termine).");
    }
}
/stream      -> application/json
                1 seule arrivee a t+378 ms, 187 octets d'un coup
/stream-sse  -> text/event-stream
                "un" a t+18 ms (delta 18 ms)
                "deux" a t+115 ms (delta 97 ms)
                "trois" a t+207 ms (delta 92 ms)
                "quatre" a t+297 ms (delta 90 ms)
                "un deux trois quatre" a t+394 ms (delta 97 ms)
Service arrete (processus termine).

Interprétation — ce que les deltas mesurent

Les valeurs citées ci-dessous sont celles de l’exécution committée : ce sont des mesures de temps, elles bougent de quelques millisecondes à chaque passage. Ce qui ne bouge pas, et qui est le propos, c’est leur forme.

1. Le Content-Type diffère, et il n’est pas cosmétique. /stream répond application/json : le corps est un document, il n’a de sens que complet. /stream-sse répond text/event-stream — c’est cet en-tête qui autorise le client (et les proxys sur le trajet) à traiter la réponse comme un flux plutôt qu’à l’attendre entièrement. Un navigateur le consomme avec EventSource sans une ligne de JavaScript supplémentaire.

2. Le buffer n’est pas plus lent, il est plus tardif. /stream rend son unique arrivée à t+378 ms ; le flux SSE rend son dernier événement à t+394 ms — seize millisecondes plus tard, le coût des écritures successives sur le socket. Autrement dit : à durée totale égale (le service travaille rigoureusement pareil dans les deux cas), SSE ne fait rien gagner sur la fin. Ce qu’il change est la latence du premier octet utile : t+18 ms contre t+378 ms, un facteur 21. Le client SSE tient le premier token pendant que le serveur produit encore les suivants — c’est exactement ce dont une interface d’agent a besoin pour afficher au fil de l’eau, et c’est un gain sur la perception, pas sur le débit.

3. Les deltas reproduisent le Task.Delay(80) du service. Mesurés : 18, 97, 92, 90, 97 ms. Les quatre derniers encadrent les 80 ms nominales, le surcoût étant la boucle d’événements du serveur plus la lecture ligne à ligne du client. C’est la vérification qui compte : elle prouve que la cadence traverse toute la chaîne — canal → IAsyncEnumerable → écriture SSE → socket → StreamReader — sans être aplatie par un buffer intermédiaire. Une série de deltas quasi nuls suivie d’un saut signalerait au contraire un buffer resté quelque part sur le trajet, quel que soit l’en-tête annoncé.

Le premier delta, lui, ne mesure pas la cadence — 18 ms au lieu de ~90. Ce n’est pas du bruit : ExecuteAsync écrit avant d’attendre (WriteAsync(token) puis Task.Delay(80)), donc le premier token est disponible dès que la requête est dépilée du canal entrant. Le dernier delta, 97 ms, confirme la symétrie de l’autre bout : l’événement done suit un Delay complet après le dernier token, il n’est pas collé à lui.

Ce que cette mesure ne dit pas. Les valeurs sont prises sur 127.0.0.1, sans proxy ni compression. Un intermédiaire qui bufferise — certains reverse-proxys le font par défaut sur text/event-stream — réintroduirait exactement le comportement de /stream sans qu’une ligne du service change. La mesure est donc à refaire côté déploiement, pas seulement côté code : c’est précisément ce que l’exercice 4 outille.

// EXERCICE 4 : distinguer un vrai flux d'un buffer, sans regarder le serveur.
//
// Le client ci-dessus fait confiance a l'en-tete `text/event-stream`. Mais un
// serveur peut annoncer cet en-tete et ecrire quand meme tout d'un coup : seule
// la MESURE tranche. Ecrire ici une fonction qui rend un verdict.
//
//   1. Prendre en entree la liste des instants d'arrivee (en ms) des lignes `data:`.
//   2. Calculer les deltas successifs, puis leur mediane.
//   3. Rendre "FLUX" si la mediane depasse un seuil passe en parametre,
//      "BUFFER" sinon -- un buffer produit des deltas quasi nuls entre les lignes,
//      puisqu'elles sortent du meme paquet.
//   4. Verifier le verdict sur les deux jeux d'instants ci-dessous.
//
// Indice : la mediane resiste au PREMIER delta, qui ne mesure pas la cadence.
// Le service ecrit son token avant d'attendre (`WriteAsync` puis `Delay`), donc
// le premier arrive nettement plus tot que les suivants (quelques dizaines de
// ms contre ~80-100). Une moyenne, elle, se laisse tirer vers le bas par lui.

// Ordre de grandeur d'un flux cadence a ~80 ms (cf. la sortie de A4 -- les
// valeurs exactes changent a chaque execution, la FORME ne change pas).
var arriveesFlux = new long[] { 30, 125, 213, 314, 402 };
// Meme charge utile rendue d'un seul paquet : les lignes se suivent sans attente.
var arriveesBuffer = new long[] { 400, 400, 401, 401, 402 };

string Verdict(long[] arrivees, long seuilMs)
{
    // TODO Etudiant : calculer les deltas, leur mediane, puis comparer au seuil.
    return null;
}

Console.WriteLine("Exercice 4 a completer : implementer Verdict(arrivees, seuilMs).");
Console.WriteLine($"  attendu pour arriveesFlux   (seuil 40) : FLUX   -- obtenu : {Verdict(arriveesFlux, 40) ?? "(non implemente)"}");
Console.WriteLine($"  attendu pour arriveesBuffer (seuil 40) : BUFFER -- obtenu : {Verdict(arriveesBuffer, 40) ?? "(non implemente)"}");
Exercice 4 a completer : implementer Verdict(arrivees, seuilMs).
  attendu pour arriveesFlux   (seuil 40) : FLUX   -- obtenu : (non implemente)
  attendu pour arriveesBuffer (seuil 40) : BUFFER -- obtenu : (non implemente)

Conclusion

Le pattern complet d’un agent streaming en C# tient en trois briques qui se composent :

Brique API Rôle
File d’événements System.Threading.Channels Writer/Reader, backpressure, flux await foreach
Cycle de vie BackgroundService ExecuteAsync, arrêt coopératif par CancellationToken
Exposition Minimal API typée .NET 10 TypedResults, handler en classe, injection DI
Conservation de la cadence TypedResults.ServerSentEvents IAsyncEnumerable<SseItem<T>>, écriture au fil de l’eau, text/event-stream

Dans le projet StreamingAgent.App/, ces trois briques forment un service réel : un AgentService hôte (canaux inbound/outbound) exposé par des endpoints typés (/health, /greet, /stream, /stream-sse), vérifiable par curl. Les deux derniers servent le même canal sous deux contrats : le buffer JSON efface la cadence, le flux SSE la conserve — et c’est la mesure des deltas d’arrivée de A4, pas la prose, qui départage. Le même contrat — flux d’événements, cycle de vie hôte, endpoints typés — gouverne les services de la pile GenAI orchestrée par Aspire (les notebooks 01 et 02).

Pour aller plus loin : brancher l’observabilité (même famille, grain #11516 A9) — ActivitySource au moment de chaque token, trace OTLP du flux complet. Le mécanisme de l’observabilité dans la série SemanticKernel s’applique tel quel à ce service.

Retour au sommet