Flux d'événements envoyés par le serveur

Cette page s'applique à Apigee et à Apigee hybrid.

Consultez la documentation d' Apigee Edge.

Apigee est compatible avec le streaming continu des réponses des points de terminaison d'événements envoyés par le serveur (SSE) aux clients en temps réel. La fonctionnalité SSE d'Apigee est utile pour gérer les API de grands modèles de langage (LLM) qui fonctionnent le plus efficacement en diffusant leurs réponses au client. Le streaming SSE réduit la latence, et les clients peuvent recevoir les données de réponse dès qu'elles sont générées par un LLM. Cette fonctionnalité permet d'utiliser des agents IA qui fonctionnent dans des environnements en temps réel, tels que des bots de service client ou des orchestrateurs de workflow.

Pour utiliser les SSE avec Apigee, il vous suffit de faire pointer un proxy d'API vers un point de terminaison cible ou proxy compatible avec les SSE. Pour contrôler plus précisément la réponse SSE, Apigee fournit un flux de point de terminaison spécial appelé EventFlow. Dans le contexte d'un EventFlow, vous pouvez ajouter un ensemble limité de règles pour effectuer des opérations sur la réponse SSE, telles que le filtrage, la modification ou la gestion des erreurs. Pour en savoir plus sur les flux de proxy, consultez Contrôler des proxys d'API avec des flux.

Créer un proxy d'API pour les SSE

L'interface utilisateur d'Apigee fournit un modèle permettant de créer un proxy incluant un EventFlow.

Pour créer un proxy d'API avec le modèle EventFlow à l'aide de l'interface utilisateur d'Apigee, procédez comme suit :

  1. Dans la console Google Cloud , accédez à la page Apigee > Développement de proxys > Proxys d'API.

    Accéder aux proxys d'API

  2. Dans le volet Proxys d'API, cliquez sur + Créer.
  3. Dans le volet Créer un proxy, sous Modèle de proxy, sélectionnez Proxy avec événements envoyés par le serveur (SSE).
  4. Sous Détails du proxy, saisissez les informations suivantes :
    • Nom du proxy : saisissez un nom pour le proxy, tel que myproxy.
    • Chemin de base : défini automatiquement sur la valeur que vous saisissez pour Nom du proxy. Le chemin de base fait partie de l'URL utilisée pour envoyer des requêtes à votre API. Apigee utilise l'URL pour mettre en correspondance et acheminer les requêtes entrantes vers le proxy d'API approprié.
    • Description (facultatif) : saisissez une description pour votre nouveau proxy d'API, par exemple "Test Apigee avec un proxy simple".
    • Cible (API existante) : saisissez l'URL cible SSE du proxy d'API. Par exemple : https://mocktarget.apigee.net/sse-events/5
    • Cliquez sur Suivant.
  5. Déployer (facultatif) :
    • Environnements de déploiement : facultatif. Cochez les cases pour sélectionner un ou plusieurs environnements dans lesquels déployer votre proxy. Si vous préférez ne pas déployer le proxy à ce stade, laissez le champ Environnements de déploiement vide. Vous pourrez toujours déployer le proxy ultérieurement.
    • Compte de service : facultatif. Un compte de service pour le proxy. Le compte de service représente l'identité du proxy déployé et détermine les autorisations dont il dispose. Il s'agit d'une fonctionnalité avancée que vous pouvez ignorer pour les besoins de ce tutoriel.

    Les proxys d'API déployés avec une configuration EventFlow seront facturés comme extensibles.

  6. Cliquez sur Créer.

Consultez également Créer un proxy d'API simple.

Configurer un EventFlow

Pour contrôler plus précisément la réponse SSE, Apigee fournit un flux de point de terminaison spécial appelé EventFlow. Dans le contexte d'un EventFlow, vous pouvez ajouter un ensemble limité de règles, listées ci-dessous, pour modifier la réponse SSE avant qu'elle ne soit renvoyée au client. Pour en savoir plus sur les flux de proxy, consultez Contrôler des proxys d'API avec des flux.

Placement d'un EventFlow

Un EventFlow comporte deux attributs :

  • name : nom permettant d'identifier le flux.
  • content-type : la valeur de cet attribut doit être text/event-stream.

Consultez également la documentation de référence sur la configuration des flux.

Un EventFlow peut être placé dans une définition TargetEndpoint ou ProxyEndpoint, comme illustré dans les exemples de code suivants :

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

Il est également important de noter que même si vous pouvez ajouter un EventFlow à un TargetEndpoint, à un ProxyEndpoint ou aux deux, un seul EventFlow est exécuté.

