O Google usa tecnologia de IA na tradução de conteúdos para seu idioma de preferência. As traduções com IA podem ter erros.

Streaming de eventos enviados pelo servidor

Esta página se aplica à Apigee e à Apigee híbrida.

Confira a documentação da Apigee Edge.

A Apigee oferece suporte ao streaming contínuo de respostas de endpoints de eventos enviados pelo servidor (SSE, na sigla em inglês) para clientes em tempo real. O recurso SSE da Apigee é útil para processar APIs de modelos de linguagem grandes (LLMs) que operam com mais eficiência ao transmitir as respostas de volta ao cliente. O streaming de SSE reduz a latência, e os clientes podem receber dados de resposta assim que eles são gerados por um LLM. Esse recurso permite o uso de agentes de IA que operam em ambientes em tempo real, como bots de atendimento ao cliente ou orquestradores de fluxo de trabalho.

Para usar SSE com a Apigee, basta apontar um proxy de API para um endpoint de destino ou proxy com SSE ativado. Para ter um controle mais refinado sobre a resposta do SSE, a Apigee oferece um fluxo de endpoint especial chamado EventFlow. No contexto de um EventFlow, é possível adicionar um conjunto limitado de políticas para realizar operações na resposta do SSE, como filtragem, modificação ou tratamento de erros. Para saber mais sobre fluxos de proxy, consulte Como controlar proxies de API com fluxos.

Criar um proxy de API para SSE

A UI da Apigee fornece um modelo para criar um novo proxy que inclui um EventFlow.

Observação: o uso do modelo de proxy SSE é opcional. Ele dá acesso a variáveis de fluxo específicas de SSE que podem ser úteis para realizar determinadas operações de política. No entanto, não é necessário usar EventFlow para usar SSE com a Apigee.

Importante: use definições de endpoint de destino separadas para destinos SSE. Misturar endpoints de destino SSE e não SSE pode resultar em um comportamento inconsistente, como variáveis de fluxo response.content vazias. Consulte Problemas conhecidos.

Siga estas etapas para criar um proxy de API com o modelo EventFlow usando a interface da Apigee:

No console do Google Cloud , acesse a página Apigee > Desenvolvimento de proxy > Proxies de API.

Acessar proxies de API
No painel Proxy de API, clique em + Criar.
No painel Criar um proxy, em Modelo de proxy, selecione Proxy com eventos enviados pelo servidor (SSE).
Em Detalhes do proxy, insira o seguinte:
- Nome do proxy: insira um nome para o proxy, como myproxy.
- Caminho base: definido automaticamente para o valor inserido em Nome do proxy. Caminho base faz parte do URL usado para fazer solicitações à API. A Apigee usa o URL para corresponder e rotear solicitações recebidas para o proxy de API apropriado.
- Descrição (opcional): insira uma descrição para o novo proxy de API, como "Como testar a Apigee com um proxy simples".
- Destino (API atual): insira o URL de destino do SSE para o proxy de API. Por exemplo: https://mocktarget.apigee.net/sse-events/5
- Clique em Próxima.
Implantar (opcional):
- Ambientes de implantação: opcional. Use as caixas de seleção para escolher um ou mais ambientes em que o proxy vai ser implantado. Se preferir não implantar o proxy neste ponto, deixe o campo Ambientes de implantação em branco. O proxy poderá ser implantado posteriormente.
- Conta de serviço: opcional Uma conta de serviço para o proxy. A conta de serviço representa a identidade do proxy implantado e determina as permissões dele. Esse é um recurso avançado e, para os fins deste tutorial, você pode ignorá-lo.
Os proxies de API implantados com uma configuração EventFlow serão faturados como Extensíveis.
Clique em Criar.

Consulte também Como criar um proxy de API simples.

Configurar um EventFlow

Para ter um controle mais refinado sobre a resposta do SSE, a Apigee oferece um fluxo de endpoint especial 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.

Posicionamento de um EventFlow

Um 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 Referência de configuração do fluxo.

