Adicionar saída de stream personalizada ao seu agente

Visão geral

Você pode configurar a saída de fluxo personalizado para instruir seu agente a emitir uma saída estruturada e legível por máquina para invocações de streaming, além de texto sem formatação. O Atlas Agent Engine envia essa saída para clientes como eventos personalizados.

Este guia mostra como executar as seguintes tarefas:

• Ativar saída de stream personalizada: defina o sinalizador features.use_custom_parser: true no seu arquivo agent.yaml.

• Registre um Analisador de Saída: Registre uma subclasse OutputParser em seu aplicativo.

• Emitir eventos personalizados: emita eventos personalizados a partir do código do seu agente .

Pré-requisitos

Antes de começar, certifique-se de ter um agente registrado com um arquivo agent.yaml. Para saber mais, consulte Introdução ao Atlas Agent Engine.

Habilitar saída de fluxo personalizado

Para ativar a saída de stream personalizada para o seu agente, adicione o seguinte código ao seu arquivo agent.yaml:

features:
  use_custom_parser: true

Se você definir o sinalizador features.use_custom_parser como true, mas não registrar um analisador de saída, o agente falhará quando uma invocação for iniciada e retornará um RuntimeError.

Registrar um analisador de saída

Registre uma subclasse OutputParser em seu aplicativo usando o instalador @app.output_parser. O Atlas Agent Engine valida a classe no momento da decoração, portanto, a passagem de uma classe que não é uma subclasse OutputParser falha quando o módulo é importado.

Para agentes LangGraph, subclasse a classeLangGraphOutputParser . Defina o atributo de classe stream_modes para os modos de fluxo do LangGraph que você deseja converter em eventos personalizados. Em seguida, implemente os métodos parse() e on_stream_error().

O exemplo a seguir registra um analisador de saída para um 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"}

A tabela a seguir descreve os membros LangGraphOutputParser:

Se o método parse() ou on_stream_error() gerar um erro, o stream falhará em vez de descartar silenciosamente a saída.

Emitir eventos personalizados

Durante uma invocação de streaming, você pode usar a função emit_custom_event() para emitir eventos personalizados diretamente de nós do LangGraph, ferramentas ou qualquer outro código que seja executado na execução. Passe um objeto JSON contendo chaves de string para emit_custom_event(), como mostrado no exemplo a seguir:

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

O Atlas Agent Engine encaminha o objeto JSON inalterado e não impõe um esquema, portanto, você pode definir a estrutura da carga útil para corresponder ao que seu cliente espera.

Ao chamar a função emit_custom_event(), as seguintes regras se aplicam:

• Chame o método somente a partir de uma invocação de streaming. Chamando-o a partir do caminho invoke síncrono gera um erro.

• Ative o sinalizador features.use_custom_parser no arquivo agent.yaml. Se o sinalizador estiver desativado, a plataforma reporta um erro.

• Aguarde o método do código assíncrono. Em código síncrono, use emit_custom_event_sync() em vez disso.

Próximos passos

Para saber como um cliente recebe a saída transmitida,consulte Transmitir a saída do seu agente no guia Invoke an Agent.