Le tableau suivant montre l'exécution des strophes EventFlow en fonction de l'emplacement du point de terminaison :

ProxyEndpoint TargetEndpoint EventFlow utilisé
EventFlow dans ProxyEndpoint EventFlow dans TargetEndpoint EventFlow dans TargetEndpoint
Aucune configuration de type "EventFlow" EventFlow dans TargetEndpoint EventFlow dans TargetEndpoint
EventFlow dans ProxyEndpoint Aucune configuration de type "EventFlow" EventFlow dans ProxyEndpoint

Ajouter des règles à un flux d'événements

Vous pouvez ajouter jusqu'à quatre règles à l'élément Response de EventFlow. Comme pour tous les flux, les règles sont exécutées dans l'ordre dans lequel elles sont ajoutées, et vous pouvez ajouter des étapes conditionnelles pour contrôler leur exécution. Il est important de noter que les types de règles que vous pouvez ajouter à un EventFlow sont limités aux suivants. Aucun autre type de stratégie n'est autorisé dans un EventFlow :

Consultez également Associer et configurer des règles dans l'UI et Associer et configurer des règles dans des fichiers XML.

Les exemples suivants montrent un EventFlow avec une étape de règle RaiseFault conditionnelle ajoutée :

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

Pour obtenir d'autres exemples de code EventFlow, consultez la section Cas d'utilisation et exemples d'EventFlow.

Variables de flux

Un EventFlow renseigne trois variables de flux de réponse. Notez que ces variables ne sont utilisables que dans le champ d'application de l'événement en cours de traitement dans le EventFlow. L'accès à ces variables ou leur définition en dehors du champ d'application EventFlow n'a aucun effet. Elles ne sont pertinentes que dans le contexte de EventFlow.

  • response.event.current.content : chaîne contenant l'intégralité de la réponse de l'événement actuel. Apigee n'analyse pas la chaîne. Il contient l'intégralité de la réponse inchangée, y compris tous les champs de données.
  • response.event.current.data : chaîne contenant la charge utile de données de l'événement actuel. Vous pouvez modifier cette variable dans EventFlow pour changer la charge utile de données envoyée au client.
  • response.event.current.count : compte de manière incrémentielle le nombre d'événements de réponse envoyés. Cette valeur est mise à jour pour chaque événement reçu. Le nombre sera de 1 pour le premier événement et s'incrémentera pour les événements suivants.

Consultez également la documentation de référence sur les variables de flux.

Cas d'utilisation et exemples d'EventFlow

Les exemples suivants montrent comment implémenter des cas d'utilisation courants pour les proxys SSE :

Modifier une réponse SSE

Cet exemple montre comment supprimer des données d'une réponse SSE EventFlow avant de la renvoyer au client. Le contenu de la réponse SSE est stocké dans une variable de flux appelée response.event.current.content. Dans ce cas, nous utilisons une règle JavaScript pour récupérer la valeur de la variable de flux, l'analyser et la modifier. Consultez également Variables de flux.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle JavaScript avec la définition suivante. Dans cet exemple, le code JavaScript est inclus directement dans la règle. Une autre option pour configurer la règle consiste à placer le code JavaScript dans un fichier de ressources. 
    <?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. Ajoutez la règle JavaScript au EventFlow du proxy. Le EventFlow est associé au TargetEndpoint ou au ProxyEndpoint par défaut. Cet exemple utilise l'API Gemini dans Vertex AI pour générer du contenu.

    <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>
  5. Enregistrez le proxy et déployez-le.
  6. Appelez le proxy déployé : 
    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."}]}]}'

    Afficher un exemple de réponse

    Voici un exemple de réponse sans filtrage. Notez que la réponse inclut un attribut 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"
  }

    Voici un autre exemple de réponse avec la règle JavaScript appliquée. L'attribut modelVersion est supprimé.

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

Filtrer une réponse SSE

Cet exemple montre comment filtrer les données d'une réponse SSE avant de la renvoyer au client. Dans ce cas, nous filtrons les données d'événement de la réponse à l'aide d'une règle JavaScript. La règle analyse la réponse de l'événement au format JSON, modifie le JSON pour supprimer les données d'événement, puis renvoie les données de réponse modifiées au client.

