Intégration de l'agent Pi : analyse de messages, réessai et annulation

19 juin 2026

Participation de l’IA

L’IA a aidé à la rédaction, à l’édition ou à la relecture de cette page.

Vérifiez soigneusement les points importants et confirmez les décisions critiques à partir de sources primaires si nécessaire.

Intégration de l’agent Pi : analyse de messages, réessai et annulation

Pour intégrer un agent AI sous forme de CLI, on ne peut éviter trois choses : comment traduire son flux d’événements privé en messages stables, qui est responsable de la réessai après échec, et comment arrêter proprement le processus lorsque l’utilisateur clique sur annuler. En réalité, ces trois choses reviennent à “distinguer les responsabilités”, mais quand on s’y met vraiment, on réalise que c’est beaucoup plus complexe que prévu.

Contexte

Récemment, je travaillais sur un projet d’assistant de code AI, et l’un des agents à intégrer était pi. C’est un coding agent TUI/CLI qui, lorsqu’il s’exécute, émet des événements JSON ligne par ligne sur stdout. Cela semble simple – lancer le processus, lire la sortie, analyser – mais quand on s’y met vraiment, on découvre que “intégrer un agent CLI” est complètement différent de “intégrer une CLI ordinaire”.

Avec une CLI ordinaire, vous lisez stdout, récupérez un code de sortie, et c’est terminé. Mais les agents CLI ont trois caractéristiques particulièrement gênantes :

Premièrement, leur flux d’événements est un protocole privé. turn_start, session, message_update, message_end, turn_end, agent_end – ce sont des définitions spécifiques à pi, pas des standards de l’industrie. Chaque couche qui veut le consommer doit le traiter séparément, ce qui revient à faire fuiter les détails internes de pi partout. C’est comme regarder une personne de loin : vous pensez la comprendre, mais vous ne voyez que ce qu’elle veut vous montrer.

Deuxièmement, leurs sémantiques d’échec sont particulièrement ambiguës. Pendant l’exécution de l’agent, il peut y avoir des fluctuations réseau, des limitations de modèle, des plantages de processus. Dans ce cas, faut-il réessayer ? Où réessayer ? Le réessai va-t-il perturber l’état de session déjà écrit en partie ? C’est une décision d’architecture, pas quelque chose qui peut être résolu en écrivant simplement une boucle for.

Troisièmement, ils sont longs et interrompibles. Un tour peut durer des dizaines de secondes ou même quelques minutes, et l’utilisateur peut vouloir annuler à tout moment. Lors de l’annulation, le processus ne doit pas devenir orphelin, les appels d’outils ne doivent pas laisser des produits semi-finis, et le contenu déjà émis ne doit pas être perdu. C’est beaucoup plus profond qu’on ne l’imagine.

Pour résoudre ces problèmes, nous avons pris le temps de structurer le chemin d’intégration. Nous en parlerons en détail plus tard, mais je vous donne déjà un aperçu : la vraie difficulté n’est pas de “lancer le processus”, mais de “distinguer les responsabilités”.

À propos de HagiCode

La solution présentée dans cet article provient du projet HagiCode – un assistant de code AI qui prend en charge plusieurs modèles et plusieurs backends d’agent CLI. Dépôt GitHub : HagiCode-org/site, n’hésitez pas à mettre une étoile. Tout le code et tous les problèmes rencontrés décrits ci-dessous sont réellement utilisés dans ce projet. En fait, c’est juste pour se souvenir de tout.

Architecture en couches

HagiCode divise l’intégration des capacités AI en deux couches :

• La couche inférieure est Hagicode.Libs, qui fournit des primitives de provider réutilisables ICliProvider<TOptions>, spécialement responsables de “lancer un agent CLI et normaliser sa sortie en flux de messages partagés”.
• La couche supérieure est hagicode-core, qui fournit un adaptateur mince au niveau du projet IAIProvider, responsable de “traduire les demandes métier en paramètres du provider, consommer le flux de messages partagés, et exposer des chunks de streaming unifiés”.