Um EventFlow pode ser colocado dentro de uma definição de TargetEndpoint ou ProxyEndpoint, conforme mostrado nos exemplos de código a seguir:

<ProxyEndpoint>

<ProxyEndpoint 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>
  <HTTPProxyConnection>
    <Properties/>
    <URL>https://httpbin.org/sse</URL>
  </HTTPProxyConnection>
</ProxyEndpoint>

<TargetEndpoint>

<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://httpbin.org/sse</URL>
  </HTTPTargetConnection>
</TargetEndpoint>

Observação: se você adicionar várias estrofes EventFlow a um endpoint, somente a última na definição do endpoint será executada. As estrofes EventFlow que aparecem antes da última são ignoradas.

Também é importante observar que, embora seja possível adicionar um EventFlow a um TargetEndpoint, ProxyEndpoint ou ambos, apenas um EventFlow é executado.

A tabela a seguir mostra a execução de estrofes EventFlow com base no posicionamento do endpoint:

ProxyEndpoint	TargetEndpoint	EventFlow usado
EventFlow em `ProxyEndpoint`	EventFlow em `TargetEndpoint`	EventFlow em `TargetEndpoint`
Não `EventFlow`	EventFlow em `TargetEndpoint`	EventFlow em `TargetEndpoint`
EventFlow em `ProxyEndpoint`	Não `EventFlow`	EventFlow em `ProxyEndpoint`

Adicionar políticas a um EventFlow

É 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 é possível adicionar 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 aos seguintes. Nenhum outro tipo de política é permitido em um EventFlow:

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

Os exemplos a seguir mostram um EventFlow com uma etapa de política RaiseFault condicional adicionada:

<ProxyEndpoint>

<ProxyEndpoint 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>
  <HTTPProxyConnection>
</ProxyEndpoint></pre>

<TargetEndpoint>

<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 do EventFlow, consulte a seção Casos de uso e exemplos do EventFlow.

Variáveis de fluxo

Um EventFlow preenche três 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 de EventFlow não tem efeito. Eles só são significativos 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. Ele contém toda a resposta inalterada, incluindo todos os campos de dados.
Observação: a Apigee limita cada evento de resposta a um máximo de 10 MB de dados.

Observação: se você definir o valor dessa variável como uma string em branco (""), o evento não será enviado ao cliente. Para ver um exemplo, consulte Filtrar uma resposta de SSE.
response.event.current.data: uma string que contém o payload de dados do evento atual. É possível modificar essa variável no EventFlow para mudar o payload de dados enviado ao cliente.
response.event.current.count: conta de forma incremental o número de eventos de resposta enviados. Esse valor é atualizado para cada evento recebido. A contagem será 1 para o primeiro evento e será incrementada para os eventos subsequentes.
Observação: se você acessar esse valor, em uma política de JavaScript, por exemplo, o primeiro evento terá um valor de 1. No entanto, o número de eventos enviados ao cliente neste momento será 0.

Consulte também 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 Model Armor da Apigee em um EventFlow
Tratamento de erros no EventFlow
Propagar mensagens de falha em um EventFlow
Usar o DataCapture para coletar contagens de tokens em análises
Usar uma política LLMTokenQuota para aplicar cotas de tokens em um EventFlow

Modificar uma resposta de SSE

Este exemplo mostra como remover dados de uma resposta EventFlow de SSE 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 de JavaScript para recuperar o valor da variável de fluxo, analisá-lo e modificá-lo. Consulte também Variáveis de fluxo.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma nova política de JavaScript com a definição a seguir. Neste exemplo, o código JavaScript está incluído diretamente na política. Colocar o código JavaScript em um arquivo de recurso é 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>

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

<ProxyEndpoint>

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

<TargetEndpoint>

<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>

Salve o proxy e implante-o.

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

Este é um exemplo de resposta sem filtragem. 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"
  }

Este é outro exemplo de resposta com a política JavaScript aplicada. O atributo modelVersion foi 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 de eventos e envia os dados de resposta modificados de volta ao cliente.

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.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

