Transmisión de eventos enviados por el servidor

Configura un flujo de eventos

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

Se debe colocar un EventFlow dentro de la definición de TargetEndpoint, como se muestra en la siguiente muestra de código:

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

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.

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 limitan a los siguientes. No se permiten otros tipos de políticas en un EventFlow:

• Javascript
• MessageLogging
• PublishMessage
• RaiseFault
• SanitizeModelResponse

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

En el siguiente ejemplo, se muestra un EventFlow con un paso de política RaiseFault condicional agregado:

<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 EventFlow, consulta la sección Casos de uso y ejemplos de EventFlow.

Variables de flujo

Un EventFlow propaga dos 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 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 toda la respuesta del evento actual. Apigee no analiza la cadena de ninguna manera. Contiene toda la respuesta sin cambios, incluidos todos los campos de datos.
• 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 aumentará para los eventos posteriores.

Consulta también la 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:

• Cómo modificar una respuesta de SSE
• Cómo filtrar una respuesta de SSE
• Cómo enviar un evento SSE a un sistema externo
• Cómo usar una política de Apigee Model Armor en un flujo de eventos
• Manejo de errores en EventFlow

Cómo modificar una respuesta de SSE

En este ejemplo, se muestra cómo quitar datos de una respuesta EventFlow de SSE 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.

1. Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
2. Abre el proxy en el editor de proxy de Apigee y haz clic en la pestaña Desarrollar.
3. Crea una nueva política de JavaScript con la siguiente definición. En este ejemplo, el código JavaScript se incluye directamente en la política. Colocar el código de 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>

4. Agrega la política de JavaScript al EventFlow del proxy. El EventFlow está conectado al TargetEndpoint predeterminado. En este ejemplo, se usa la API de Gemini en Vertex AI para generar contenido.

   <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-1.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse</URL>
     </HTTPTargetConnection>
   </TargetEndpoint>
   

5. Guarda el proxy y, luego, impleméntalo.
6. 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."}]}]}'

   Muestra 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-1.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-1.5-flash"
     }

   Esta es otra respuesta de muestra 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
           }
         ]
       }
     }

Cómo filtrar una respuesta de SSE

En este ejemplo, se muestra cómo filtrar datos de una respuesta de SSE antes de devolverlos 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, lo modifica para quitar los datos del evento y, luego, le envía los datos de respuesta modificados al cliente.

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

1. Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
2. Abre el proxy en el editor de proxy de Apigee y haz clic en la pestaña Desarrollar.
3. Crea una nueva política de JavaScript con la siguiente definición. En este ejemplo, el código JavaScript se incluye directamente en la política. Colocar el código de JavaScript en un archivo de recursos es otra opción para configurar la 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. Agrega la política de JavaScript al EventFlow del proxy. El EventFlow está conectado al TargetEndpoint predeterminado. En este ejemplo, se usa la API de Gemini en Vertex AI para generar contenido.

   <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-1.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse	</URL>
     </HTTPTargetConnection>
   </TargetEndpoint>
   

5. Guarda el proxy y, luego, impleméntalo.
6. 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."}]}]}'

   Muestra 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-1.5-flash"
     }
       data: {
       "error": "Service temporarily unavailable. We are experiencing high traffic.",
       "modelVersion": "gemini-1.5-flash"
       }

   Esta es otra respuesta de muestra después de aplicar el filtrado con el mensaje de error borrado.

   data: {
     "candidates": [
       {
         "content": {
           "parts": [
             {
               "text": "El"
             }
           ],
           "role": "model"
         }
       }
     ],
     "usageMetadata": {
       "promptTokenCount": 8,
       "totalTokenCount": 8
     },
     "modelVersion": "gemini-1.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-1.5-flash"
   }

Cómo enviar un evento SSE a un sistema externo

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

1. Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
2. Abre el proxy en el editor de proxy de Apigee y haz clic en la pestaña Desarrollar.
3. 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>

4. Agrega la política PublishMessage como un paso en el EventFlow del 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. Implementa y prueba el proxy de API.
6. Con el contenido generado agregado al tema de Pub/Sub, puedes, por ejemplo, crear una función de Cloud Run para procesar los mensajes del tema.

Usa una política de Model Armor de Apigee en un flujo de eventos

Puedes usar la política SanitizeModelResponse para limpiar los eventos entrantes enviados por el servidor en un EventFlow. Esta política protege tus aplicaciones de IA, ya que limpia las respuestas de los modelos de lenguaje grandes (LLM). Para obtener información sobre Model Armor, consulta la descripción general de Model Armor. Para obtener información sobre las políticas de Apigee Model Armor, consulta Comienza a usar las políticas de Apigee Model Armor.

1. Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
2. Abre el proxy en el editor de proxy de Apigee y haz clic en la pestaña Desarrollar.
3. Crea una nueva política SanitizeModelResponse con la siguiente definición:

     <?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) Agrega una política de JavaScript para agrupar eventos antes de enviarlos a la política de Model Armor de Apigee.

   <?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. Agrega las políticas de JavaScript y ModelArmor a un paso en el EventFlow del 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. 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.

1. Crea un proxy nuevo con la plantilla de proxy de SSE. Consulta Crea un proxy de API con eventos enviados por el servidor (SSE).
2. Abre el proxy en el editor de proxy de Apigee y haz clic en la pestaña Desarrollar.
3. Crea una nueva política de 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>

4. Adjunta la política de 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>

5. Crea una política de 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>

6. Agrega la política de MessageLogging a las FaultRules del extremo 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. Implementa y prueba el proxy de API.

Cómo ver datos de SSE en Apigee Analytics

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.

Depuración de proxies de SSE

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

Soluciona problemas

Si tienes 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:

• Debido a que los datos de estadísticas 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 estadísticas.
• Las fallas dentro de un EventFlow hacen que la transmisión se cierre de inmediato y no se genere ningún evento de error en particular para el cliente final. Si deseas obtener información para registrar manualmente este tipo de errores, consulta los casos de uso y ejemplos de EventFlow.
• Un cliente que recibe respuestas SSE transmitidas recibirá los encabezados HTTP, incluidos los códigos de estado, al comienzo de la transmisión de eventos. Como resultado, si el flujo de eventos entra en un estado de error, el código de estado que se recibió 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 comienzo del flujo de eventos. La sesión de depuración puede reflejar el código de falla que genera el error, mientras que el cliente solo ve el estado 2xx que se recibió inicialmente en los encabezados.