L’intégration de Pi suit ce chemin. La couche inférieure PiProvider lance le processus pi, lit le flux d’événements JSON, et le normalise en messages partagés ; la couche supérieure PiCliProvider traduit AIRequest en PiOptions, consomme CliMessage, et émet AIStreamingChunk.

Ces trois choses – analyse de messages, réessai, annulation – sont respectivement situées à trois endroits différents : PiJsonEventMapper, une proposition d’archivage qui semble étrange, et CliProcessManager. Parlons-en un par un.

Analyse de messages : comment transformer les événements privés Pi en messages partagés

pi émet des événements JSON ligne par ligne en mode --mode json --print. Cet ensemble d’événements est privé à pi et ne doit pas fuir directement vers la couche supérieure, sinon chaque consommateur devrait coupler les détails internes de pi, et si pi met à jour sa structure d’événements, tout le projet devrait être modifié. En réalité, cette fuite est comme écrire ses pensées sur son visage – c’est fatigant pour les autres et pas forcément confortable pour soi-même.

Nous avons utilisé PiJsonEventMapper pour faire une couche de traduction, normalisant les événements pi en CliMessage partagés. CliMessage est défini dans HagiCode.Libs.Core/Transport/CliMessage.cs, sa structure est très simple, c’est juste un record (Type, Content). La relation de mappage est approximativement la suivante :

événement pi	message partagé	usage
session	session.started / session.resumed	cycle de vie de session
message_update (type text)	assistant	incréments de corps de streaming
message_update (type thinking)	assistant.thought	chaîne de pensée
message_update (type tool)	tool.call / tool.update	initiation d’appel d’outil
message_end / turn_end (toolResult)	tool.completed / tool.failed	résultats d’outil
turn_end / agent_end	terminal.completed	fin de ce tour
sortie non nulle / échec d’analyse	terminal.failed	échec final

Ce tableau n’est qu’un guide rapide, il contient deux techniques clés qui ont été découvertes après des erreurs, méritant d’être détaillées.

Technique 1 : convertir cumulative snapshot en delta

C’est le point le plus facile où on plante. L’événement message_update de pi n’émet pas d’incréments, mais le texte complet cumulé – chaque fois qu’un token arrive, il réémet le “texte complet jusqu’à maintenant”.

Si vous transmettez directement le contenu reçu au frontend, les utilisateurs verront le contenu se répéter : le premier est “你”, le deuxième est “你好”, le troisième est “你好，”, le quatrième est “你好，世”… Le frontend pensera que ce sont quatre sorties indépendantes. En réalité, ce genre de répétition, vu une fois c’est nouveau, vu dix fois c’est fatiguant.

La solution est la comparaison de préfixes, pour calculer le vrai incrément :

// Clé : pi envoie un snapshot cumulé, pas un incrément
// Utilisez la comparaison de préfixes pour extraire l'incrément, sinon le frontend verra du contenu répété
if (text.StartsWith(_lastAssistantTextSnapshot, StringComparison.Ordinal))
{
    var delta = text[_lastAssistantTextSnapshot.Length..];
    _lastAssistantTextSnapshot = text;
    return delta.Length == 0 ? null : delta;
}

Il y a aussi un piège caché : relecture de préfixe entre tours. Après la fin de l’appel d’outil, quand assistant recommence à parler, pi va réémettre le texte précédent depuis le début. Si vous ne gardez qu’un snapshot global, vous traiterez le contenu relu comme un incrément, ce qui provoquera une répétition après l’appel d’outil. PiProviderTests a un cas de test spécifique ExecuteAsync_deduplicates_replayed_assistant_prefix_after_tool_turns qui couvre ce scénario. En d’autres termes, les snapshots avant et après l’appel d’outil doivent être alignés, ils ne peuvent pas être traités indépendamment.

Technique 2 : buffer thinking jusqu’à la fin du tour

La chaîne de pensée (thinking) ne doit pas être émise à chaque token reçu. pi va insérer beaucoup de fragments de pensée pendant les appels d’outils, si vous les transmettez en temps réel, l’ordre du flux deviendra un désordre – tantôt le corps assistant, tantôt les fragments de pensée, tantôt tool.call. Est-ce que ça a du sens ? En réalité, ça n’a pas beaucoup de sens, ça ajoute juste du chaos.