<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>

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

<ProxyEndpoint>

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

<TargetEndpoint>

<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>

Salve o proxy e implante-o.

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

Veja um exemplo de como a resposta pode ser sem aplicar nenhum filtro. Observe que ele inclui dados de erro:

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"
    }

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

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 da Apigee ao EventFlow para enviar um evento SSE a um tópico do Pub/Sub.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma nova 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>

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

<ProxyEndpoint>

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

<TargetEndpoint>

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

Implantar e testar o proxy de API.
Com o conteúdo gerado adicionado ao tópico do Pub/Sub, é possível, por exemplo, criar uma função do Cloud Run para processar mensagens do tópico.

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

Use a política SanitizeModelResponse para limpar eventos enviados pelo servidor em um EventFlow. Defina <ContinuousScan> como true para verificar todo o stream, como nas etapas a seguir, ou deixe como false (o padrão) para verificar cada evento separadamente. Com true, a política retém cada evento até que o Model Armor confirme que o texto está limpo e, em seguida, o libera inteiro, inalterado e em ordem para as etapas posteriores e para o cliente.

Para mais informações sobre como a política lida com correspondências, erros e cotas, consulte Limpar respostas de eventos enviados pelo servidor (SSE) de streaming. Para mais informações sobre o Model Armor, consulte a visão geral do Model Armor.

Para configurar a política, siga estas etapas:

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma política SanitizeModelResponse com <ContinuousScan>true</ContinuousScan> e <LLMResponseSource> configurados para extrair o delta de texto de response.event.current.data. O exemplo a seguir é para streams de SSE do Gemini:

<?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>{jsonPath('$.candidates[-1].content.parts[-1].text',response.event.current.data,true)}</LLMResponseSource>
  <ContinuousScan>true</ContinuousScan>
</SanitizeModelResponse>

Para streams SSE compatíveis com a OpenAI, defina <LLMResponseSource> como {jsonPath('$.choices[-1].delta.content',response.event.current.data,true)}. Para streams SSE do Anthropic Claude, defina <LLMResponseSource> como {jsonPath('$.delta.text',response.event.current.data,true)}.

Anexe a política SanitizeModelResponse ao elemento <Response> do EventFlow no proxy. Se você também usar LLMTokenQuota ou DataCapture no EventFlow, coloque SanitizeModelResponse primeiro para que essas políticas recebam apenas eventos verificados:
```
<EventFlow name="EventFlow" content-type="text/event-stream">
  <Request/>
  <Response>
    <Step>
      <Name>SMR-modelresponse</Name>
    </Step>
  </Response>
</EventFlow>
```
Implantar e testar 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 uma depuração extra, envie informações de falha para o Cloud Logging, conforme mostrado neste exemplo.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

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>

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>

Criar 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>

Adicione a política MessageLogging às FaultRules do endpoint de destino ou de proxy:

<TargetEndpoint>

<?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>

<ProxyEndpoint>

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

Implantar e testar o proxy de API.

Propagar erros de falha em um EventFlow

Neste exemplo, mostramos como usar um EventFlow para propagar erros de falha para o cliente. O processo funciona para notificar o cliente sobre erros imediatamente durante a execução da política.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma nova política JavaScript com a seguinte definição:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
  <Javascript continueOnError="false" enabled="true" timeLimit="200" name="js-error">
    <DisplayName>js-error</DisplayName>
    <Properties/>
    <Source>
      if(context.getVariable("response.event.current.count")=="2") {
        throw new Error("Internal Error");
      }
      context.setVariable("response.event.current.content", context.getVariable("response.event.current.content"));
    </Source>
  </Javascript>

Essa política foi projetada para gerar um erro quando uma condição específica é atendida.

Anexe a política JavaScript ao EventFlow do proxy SSE na configuração TargetEndpoint ou ProxyEndpoint. Essa etapa garante que o EventFlow processe a política durante o processamento da resposta:

<TargetEndpoint>

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<TargetEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>js-error</Name>
      </Step>
    </Response>
  </EventFlow>
  <HTTPTargetConnection>
    <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent</URL>
  </HTTPTargetConnection>
