Implementación de la integración de Pi Agent: análisis de mensajes, reintentos y cancelación

19 jun 2026

Participación de IA

La IA ayudó a redactar, editar o revisar esta página.

Evalúa con cuidado los detalles importantes y confirma las decisiones críticas con fuentes primarias cuando sea necesario.

Implementación de la integración de Pi Agent: análisis de mensajes, reintentos y cancelación

Al integrar un AI agent con forma de CLI, hay tres cosas inevitables: cómo traducir su flujo de eventos privado en mensajes estables, quién se encarga del reintento cuando falla, y cómo detener el proceso de forma limpia cuando el usuario hace clic en cancelar. En realidad, estas tres cosas se reducen a “separar responsabilidades”, pero solo cuando las haces te das cuenta de la profundidad del problema.

Antecedentes

Recientemente estoy trabajando en un proyecto de asistente de código con IA, y uno de los agentes con los que necesito integrar es pi. Es en sí mismo un coding agent TUI/CLI que, al ejecutarse, emite eventos JSON línea por línea en stdout. Suena simple—iniciar el proceso, leer la salida, analizar—pero cuando te pones manos a la obra descubres que “integrar un agent CLI” y “integrar un CLI normal” son cosas completamente diferentes.

Con un CLI normal, lees stdout, obtienes un código de salida y listo. Pero un agent CLI lamentablemente tiene tres características que causan dolores de cabeza:

Primero, su flujo de eventos es un protocolo privado. turn_start, session, message_update, message_end, turn_end, agent_end—estos están definidos por pi mismo, no son estándares de la industria. Cada capa superior que quiera consumirlo tiene que procesarlo por su cuenta, lo que equivale a exponer los detalles internos de pi por todas partes. Es como ver a una persona a distancia, crees que la ves claramente, pero en realidad solo estás viendo el lado que ella quiere mostrarte.

Segundo, su semántica de fallo es especialmente ambigua. El agent puede fallar por un temblor de red, limitación del modelo, o que el proceso se bloquee durante su ejecución. ¿En ese caso, se debe reintentar o no? ¿Dónde se reintenta? ¿El reintento va a desordenar el estado de sesión que ya se escribió parcialmente? Esta es una decisión de arquitectura, no algo que se resuelva escribiendo simplemente un bucle for.

Tercero, es largo e interrumpible. Un turno puede ejecutarse durante decenas de segundos o incluso minutos, y el usuario puede querer cancelarlo en cualquier momento. Al cancelar, el proceso no debe convertirse en huérfano, las llamadas a herramientas no deben quedar a medias, y el contenido ya emitido no se debe perder. El agua aquí es mucho más profunda de lo que imaginabas.

Para resolver estos puntos dolorosos, dedicamos tiempo a ordenar la ruta de integración. Lo explicaremos en detalle más adelante, pero primero un spoiler: la verdadera dificultad no está en “iniciar el proceso”, sino en “separar responsabilidades”.

Sobre HagiCode

La solución compartida en este artículo proviene del proyecto HagiCode—un asistente de código con IA que soporta múltiples modelos y múltiples backends de agent CLI. Repositorio GitHub: HagiCode-org/site, bienvenido a darle una estrella. Todo el código a continuación, todos los huecos que pisamos, se están ejecutando realmente en este proyecto. De hecho, escribirlo es solo para dejarse un recuerdo.

Estratificación general

HagiCode divide la integración de capacidades de IA en dos capas:

• La capa inferior es Hagicode.Libs, que proporciona el primitivo de proveedor reutilizable ICliProvider<TOptions>, especializado en “iniciar un agent CLI y normalizar su salida en un flujo de mensajes compartido”.
• La capa superior es hagicode-core, que proporciona un thin adapter a nivel de proyecto IAIProvider, responsable de “traducir solicitudes de negocio en parámetros del proveedor, consumir el flujo de mensajes compartido y exponer un flujo de chunk unificado”.

La integración de Pi sigue este camino. La capa inferior PiProvider inicia el proceso pi, lee el flujo de eventos JSON, lo normaliza en mensajes compartidos; la capa superior PiCliProvider traduce AIRequest en PiOptions, consume CliMessage y emite AIStreamingChunk.