Notre approche est : quand on reçoit un événement thinking, on le met d’abord dans BufferThinkingSnapshot pour stockage temporaire, et on attend message_end ou turn_end avec stopReason != "toolUse" pour DrainBufferedThinkingMessages uniformément. Ainsi, les fragments de pensée pendant les appels d’outils ne pollueront pas le flux principal, et à la fin du tour, on donne le processus de pensée complet en une seule fois.

Tolérance aux erreurs : les mauvaises lignes ne doivent pas faire planter le flux

Les agents CLI ne sont pas des systèmes idéaux des manuels, ils émettent parfois une ligne non JSON, ou un JSON sans champ type. Si vous lancez une exception ici, tout le flux meurt, les utilisateurs ne voient rien. Après tout, le monde réel n’est pas toujours parfait, qui peut garantir que chaque ligne est régulière ?

Notre stratégie est : si une ligne échoue à être analysée, le flux n’est pas interrompu, mais collectée dans _invalidOutputLines. Une fois le processus terminé, dans Complete(), ces “mauvaises lignes” sont assemblées dans le texte de diagnostic de terminal.failed. Ainsi, quand les utilisateurs voient une erreur, ils peuvent voir directement ce que pi a émis de chaotique, plutôt qu’un simple “erreur d’analyse”.

Réessai : provider ne le fait pas, qui le fait ?

C’est le piège le plus facile à éviter dans toute l’intégration. Intuitivement, “intégrer une CLI devrait inclure le réessai”, mais HagiCode a activement supprimé tout réessai automatique au niveau provider dans une proposition d’archivage. La proposition s’appelle remove-provider-auto-retry-support.

Pourquoi pas de réessai automatique

Le contexte de la proposition est très direct. La logique de réessai était à l’origine dispersée à deux endroits : il y en avait une dans Hagicode.Libs (replay fresh-runtime style OpenCode), et une autre dans hagicode-core (ProviderErrorAutoRetryCoordinator). Les deux faisaient leur propre chose, ce qui faisait que “réessayer ou pas” devenait un comportement implicite caché dans le provider, qui changeait furtivement le timing d’échec, le mode de reprise de session et le flux d’état de chat.

Réfléchissez un peu : l’utilisateur envoie un message, le provider réessaie trois fois en interne, les deux premiers échouent, le troisième réussit. La couche supérieure ne sait pas ce qui s’est passé entre-temps, l’état de session, le comptage de tokens, la progression UI ne correspondent pas. Ce genre de comportement implicite, comment dire, c’est un poison lent dans l’architecture.

Ainsi la frontière a été réduite à une phrase :

Le provider converge vers une sémantique de tentative unique, l’appelant doit considérer l’état sans réessai comme résultat normal d’exécution unique.

À quoi ça ressemble pour PiProvider

En code, ce sont trois choses :

• Dans PiOptions, aucun champ lié au réessai – pas de maxAttempts, pas de retryDelay, pas de retryClassifier.
• ExecuteAsync se termine après l’exécution d’un processus pi, l’échec donne directement terminal.failed.
• Les classificateurs précédemment au service du réessai automatique (ClaudeCodeRetryableTerminalFailureClassifier, CodexRetryableTerminalFailureClassifier, etc.) tant qu’ils servent uniquement le réessai automatique, tous sont supprimés du chemin actif.

Mais notez, la capacité de réessai n’a pas disparu, elle a juste été déplacée vers le haut. La proposition écrit explicitement “laisser une frontière stable pour une reprise ultérieure unifiée par une couche supérieure”. La DTO de configuration providerErrorAutoRetry, la normalisation, la sérialisation, le round-trip de la page de paramètres frontend sont tous conservés, mais elle ne pilote plus l’exécution du provider. Après tout, certaines choses ne sont pas vraiment abandonnées, juste conservées d’une autre manière.

Et si on veut réessayer

