Streaming de eventos enviados pelo servidor

Configurar um EventFlow

Para ter um controle mais refinado sobre a resposta do SSE, a Apigee fornece um fluxo especial de endpoint de destino chamado EventFlow. No contexto de um EventFlow, é possível adicionar um conjunto limitado de políticas, listadas abaixo, para modificar a resposta do SSE antes que ela seja transmitida de volta ao cliente. Para saber mais sobre fluxos de proxy, consulte Como controlar proxies de API com fluxos.

Um EventFlow precisa ser colocado dentro da definição de TargetEndpoint, conforme mostrado na exemplo de código a seguir:

<TargetEndpoint name="default">
  <Description/>
  <FaultRules/>
  <PreFlow name="PreFlow">
    <Request/>
    <Response/>
  </PreFlow>
  <PostFlow name="PostFlow">
    <Request/>
    <Response/>
  </PostFlow>
  <Flows/>
  <EventFlow name="EventFlow" content-type="text/event-stream">
    <Response/>
  </EventFlow>
  <HTTPTargetConnection>
    <Properties/>
    <URL>https://httpbun.org/sse</URL>
  </HTTPTargetConnection>
</TargetEndpoint>

O EventFlow tem dois atributos:

• name: um nome para identificar o fluxo.
• content-type: o valor desse atributo precisa ser text/event-stream.

Consulte também a referência de configuração do fluxo.

É possível adicionar até quatro políticas ao elemento Response do EventFlow. Como em todos os fluxos, as políticas são executadas na ordem em que são adicionadas, e você pode incluir etapas condicionais para controlar a execução delas. É importante observar que os tipos de políticas que podem ser adicionadas a um EventFlow são restritos ao seguinte. Nenhum outro tipo de política é permitido em um EventFlow:

• JavaScript
• MessageLogging
• PublishMessage
• RaiseFault
• SanitizeModelResponse

Consulte também Como anexar e configurar políticas na interface e Como anexar e configurar políticas em arquivos XML.

O exemplo a seguir mostra um EventFlow com uma etapa de política RaiseFault condicional adicionada:

<TargetEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>Raise-Fault-Cred-Invalid</Name>
        <Condition>fault.name equals "invalid_access_token"</Condition>
      </Step>
    </Response>
  </EventFlow>
  <HTTPTargetConnection>
</TargetEndpoint></pre>

Para mais exemplos de código EventFlow, consulte a seção Casos de uso e exemplos do EventFlow.

Variáveis de fluxo

Um EventFlow preenche duas variáveis de fluxo de resposta. Essas variáveis só podem ser usadas no escopo do evento atual que está sendo processado no EventFlow. Acessar ou definir essas variáveis fora do escopo EventFlow não tem efeito. Elas só fazem sentido no contexto do EventFlow.

• response.event.current.content: uma string que contém toda a resposta do evento atual. A Apigee não analisa a string de forma alguma. Ela contém toda a resposta inalterada, incluindo todos os campos de dados.
• response.event.current.count: conta incrementalmente o número de eventos de resposta enviados. Esse valor é atualizado para cada evento recebido. A contagem será 1 para o primeiro evento e aumentará para os eventos subsequentes.

Consulte também a referência de variáveis de fluxo.

Casos de uso e exemplos do EventFlow

Os exemplos a seguir mostram como implementar casos de uso comuns para proxies SSE:

• Modificar uma resposta de SSE
• Filtrar uma resposta de SSE
• Enviar um evento SSE para um sistema externo
• Usar uma política do Apigee Model Armor em um EventFlows
• Tratamento de erros no EventFlow

Modificar uma resposta de SSE

Este exemplo mostra como remover dados de uma resposta de SSE EventFlow antes de retorná-la ao cliente. O conteúdo da resposta do SSE é armazenado em uma variável de fluxo chamada response.event.current.content. Nesse caso, usamos uma política JavaScript para recuperar o valor da variável de fluxo, analisar e modificar. Consulte também Variáveis de fluxo.

1. Crie um novo proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
2. Abra o proxy no editor de proxy do Apigee e clique na guia Desenvolver.
3. Crie uma política de JavaScript com a seguinte definição. Neste exemplo, o código JavaScript é incluído diretamente na política. Colocar o código JavaScript em um arquivo de recursos é outra opção para configurar a política.

   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <Javascript continueOnError="false" enabled="true" timeLimit="200" name="js-update-resp">
     <DisplayName>js-update-resp</DisplayName>
     <Properties/>
     <Source>
       var event = JSON.parse(context.getVariable("response.event.current.content"));
       event.modelVersion = null;
       context.setVariable("response.event.current.content",JSON.stringify(event));
     </Source>
   </Javascript>