Estas tres cosas—análisis de mensajes, reintentos, cancelación—caen respectivamente en tres lugares diferentes: PiJsonEventMapper, una propuesta de archivo que parece extraña, y CliProcessManager. Hablemos de cada una.

Análisis de mensajes: cómo convertir los eventos privados de Pi en mensajes compartidos

pi emite eventos JSON línea por línea en --mode json --print. Este conjunto de eventos es privado de pi, y absolutamente no debe filtrarse directamente a la capa superior, de lo contrario cada consumidor tendría que acoplarse a los detalles internos de pi, y si pi actualiza la estructura de eventos, todo el proyecto seguiría el cambio. De hecho, este tipo de fuga no es muy diferente a escribir los pensamientos en la cara—los demás se cansan de mirar, y uno mismo tampoco se siente necesariamente cómodo.

Usamos PiJsonEventMapper para hacer una capa de traducción, normalizando los eventos de pi en CliMessage compartidos. CliMessage se define en HagiCode.Libs.Core/Transport/CliMessage.cs, su estructura es muy simple, solo es un record (Type, Content). La relación de mapeo es aproximadamente la siguiente:

evento pi	mensaje compartido	uso
session	session.started / session.resumed	ciclo de vida de sesión
message_update (tipo text)	assistant	incremento de texto de streaming
message_update (tipo thinking)	assistant.thought	cadena de pensamiento
message_update (tipo tool)	tool.call / tool.update	iniciación de llamada a herramienta
message_end / turn_end (toolResult)	tool.completed / tool.failed	resultado de herramienta
turn_end / agent_end	terminal.completed	final de turno actual
salida no cero / fallo de análisis	terminal.failed	fallo de estado final

Esta tabla es solo una referencia rápida, pero contiene dos técnicas clave que descubrimos después de pisar huecos, y vale la pena explicarlas.

Técnica 1: convertir cumulative snapshot en delta

Este es el punto donde más fácil es tener un accidente. El evento message_update de pi no emite incrementos, sino texto completo acumulado—cada vez que llega un token, vuelve a emitir “el texto completo hasta el momento”.

Si reenvías directamente el contenido recibido al frontend, el usuario verá el contenido repetirse una y otra vez: la primera es “你”, la segunda es “你好”, la tercera es “你好，”, la cuarta es “你好，世”… El frontend pensará que son cuatro salidas independientes. De hecho, la repetición, vista una vez es frescura, vista diez veces es solo aburrimiento.

La solución es la comparación de prefijos, calculando el verdadero incremento:

// Clave: pi emite snapshots acumulados, no incrementos
// Usa comparación de prefijos para sacar el incremento, de lo contrario el frontend verá contenido repetido
if (text.StartsWith(_lastAssistantTextSnapshot, StringComparison.Ordinal))
{
    var delta = text[_lastAssistantTextSnapshot.Length..];
    _lastAssistantTextSnapshot = text;
    return delta.Length == 0 ? null : delta;
}

Aquí también hay un hueco oculto: reproducción de prefijos entre turnos. pi, cuando finaliza una llamada a herramienta y el assistant continúa hablando, volverá a emitir desde el principio el texto anterior. Si solo guardas un snapshot global, tratarás el contenido reproducido como un incremento, lo que causará que aparezca un segmento repetido después de la llamada a herramienta. PiProviderTests tiene un caso de uso dedicado ExecuteAsync_deduplicates_replayed_assistant_prefix_after_tool_turns que cubre este escenario. En otras palabras, los snapshots antes y después de la llamada a herramienta deben alinearse en el procesamiento, no cada uno por su lado.

Técnica 2: thinking se debe bufferizar hasta el final del turno

La cadena de pensamiento (thinking) no se puede emitir hacia afuera cada vez que se recibe un token. pi insertará muchos fragmentos de pensamiento durante una llamada a herramienta, y si se reenvían en tiempo real, el orden del flujo se convertirá en una sopa—un momento es el texto del assistant, otro momento son fragmentos de pensamiento, otro momento又是 tool.call. ¿Tiene sentido esto? De hecho, no tiene mucho sentido, solo aumenta el caos.