Si vous voulez ajouter un réessai au-dessus de pi, la bonne façon de faire est dans l’appelant de PiCliProvider – par exemple, votre couche d’orchestration de session (dans HagiCode c’est SessionGrain d’Orleans, le frontend peut être une couche d’orchestration de chat). Après avoir obtenu terminal.failed, vous jugez vous-même si c’est réessai, vous décidez du délai et du nombre, puis vous envoyez encore un ExecuteAsync.

Un modèle minimal viable ressemble à ceci :

// La logique de réessai est placée chez l'appelant, ne la remettez pas dans PiProvider
// Sinon ça détruit la frontière "tentative unique" que le provider vient d'établir
async Task<AIResponse> ExecuteWithRetryAsync(AIRequest req, int maxAttempts, CancellationToken ct)
{
    for (var attempt = 1; ; attempt++)
    {
        var response = await provider.ExecuteAsync(req, ct);

        // Succès ou limite atteinte, on retourne
        if (response.FinishReason != FinishReason.Unknown || attempt >= maxAttempts)
            return response;

        // Réessai uniquement pour les échecs finaux réessaiables (réseau, 5xx, plantage de processus)
        // model rejected, auth failure, ce genre de réessai n'a pas de sens, ne réessayez pas
        await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), ct);
    }
}

La logique de classification “réessai” n’est maintenant plus dans le provider, l’appelant la définit lui-même. La configuration providerErrorAutoRetry (maxAttempts, retryDelay, enabled) peut toujours être lue depuis la page de paramètres frontend, mais ce qui pilote vraiment le réessai est votre couche d’orchestration, pas PiProvider. Répétez ceci trois fois.

Annulation : transmission de token + arrêt en trois phases

Pour l’annulation, PiProvider n’implémente presque rien, déléguant tout à CliProcessManager, PiProvider n’est responsable que de deux choses : passer le CancellationToken, et faire le nettoyage en cas d’exception.

Transmission sur toute la chaîne

La chaîne est comme ça, transmise jusqu’au bout :

CancellationToken de l'appelant
  → PiCliProvider.StreamCoreAsync(cancellationToken)
  → PiProvider.ExecuteProcessAsync([EnumeratorCancellation] cancellationToken)
  → ReadLineAsync(cancellationToken) / WaitForExitAsync(cancellationToken)
  → En cas d'exception _processManager.StopAsync(handle, CancellationToken.None)

Notez la dernière ligne : pour le nettoyage, on utilise CancellationToken.None, pas le token transmis par l’utilisateur. C’est un détail, mais extrêmement important.

La raison est : le token de l’utilisateur est déjà annulé. Si vous utilisez ce token déjà annulé pour faire le nettoyage, la tâche de nettoyage sera immédiatement annulée, le processus deviendra orphelin – pi continue de tourner en arrière-plan, personne ne le récupère, CPU et mémoire occupés inutilement. Donc le nettoyage doit utiliser CancellationToken.None, pour s’assurer que l’action de nettoyage s’exécute certainement jusqu’au bout. En fait, c’est comme avec les gens, certaines choses doivent être bien terminées après qu’il se soit complètement arrêté, sinon on laisse juste un gâchis.

Arrêt progressif en trois phases

CliProcessManager.StopProcessAsync est un processus d’arrêt progressif en trois phases, les constantes de temps sont définies en haut du fichier :

// Patience pour l'arrêt gracieux : donnez d'abord au processus le temps de terminer lui-même
private static readonly TimeSpan GracefulStopTimeout = TimeSpan.FromSeconds(2);
// Patience après le kill forcé pour attendre que le processus se termine vraiment
private static readonly TimeSpan StopWaitTimeout = TimeSpan.FromSeconds(5);

Les trois phases progressent ainsi :

1. Signal d’interruption. TryInterruptAsync écrit d’abord un \u0003 dans stdin (c’est le caractère Ctrl+C), et sous Unix en plus kill -INT <pid>. Cette étape est pour laisser pi terminer gracieusement lui-même – il peut percevoir l’interruption et terminer ce qu’il est en train d’écrire.
2. Attente gracieuse. Attendez au plus 2 secondes, voyez si le processus se termine lui-même.
3. Kill forcé. S’il ne s’est pas terminé, faites directement Process.Kill(entireProcessTree: true), tuez tout l’arbre de processus ensemble, et attendez au plus 5 secondes pour confirmer qu’il est vraiment mort.