</TargetEndpoint>

<ProxyEndpoint>

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<ProxyEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>js-error</Name>
      </Step>
    </Response>
  </EventFlow>
  <HTTPProxyConnection>
    <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent</URL>
  </HTTPProxyConnection>
</ProxyEndpoint>

Implante o proxy de API.

Teste com o comportamento do proxy usando o seguinte comando curl:

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

Substitua ENVIRONMENT_GROUP_NAME pelo nome do grupo de ambientes.

A saída será semelhante ao exemplo a seguir:

data: {"candidates": [{"content": {"parts": [{"text": "El"}],"role": "model"}}],"usageMetadata": {"promptTokenCount": 8,"totalTokenCount": 8},"modelVersion": "gemini-2.5-flash"}
data: {"fault":{"faultstring":"Execution of JS-error failed with error: Exception thrown from JavaScript : Error: Internal Error (Resource_1_js#2)","detail":{"errorcode":"steps.javascript.ScriptExecutionFailed"}}}

A saída mostra o fluxo de dados inicial seguido por uma mensagem fault. No caso de um erro, a Apigee captura as informações de falha e as envia ao cliente como um evento.

Para mais informações sobre o tratamento de falhas na Apigee, consulte Como lidar com falhas.

Usar o DataCapture para coletar contagens de tokens em análises

Este exemplo mostra como usar a política DataCapture em um EventFlow para coletar contagens de tokens de eventos SSE para uso na análise da Apigee.

Criar coletores de dados para contagens de tokens. Execute os comandos a seguir para criar os coletores de dados necessários na sua organização:

curl -X POST "https://apigee.googleapis.com/v1/organizations/ORG_NAME/datacollectors" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dc_tokenCount",
    "description": "Collects token count",
    "type": "INTEGER"
  }'

curl -X POST "https://apigee.googleapis.com/v1/organizations/ORG_NAME/datacollectors" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dc_thoughtsTokenCount",
    "description": "Collects thought token count",
    "type": "INTEGER"
  }'

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma nova política de JavaScript para analisar as contagens de tokens dos dados de eventos SSE. No editor de proxy da Apigee, clique em + ao lado de Políticas, selecione JavaScript como o tipo de política e insira um nome para ela (por exemplo, JS-parse-token). O XML da política resultante será semelhante ao seguinte:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<Javascript continueOnError="false" enabled="true" timeLimit="200" name="JS-parse-token">
  <DisplayName>JS-parse-token</DisplayName>
  <Properties/>
  <Source>
    const event=context.getVariable('response.event.current.content')
    try {
      const jsonString = event.replace(/^data:\s*/, '').trim();
      const parsed = JSON.parse(jsonString);
      const usage = parsed.usageMetadata || {};
      context.setVariable("thoughtsToken", usage.thoughtsTokenCount);
      context.setVariable("totalToken", usage.totalTokenCount);
    } catch (e) {
      context.setVariable("js_error_occurred", true);
    }
  </Source>
</Javascript>

Crie uma nova política DataCapture para capturar as contagens de tokens. No editor de proxy da Apigee, clique em + ao lado de Políticas, selecione DataCapture como o tipo de política e insira um nome para a política (por exemplo, DC-capture-tokencount). O XML da política resultante será semelhante ao seguinte:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<DataCapture name="DC-capture-tokencount" continueOnError="false" enabled="true">
  <DisplayName>DC-capture-tokencount</DisplayName>
  <IgnoreUnresolvedVariables>false</IgnoreUnresolvedVariables>
  <Capture>
    <Collect ref="thoughtsToken" default="-1"/>
    <DataCollector>dc_thoughtsTokenCount</DataCollector>
  </Capture>
  <Capture>
    <Collect ref="totalToken" default="-1"/>
    <DataCollector>dc_tokenCount</DataCollector>
  </Capture>
</DataCapture>