Nuestro enfoque es: cuando se reciben eventos thinking, primero se colocan en BufferThinkingSnapshot para almacenamiento temporal, y cuando llega message_end o turn_end y stopReason != "toolUse", se hace DrainBufferedThinkingMessages de manera unificada. Así los fragmentos de pensamiento durante las llamadas a herramientas no contaminan el flujo principal, y al final del turno se da todo el proceso de pensamiento completo de una vez.

Tolerancia a fallos: las líneas malas no deben hacer que el flujo se bloquee

El agent CLI no es el sistema ideal de los libros de texto, ocasionalmente emitirá una línea que no es JSON, o un JSON que no tiene el campo type. Si lanzas una excepción aquí, todo el flujo muere, y el usuario no ve nada. Después de todo, el mundo real siempre tiene algunas imperfecciones, ¿quién puede garantizar que cada línea sea ordenada y regulada?

Nuestra estrategia es: cualquier línea que falle el análisis, no interrumpe el flujo, sino que se recopila en _invalidOutputLines. Cuando el proceso termina, en Complete() se ensamblan estas “líneas malas” en el texto de diagnóstico de terminal.failed. Así cuando el usuario ve el error, puede ver directamente qué cosas extrañas emitió pi, en lugar de un seco “parse error”.

Reintentos: ¿quién los hace si la capa de proveedor no lo hace?

Este es el hueco más fácil de pisar en toda la integración. Intuitivamente “integrar un CLI debería venir con reintentos”, pero HagiCode en una propuesta de archivo eliminó activamente todos los reintentos automáticos de la capa de proveedor. La propuesta se llama remove-provider-auto-retry-support.

Por qué no reintento automático

El fondo de la propuesta está escrito muy directamente. La lógica de reintento originalmente estaba dispersa en dos lugares: había una en Hagicode.Libs (fresh replay de runtime estilo OpenCode), y otra en hagicode-core (ProviderErrorAutoRetryCoordinator). Ambos lados hacían lo suyo, lo que hacía que “si se reintenta o no” se convirtiera en un comportamiento implícito oculto dentro del proveedor, que cambiaría silenciosamente el momento de fallo, la forma de continuar la sesión y el flujo de estado del chat.

Piénsalo y te da dolor de cabeza: el usuario envía un mensaje, el proveedor reintenta internamente tres veces por su cuenta, las dos primeras fallan, la tercera tiene éxito. La capa superior no sabe en absoluto qué pasó en el medio, el estado de sesión, el conteo de tokens, el progreso de la UI, todo está desalineado. Este comportamiento implícito, ¿cómo decirlo? Es veneno lento en la arquitectura.

Así que el límite se redujo a una frase:

El proveedor se reduce a la semántica de intento único, el llamador debe tratar el estado sin reintentos como resultado de ejecución única normal.

¿Cómo queda en PiProvider

Llevado al código, son tres cosas:

• PiOptions no tiene ningún campo relacionado con retry—no hay maxAttempts, no hay retryDelay, no hay retryClassifier.
• ExecuteAsync termina después de que se ejecuta un proceso pi, si falla directamente da terminal.failed.
• Los clasificadores (ClaudeCodeRetryableTerminalFailureClassifier, CodexRetryableTerminalFailureClassifier, etc.) que antes servían al reintento automático, mientras que sirvan puramente al reintento automático, se eliminan todos del camino activo.

Pero por favor nota, la capacidad de reintento no desapareció, solo se movió hacia arriba. La propuesta escribe explícitamente “para dejar un límite estable para que capas más altas tomen el control unificado de reintentos más adelante”. El DTO de la opción de configuración providerErrorAutoRetry, su normalización, serialización, el round-trip de la página de configuración del frontend, todo se mantiene, solo ya no impulsa la ejecución del proveedor. Después de todo, algunas cosas no es que realmente ya no se quieran, solo que se guardan de otra forma.