4. Adicione a política JavaScript ao EventFlow do proxy. O EventFlow é anexado ao TargetEndpoint padrão. Este exemplo usa a API Gemini na Vertex AI para gerar conteúdo.

   <TargetEndpoint name="default">
     <EventFlow content-type="text/event-stream">
       <Response>
         <Step>
           <Name>js-update-resp</Name>
         </Step>
       </Response>
     </EventFlow>
     <HTTPTargetConnection>
       <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse</URL>
     </HTTPTargetConnection>
   </TargetEndpoint>
   

5. Salve e implante o proxy.
6. Chame o proxy implantado:

   curl -X POST -H 'Content-Type: application/json'  \
     "https://YOUR_APIGEE_ENVIRONMENT_GROUP_HOSTNAME/YOUR_API_PATH" \
     -d '{ "contents":[{"parts":[{"text": "Write a story about a magic pen."}]}]}'

   Mostrar um exemplo de resposta

   Esta é uma resposta de exemplo sem filtros aplicados. A resposta inclui um atributo modelVersion": "gemini-2.5-flash".

   data: {
       "candidates": [
         {
           "content": {
             "parts": [
               {
                 "text": "ara found the pen tucked away in a dusty antique shop, nestled amongst chipped tea"
               }
             ],
             "role": "model"
           }
         }
       ],
       "usageMetadata": {
         "promptTokenCount": 8,
         "totalTokenCount": 8
       },
       "modelVersion": "gemini-2.5-flash"
     }

   Esta é outra resposta de exemplo com a política de JavaScript aplicada. O atributo modelVersion é removido.

   data: {
       "candidates": [
         {
           "content": {
             "parts": [
               {
                 "text": " the fantastical creatures of her imagination.  The quiet beauty of a simple life was a magic all its own.\n"
               }
             ],
             "role": "model"
           },
           "finishReason": "STOP"
         }
       ],
       "usageMetadata": {
         "promptTokenCount": 8,
         "candidatesTokenCount": 601,
         "totalTokenCount": 609,
         "promptTokensDetails": [
           {
             "modality": "TEXT",
             "tokenCount": 8
           }
         ],
         "candidatesTokensDetails": [
           {
             "modality": "TEXT",
             "tokenCount": 601
           }
         ]
       }
     }

Filtrar uma resposta de SSE

Este exemplo mostra como filtrar dados de uma resposta de SSE antes de retorná-los ao cliente. Nesse caso, filtramos os dados de eventos da resposta usando uma política JavaScript. A política analisa a resposta do evento em JSON, modifica o JSON para remover os dados do evento e envia os dados de resposta modificados de volta ao cliente.

Assim como no exemplo anterior, este exemplo recupera o valor da variável de fluxo response.event.current.content e o analisa em JSON. Em seguida, aplica a lógica para implementar a filtragem pretendida.

1. Crie um novo proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
2. Abra o proxy no editor de proxy do Apigee e clique na guia Desenvolver.
3. Crie uma política de JavaScript com a seguinte definição. Neste exemplo, o código JavaScript é incluído diretamente na política. Colocar o código JavaScript em um arquivo de recursos é outra opção para configurar a política.

   <Javascript continueOnError="false" enabled="true" timeLimit="200" name="js-filter-resp">
     <DisplayName>js-filter-resp</DisplayName>
     <Properties/>
     <Source>
       var event = JSON.parse(context.getVariable("response.event.current.content"));
       if("error" in event){
         // Do not send event to customer
         context.setVariable("response.event.current.content", "");
       }
     </Source>
   </Javascript>

4. Adicione a política JavaScript ao EventFlow do proxy. O EventFlow é anexado ao TargetEndpoint padrão. Este exemplo usa a API Gemini na Vertex AI para gerar conteúdo.

   <TargetEndpoint name="default">
     <EventFlow content-type="text/event-stream">
       <Response>
         <Step>
           <Name>js-filter-resp</Name>
         </Step>
       </Response>
      </EventFlow>
     <HTTPTargetConnection>
   	  <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse	</URL>
     </HTTPTargetConnection>
   </TargetEndpoint>
   