Comme dans l'exemple précédent, cet exemple récupère la valeur de la variable de flux response.event.current.content et l'analyse au format JSON, puis applique une logique pour implémenter le filtrage prévu.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle JavaScript avec la définition suivante. Dans cet exemple, le code JavaScript est inclus directement dans la règle. Une autre option pour configurer la règle consiste à placer le code JavaScript dans un fichier de ressources. 
    <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. Ajoutez la règle JavaScript au EventFlow du proxy. Le EventFlow est associé au TargetEndpoint ou au ProxyEndpoint par défaut. Cet exemple utilise l'API Gemini dans Vertex AI pour générer du contenu.

    <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>
  5. Enregistrez le proxy et déployez-le.
  6. Appelez le proxy déployé : 
    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."}]}]}'

    Afficher un exemple de réponse

    Voici un exemple de ce à quoi pourrait ressembler la réponse sans appliquer de filtrage. Notez qu'il inclut des données d'erreur :

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

    Voici un autre exemple de réponse après l'application du filtrage, avec le message d'erreur nettoyé.

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

Envoyer un événement SSE à un système externe

Dans cet exemple, nous associons la règle PublishMessage d'Apigee à EventFlow pour envoyer un événement SSE à un sujet Pub/Sub.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle PublishMessage avec la définition suivante : 
    <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. Ajoutez la règle PublishMessage en tant qu'étape dans le EventFlow du proxy d'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>
  5. Déployez et testez le proxy d'API.
  6. Une fois votre contenu généré ajouté au sujet Pub/Sub, vous pouvez, par exemple, créer une fonction Cloud Run pour traiter les messages du sujet.

Utiliser une règle Apigee Model Armor dans un flux d'événements

Vous pouvez utiliser la règle SanitizeModelResponse pour nettoyer les événements envoyés par le serveur dans un EventFlow. Définissez <ContinuousScan> sur true pour vérifier l'intégralité du flux, comme dans les étapes suivantes, ou laissez-le sur false (valeur par défaut) pour vérifier chaque événement séparément. Avec true, la règle conserve chaque événement jusqu'à ce que Model Armor confirme que son texte est propre, puis le libère dans son intégralité, sans modification et dans l'ordre, pour les étapes ultérieures et pour le client.

Pour en savoir plus sur la façon dont la règle gère les correspondances, les erreurs et les quotas, consultez Nettoyer les réponses des événements envoyés par le serveur (SSE) en streaming. Pour en savoir plus sur Model Armor, consultez la présentation de Model Armor.

Pour configurer la règle, procédez comme suit :

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle SanitizeModelResponse avec <ContinuousScan>true</ContinuousScan> et <LLMResponseSource> configurés pour extraire le delta de texte de response.event.current.data. L'exemple suivant concerne les flux SSE 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>

    Pour les flux SSE compatibles avec OpenAI, définissez <LLMResponseSource> sur {jsonPath('$.choices[-1].delta.content',response.event.current.data,true)}. Pour les flux SSE Anthropic Claude, définissez <LLMResponseSource> sur {jsonPath('$.delta.text',response.event.current.data,true)}.

  4. Associez la règle SanitizeModelResponse à l'élément <Response> du EventFlow dans votre proxy. Si vous utilisez également LLMTokenQuota ou DataCapture dans EventFlow, placez d'abord SanitizeModelResponse afin que ces règles ne reçoivent que les événements vérifiés : 
    <EventFlow name="EventFlow" content-type="text/event-stream">
  <Request/>
  <Response>
    <Step>
      <Name>SMR-modelresponse</Name>
    </Step>
  </Response>
</EventFlow>
  5. Déployez et testez le proxy d'API.

Gestion des exceptions dans EventFlow

Par défaut, le flux d'événements se termine lorsqu'une erreur se produit. Toutefois, si vous souhaitez effectuer un débogage supplémentaire, vous pouvez envoyer des informations sur les erreurs à Cloud Logging, comme illustré dans cet exemple.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle RaiseFault avec la définition suivante : 
    <?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. Associez la règle RaiseFault au EventFlow du proxy SSE : 
    <EventFlow content-type="text/event-stream">
  <Response>
    <Step>
      <Name>RF-Empty-Event</Name>
      <Condition>response.event.current.content ~ "data: "</Condition>
    </Step>
  </Response>
</EventFlow>
  5. Créer une règle MessageLogging pour consigner les erreurs Exemple : 
    <?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. Ajoutez la règle MessageLogging aux FaultRules du point de terminaison cible ou du 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>
  7. Déployez et testez le proxy d'API.

Propager les erreurs d'erreur dans un EventFlow