¿Entonces qué hacer si quieres reintentar

Si quieres agregar reintentos encima de pi, la forma correcta es hacerlo en el llamador de PiCliProvider—por ejemplo, tu capa de orquestación de sesión (en HagiCode es SessionGrain de Orleans, el frontend podría ser la capa de orquestación de chat). Después de obtener terminal.failed, tú mismo juzgas si es reintentable, tú mismo decides el retraso y el número de veces, y envías ExecuteAsync otra vez.

Un patrón mínimamente viable se ve así:

// La lógica de reintento se coloca en el llamador, no se mete de vuelta en PiProvider
// De lo contrario se destruirá el límite de "intento único" que acaba de establecer el proveedor
async Task<AIResponse> ExecuteWithRetryAsync(AIRequest req, int maxAttempts, CancellationToken ct)
{
    for (var attempt = 1; ; attempt++)
    {
        var response = await provider.ExecuteAsync(req, ct);

        // Éxito o alcanzar el límite y regresar
        if (response.FinishReason != FinishReason.Unknown || attempt >= maxAttempts)
            return response;

        // Solo reintentar fallos de estado final reintentables (red, 5xx, colapso de proceso)
        // model rejected, auth failure este tipo de reintento tampoco tiene sentido, no reintentes
        await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, attempt)), ct);
    }
}

La lógica de clasificación de “reintentable” ahora no está dentro del proveedor, el llamador la define por sí mismo. La configuración providerErrorAutoRetry (maxAttempts, retryDelay, enabled) todavía se puede leer desde la página de configuración del frontend, pero lo que realmente impulsa el reintento es tu capa de orquestación, no PiProvider. Por favor repite esto tres veces.

Cancelación: transmisión de token + parada en tres etapas

Para la cancelación, PiProvider casi no la implementa, todo se delega a CliProcessManager, PiProvider solo es responsable de dos cosas: pasar el CancellationToken hacia abajo, y hacer la limpieza cuando hay excepciones.

Transmisión completa a través de la cadena

La cadena es así, se transmite hasta el fondo:

CancellationToken del llamador
  → PiCliProvider.StreamCoreAsync(cancellationToken)
  → PiProvider.ExecuteProcessAsync([EnumeratorCancellation] cancellationToken)
  → ReadLineAsync(cancellationToken) / WaitForExitAsync(cancellationToken)
  → Cuando hay excepción _processManager.StopAsync(handle, CancellationToken.None)

Fíjate en la última línea: al limpiar se usa CancellationToken.None, no el token que el usuario pasó. Este es un detalle, pero es extremadamente importante.

La razón es: el token del usuario ya está cancelado. Si usas este token ya cancelado para hacer la limpieza, la tarea de limpieza se cancelará inmediatamente, y el proceso se convertirá en huérfano—pi sigue ejecutándose en segundo plano, nadie lo recoge, CPU y memoria ocupadas en vano. Así que la limpieza debe usar CancellationToken.None, asegurando que la acción de limpieza definitivamente se pueda ejecutar completamente. De hecho es igual que con las personas, algunas cosas se deben terminar bien después de que se detenga por completo, de lo contrario solo se deja un desastre.

Parada progresiva en tres etapas

CliProcessManager.StopProcessAsync es un proceso de parada progresiva en tres etapas, las constantes de tiempo se definen en la parte superior del archivo:

// Paciencia de parada elegante: primero da tiempo al proceso para que termine solo
private static readonly TimeSpan GracefulStopTimeout = TimeSpan.FromSeconds(2);
// Paciencia de esperar a que el proceso realmente salga después de matarlo por la fuerza
private static readonly TimeSpan StopWaitTimeout = TimeSpan.FromSeconds(5);

Las tres etapas progresan así:

1. Señal de interrupción. TryInterruptAsync primero escribe un \u0003 en stdin (es el carácter Ctrl+C), y en Unix adicionalmente hace kill -INT <pid>. Este paso es para dejar que pi termine elegantemente por sí mismo—puede percibir la interrupción y terminar lo que está escribiendo.
2. Espera elegante. Espera hasta 2 segundos, a ver si el proceso salió por sí mismo.
3. Matar por la fuerza. Si aún no sale, directamente Process.Kill(entireProcessTree: true), matando todo el árbol de procesos junto, y luego espera hasta 5 segundos para confirmar que realmente murió.