Anexe primeiro a política de JavaScript e, em seguida, a política DataCapture, dentro do EventFlow do proxy SSE na configuração TargetEndpoint ou ProxyEndpoint. Essa etapa garante que o EventFlow processe as políticas durante o processamento da resposta. O XML resultante será semelhante ao seguinte:

<TargetEndpoint>

<TargetEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>JS-parse-token</Name>
      </Step>
      <Step>
        <Name>DC-capture-tokencount</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>

<ProxyEndpoint>

<ProxyEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>JS-parse-token</Name>
      </Step>
      <Step>
        <Name>DC-capture-tokencount</Name>
      </Step>
    </Response>
  </EventFlow>
  <HTTPProxyConnection>
    <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse</URL>
  </HTTPProxyConnection>
</ProxyEndpoint>

Salve e implante o proxy. Para mais detalhes, consulte Como implantar proxies de API.
Envie solicitações de teste para o proxy. Depois que o proxy processa as respostas de SSE, as contagens de tokens capturadas ficam disponíveis no Apigee Analytics pelos coletores de dados dc_tokenCount e dc_thoughtsTokenCount.

Usar uma política LLMTokenQuota para aplicar cotas de tokens em um EventFlow

Este exemplo mostra como usar a política LLMTokenQuota em um EventFlow para contar e aplicar cotas de token para um stream SSE. Ao processar o stream de eventos, a contagem de tokens é executada apenas quando os metadados de uso de tokens da resposta do LLM são encontrados no evento. Quando os metadados de uso do token são descobertos, eles são extraídos e a política é executada. Para todos os outros eventos, a política resulta em uma operação nula.

Observação: quando a política LLMTokenQuota faz parte de um EventFlow, os elementos <LLMTokenUsageSource> e <LLMModelSource> precisam ler os dados do evento atual usando a variável de fluxo response.event.current.data, com o terceiro argumento da função jsonPath definido como false.

Crie um proxy com o modelo de proxy SSE. Consulte Criar um proxy de API com eventos enviados pelo servidor (SSE).
Abra o proxy no editor de proxy da Apigee e clique na guia Desenvolver.

Crie uma nova política LLMTokenQuota para contar os tokens consumidos pelo stream de SSE. No editor de proxy da Apigee, clique em + ao lado de Políticas, selecione LLMTokenQuota como o tipo de política e insira um nome para ela (por exemplo, LTQ-Count-Only). O XML da política resultante será semelhante ao seguinte:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<LLMTokenQuota name="LTQ-Count-Only" type="rollingwindow">
  <SharedName>common-counter</SharedName>
  <CountOnly>true</CountOnly>
  <Allow count="15000"/>
  <Interval>30</Interval>
  <TimeUnit>minute</TimeUnit>
  <Distributed>true</Distributed>
  <LLMTokenUsageSource>{jsonPath('$.usageMetadata.candidatesTokenCount',response.event.current.data,false)}</LLMTokenUsageSource>
  <LLMModelSource>{jsonPath('$.modelVersion',response.event.current.data,false)}</LLMModelSource>
</LLMTokenQuota>

Crie uma segunda política LLMTokenQuota para aplicar a cota de tokens nas solicitações recebidas. No editor de proxy da Apigee, clique em + ao lado de Políticas, selecione LLMTokenQuota como o tipo de política e insira um nome para a política (por exemplo, LTQ-Enforce-Only). Essa política usa o mesmo <SharedName> que a política de contagem, para que ambas compartilhem o mesmo contador de cotas. O XML da política resultante será semelhante ao seguinte:
```
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<LLMTokenQuota name="LTQ-Enforce-Only" type="rollingwindow">
  <SharedName>common-counter</SharedName>
  <EnforceOnly>true</EnforceOnly>
  <Allow count="15000"/>
  <Interval>30</Interval>
  <TimeUnit>minute</TimeUnit>
  <Distributed>true</Distributed>
</LLMTokenQuota>
```
Adicione a política LTQ-Enforce-Only como uma etapa no Request do PreFlow do ProxyEndpoint. Essa etapa garante que a cota seja verificada em cada solicitação recebida. O XML resultante será semelhante a este:
```
<ProxyEndpoint name="default">
  <PreFlow name="PreFlow">
    <Request>
      <Step>
        <Name>LTQ-Enforce-Only</Name>
      </Step>
    </Request>
    <Response/>
  </PreFlow>
</ProxyEndpoint>
```