Dans cet exemple, nous vous montrons comment utiliser un EventFlow pour propager les erreurs d'erreur au client. Le processus permet d'informer immédiatement le client des erreurs lors de l'exécution de la règle.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle JavaScript avec la définition suivante : 
    <?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>

    Cette règle est conçue pour générer une erreur lorsqu'une condition spécifique est remplie.

  4. Associez la règle JavaScript au EventFlow du proxy SSE dans la configuration TargetEndpoint ou ProxyEndpoint. Cette étape garantit que l'EventFlow traite la règle lors du traitement de la réponse :

    <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>
  5. Déployez le proxy d'API.
  6. Testez le comportement du proxy à l'aide de la commande curl suivante : 
    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."}]}]}'

    Remplacez ENVIRONMENT_GROUP_NAME par le nom de votre groupe d'environnements.

    Le résultat doit ressembler à l'exemple ci-dessous :

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

    Le résultat affiche le flux de données initial suivi d'un message fault. En cas d'erreur, Apigee capture les informations sur l'erreur et les envoie au client en tant qu'événement.

Pour en savoir plus sur la gestion des erreurs dans Apigee, consultez Gérer les erreurs.

Utiliser DataCapture pour collecter le nombre de jetons dans Analytics

Cet exemple vous montre comment utiliser la règle DataCapture dans un EventFlow pour collecter le nombre de jetons à partir d'événements SSE afin de les utiliser dans Apigee Analytics.

  1. Créez des collecteurs de données pour le nombre de jetons. Exécutez les commandes suivantes pour créer les collecteurs de données requis dans votre organisation : 
    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"
  }'
  2. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  3. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  4. Créez une règle JavaScript pour analyser le nombre de jetons à partir des données d'événement SSE. Dans l'éditeur de proxy Apigee, cliquez sur + à côté de Policies (Règles), sélectionnez JavaScript comme type de règle, puis saisissez un nom pour la règle (par exemple, JS-parse-token). Le code XML de la règle obtenue doit ressembler à ceci : 
    <?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>
  5. Créez une règle DataCapture pour capturer le nombre de jetons. Dans l'éditeur de proxy Apigee, cliquez sur + à côté de Règles, sélectionnez DataCapture comme type de règle, puis saisissez un nom pour la règle (par exemple, DC-capture-tokencount). Le code XML de la règle obtenue doit se présenter comme suit : 
    <?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>
  6. Associez d'abord la règle JavaScript, puis la règle DataCapture, dans le EventFlow du proxy SSE au sein de la configuration TargetEndpoint ou ProxyEndpoint. Cette étape garantit que EventFlow traite les règles lors du traitement de la réponse. Le code XML obtenu doit se présenter comme suit :

    <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>
  7. Enregistrez et déployez le proxy. Pour en savoir plus, consultez Déployer des proxys d'API.
  8. Envoyez des requêtes de test au proxy. Une fois que le proxy a traité les réponses SSE, le nombre de jetons capturés est disponible dans Apigee Analytics via les collecteurs de données dc_tokenCount et dc_thoughtsTokenCount.

Utiliser une règle LLMTokenQuota pour appliquer des quotas de jetons dans un EventFlow

Cet exemple montre comment utiliser la règle LLMTokenQuota dans un EventFlow pour compter et appliquer des quotas de jetons pour un flux SSE. Lors du traitement du flux d'événements, le comptage des jetons n'est exécuté que lorsque les métadonnées d'utilisation des jetons de la réponse du LLM sont trouvées dans l'événement. Lorsque les métadonnées d'utilisation du jeton sont découvertes, elles sont extraites et la règle est exécutée. Pour tous les autres événements, la règle n'entraîne aucune opération.

  1. Créez un proxy avec le modèle de proxy SSE. Consultez Créer un proxy d'API avec des événements envoyés par le serveur (SSE).
  2. Ouvrez le proxy dans l'éditeur de proxy Apigee, puis cliquez sur l'onglet Développer.
  3. Créez une règle LLMTokenQuota pour comptabiliser les jetons consommés par le flux SSE. Dans l'éditeur de proxy Apigee, cliquez sur + à côté de Règles, sélectionnez LLMTokenQuota comme type de règle, puis saisissez un nom pour la règle (par exemple, LTQ-Count-Only). Le code XML de la règle obtenue doit ressembler à ceci : 
    <?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>
  4. Créez une deuxième règle LLMTokenQuota pour appliquer le quota de jetons aux requêtes entrantes. Dans l'éditeur de proxy Apigee, cliquez sur + à côté de Règles, sélectionnez LLMTokenQuota comme type de règle, puis saisissez un nom pour la règle (par exemple, LTQ-Enforce-Only). Cette règle utilise le même <SharedName> que la règle de décompte afin que les deux règles partagent le même compteur de quota. Le fichier XML de la règle obtenue doit se présenter comme suit : 
    <?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>
  5. Ajoutez la règle LTQ-Enforce-Only en tant qu'étape dans le Request du PreFlow du ProxyEndpoint. Cette étape garantit que le quota est vérifié à chaque requête entrante. Le code XML obtenu doit se présenter comme suit : 
    <ProxyEndpoint name="default">
  <PreFlow name="PreFlow">
    <Request>
      <Step>
        <Name>LTQ-Enforce-Only</Name>
      </Step>
    </Request>
    <Response/>
  </PreFlow>