¿Por qué entireProcessTree: true? Porque cuando pi ejecuta herramientas generará procesos hijos—por ejemplo el proceso de modelo local al que enruta el proveedor, el proceso hijo bash que ejecuta. Si solo matas el proceso padre, el proceso hijo se convertirá en huérfano y seguirá ejecutándose. Matando todo el árbol junto queda limpio.

En Windows no existe SIGINT, solo se puede confiar en el carácter Ctrl+C, así que el comportamiento multiplataforma tendrá diferencias, esto hay que tenerlo presente.

Limpieza de excepciones de PiProvider

El ExecuteProcessAsync de PiProvider, cuando ReadLineAsync lanza una excepción, usará ExceptionDispatchInfo.Capture para guardar temporalmente la excepción, después de salir del bucle llama a StopAsync para limpiar el proceso, y luego pendingException.Throw() para volver a lanzar la excepción original a la capa superior.

¿Por qué guardar y luego lanzar? Porque si se lanza directamente, el proceso no tuvo tiempo de ser reciclado y se convierte en huérfano; si se lanza antes de StopAsync, la lógica de limpieza simplemente no se ejecuta. Guardarlo temporalmente, primero asegurar que el proceso definitivamente sea reciclado, y luego preservar completamente la semántica de la OperationCanceledException original para el llamador—el llamador al obtener esta excepción, puede juzgar “ah, es cancelación activa del usuario”, no “ocurrió un error”.

Contrato unificado de fallo de inicio

Hay otro detalle que vale la pena mencionar por separado. Fallo de inicio del proceso—por ejemplo el ejecutable pi no existe, los permisos no son correctos—PiProvider no lanza una excepción, sino que sintetiza un mensaje terminal.failed, y luego yield break.

¿Por qué hacerlo así? Porque si se lanza una excepción, el consumidor de la capa superior tendría que manejar dos semánticas completamente diferentes: una es “mensajes normales durante el consumo de streaming”, otra es “excepción lanzada antes de que el streaming comenzara”. Esto haría que el await foreach del consumidor sea especialmente difícil de escribir.

Después de unificar en “primero siempre te doy el mensaje, luego termino el flujo”, la lógica del consumidor se vuelve consistente: obtener terminal.failed cuenta como fallo, obtener terminal.completed cuenta como éxito, no necesita manejo ramificado con try/catch. Esta es una decisión de diseño pequeña pero importante, que estabiliza el contrato.

Práctica: postura correcta para consumir el flujo

Referenciando PiScenarioMessageReader en HagiCode (escenario de prueba de console de libs) y PiCliProvider.StreamCoreAsync (thin adapter de core), el consumidor se ve aproximadamente así:

await foreach (var message in provider.ExecuteAsync(options, prompt, cancellationToken))
{
    // 1. El fallo debe tener prioridad de cortocircuito, no procesar mensajes posteriores
    if (NormalizedAcpCliAdapter.TryGetFailureMessage(message.Content, out var failure))
    {
        yield return new AIStreamingChunk { Type = StreamingChunkType.Error, ErrorMessage = failure };
        yield break;   // después de terminal.failed el flujo termina
    }

    // 2. El texto del assistant es cumulative snapshot, haz otra vez el cálculo de incremento
    if (message.Type == "assistant" && TryGetText(message.Content, out var text))
    {
        var delta = ReconcileSnapshot(text);  // comparación de prefijos
        if (!string.IsNullOrEmpty(delta)) yield return Chunk(delta);
    }

    // 3. terminal.completed es la única señal confiable de "finalización"
    if (message.Type == "terminal.completed") break;
}

Referencia rápida de huecos comunes

Organiza los huecos que pisamos en este camino en una tabla, conveniente para quienes vienen después:

