Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Transmisión de eventos enviados por el servidor

Esta página se aplica a Apigee y Apigee Hybrid.

Consulta la documentación de Apigee Edge.

Apigee admite la transmisión continua de respuestas desde extremos de eventos enviados por el servidor (SSE) a los clientes en tiempo real. La función SSE de Apigee es útil para controlar las APIs de modelos de lenguaje grandes (LLM) que funcionan de manera más eficaz transmitiendo sus respuestas al cliente. La transmisión de SSE reduce la latencia, y los clientes pueden recibir datos de respuesta en cuanto los genera un LLM. Esta función admite el uso de agentes de IA que operan en entornos en tiempo real, como bots de atención al cliente o coordinadores de flujos de trabajo.

Para usar SSE con Apigee, solo debes apuntar un proxy de API a un extremo de destino o proxy habilitado para SSE. Para lograr un control más detallado sobre la respuesta de SSE, Apigee proporciona un flujo de extremo especial llamado EventFlow. En el contexto de un EventFlow, puedes agregar un conjunto limitado de políticas para realizar operaciones en la respuesta de SSE, como filtrar, modificar o controlar errores. Para obtener más información sobre los flujos de proxy, consulta Controla proxies de API con flujos.

Crea un proxy de API para SSE

La IU de Apigee proporciona una plantilla para crear un proxy nuevo que incluya un EventFlow.

Nota: El uso de la plantilla de proxy de SSE es opcional. Te brinda acceso a variables de flujo específicas de SSE que pueden ser útiles para realizar ciertas operaciones de políticas. Sin embargo, no es necesario que uses EventFlow para usar SSE con Apigee.

Importante: Usa definiciones de extremos de destino independientes para los destinos de SSE. Mezclar extremos de destino de SSE y que no sean de SSE puede generar un comportamiento incoherente, como variables de flujo response.content vacías. Consulta Problemas conocidos.

Sigue estos pasos para crear un proxy de API con la plantilla EventFlow usando la IU de Apigee:

En la consola de Google Cloud , ve a la página Apigee > Desarrollo de proxies > Proxies de API.

Ir a Proxies de API
En el panel Proxies de API, haz clic en + Crear.
En el panel Crear un proxy, en Plantilla de proxy, selecciona Proxy con eventos enviados por el servidor (SSE).
En Detalles del proxy, ingresa lo siguiente:
- Nombre del proxy: Ingresa un nombre para el proxy, como myproxy.
- Ruta de acceso base: Se establece automáticamente en el valor que ingresas para Nombre del proxy. La ruta base es parte de la URL que se usa para realizar solicitudes a la API. Edge usa la URL para hacer coincidir solicitudes entrantes y enrutarlas al proxy de API adecuado.
- Descripción (Opcional): Ingresa una descripción para tu nuevo proxy de API, como "Prueba de Apigee con un proxy simple".
- Destino (API existente): Ingresa la URL de destino de SSE para el proxy de API. Por ejemplo: https://mocktarget.apigee.net/sse-events/5.
- Haz clic en Siguiente.
Implementa (opcional):
- Entornos de implementación: Opcional. Usa las casillas de verificación para seleccionar uno o más entornos en los que implementarás tu proxy. Si prefieres no implementar el proxy en este punto, deja vacío el campo Entornos de implementación. Siempre puedes implementar el proxy más adelante.
- Cuenta de servicio: (opcional) Una cuenta de servicio para el proxy. La cuenta de servicio representa la identidad del proxy implementado y determina qué permisos tiene. Esta es una función avanzada y, para los fines de este instructivo, puedes ignorarla.
Los proxies de API implementados con una configuración de EventFlow se facturarán como Extensible.
Haz clic en Crear.

Consulta también Compila un proxy de API simple.

Configura un EventFlow

Para lograr un control más detallado sobre la respuesta de SSE, Apigee proporciona un flujo de extremo especial llamado EventFlow. En el contexto de un EventFlow, puedes agregar un conjunto limitado de políticas, que se enumeran a continuación, para modificar la respuesta de SSE antes de que se transmita de vuelta al cliente. Para obtener más información sobre los flujos de proxy, consulta Controla proxies de API con flujos.

Ubicación de un EventFlow

Un EventFlow tiene dos atributos:

name: Es un nombre para identificar el flujo.
content-type: El valor de este atributo debe ser text/event-stream.

Consulta también la Referencia de configuración de flujo.

Se puede colocar un EventFlow dentro de una definición de TargetEndpoint o ProxyEndpoint, como se muestra en las siguientes muestras de código:

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

