Construire un agent de coding en Zig : le streaming SSE

Cinq outils, une bonne mécanique d'agent — et pourtant, à chaque tour, l'utilisateur attendait silencieusement 3, 5, parfois 15 secondes avant que la réponse arrive d'un bloc. Cet article résout ça : le streaming SSE. Les tokens apparaissent sur stdout à mesure que le modèle les produit. C'est ce qui transforme Hypercode d'une démo en quelque chose qu'on a envie d'utiliser.

Le code reste sur github.com/alexisbchz/hypercode.

Pourquoi le streaming, vraiment

L'argument évident — « c'est plus réactif » — n'est pas le seul. Streamer change la perception du temps. Sans streaming, l'utilisateur ne sait pas si le modèle a bloqué, si on attend les outils, si le réseau est mort. Avec streaming, chaque token est un signe de vie. Une réponse de 10 secondes en bloc est insupportable ; la même réponse en flux est confortable.

Côté agent, c'est aussi un signal de finish_reason. Quand on voit data: [DONE], on sait que ce tour est fini. On peut passer à la suite — exécuter des outils, demander à l'utilisateur — sans deviner.

La forme du flux SSE

OpenRouter, comme OpenAI, envoie un Server-Sent Events stream quand on passe stream: true. Chaque événement est une ligne préfixée par data: :

data: {"id":"...","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}

data: {"id":"...","choices":[{"delta":{"content":" world"},"finish_reason":null}]}

data: {"id":"...","choices":[{"delta":{},"finish_reason":"stop"}]}

data: [DONE]

Trois choses changent par rapport au non-streaming :

1. Le champ s'appelle delta (pas message) — chaque chunk contient l'incrément.
2. Les événements sont séparés par une ligne vide (\n\n).
3. La fin du stream est marquée par data: [DONE].

Pour les tool calls, c'est plus tordu. Au lieu d'un chunk unique qui contient le tool call entier, le modèle envoie des fragments, chacun marqué par un index :

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_abc","function":{"name":"read","arguments":""}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"path"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\":\"src/main.zig\"}"}}]}}]}
data: {"choices":[{"finish_reason":"tool_calls"}]}
data: [DONE]

L'id, le name, et les arguments arrivent en morceaux. Il faut les réassembler par index avant de pouvoir exécuter l'outil.

Heureusement, OpenAI/OpenRouter ne mélange pas dans la pratique : un tour est soit du texte streamé, soit des tool_calls réassemblés. Pas les deux.

Activer le streaming

Une ligne dans le builder de body :

try s.objectField("stream");
try s.write(true);

Et la requête repart. À partir de là, la réponse est un flux SSE au lieu d'un JSON unique.

De client.fetch à client.request

client.fetch est confortable mais il attend la fin avant de remplir notre buffer. Pour streamer, on descend d'un cran et on parle directement au reader de la réponse :

var req = client.request(.POST, uri, .{
    .headers = .{ .accept_encoding = .omit },
    .extra_headers = &.{
        .{ .name = "authorization", .value = auth },
        .{ .name = "content-type", .value = "application/json" },
        .{ .name = "accept", .value = "text/event-stream" },
    },
}) catch return .network_error;
defer req.deinit();

req.sendBodyComplete(payload) catch return .network_error;

var redirect_buf: [constants.stream_redirect_buffer_bytes]u8 = undefined;
var response = req.receiveHead(&redirect_buf) catch return .network_error;
if (response.head.status != .ok) return .{ .http_status = @intFromEnum(response.head.status) };

var transfer_buf: [constants.stream_transfer_buffer_bytes]u8 = undefined;
const reader = response.reader(&transfer_buf);

Boucle de lecture, ligne par ligne

