Agregue una salida de flujo personalizada a su agente.

Overview

Puede configurar una salida de flujo personalizada para indicarle a su agente que emita una salida estructurada y legible por máquina para las invocaciones de flujo, además de texto sin formato. El motor del agente Atlas envía esta salida a los clientes como eventos personalizados.

Esta guía muestra cómo realizar las siguientes tareas:

Habilitar la salida de flujo personalizada: Establezca la bandera features.use_custom_parser: true en su archivo agent.yaml.
Registra un analizador de salida: Registra una subclase de OutputParser en tu aplicación.
Emitir eventos personalizados: Emite eventos personalizados desde el código de tu agente.

Requisitos previos

Antes de comenzar, asegúrese de tener un agente registrado con un archivo agent.yaml. Para obtener más información, consulte la sección "Primeros pasos con el motor de agentes de Atlas".

Habilitar salida de flujo personalizada

Para habilitar la salida de flujo personalizada para su agente, agregue el siguiente código a su archivo agent.yaml:

features:
  use_custom_parser: true

Si establece el indicador features.use_custom_parser en true pero no registra un analizador de salida, el agente falla cuando se inicia una invocación y devuelve un RuntimeError.

Registrar un analizador de salida

Registra una subclase OutputParser en tu aplicación usando el decorador @app.output_parser. El motor del agente Atlas valida la clase en el momento de la decoración, por lo que pasar una clase que no sea una subclase OutputParser provocará un error cuando el módulo la importe.

Para los agentes LangGraph, cree una subclase de la clase LangGraphOutputParser. Establezca el atributo de clase stream_modes en los modos de flujo de LangGraph que desea convertir en eventos personalizados. A continuación, implemente los métodos parse() y on_stream_error().

El siguiente ejemplo registra un analizador de salida para un agente LangGraph:

from agent_engine_sdk_langgraph.output_parser import LangGraphOutputParser
@app.output_parser
class BriefParser(LangGraphOutputParser):
    stream_modes = ("messages", "values")
    async def parse(self, item, ctx):
        ...
        yield {"event": "brief", "data": new_brief}
    async def on_stream_error(self, ctx, error):
        ...
        return {"event": "error"}

La siguiente tabla describe los miembros LangGraphOutputParser:

nodo	Descripción
`stream_modes`	Secuencia de nombres de modos de flujo de LangGraph, como `("messages", "values")`. El motor del agente Atlas pasa un elemento de flujo normalizado para cada modo listado al método `parse()`.
`parse(item, ctx)`	Determina los eventos personalizados que se emitirán para un elemento del flujo. Implementa `parse()` como un método asíncrono que devuelve objetos serializables en JSON. Para omitir un elemento, devuelve un valor sin devolver ningún valor.
`on_stream_error(ctx, error)`	Devuelve un evento personalizado opcional serializable en formato JSON para emitir antes de que la plataforma muestre un error de flujo.

Si el método parse() o on_stream_error() genera un error, el flujo falla en lugar de descartar silenciosamente la salida.

Emitir eventos personalizados

Durante una invocación de transmisión, puede usar la función emit_custom_event() para emitir eventos personalizados directamente desde nodos, herramientas o cualquier otro código de LangGraph que se ejecute. Pase un objeto JSON que contenga claves de cadena a emit_custom_event(), como se muestra en el siguiente ejemplo:

from runner_shared import emit_custom_event
await emit_custom_event({"event": "todo", "items": ["a", "b"]})

El motor del agente Atlas reenvía el objeto JSON sin modificaciones y no impone un esquema, por lo que puede definir la estructura de la carga útil para que coincida con lo que espera su cliente.

Al llamar a la función emit_custom_event(), se aplican las siguientes reglas:

Llame al método únicamente desde una invocación de transmisión. Llamarlo desde la ruta síncrona invoke genera un error.
Habilite la bandera features.use_custom_parser en el archivo agent.yaml. Si la bandera está deshabilitada, la plataforma generará un error.
Espere a que finalice el método desde código asíncrono. Desde código síncrono, utilice emit_custom_event_sync() en su lugar.

Próximos pasos

Para saber cómo recibe un cliente la salida transmitida, consulte la sección "Transmitir la salida de su agente" en la guía "Invocar un agente".

Volver

Servidores MCP remotos