Nota: Si agregas varias estrofas EventFlow a un extremo, solo se ejecutará la última en la definición del extremo. Se ignoran las estrofas EventFlow que aparecen antes de la última.

También es importante tener en cuenta que, aunque puedes agregar un EventFlow a un TargetEndpoint, un ProxyEndpoint o ambos, solo se ejecuta un EventFlow.

En la siguiente tabla, se muestra la ejecución de las estrofas de EventFlow según la ubicación del extremo:

ProxyEndpoint	TargetEndpoint	EventFlow usado
EventFlow en `ProxyEndpoint`	EventFlow en `TargetEndpoint`	EventFlow en `TargetEndpoint`
No`EventFlow`	EventFlow en `TargetEndpoint`	EventFlow en `TargetEndpoint`
EventFlow en `ProxyEndpoint`	No`EventFlow`	EventFlow en `ProxyEndpoint`

Agrega políticas a un EventFlow

Puedes agregar hasta un total de cuatro políticas al elemento Response de EventFlow. Al igual que con todos los flujos, las políticas se ejecutan en el orden en que se agregan, y puedes agregar pasos condicionales para controlar su ejecución. Es importante tener en cuenta que los tipos de políticas que puedes agregar a un EventFlow se restringen a los siguientes. No se permiten otros tipos de políticas en un EventFlow:

Consulta también Adjunta y configura políticas en la IU y Adjunta y configura políticas en archivos en formato XML.

En los siguientes ejemplos, se muestra un EventFlow con un paso de política RaiseFault condicional agregado:

<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 ver más ejemplos de código de EventFlow, consulta la sección Casos de uso y ejemplos de EventFlow.

Variables de flujo

Un EventFlow propaga tres variables de flujo de respuesta. Ten en cuenta que estas variables solo se pueden usar dentro del alcance del evento actual que se procesa en el EventFlow. Acceder a estas variables o establecerlas fuera del alcance de EventFlow no tiene ningún efecto. Solo tienen sentido en el contexto de EventFlow.

response.event.current.content: Es una cadena que contiene la respuesta completa del evento actual. Apigee no analiza la cadena de ninguna manera. Contiene la respuesta completa sin cambios, incluidos todos los campos de datos.
Nota: Apigee limita cada evento de respuesta a un máximo de 10 MB de datos.

Nota: Si estableces el valor de esta variable en una cadena vacía (""), se evita que el evento se envíe al cliente. Para ver un ejemplo, consulta Filtra una respuesta de SSE.
response.event.current.data: Es una cadena que contiene la carga útil de datos del evento actual. Puedes modificar esta variable en EventFlow para cambiar la carga útil de datos que se envía al cliente.
response.event.current.count: Cuenta de forma incremental la cantidad de eventos de respuesta enviados. Este valor se actualiza para cada evento recibido. El recuento será 1 para el primer evento y se incrementará para los eventos posteriores.
Nota: Si accedes a este valor, en una política de JavaScript, por ejemplo, el primer evento tendrá un valor de 1; sin embargo, la cantidad de eventos enviados al cliente en este punto será 0.

Consulta también Referencia de variables de flujo.

Casos de uso y ejemplos de EventFlow

En los siguientes ejemplos, se muestra cómo implementar casos de uso comunes para los proxies de SSE:

Modifica una respuesta de SSE
Filtra una respuesta de SSE
Envía un evento de SSE a un sistema externo
Usa una política de Model Armor de Apigee en un EventFlow
Manejo de errores en EventFlow
Propaga mensajes de error en un EventFlow
Usa DataCapture para recopilar recuentos de tokens en las estadísticas
Usa una política LLMTokenQuota para aplicar cuotas de tokens en un EventFlow

Modifica una respuesta de SSE

En este ejemplo, se muestra cómo quitar datos de una respuesta de SSE EventFlow antes de devolverla al cliente. El contenido de la respuesta de SSE se almacena en una variable de flujo llamada response.event.current.content. En este caso, usamos una política de JavaScript para recuperar el valor de la variable de flujo, analizarla y modificarla. Consulta también Variables de flujo.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política de JavaScript con la siguiente definición. En este ejemplo, el código de JavaScript se incluye directamente en la política. Colocar el código JavaScript en un archivo de recursos es otra opción para configurar la 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>

Agrega la política de JavaScript al EventFlow del proxy. La EventFlow se adjunta a la TargetEndpoint o ProxyEndpoint predeterminada. En este ejemplo, se usa la API de Gemini en Vertex AI para generar contenido.

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

Guarda el proxy y, luego, impleméntalo.

Llama al proxy implementado:

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 una respuesta de ejemplo