Pourquoi entireProcessTree: true ? Parce que pi peut générer des processus enfants quand il exécute des outils – par exemple le processus de modèle local vers lequel le provider route, le sous-processus bash qui tourne. Si vous ne tuez que le processus parent, les processus enfants deviendront orphelins et continueront à tourner. Tuer tout l’arbre ensemble est propre.

Sous Windows il n’y a pas de SIGINT, on ne peut compter que sur le caractère Ctrl+C, donc le comportement multi-plateforme différera, il faut s’en souvenir.

Nettoyage des exceptions de PiProvider

Le ExecuteProcessAsync de PiProvider, quand ReadLineAsync lance une exception, utilise ExceptionDispatchInfo.Capture pour stocker temporairement l’exception, après avoir sorti de la boucle appelle StopAsync pour nettoyer le processus, puis pendingException.Throw() relance l’exception originale vers la couche supérieure.

Pourquoi stocker puis relancer ? Parce que si vous lancez directement, le processus n’a pas le temps d’être récupéré, il devient orphelin ; si vous lancez avant StopAsync, la logique de nettoyage ne sera jamais atteinte. Stockez temporairement, assurez d’abord que le processus est certainement récupéré, puis préservez complètement la sémantique originale de OperationCanceledException pour l’appelant – l’appelant reçoit cette exception, peut juger “ah, c’est l’utilisateur qui a annulé activement”, pas “il y a une erreur”.

Contrat unifié pour les échecs de démarrage

Il y a un détail qui mérite d’être mentionné séparément. Échec de démarrage de processus – par exemple l’exécutable pi n’existe pas, les permissions sont incorrectes – PiProvider ne lance pas d’exception, mais synthétise un message terminal.failed, puis yield break.

Pourquoi faire ainsi ? Parce que si vous lancez une exception, l’appelant de la couche supérieure doit gérer deux sémantiques complètement différentes : l’une est “messages normaux pendant la consommation de flux”, l’autre est “exception lancée avant même de commencer le flux”. Cela rendra await foreach de l’appelant particulièrement difficile à écrire.

Après unifier en “toujours d’abord vous donner des messages, puis terminer le flux”, la logique de l’appelant devient cohérente : obtenir terminal.failed c’est un échec, obtenir terminal.completed c’est un succès, pas besoin de try/catch en branchements séparés. C’est une petite mais importante décision de conception, qui stabilise le contrat.

Pratique : la bonne façon de consommer le flux

Référez-vous à PiScenarioMessageReader (scénario de test console de libs) et PiCliProvider.StreamCoreAsync (adaptateur mince de core) dans HagiCode, l’appelant ressemble approximativement à ceci :

await foreach (var message in provider.ExecuteAsync(options, prompt, cancellationToken))
{
    // 1. En cas d'échec, priorité au court-circuit, ne pas traiter les messages suivants
    if (NormalizedAcpCliAdapter.TryGetFailureMessage(message.Content, out var failure))
    {
        yield return new AIStreamingChunk { Type = StreamingChunkType.Error, ErrorMessage = failure };
        yield break;   // après terminal.failed le flux se termine
    }

    // 2. Le texte assistant est un snapshot cumulé, faire un autre calcul d'incrément soi-même
    if (message.Type == "assistant" && TryGetText(message.Content, out var text))
    {
        var delta = ReconcileSnapshot(text);  // comparaison de préfixes
        if (!string.IsNullOrEmpty(delta)) yield return Chunk(delta);
    }

    // 3. terminal.completed est le seul signal fiable de "fin"
    if (message.Type == "terminal.completed") break;
}

Guide rapide des pièges courants

Organisons les pièges rencontrés dans ce parcours en un tableau, pour faciliter aux suivants :

