Vom Server gesendete Ereignisse streamen

EventFlow konfigurieren

Für eine detailliertere Kontrolle über die SSE-Antwort bietet Apigee einen speziellen Zielendpunkt-Vorgang namens EventFlow. Im Kontext eines EventFlow können Sie eine begrenzte Anzahl der unten aufgeführten Richtlinien hinzufügen, um die SSE-Antwort zu ändern, bevor sie an den Client zurückgestreamt wird. Weitere Informationen zu Proxy-Abläufen finden Sie unter API-Proxys mit Abläufen steuern.

Ein EventFlow muss in der TargetEndpoint-Definition platziert werden, wie im folgenden Codebeispiel gezeigt:

<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 hat zwei Attribute:

• name: Ein Name, der den Ablauf identifiziert.
• content-type: Der Wert dieses Attributs muss text/event-stream sein.

Weitere Informationen finden Sie in der Ablaufkonfigurationsreferenz.

Dem Response-Element der EventFlow können insgesamt bis zu vier Richtlinien hinzugefügt werden. Wie bei allen Abläufen werden Richtlinien in der Reihenfolge ausgeführt, in der sie hinzugefügt werden. Sie können bedingte Schritte hinzufügen, um die Ausführung zu steuern. Die folgenden Richtlinientypen können einem EventFlow hinzugefügt werden: In einer EventFlow sind keine anderen Richtlinientypen zulässig:

• JavaScript
• MessageLogging
• PublishMessage
• RaiseFault
• SanitizeModelResponse

Weitere Informationen finden Sie unter Richtlinien in der Benutzeroberfläche anhängen und konfigurieren und Richtlinien in XML-Dateien anhängen und konfigurieren.

Das folgende Beispiel zeigt ein EventFlow mit einem bedingten Schritt der RaiseFault-Richtlinie:

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

Weitere EventFlow-Codebeispiele finden Sie im Abschnitt EventFlow-Anwendungsfälle und ‑Beispiele.

Ablaufvariablen

Mit einer EventFlow werden zwei Ablaufvariablen für die Antwort ausgefüllt. Diese Variablen können nur im Rahmen des aktuellen Ereignisses verwendet werden, das in EventFlow verarbeitet wird. Der Zugriff auf oder die Einstellung dieser Variablen außerhalb des Gültigkeitsbereichs von EventFlow hat keine Auswirkungen. Sie sind nur im Kontext der EventFlow sinnvoll.

• response.event.current.content: Ein String, der die gesamte Antwort des aktuellen Ereignisses enthält. Apigee parst den String nicht. Sie enthält die gesamte Antwort unverändert, einschließlich aller Datenfelder.
• response.event.current.count: Die Anzahl der gesendeten Antwortereignisse wird inkrementell gezählt. Dieser Wert wird für jedes empfangene Ereignis aktualisiert. Für das erste Ereignis wird „1“ gezählt, für nachfolgende Ereignisse wird der Wert erhöht.

Weitere Informationen finden Sie unter Referenz zu Ablaufvariablen.

Anwendungsfälle und Beispiele für EventFlow

In den folgenden Beispielen wird gezeigt, wie gängige Anwendungsfälle für SSE-Proxys implementiert werden:

• SSE-Antwort ändern
• SSE-Antwort filtern
• SSE-Ereignisse an ein externes System senden
• Apigee Model Armor-Richtlinie in einem EventFlow verwenden
• Fehlerbehandlung im EventFlow

SSE-Antwort ändern

In diesem Beispiel wird gezeigt, wie Daten aus einer SSE EventFlow-Antwort entfernt werden, bevor sie an den Client zurückgegeben werden. Der Inhalt der SSE-Antwort wird in einer Ablaufvariablen namens response.event.current.content gespeichert. In diesem Fall verwenden wir eine JavaScript-Richtlinie, um den Wert der Ablaufvariablen abzurufen, zu parsen und zu ändern. Weitere Informationen finden Sie unter Ablaufvariablen.

1. Erstellen Sie mit der SSE-Proxy-Vorlage einen neuen Proxy. Weitere Informationen finden Sie unter API-Proxy mit vom Server gesendeten Ereignissen (SSE) erstellen.
2. Öffnen Sie den Proxy im Apigee-Proxy-Editor und klicken Sie auf den Tab Entwickeln.
3. Erstellen Sie eine neue JavaScript-Richtlinie mit der folgenden Definition. In diesem Beispiel ist der JavaScript-Code direkt in der Richtlinie enthalten. Eine weitere Möglichkeit zur Konfiguration der Richtlinie besteht darin, den JavaScript-Code in einer Ressourcendatei zu platzieren.

   <?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. Fügen Sie die JavaScript-Richtlinie der EventFlow des Proxys hinzu. Die EventFlow ist an das Standard-TargetEndpoint angehängt. In diesem Beispiel wird die Gemini API in Vertex AI verwendet, um Inhalte zu generieren.

   <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. Speichern Sie den Proxy und stellen Sie ihn bereit.