5. Salve e implante o proxy.
6. Chame o proxy implantado:

   curl -X POST -H 'Content-Type: application/json'  \
       "https://YOUR_APIGEE_ENVIRONMENT_GROUP_HOSTNAME/YOUR_API_PATH" \
       -d '{ "contents":[{"parts":[{"text": "Write a story about a magic pen."}]}]}'

   Mostrar um exemplo de resposta

   Confira um exemplo de como a resposta pode aparecer sem aplicar filtros. Ele inclui dados de erros:

   data: {
       "candidates": [
         {
           "content": {
             "parts": [
               {
                 "text": "El"
               }
             ],
             "role": "model"
           }
         }
       ],
       "usageMetadata": {
         "promptTokenCount": 8,
         "totalTokenCount": 8
       },
       "modelVersion": "gemini-2.5-flash"
     }
       data: {
       "error": "Service temporarily unavailable. We are experiencing high traffic.",
       "modelVersion": "gemini-2.5-flash"
       }

   Confira outro exemplo de resposta após a aplicação da filtragem com a mensagem de erro limpa.

   data: {
     "candidates": [
       {
         "content": {
           "parts": [
             {
               "text": "El"
             }
           ],
           "role": "model"
         }
       }
     ],
     "usageMetadata": {
       "promptTokenCount": 8,
       "totalTokenCount": 8
     },
     "modelVersion": "gemini-2.5-flash"
   }
   data: {
     "candidates": [
       {
         "content": {
           "parts": [
             {
               "text": "ara found the pen tucked away in a dusty antique shop, nestled amongst chipped tea"
             }
           ],
           "role": "model"
         }
       }
     ],
     "usageMetadata": {
       "promptTokenCount": 8,
       "totalTokenCount": 8
     },
     "modelVersion": "gemini-2.5-flash"
   }

Enviar um evento SSE para um sistema externo

Neste exemplo, anexamos a política PublishMessage do Apigee ao EventFlow para enviar um evento SSE a um tópico do Pub/Sub.

1. Crie um novo proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
2. Abra o proxy no editor de proxy do Apigee e clique na guia Desenvolver.
3. Crie uma política PublishMessage com a seguinte definição:

   <PublishMessage continueOnError="false" enabled="true" name="PM-record-event">
     <DisplayName>PM-record-event</DisplayName>
     <Source>{response.event.current.content}</Source>
     <CloudPubSub>
       <Topic>projects/<customer_project>/topics/<topic_name></Topic>
     </CloudPubSub>
   </PublishMessage>

4. Adicione a política PublishMessage como uma etapa no EventFlow do proxy de API.

   <TargetEndpoint name="default">
     <EventFlow content-type="text/event-stream">
       <Response>
         <Step>
           <Name>PM-record-event</Name>
         </Step>
       </Response>
     </EventFlow>
     <HTTPTargetConnection>
   </TargetEndpoint>

5. Implante e teste o proxy de API.
6. Com o conteúdo gerado adicionado ao tópico do Pub/Sub, você pode, por exemplo, criar uma função do Cloud Run para processar mensagens do tópico.

Usar uma política do Apigee Model Armor em um EventFlow

Você pode usar a política SanitizeModelResponse para limpar eventos enviados pelo servidor em um EventFlow. Essa política protege seus aplicativos de IA ao limpar as respostas de modelos de linguagem grandes (LLMs). Para saber mais sobre o Model Armor, consulte Visão geral do Model Armor. Para informações sobre as políticas do Apigee Model Armor, consulte Começar a usar as políticas do Apigee Model Armor.

1. Crie um novo proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
2. Abra o proxy no editor de proxy do Apigee e clique na guia Desenvolver.
3. Crie uma política SanitizeModelResponse com a seguinte definição:

     <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
     <SanitizeModelResponse async="false" continueOnError="false" enabled="true" name="SMR-modelresponse">
       <IgnoreUnresolvedVariables>true</IgnoreUnresolvedVariables>
       <DisplayName>SMR-modelresponse</DisplayName>
       <ModelArmor>
         <TemplateName>projects/{project}/locations/{location}/templates/{template-name}</TemplateName>
       </ModelArmor>
       <LLMResponseSource>{response_partial}</LLMResponseSource>
       <!-- Use the below settings if you want to call a Model Armor policy on every event -->
       <LLMResponseSource>{response.event.current.content}</LLMResponseSource>
     </SanitizeModelResponse>

4. (Opcional) Adicione uma política JavaScript para agrupar eventos antes de enviá-los à política do Apigee Model Armor.

   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <Javascript continueOnError="false" enabled="true" timeLimit="200" name="JS-combine-resp">
     <DisplayName>JS-combine-events</DisplayName>
     <Properties/>
     <Source>
       var eventText = JSON.parse(context.getVariable("response.event.current.content").substring(5)).candidates[0].content.parts[0].text;
       var finishReason = JSON.parse(context.getVariable("response.event.current.content").substring(5)).candidates[0].finishReason;
       var idx = context.getVariable("response.event.current.count");
       if(idx%5==0 || finishReason=="STOP") {
         context.setVariable("response_partial", context.getVariable("tmp_buffer_pre"));
         context.setVariable("buff_ready", true);
         context.setVariable("tmp_buffer_pre", "");
       } else {
         context.setVariable("buff_ready", false);
         context.setVariable("response_partial", "");
         var previousBufferVal = context.getVariable("tmp_buffer_pre");
         if(previousBufferVal) {
           context.setVariable("tmp_buffer_pre", previousBufferVal+eventText);
         } else {
           context.setVariable("tmp_buffer_pre", eventText);
         }
       }
     </Source>
   </Javascript>