while (true) {
    const raw = reader.takeDelimiterInclusive('\n') catch |err| switch (err) {
        error.EndOfStream => break,
        else => return .network_error,
    };
    const line = std.mem.trim(u8, raw, "\r\n");
    if (line.len == 0) continue;
    if (!std.mem.startsWith(u8, line, "data: ")) continue;
    const data = line[6..];
    if (std.mem.eql(u8, data, "[DONE]")) break;

    const parsed = std.json.parseFromSlice(Delta, gpa, data, .{ .ignore_unknown_fields = true }) catch continue;
    defer parsed.deinit();
    if (parsed.value.choices.len == 0) continue;
    const delta = parsed.value.choices[0].delta;

    if (delta.content) |c| if (c.len > 0) {
        try out_writer.writeAll(c);
        try out_writer.flush();
        try text_acc.writer.writeAll(c);
    };

    if (delta.tool_calls) |calls| {
        for (calls) |c| try absorb_chunk(&accumulators, c);
    }
}

Six points de pédagogie :

1. takeDelimiterInclusive('\n') au lieu de takeDelimiterExclusive.

Dans le Post 04 sur la conversation, on a découvert que takeDelimiterExclusive retourne le contenu sans le délimiteur — mais ne consomme pas le \n non plus. Sur une ligne vide, ça produit "" à chaque appel sans avancer. Boucle infinie. Le variant Inclusive consomme le \n proprement ; on l'enlève après avec std.mem.trim.

2. if (line.len == 0) continue;

Les lignes vides sont les séparateurs d'événements SSE. On les ignore.

3. if (!std.mem.startsWith(u8, line, "data: ")) continue;

SSE permet aussi des commentaires (:), des event: et id: qu'on n'utilise pas. On filtre.

4. [DONE] est une chaîne littérale, pas du JSON.

C'est un marqueur de fin de stream propre à OpenAI. On en sort.

5. parseFromSlice ... catch continue;

Un chunk malformé ne fait pas planter le tour entier. On l'ignore et on continue. C'est rare en pratique, mais l'API publique d'OpenRouter peut servir n'importe quoi.

6. out_writer.flush() après chaque chunk.

Sans flush, le buffer de stdout (std.Io.File.Writer) retient les bytes. Avec flush, ils partent au terminal immédiatement. C'est ce qui rend l'effet « tokens qui apparaissent » visible.

Réassembler les tool calls

Une fonction pour absorber un fragment :

const ToolAccumulator = struct {
    used: bool = false,
    id: std.Io.Writer.Allocating,
    name: std.Io.Writer.Allocating,
    arguments: std.Io.Writer.Allocating,
};

fn absorb_chunk(accumulators: *[constants.tool_calls_per_response_max]ToolAccumulator, c: ChunkToolCall) !void {
    if (c.index >= accumulators.len) return;
    const acc = &accumulators[c.index];
    acc.used = true;
    if (c.id) |id| try acc.id.writer.writeAll(id);
    if (c.function) |f| {
        if (f.name) |n| try acc.name.writer.writeAll(n);
        if (f.arguments) |a| try acc.arguments.writer.writeAll(a);
    }
}

C'est un tableau de 16 accumulateurs (tool_calls_per_response_max). Chaque chunk a un index ; on l'utilise pour router le fragment vers le bon accumulateur. Trois Allocating writers par accumulateur : id, name, arguments. Chacun grandit au rythme des fragments.

À la fin du stream, on collecte ceux qui ont vu au moins un fragment :

fn collect_tool_calls(gpa, accumulators) ![]const ToolCall {
    var count: usize = 0;
    for (accumulators.*) |acc| if (acc.used) { count += 1; };
    if (count == 0) return &.{};

    const owned = try gpa.alloc(ToolCall, count);
    var i: usize = 0;
    for (accumulators) |*acc| {
        if (!acc.used) continue;
        owned[i] = .{
            .id = try acc.id.toOwnedSlice(),
            .name = try acc.name.toOwnedSlice(),
            .arguments_json = try acc.arguments.toOwnedSlice(),
        };
        i += 1;
    }
    return owned;
}

toOwnedSlice transfère la propriété de l'accumulateur vers le slice retourné — pas de copie. Les accumulateurs deviennent vides, deinit au retour est gratuit.