6. Rufen Sie den bereitgestellten Proxy auf:

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

   Beispielantwort anzeigen

   Dies ist eine Beispielantwort ohne Filterung. Beachten Sie, dass die Antwort ein modelVersion": "gemini-1.5-flash"-Attribut enthält.

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

   Dies ist eine weitere Beispielantwort mit der angewendeten JavaScript-Richtlinie. Das modelVersion-Attribut wird entfernt.

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

SSE-Antwort filtern

In diesem Beispiel wird gezeigt, wie Daten aus einer SSE-Antwort gefiltert werden, bevor sie an den Client zurückgegeben werden. In diesem Fall filtern wir Ereignisdaten aus der Antwort mithilfe einer JavaScript-Richtlinie. Die Richtlinie analysiert die Ereignisantwort in JSON, ändert die JSON, um die Ereignisdaten zu entfernen, und sendet die geänderten Antwortdaten dann an den Client zurück.

Wie im vorherigen Beispiel wird auch hier der Wert der Ablaufvariablen response.event.current.content abgerufen und in JSON geparst. Anschließend wird Logik angewendet, um die beabsichtigte Filterung zu implementieren.

1. Erstellen Sie mit der SSE-Proxy-Vorlage einen neuen Proxy. Weitere Informationen finden Sie unter API-Proxy mit vom Server gesendeten Ereignissen (SSE) erstellen.
2. Öffnen Sie den Proxy im Apigee-Proxy-Editor und klicken Sie auf den Tab Entwickeln.
3. Erstellen Sie eine neue JavaScript-Richtlinie mit der folgenden Definition. In diesem Beispiel ist der JavaScript-Code direkt in der Richtlinie enthalten. Eine weitere Möglichkeit zur Konfiguration der Richtlinie besteht darin, den JavaScript-Code in einer Ressourcendatei zu platzieren.

   <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. Fügen Sie die JavaScript-Richtlinie der EventFlow des Proxys hinzu. Die EventFlow ist an das Standard-TargetEndpoint angehängt. In diesem Beispiel wird die Gemini API in Vertex AI verwendet, um Inhalte zu generieren.

   <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. Speichern Sie den Proxy und stellen Sie ihn bereit.
6. Rufen Sie den bereitgestellten Proxy auf:

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

   Beispielantwort anzeigen

   Hier ein Beispiel für die Antwort ohne Filterung: Beachten Sie, dass es Fehlerdaten enthält:

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

   Hier ist ein weiteres Beispiel für eine Antwort nach dem Filtern, bei der die Fehlermeldung entfernt wurde.

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

SSE-Ereignisse an ein externes System senden

In diesem Beispiel fügen wir der EventFlow die PublishMessage-Richtlinie von Apigee hinzu, um ein SSE-Ereignis an ein Pub/Sub-Thema zu senden.

1. Erstellen Sie mit der SSE-Proxy-Vorlage einen neuen Proxy. Weitere Informationen finden Sie unter API-Proxy mit vom Server gesendeten Ereignissen (SSE) erstellen.
2. Öffnen Sie den Proxy im Apigee-Proxy-Editor und klicken Sie auf den Tab Entwickeln.
3. Erstellen Sie eine neue PublishMessage-Richtlinie mit der folgenden Definition:

   <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. Fügen Sie die PublishMessage-Richtlinie als Schritt in den EventFlow des API-Proxy hinzu.

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

5. Stellen Sie den API-Proxy bereit und testen Sie ihn.
6. Nachdem Sie dem Pub/Sub-Thema Ihre generierten Inhalte hinzugefügt haben, können Sie beispielsweise eine Cloud Run-Funktion erstellen, um Nachrichten aus dem Thema zu verarbeiten.

Apigee Model Armor-Richtlinie in einem EventFlow verwenden

Mit der Richtlinie „SanitizeModelResponse“ können Sie eingehende vom Server gesendete Ereignisse in einem EventFlow bereinigen. Diese Richtlinie schützt Ihre KI-Anwendungen, indem Antworten von Large Language Models (LLMs) gesäubert werden. Informationen zu Model Armor finden Sie unter Model Armor – Übersicht. Informationen zu den Apigee Model Armor-Richtlinien finden Sie unter Erste Schritte mit Apigee Model Armor-Richtlinien.