5. Adicione as políticas JavaScript e ModelArmor a uma etapa do EventFlow do proxy:

   <EventFlow name="EventFlow" content-type="text/event-stream">
     <Request/>
     <Response>
       <Step>
         <Name>JS-combine-resp</Name>
       </Step>
       <Step>
         <!-- Remove below Condition if you want to call model armor policy on every event -->
         <Condition> buff_ready = true </Condition>
         <Name>SMR-modelresponse</Name>
       </Step>
     </Response>
   </EventFlow>

6. Implante e teste o proxy de API.

Tratamento de erros no EventFlow

Por padrão, o fluxo de eventos termina quando ocorre uma falha. No entanto, se você quiser fazer mais depuração, envie informações de falha para o Cloud Logging, conforme mostrado neste exemplo.

1. Crie um novo proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
2. Abra o proxy no editor de proxy do Apigee e clique na guia Desenvolver.
3. Crie uma política RaiseFault com a seguinte definição:

   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <RaiseFault continueOnError="false" enabled="true" name="RF-Empty-Event">
     <DisplayName>RF-Empty-Event</DisplayName>
     <Properties/>
     <FaultResponse>
       <AssignVariable>
         <Name>faultReason</Name>
         <Value>empty-event</Value>
       </AssignVariable>
     </FaultResponse>
     <IgnoreUnresolvedVariables>true</IgnoreUnresolvedVariables>
   </RaiseFault>

4. Anexe a política RaiseFault ao EventFlow do proxy SSE:

   <EventFlow content-type="text/event-stream">
     <Response>
       <Step>
         <Name>RF-Empty-Event</Name>
         <Condition>response.event.current.content ~ "data: "</Condition>
       </Step>
     </Response>
   </EventFlow>

5. Crie uma política MessageLogging para registrar erros. Exemplo:

   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <MessageLogging continueOnError="false" enabled="true" name="ML-log-error">
     <DisplayName>ML-log-error</DisplayName>
     <CloudLogging>
       <LogName>projects/{organization.name}/logs/apigee_errors</LogName>
       <Message contentType="text/plain">Request failed due to {faultReason}.</Message>
       <ResourceType>api</ResourceType>
     </CloudLogging>
     <logLevel>ALERT</logLevel>
   </MessageLogging>

6. Adicione a política MessageLogging às FaultRules do endpoint de destino:

   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <TargetEndpoint name="TargetEndpoint-1">
     <Description/>
     <FaultRules>
       <FaultRule name="default-fault">
         <Step>
           <Name>ML-log-error</Name>
         </Step>
       </FaultRule>
     </FaultRules>
     ...
   </TargetEndpoint>

7. Implante e teste o proxy de API.

Como visualizar dados de SSE no Apigee Analytics

Os dados dos proxies de SSE aparecem na Análise da Apigee como esperado para qualquer proxy de API. No console do Cloud, acesse Analytics > Métricas de API.

Como depurar proxies SSE

Use a ferramenta de depuração do Apigee para depurar proxies SSE. Os dados de depuração são capturados para EventFlow da mesma forma que para os outros tipos de fluxo.

Solução de problemas

Para problemas de tráfego em tempo real, verifique os registros de acesso da Apigee para determinar a causa.

Limitações

As seguintes limitações se aplicam aos proxies SSE:

• Como os dados de análise são registrados depois que a sessão SSE é fechada, pode haver um atraso na geração de relatórios.
• Falhas em um EventFlow fazem com que o stream seja encerrado imediatamente, e nenhum evento de erro específico é gerado para o cliente final. Para informações sobre como registrar manualmente esses tipos de erros, consulte Casos de uso e exemplos do EventFlow.
• Um cliente que recebe respostas SSE transmitidas vai receber os cabeçalhos HTTP, incluindo códigos de status, no início do fluxo de eventos. Como resultado, se o fluxo de eventos entrar em um estado de erro, o código de status recebido inicialmente não vai refletir o estado de erro.

  Essa limitação pode ser vista ao visualizar uma sessão de depuração. Na sessão, você pode notar que o código de status HTTP para streams que entram no estado de erro é diferente dos códigos de status enviados ao cliente. Isso pode acontecer porque a entrada da sessão de depuração é gerada depois que toda a solicitação é processada, e não no início do fluxo de eventos. A sessão de depuração pode refletir o código de falha gerado pelo erro, enquanto o cliente só vê o status 2xx recebido inicialmente nos cabeçalhos.