phénomène	cause	traitement
Le frontend voit du texte assistant répété	Pas fait cumulative vers delta	Utilisez ReconcileAssistantTextSnapshot pour la comparaison de préfixes
Le processus continue de tourner après annulation	Nettoyage avec un token déjà annulé	Changez pour CancellationToken.None pour le nettoyage
Le réessai ne fonctionne pas	A mis le réessai dans PiProvider, mais provider a sémantique de tentative unique	Déplacez vers la couche d’orchestration de l’appelant
Perte des messages d’erreur de pi	Pas lu les champs de diagnostic de terminal.failed	Transmettez complètement text / invalid_output_lines / stderr
Reçoit des fragments de pensée pendant les appels d’outils	A transmis directement les événements thinking	Buffer jusqu’à la fin du tour puis DrainBufferedThinkingMessages

Comment vérifier

La couche libs utilise StubCliProcessManager pour mock le processus, les tests unitaires couvrent la construction de paramètres, la normalisation d’événements, la déduplication d’incréments, la transmission d’échecs pour ces logiques pures. Le chemin CLI réel utilise la variable d’environnement HAGICODE_REAL_CLI_TESTS pour opt-in, et utilise des modèles réels pour les scénarios trip. Les PiCliProviderTests de la couche core vérifient la projection AIStreamingChunk et le binding de session de l’adaptateur mince.

# Exécuter les tests unitaires liés à Pi dans le dépôt Hagicode.Libs
dotnet test --filter "FullyQualifiedName~PiProviderTests"

# Exécuter les tests d'intégration CLI réels (pi doit être installé localement)
HAGICODE_REAL_CLI_TESTS=1 dotnet test --filter "FullyQualifiedName~PiProviderTests.RealCli"

Conclusion

En reliant ces trois choses, le modèle mental de l’intégration de pi est en fait une phrase : laissez chaque couche faire son propre travail.

• L’analyse de messages est confiée à PiJsonEventMapper : événements privés normalisés en CliMessage partagés, cumulative snapshot converti en delta, thinking bufferisé jusqu’à la fin du tour.
• Le réessai est confié à l’appelant : provider tentative unique, qui veut réessayer le fait soi-même dans la couche supérieure, la configuration est conservée mais ne pilote plus le provider.
• L’annulation est confiée à CliProcessManager : CancellationToken transmis sur toute la chaîne, nettoyage avec CancellationToken.None, arrêt progressif en trois phases (signal d’interruption → attente gracieuse → kill forcé de tout l’arbre de processus).

Une fois ces frontières clairement définies, intégrer un nouvel agent CLI devient presque un travail de production en série – vous n’avez qu’à écrire un nouveau XxxProvider et XxxJsonEventMapper, toutes les logiques transversales de réessai, annulation, contrat de messages, gestion d’erreurs sont réutilisées. C’est aussi la raison fondamentale pour laquelle HagiCode peut prendre en charge plusieurs backends d’agent CLI (claude code, codex, pi, gemini cli, etc.) sans devenir chaotique.

Disons encore une fois cette frontière la plus importante : n’ajoutez pas de réessai au niveau provider. Une fois que vous comprenez ce point, l’intégration d’un agent CLI est déjà à moitié faite…

Conclusion

Revenant au thème “Intégration de l’agent Pi : analyse de messages, réessai et annulation”, ce qui vaut vraiment d’être confirmé encore et encore ne sont pas des techniques dispersées, mais si les contraintes, les frontières d’implémentation et les compromis d’ingénierie ont été clairement vus.

Tant que vous consolidez les critères de jugement décrits dans l’article en points de contrôle stables, vous pourrez prendre des décisions fiables plus rapidement face à des problèmes similaires à l’avenir.

Traduction IA

Cette page a été traduite depuis la version originale en chinois simplifié avec l’aide de l’IA.

Consultez l’original si vous devez vérifier la formulation, les nuances techniques ou les dernières mises à jour.

Voir l’original en chinois simplifié

Tags:

• hagicode
• pi
• cli
• provider

MonoSpecs qu'est-ce que c'est : pourquoi c'est une évolution et une extension supplémentaire d'OpenSpec

Intégration de Reasonix 1.x avec DeepSeek V4 : Guide pratique du sélecteur de modèles ACP