</ProxyEndpoint>
  6. Ajoutez la règle LTQ-Count-Only en tant qu'étape dans le EventFlow du proxy SSE dans la configuration TargetEndpoint ou ProxyEndpoint. Cette étape garantit que EventFlow traite la règle lors du traitement de la réponse et incrémente le compteur de quota partagé. Le code XML obtenu doit se présenter comme suit :

    <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>
  7. Enregistrez et déployez le proxy. Pour en savoir plus, consultez Déployer des proxys d'API.
  8. Envoyez des requêtes de test au proxy. À chaque requête entrante, la règle LTQ-Enforce-Only vérifie le compteur de quota partagé et rejette la requête si la limite configurée a déjà été atteinte. Lorsque le proxy traite les réponses SSE, la règle LTQ-Count-Only comptabilise les jetons signalés dans les métadonnées d'utilisation des jetons de chaque événement et incrémente le même compteur de quota partagé. Lorsque le compteur atteint la limite configurée, les requêtes suivantes consommant des jetons sont rejetées jusqu'à ce que le compteur soit réinitialisé.

Pour en savoir plus sur la règle LLMTokenQuota, y compris sur son utilisation avec les réponses SSE et sur la configuration des compteurs de quotas partagés, consultez la page Règle LLMTokenQuota.

Afficher les données SSE dans Apigee Analytics

Les données des proxys SSE s'affichent dans Apigee Analytics comme prévu pour n'importe quel proxy d'API. Dans la console Cloud, accédez à Analyse > Métriques de l'API.

Déboguer des proxys SSE

Utilisez l'outil de débogage Apigee pour déboguer les proxys SSE. Les données de débogage sont capturées pour EventFlow, tout comme pour les autres types de flux.

Dépannage

Pour les problèmes de trafic en temps réel, consultez les journaux d'accès Apigee afin d'en déterminer la cause.

Limites

Les limites suivantes s'appliquent aux proxys SSE :

  • Si la règle DataCapture est appliquée à chaque événement, seule la dernière valeur sera enregistrée.
  • Bien qu'une connexion SSE puisse être maintenue ouverte indéfiniment, sa durée est en fin de compte limitée par le paramètre de délai avant expiration de l'équilibreur de charge en aval d'Apigee. Par défaut, le délai avant expiration de l'équilibreur de charge Apigee est défini sur 30 secondes. Pour les connexions de plus longue durée, nous vous recommandons d' augmenter ce délai avant expiration ou de créer un service de backend distinct avec une valeur de délai avant expiration plus élevée pour gérer le trafic SSE.
  • Étant donné que les données d'analyse sont enregistrées après la fermeture de la session SSE, vous pouvez constater un certain retard dans la création de rapports sur les données d'analyse.
  • Les erreurs dans un EventFlow entraînent l'arrêt immédiat du flux et génèrent une erreur. Pour savoir comment consigner manuellement ces types d'erreurs ou les envoyer au client, consultez Cas d'utilisation et exemples d'EventFlow.
  • Un client recevant des réponses SSE en streaming recevra les en-têtes HTTP, y compris les codes d'état, au début du flux d'événements. Par conséquent, si le flux d'événements passe à un état d'erreur, le code d'état initialement reçu ne reflète pas l'état d'erreur.

    Cette limitation est visible lors de l'affichage d'une session de débogage. Dans la session, vous remarquerez peut-être que le code d'état HTTP pour les flux qui passent à l'état d'erreur diffère des codes d'état envoyés au client. Cela peut se produire, car l'entrée de la session de débogage est générée une fois que l'intégralité de la requête a été traitée, plutôt qu'au début du flux d'événements. La session de débogage peut refléter le code d'erreur généré par l'erreur, tandis que le client ne voit que l'état 2xx initialement reçu dans les en-têtes.