1. Erstellen Sie mit der SSE-Proxy-Vorlage einen neuen Proxy. Weitere Informationen finden Sie unter API-Proxy mit vom Server gesendeten Ereignissen (SSE) erstellen.
2. Öffnen Sie den Proxy im Apigee-Proxy-Editor und klicken Sie auf den Tab Entwickeln.
3. Erstellen Sie eine neue Richtlinie für die SanitizeModelResponse-Funktion mit der folgenden Definition:

     <?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. Optional: Fügen Sie eine JavaScript-Richtlinie hinzu, um Ereignisse zu gruppieren, bevor sie an die Apigee Model Armor-Richtlinie gesendet werden.

   <?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. Fügen Sie die JavaScript- und ModelArmor-Richtlinien einem Schritt in der EventFlow des Proxys hinzu:

   <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. Stellen Sie den API-Proxy bereit und testen Sie ihn.

Fehlerbehandlung im EventFlow

Standardmäßig endet der Ereignisstream, wenn ein Fehler auftritt. Wenn Sie jedoch zusätzliche Fehlerbehebungen durchführen möchten, können Sie Fehlerinformationen wie in diesem Beispiel an Cloud Logging senden.

1. Erstellen Sie mit der SSE-Proxy-Vorlage einen neuen Proxy. Weitere Informationen finden Sie unter API-Proxy mit vom Server gesendeten Ereignissen (SSE) erstellen.
2. Öffnen Sie den Proxy im Apigee-Proxy-Editor und klicken Sie auf den Tab Entwickeln.
3. Erstellen Sie eine neue RaiseFault-Richtlinie mit der folgenden Definition:

   <?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. Hängen Sie die RaiseFault-Richtlinie an die EventFlow des SSE-Proxys an:

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

5. Erstellen Sie eine MessageLogging-Richtlinie, um Fehler zu protokollieren. Beispiel:

   <?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. Fügen Sie die MessageLogging-Richtlinie den FaultRules des Zielendpunkts hinzu:

   <?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. Stellen Sie den API-Proxy bereit und testen Sie ihn.

SSE-Daten in Apigee Analytics ansehen

Daten für SSE-Proxys werden in Apigee Analytics wie bei jedem API-Proxy angezeigt. Gehen Sie in der Cloud Console zu Analysen > API-Messwerte.

SSE-Proxys debuggen

Verwenden Sie das Apigee-Debugging-Tool, um SSE-Proxys zu debuggen. Debug-Daten werden für EventFlow genau wie für die anderen Ablauftypen erfasst.

Fehlerbehebung

Wenn Probleme mit Echtzeit-Traffic auftreten, prüfen Sie die Apigee-Zugriffsprotokolle, um die Ursache zu ermitteln.

Beschränkungen

Für SSE-Proxys gelten die folgenden Einschränkungen:

• Da Analysedaten erst nach dem Ende der SSE-Sitzung erfasst werden, kann es zu einer Verzögerung bei der Berichterstellung kommen.
• Fehler innerhalb eines EventFlows führen dazu, dass der Stream sofort beendet wird und kein bestimmtes Fehlerereignis an den Endclient gesendet wird. Informationen zum manuellen Protokollieren dieser Art von Fehlern finden Sie unter EventFlow-Anwendungsfälle und ‑Beispiele.
• Ein Client, der gestreamte SSE-Antworten empfängt, erhält die HTTP-Header, einschließlich aller Statuscodes, am Anfang des Ereignisstreams. Wenn der Ereignisstream also in einen Fehlerstatus gerät, entspricht der ursprünglich empfangene Statuscode nicht dem Fehlerstatus.

  Diese Einschränkung wird in einer Fehlerbehebungssitzung angezeigt. Während der Sitzung fällt dir möglicherweise auf, dass sich der HTTP-Statuscode für Streams, die den Fehlerstatus erreichen, von den Statuscodes unterscheidet, die an den Client gesendet werden. Das kann passieren, weil der Eintrag für die Debug-Sitzung erst nach der Verarbeitung der gesamten Anfrage und nicht am Anfang des Ereignisstreams generiert wird. Die Debug-Sitzung kann den vom Fehler generierten Fehlercode widerspiegeln, während der Client nur den 2xx-Status sieht, der ursprünglich in den Headern empfangen wurde.