Esta es una respuesta de ejemplo sin ningún filtro aplicado. Ten en cuenta que la respuesta incluye un 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 es otra respuesta de ejemplo con la política de JavaScript aplicada. Se quitó el atributo modelVersion.

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

Filtra una respuesta de SSE

En este ejemplo, se muestra cómo filtrar datos de una respuesta de SSE antes de devolverla al cliente. En este caso, filtramos los datos de eventos de la respuesta con una política de JavaScript. La política analiza la respuesta del evento en JSON, modifica el JSON para quitar los datos del evento y, luego, envía los datos de respuesta modificados de vuelta al cliente.

Al igual que en el ejemplo anterior, en este ejemplo, se recupera el valor de la variable de flujo response.event.current.content y se analiza en JSON. Luego, se aplica la lógica para implementar el filtrado previsto.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

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

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

Guarda el proxy y, luego, impleméntalo.

Llama al proxy implementado:

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 una respuesta de ejemplo

Este es un ejemplo de cómo se vería la respuesta sin aplicar ningún filtro. Observa que incluye datos de error:

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

Esta es otra respuesta de ejemplo después de que se aplica el filtrado con el mensaje de error borrado.

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

Envía un evento de SSE a un sistema externo

En este ejemplo, adjuntamos la política PublishMessage de Apigee al EventFlow para enviar un evento de SSE a un tema de Pub/Sub.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política PublishMessage con la siguiente definición:

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

Agrega la política PublishMessage como un paso en el EventFlow del 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>

Implementa y prueba el proxy de API.
Con el contenido generado agregado al tema de Pub/Sub, puedes, por ejemplo, crear una función de Cloud Run para procesar mensajes del tema.

Usa una política de Model Armor de Apigee en un EventFlow

Puedes usar la política SanitizeModelResponse para limpiar los eventos enviados por el servidor en un EventFlow. Establece <ContinuousScan> en true para verificar todo el flujo, como en los siguientes pasos, o déjalo en false (el valor predeterminado) para verificar cada evento por separado. Con true, la política retiene cada evento hasta que Model Armor confirma que su texto está limpio y, luego, lo libera completo, sin cambios y en orden para los pasos posteriores y para el cliente.

Para obtener más información sobre cómo la política controla las coincidencias, los errores y las cuotas, consulta Limpia las respuestas de eventos enviados por el servidor (SSE) de transmisión. Para obtener más información sobre Model Armor, consulta Descripción general de Model Armor.

Para configurar la política, completa los siguientes pasos:

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una política SanitizeModelResponse con <ContinuousScan>true</ContinuousScan> y <LLMResponseSource> configurados para extraer el delta de texto de response.event.current.data. El siguiente ejemplo es para los flujos de SSE de 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 las transmisiones de SSE compatibles con OpenAI, establece <LLMResponseSource> en {jsonPath('$.choices[-1].delta.content',response.event.current.data,true)}. Para los flujos de SSE de Anthropic Claude, establece <LLMResponseSource> en {jsonPath('$.delta.text',response.event.current.data,true)}.

Adjunta la política SanitizeModelResponse al elemento <Response> de EventFlow en tu proxy. Si también usas LLMTokenQuota o DataCapture en EventFlow, coloca SanitizeModelResponse primero para que esas políticas solo reciban eventos verificados:
```
<EventFlow name="EventFlow" content-type="text/event-stream">
  <Request/>
  <Response>
    <Step>
      <Name>SMR-modelresponse</Name>
    </Step>
  </Response>
</EventFlow>
```
Implementa y prueba el proxy de API.

Manejo de errores en EventFlow

De forma predeterminada, el flujo de eventos finaliza cuando se produce una falla. Sin embargo, si quieres realizar una depuración adicional, puedes enviar información de fallas a Cloud Logging, como se muestra en este ejemplo.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política RaiseFault con la siguiente definición:

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

Adjunta la política RaiseFault al EventFlow del proxy de SSE:

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

Crea una política MessageLogging para registrar errores. Por ejemplo:

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

Agrega la política MessageLogging a las FaultRules del extremo de destino o del 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>

Implementa y prueba el proxy de API.

Propaga errores de falla en un EventFlow

En este ejemplo, te mostramos cómo usar un EventFlow para propagar errores de falla al cliente. El proceso funciona para notificar al cliente sobre los errores de inmediato durante la ejecución de la política.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política de JavaScript con la siguiente definición:

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

Esta política está diseñada para arrojar un error cuando se cumple una condición específica.

Adjunta la política de JavaScript al EventFlow del proxy de SSE dentro de la configuración de TargetEndpoint o ProxyEndpoint. Este paso garantiza que EventFlow procese la política durante el control de la respuesta:

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