Le contrat de call change

Avant, call retournait un Result.text: []const u8 à la fin. Maintenant, le texte est écrit au fur et à mesure sur un writer qu'on lui passe. À la fin il rend quand même la version complète, pour qu'on l'ajoute à la session.

pub fn call(
    gpa: std.mem.Allocator,
    io: std.Io,
    api_key: []const u8,
    model: []const u8,
    messages: []const session.Message,
    out_writer: *std.Io.Writer,  // ← nouveau
) !Result {

Le caller (main) passe stdout. Pendant que call tourne, les tokens apparaissent en direct. Quand la fonction retourne avec .text, on a le texte complet — utilisé par session.append_assistant_text pour la mémoire multi-turn.

const result = try openrouter.call(gpa, io, cfg.api_key, cfg.model, session.messages.items, stdout);
switch (result) {
    .text => |text| {
        try stdout.writeAll("\n");
        try stdout.flush();
        try session.append_assistant_text(text);
        return;
    },
    .tool_calls => |calls| {
        try session.append_assistant_tool_calls(calls);
        for (calls) |c| try run_one_tool(gpa, io, session, stderr, c);
    },
    // ...
}

Le \n final est nécessaire — le modèle ne termine pas ses réponses par un newline, alors le prompt suivant collerait à la dernière ligne.

Pourquoi accumuler le texte aussi

L'alternative — passer stdout et récupérer le texte écrit dessus — demanderait un tee writer (un writer qui écrit vers deux destinations). Faisable, mais un vtable de plus à implémenter. Plus simple : openrouter.call écrit sur out_writer et dans son propre buffer, double l'I/O en mémoire (négligeable), et garde le code linéaire.

if (delta.content) |c| if (c.len > 0) {
    try out_writer.writeAll(c);    // visible à l'utilisateur
    try out_writer.flush();
    try text_acc.writer.writeAll(c);  // pour le retour
};

C'est le coût de la simplicité : trois lignes claires plutôt qu'un Writer custom.

Test à chaud

./zig-out/bin/hypercode "Say hi in 3 short words."

Le texte apparaît mot par mot dans le terminal. Pas d'attente.

Hi there friend!

Avec un outil :

./zig-out/bin/hypercode "Read src/main.zig and tell me what the answer function does, in 2 sentences."

→ read({"path": "src/main.zig"})
The `answer` function runs an agent loop that calls the OpenRouter API to send
the conversation history to an LLM and receive a response. If the LLM returns
tool calls, it executes them via `run_one_tool` and continues the loop; if it
returns text, it outputs the answer and returns.

Le tool call s'affiche immédiatement (un seul chunk, non-stream), puis le texte de la réponse arrive en streaming. Confortable.

Les commits

969473f feat(openrouter): stream chat completions; accumulate tool_calls by index

Un commit pour le streaming complet. La refacto précédente — Message en union étiquetée, ownership transfer — a rendu la transition propre : on ne touche qu'à openrouter.zig et à six lignes de main.zig. Le reste du code ne sait même pas que la réponse arrive en flux.

Conclusion

Hypercode est maintenant fluide. Le modèle parle, on lit. Si le modèle veut un outil, on l'exécute, et le cycle continue. C'est l'UX qu'un assistant de coding doit avoir.

Pour les cinq outils (read, write, edit, bash, grep) plus le streaming, la base d'un agent fonctionnel est là. Il reste deux gros morceaux dans la série :

• La persistance : sauvegarder l'historique de conversation pour pouvoir reprendre après un crash ou un redémarrage. Append-only journal, replay au démarrage.
• La sandbox : protéger l'utilisateur d'un bash mal intentionné. Probablement bwrap sur Linux, sandbox-exec sur macOS.

Dans le prochain article, on attaque la persistance. C'est ce qui transforme un agent stateless en quelque chose qui ressemble à une session de travail.

Bloqué ou envie de partager vos notes ? Rejoins le serveur Discord.