Adicione a política LTQ-Count-Only como uma etapa dentro do EventFlow do proxy SSE na configuração TargetEndpoint ou ProxyEndpoint. Essa etapa garante que o EventFlow processe a política durante o processamento da resposta e incremente o contador de cotas compartilhado. O XML resultante será semelhante ao seguinte:

<TargetEndpoint>

<TargetEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>LTQ-Count-Only</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>

<ProxyEndpoint>

<ProxyEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>LTQ-Count-Only</Name>
      </Step>
    </Response>
  </EventFlow>
  <HTTPProxyConnection>
    <URL>https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse</URL>
  </HTTPProxyConnection>
</ProxyEndpoint>

Salve e implante o proxy. Para mais detalhes, consulte Como implantar proxies de API.
Envie solicitações de teste para o proxy. Em cada solicitação recebida, a política LTQ-Enforce-Only verifica o contador de cotas compartilhadas e rejeita a solicitação se o limite configurado já tiver sido atingido. À medida que o proxy processa as respostas de SSE, a política LTQ-Count-Only conta os tokens informados nos metadados de uso de tokens de cada evento e incrementa o mesmo contador de cota compartilhado. Quando o contador atinge o limite configurado, as solicitações subsequentes que consomem tokens são rejeitadas até que o contador seja redefinido.

Para mais informações sobre a política LLMTokenQuota, incluindo como usá-la com respostas SSE e como configurar contadores de cota compartilhados, consulte a política LLMTokenQuota.

Como visualizar dados de SSE na análise da Apigee

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

Como depurar proxies SSE

Use a ferramenta de depuração da 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:

Se a política DataCapture for aplicada a todos os eventos, somente o valor mais recente será registrado.
Embua uma conexão SSE possa ser mantida aberta indefinidamente, a duração dela é limitada pela configuração de tempo limite do balanceador de carga downstream da Apigee. Por padrão, o tempo limite do balanceador de carga da Apigee é definido como 30 segundos. Para conexões de longa duração, recomendamos aumentar esse tempo limite ou criar um serviço de back-end separado com um valor de tempo limite maior para processar o tráfego SSE.
Como os dados de análise são registrados após o encerramento da sessão de SSE, pode haver algum atraso na geração de relatórios de dados de análise.
As falhas em um EventFlow fazem com que o fluxo seja encerrado imediatamente e geram um erro. Para saber como registrar manualmente esses tipos de erros ou enviá-los ao cliente, consulte Casos de uso e exemplos do EventFlow.
Um cliente que recebe respostas SSE transmitidas vai receber os cabeçalhos HTTP, incluindo todos os códigos de status, no início do fluxo de eventos. Como resultado, se o stream 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 ocorrer 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 inicialmente recebido nos cabeçalhos.

Streaming de eventos enviados pelo servidor Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

Criar um proxy de API para SSE

Configurar um EventFlow

Posicionamento de um EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Adicionar políticas a um EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Variáveis de fluxo

Casos de uso e exemplos do EventFlow

Modificar uma resposta de SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostrar um exemplo de resposta

Filtrar uma resposta de SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostrar um exemplo de resposta

Enviar um evento SSE para um sistema externo

<ProxyEndpoint>

<TargetEndpoint>

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

Tratamento de erros no EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Propagar erros de falha em um EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Usar o DataCapture para coletar contagens de tokens em análises

<TargetEndpoint>

<ProxyEndpoint>

Usar uma política LLMTokenQuota para aplicar cotas de tokens em um EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Como visualizar dados de SSE na análise da Apigee

Como depurar proxies SSE

Solução de problemas

Limitações

Streaming de eventos enviados pelo servidor