Implementa el proxy de API.

Prueba con el comportamiento del proxy con el siguiente comando de 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."}]}]}'

Reemplaza ENVIRONMENT_GROUP_NAME por el nombre de tu grupo de entornos.

El resultado debería ser similar al siguiente ejemplo:

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

El resultado muestra el flujo de datos inicial seguido de un mensaje fault. En caso de que se produzca un error, Apigee captura la información de la falla y la envía al cliente como un evento.

Para obtener más información sobre el manejo de fallas en Apigee, consulta Manejo de fallas.

Usa DataCapture para recopilar recuentos de tokens en las estadísticas

En este ejemplo, se muestra cómo usar la política DataCapture dentro de un EventFlow para recopilar recuentos de tokens de eventos de SSE y usarlos en las estadísticas de Apigee.

Crea recopiladores de datos para los recuentos de tokens. Ejecuta los siguientes comandos para crear los recopiladores de datos necesarios en tu organización:

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

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política de JavaScript para analizar los recuentos de tokens de los datos de eventos de SSE. En el editor de proxies de Apigee, haz clic en + junto a Políticas, selecciona JavaScript como el tipo de política y, luego, ingresa un nombre para la política (por ejemplo, JS-parse-token). El XML de la política resultante debería verse de la siguiente manera:

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

Crea una nueva política de DataCapture para capturar los recuentos de tokens. En el editor de proxies de Apigee, haz clic en + junto a Políticas, selecciona DataCapture como el tipo de política y, luego, ingresa un nombre para la política (por ejemplo, DC-capture-tokencount). El XML de la política resultante debería verse de la siguiente manera:

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

Adjunta primero la política de JavaScript y, luego, la política de DataCapture, dentro del EventFlow del proxy de SSE en la configuración de TargetEndpoint o ProxyEndpoint. Este paso garantiza que el EventFlow procese las políticas durante el manejo de la respuesta. El XML resultante debería verse de la siguiente manera:

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

Guarda e implementa el proxy. Para obtener más información, consulta Implementa proxies de API.
Enviar solicitudes de prueba al proxy Después de que el proxy procesa las respuestas de SSE, los recuentos de tokens capturados están disponibles en Apigee Analytics a través de los recopiladores de datos dc_tokenCount y dc_thoughtsTokenCount.

Usa una política LLMTokenQuota para aplicar cuotas de tokens en un EventFlow

En este ejemplo, se muestra cómo usar la política LLMTokenQuota dentro de un EventFlow para contar y aplicar cuotas de tokens para un flujo de SSE. Durante el procesamiento del flujo de eventos, el recuento de tokens se ejecuta solo cuando los metadatos de uso de tokens de la respuesta del LLM se encuentran en el evento. Cuando se descubren los metadatos de uso del token, se extraen y se ejecuta la política. Para todos los demás eventos, la política genera una operación nula.

Nota: Cuando la política LLMTokenQuota forma parte de un EventFlow, los elementos <LLMTokenUsageSource> y <LLMModelSource> deben leer los datos del evento actual con la variable de flujo response.event.current.data, con el tercer argumento de la función jsonPath establecido en false.

Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
Abre el proxy en el editor de proxies de Apigee y haz clic en la pestaña Desarrollar.

Crea una nueva política LLMTokenQuota para contar los tokens que consume la transmisión de SSE. En el editor de proxies de Apigee, haz clic en + junto a Políticas, selecciona LLMTokenQuota como el tipo de política y, luego, ingresa un nombre para la política (por ejemplo, LTQ-Count-Only). El XML de la política resultante debería verse de la siguiente manera:

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

Crea una segunda política LLMTokenQuota para aplicar la cuota de tokens en las solicitudes entrantes. En el editor de proxies de Apigee, haz clic en + junto a Políticas, selecciona LLMTokenQuota como el tipo de política y, luego, ingresa un nombre para la política (por ejemplo, LTQ-Enforce-Only). Esta política usa el mismo <SharedName> que la política de recuento para que ambas compartan el mismo contador de cuotas. El XML de la política resultante debería verse de la siguiente manera:
```
<?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>
```
Agrega la política LTQ-Enforce-Only como un paso en el Request del PreFlow del ProxyEndpoint. Este paso garantiza que se verifique la cuota en cada solicitud entrante. El XML resultante debería verse de la siguiente manera:
```
<ProxyEndpoint name="default">
  <PreFlow name="PreFlow">
    <Request>
      <Step>
        <Name>LTQ-Enforce-Only</Name>
      </Step>
    </Request>
    <Response/>
  </PreFlow>
</ProxyEndpoint>
```