fenómeno	causa	tratamiento
El frontend ve texto de assistant repetido	No se hizo cumulative a delta	Usa ReconcileAssistantTextSnapshot para comparación de prefijos
Después de cancelar el proceso sigue ejecutándose	La limpieza usó un token ya cancelado	Cambia a usar CancellationToken.None para la limpieza
El reintento no entra en vigor	Se escribió el reintento en PiProvider, pero el proveedor tiene semántica de intento único	Mover hacia arriba a la capa de orquestación del llamador
Se pierde el mensaje de error de pi	No se leyó el campo de diagnóstico de terminal.failed	Transmitir completamente text / invalid_output_lines / stderr
Se reciben fragmentos de pensamiento durante llamadas a herramientas	Se reenviaron directamente eventos thinking	Bufferizar hasta el final del turno y luego DrainBufferedThinkingMessages

Cómo verificar

La capa libs usa StubCliProcessManager para mockear el proceso, los unit tests cubren la construcción de parámetros, normalización de eventos, deduplicación de incrementos, transmisión de fallos, etc. La ruta de CLI real usa la variable de entorno HAGICODE_REAL_CLI_TESTS para opt-in, ejecuta escenarios de trip con modelos reales. PiCliProviderTests en la capa core verifica la proyección de AIStreamingChunk del thin adapter y el binding de sesión.

# Ejecuta unit tests relacionados con Pi en el repositorio Hagicode.Libs
dotnet test --filter "FullyQualifiedName~PiProviderTests"

# Ejecuta tests de integración de CLI real (necesita tener pi instalado localmente)
HAGICODE_REAL_CLI_TESTS=1 dotnet test --filter "FullyQualifiedName~PiProviderTests.RealCli"

Resumen

Conectando estas tres cosas, el modelo mental para integrar pi en realidad se reduce a una frase: dejar que cada capa haga solo su propio trabajo.

• El análisis de mensajes se deja a PiJsonEventMapper: eventos privados normalizados en CliMessage compartidos, cumulative snapshot convertido en delta, thinking bufferizado hasta el final del turno.
• Los reintentos se dejan al llamador: el proveedor hace intento único, quien quiere reintentar lo hace en la capa superior por su cuenta, la configuración se mantiene pero ya no impulsa al proveedor.
• La cancelación se deja a CliProcessManager: CancellationToken se transmite a través de toda la cadena, la limpieza usa CancellationToken.None, parada progresiva en tres etapas (señal de interrupción → espera elegante → matar por la fuerza todo el árbol de procesos).

Después de aclarar estos límites, integrar un nuevo agent CLI casi se convierte en trabajo de línea de ensamblaje—solo necesitas escribir un nuevo XxxProvider y XxxJsonEventMapper, y toda la lógica transversal de reintentos, cancelación, contrato de mensajes, manejo de errores se reutiliza. Esta es también la razón fundamental por la que HagiCode puede soportar simultáneamente múltiples backends de agent CLI (claude code, codex, pi, gemini cli, etc.) sin volverse un lío.

Por último, repito ese límite más importante: no agregues reintentos en la capa de proveedor. Una vez que entiendas esto, integrar un agent CLI ya ha pasado más de la mitad…

Resumen

Volviendo al tema “Implementación de la integración de Pi Agent: análisis de mensajes, reintentos y cancelación”, lo que realmente vale la pena confirmar repetidamente no son las técnicas dispersas, sino si las condiciones de restricción, los límites de implementación y las compensaciones de ingeniería ya se han visto con claridad.

Mientras沉淀es los criterios de juicio en el artículo en ítems de verificación estables, al enfrentar problemas similares en el futuro podrás tomar decisiones confiables más rápidamente.

Traducción con IA

Esta página se tradujo desde la versión original en chino simplificado con ayuda de IA.

Consulta el original si necesitas verificar la redacción, los matices técnicos o las actualizaciones más recientes.

Ver original en chino simplificado

Tags:

• hagicode
• pi
• cli
• provider

MonoSpecs ¿Qué es y por qué se dice que es una actualización y extensión de OpenSpec

Integración de Reasonix 1.x con DeepSeek V4: Práctica de integración del selector de modelos ACP