Agrega la política LTQ-Count-Only como un paso dentro de la EventFlow del proxy de SSE en la configuración de TargetEndpoint o ProxyEndpoint. Este paso garantiza que el EventFlow procese la política durante el manejo de la respuesta y aumente el contador de cuota compartido. El XML resultante debería verse de la siguiente manera:

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

Guarda e implementa el proxy. Para obtener más información, consulta Implementa proxies de API.
Enviar solicitudes de prueba al proxy En cada solicitud entrante, la política LTQ-Enforce-Only verifica el contador de cuota compartida y rechaza la solicitud si ya se alcanzó el límite configurado. A medida que el proxy procesa las respuestas de SSE, la política LTQ-Count-Only cuenta los tokens informados en los metadatos de uso de tokens de cada evento y aumenta el mismo contador de cuota compartido. Cuando el contador alcanza el límite configurado, se rechazan las solicitudes posteriores que consumen tokens hasta que se restablece el contador.

Para obtener más información sobre la política LLMTokenQuota, incluido cómo usarla con respuestas de SSE y cómo configurar contadores de cuotas compartidos, consulta la política LLMTokenQuota.

Visualiza datos de SSE en las estadísticas de Apigee

Los datos de los proxies de SSE aparecen en las estadísticas de Apigee, como se espera para cualquier proxy de API. En la consola de Cloud, ve a Analytics > Métricas de API.

Depura proxies de SSE

Usa la herramienta de depuración de Apigee para depurar proxies de SSE. Los datos de depuración se capturan para EventFlow de la misma manera que para los otros tipos de flujos.

Soluciona problemas

En el caso de problemas de tráfico en tiempo real, consulta los registros de acceso de Apigee para determinar la causa.

Limitaciones

Se aplican las siguientes limitaciones a los proxies de SSE:

Si la política DataCapture se aplica en cada evento, solo se registrará el valor más reciente.
Si bien una conexión SSE se puede mantener abierta de forma indefinida, su duración se ve limitada por el parámetro de configuración de tiempo de espera del balanceador de cargas descendente de Apigee. De forma predeterminada, el tiempo de espera del balanceador de cargas de Apigee se establece en 30 segundos. Para las conexiones de mayor duración, recomendamos aumentar este tiempo de espera o crear un servicio de backend independiente con un valor de tiempo de espera más alto para controlar el tráfico de SSE.
Debido a que los datos de Analytics se registran después de que se cierra la sesión de SSE, es posible que notes cierta demora en los informes de datos de Analytics.
Las fallas dentro de un EventFlow hacen que la transmisión salga de inmediato y arroje un error. Para obtener información sobre cómo registrar manualmente estos tipos de errores o enviarlos al cliente, consulta Casos de uso y ejemplos de EventFlow.
Un cliente que reciba respuestas de SSE transmitidas recibirá los encabezados HTTP, incluidos los códigos de estado, al comienzo de la transmisión de eventos. Como resultado, si la transmisión de eventos entra en un estado de error, el código de estado recibido inicialmente no reflejará el estado de error.
Esta limitación se puede ver cuando se visualiza una sesión de depuración. En la sesión, es posible que notes que el código de estado HTTP de las transmisiones que ingresan al estado de error difiere de los códigos de estado que se envían al cliente. Esto puede ocurrir porque la entrada de la sesión de depuración se genera después de que se procesa toda la solicitud, en lugar de al principio del flujo de eventos. La sesión de depuración puede reflejar el código de error generado por el error, mientras que el cliente solo ve el estado 2xx que se recibió inicialmente en los encabezados.

Transmisión de eventos enviados por el servidor Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Crea un proxy de API para SSE

Configura un EventFlow

Ubicación de un EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Agrega políticas a un EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Variables de flujo

Casos de uso y ejemplos de EventFlow

Modifica una respuesta de SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostrar una respuesta de ejemplo

Filtra una respuesta de SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostrar una respuesta de ejemplo

Envía un evento de SSE a un sistema externo

<ProxyEndpoint>

<TargetEndpoint>

Usa una política de Model Armor de Apigee en un EventFlow

Manejo de errores en EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Propaga errores de falla en un EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Usa DataCapture para recopilar recuentos de tokens en las estadísticas

<TargetEndpoint>

<ProxyEndpoint>

Usa una política LLMTokenQuota para aplicar cuotas de tokens en un EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Visualiza datos de SSE en las estadísticas de Apigee

Depura proxies de SSE

Soluciona problemas

Limitaciones

Transmisión de eventos enviados por el servidor