Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Eventi inviati dal server in streaming

Questa pagina si applica ad Apigee e ad Apigee hybrid.

Visualizza la documentazione di Apigee Edge.

Apigee supporta lo streaming continuo delle risposte dagli endpoint di eventi inviati dal server (SSE) ai client in tempo reale. La funzionalità SSE di Apigee è utile per gestire le API dei modelli linguistici di grandi dimensioni (LLM) che operano in modo più efficace trasmettendo le risposte al client. Lo streaming SSE riduce la latenza e i client possono ricevere i dati di risposta non appena vengono generati da un LLM. Questa funzionalità supporta l'utilizzo di agenti AI che operano in ambienti in tempo reale, come bot di assistenza clienti o orchestratori di workflow.

Per utilizzare SSE con Apigee, basta indirizzare un proxy API a un endpoint di destinazione o proxy abilitato per SSE. Per ottenere un controllo più granulare della risposta SSE, Apigee fornisce un flusso di endpoint speciale chiamato EventFlow. Nel contesto di un EventFlow, puoi aggiungere un insieme limitato di policy per eseguire operazioni sulla risposta SSE, come il filtro, la modifica o la gestione degli errori. Per scoprire di più sui flussi dei proxy, consulta Controllo dei proxy API con i flussi.

Crea un proxy API per SSE

La UI di Apigee fornisce un modello per la creazione di un nuovo proxy che include un EventFlow.

Nota: l'utilizzo del modello di proxy SSE è facoltativo. Ti consente di accedere alle variabili di flusso specifiche per SSE che possono essere utili per eseguire determinate operazioni delle policy. Tuttavia, non è necessario utilizzare EventFlow per utilizzare SSE con Apigee.

Importante: utilizza definizioni di endpoint di destinazione separate per le destinazioni SSE. La combinazione di endpoint di destinazione SSE e non SSE potrebbe comportare un comportamento incoerente, ad esempio variabili di flusso response.content vuote. Vedi Problemi noti.

Segui questi passaggi per creare un proxy API con il template EventFlow utilizzando la UI di Apigee:

Nella console Google Cloud , vai alla pagina Apigee > Sviluppo proxy > Proxy API.

Vai a Proxy API
Nel riquadro Proxy API, fai clic su + Crea.
Nel riquadro Crea un proxy, in Template di proxy, seleziona Proxy con Server-Sent Events (SSE).
Nella sezione Dettagli proxy, inserisci quanto segue:
- Nome proxy: inserisci un nome per il proxy, ad esempio myproxy.
- Base Path (Percorso di base): impostato automaticamente sul valore inserito per Proxy name (Nome proxy). Il percorso di base fa parte dell'URL utilizzato per inviare richieste alla tua API. Apigee utilizza l'URL per trovare una corrispondenza e instradare le richieste in entrata al proxy API appropriato.
- Descrizione (facoltativa): inserisci una descrizione per il nuovo proxy API, ad esempio "Test di Apigee con un proxy semplice".
- Target (Existing API) (Destinazione (API esistente)): inserisci l'URL di destinazione SSE per il proxy API. Ad esempio: https://mocktarget.apigee.net/sse-events/5
- Fai clic su Avanti.
(Facoltativo) Deployment:
- Ambienti di deployment: facoltativo. Utilizza le caselle di controllo per selezionare uno o più ambienti in cui eseguire il deployment del proxy. Se preferisci non eseguire il deployment del proxy in questo momento, lascia vuoto il campo Ambienti di deployment. Puoi sempre eseguire il deployment del proxy in un secondo momento.
- Service account: facoltativo. Un service account per il proxy. Il account di servizio rappresenta l'identità del proxy di cui è stato eseguito il deployment e determina le autorizzazioni di cui dispone. Si tratta di una funzionalità avanzata e, ai fini di questo tutorial, puoi ignorarla.
I proxy API di cui è stato eseguito il deployment con una configurazione EventFlow verranno fatturati come Extensible.
Fai clic su Crea.

Vedi anche Creazione di un semplice proxy API.

Configura un EventFlow

Per ottenere un controllo più granulare della risposta SSE, Apigee fornisce un flusso di endpoint speciale chiamato EventFlow. Nel contesto di un EventFlow, puoi aggiungere un insieme limitato di policy, elencate di seguito, per modificare la risposta SSE prima che venga trasmessa in streaming al client. Per scoprire di più sui flussi dei proxy, consulta Controllo dei proxy API con i flussi.

Posizionamento di un EventFlow

UnEventFlow ha due attributi:

name: un nome per identificare il flusso.
content-type: il valore di questo attributo deve essere text/event-stream.

Consulta anche il Riferimento per la configurazione dei flussi.

Un EventFlow può essere inserito all'interno di una definizione TargetEndpoint o ProxyEndpoint, come mostrato nei seguenti esempi di codice:

<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: se aggiungi più istruzioni EventFlow a un endpoint, viene eseguita solo l'ultima nella definizione dell'endpoint. Eventuali sezioni EventFlow che compaiono prima dell'ultima vengono ignorate.

È inoltre importante notare che, sebbene sia possibile aggiungere un EventFlow a un TargetEndpoint, a un ProxyEndpoint o a entrambi, viene eseguito un solo EventFlow.

La tabella seguente mostra l'esecuzione delle istruzioni EventFlow in base al posizionamento degli endpoint:

ProxyEndpoint	TargetEndpoint	EventFlow utilizzato
EventFlow in `ProxyEndpoint`	EventFlow in `TargetEndpoint`	EventFlow in `TargetEndpoint`
Nessuna `EventFlow`	EventFlow in `TargetEndpoint`	EventFlow in `TargetEndpoint`
EventFlow in `ProxyEndpoint`	Nessuna `EventFlow`	EventFlow in `ProxyEndpoint`

Aggiungi policy a un EventFlow

Puoi aggiungere fino a un totale di quattro policy all'elemento Response di EventFlow. Come per tutti i flussi, le policy vengono eseguite nell'ordine in cui vengono aggiunte e puoi aggiungere passaggi condizionali per controllarne l'esecuzione. È importante notare che i tipi di policy che puoi aggiungere a un EventFlow sono limitati ai seguenti. In un EventFlow non sono consentiti altri tipi di policy:

Consulta anche Collegamento e configurazione delle policy nell'interfaccia utente e Collegamento e configurazione delle policy nei file XML.

Gli esempi seguenti mostrano un EventFlow con un passaggio della policy RaiseFault condizionale aggiunto:

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

Per altri esempi di codice EventFlow, consulta la sezione Casi d'uso ed esempi di EventFlow.

Variabili di flusso

Un EventFlow compila tre variabili del flusso di risposta. Tieni presente che queste variabili sono utilizzabili solo nell'ambito dell'evento corrente in fase di elaborazione all'interno di EventFlow. L'accesso o l'impostazione di queste variabili al di fuori dell'ambito EventFlow non ha alcun effetto. Sono significativi solo nel contesto di EventFlow.

response.event.current.content: una stringa contenente l'intera risposta dell'evento corrente. Apigee non analizza la stringa in alcun modo. Contiene l'intera risposta invariata, inclusi tutti i campi di dati.
Nota: Apigee limita ogni evento di risposta a un massimo di 10 MB di dati.

Nota: Impostando il valore di questa variabile su una stringa vuota (""), si impedisce l'invio dell'evento al client. Per un esempio, consulta Filtra una risposta SSE.
response.event.current.data: una stringa contenente il payload dei dati dell'evento corrente. Puoi modificare questa variabile in EventFlow per cambiare il payload dei dati inviato al client.
response.event.current.count: conteggia in modo incrementale il numero di eventi di risposta inviati. Questo valore viene aggiornato per ogni evento ricevuto. Il conteggio sarà 1 per il primo evento e aumenterà per gli eventi successivi.
Nota: se accedi a questo valore, ad esempio in una policy JavaScript, il primo evento avrà un valore pari a 1; tuttavia, il numero di eventi inviati al client in questo momento sarà 0.

Consulta anche il Riferimento alle variabili di flusso.

Casi d'uso ed esempi di EventFlow

Gli esempi seguenti mostrano come implementare casi d'uso comuni per i proxy SSE:

Modifica una risposta SSE
Filtra una risposta SSE
Invia un evento SSE a un sistema esterno
Utilizza una policy Apigee Model Armor in un EventFlow
Gestione degli errori in EventFlow
Propaga i messaggi di errore in un EventFlow
Utilizza DataCapture per raccogliere i conteggi dei token in Analytics
Utilizza una policy LLMTokenQuota per applicare le quote di token in un EventFlow

Modifica una risposta SSE

Questo esempio mostra come rimuovere i dati da una risposta SSE EventFlow prima di restituirla al client. I contenuti della risposta SSE vengono archiviati in una variabile di flusso denominata response.event.current.content. In questo caso, utilizziamo una policy JavaScript per recuperare il valore della variabile di flusso, analizzarlo e modificarlo. Consulta anche Variabili di flusso.

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy JavaScript con la seguente definizione. In questo esempio, il codice JavaScript è incluso direttamente nella policy. L'inserimento del codice JavaScript in un file di risorse è un'altra opzione per la configurazione della policy.

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

Aggiungi la policy JavaScript al EventFlow del proxy. EventFlow è collegato alla TargetEndpoint o alla ProxyEndpoint predefinita. Questo esempio utilizza l'API Gemini in Vertex AI per generare contenuti.

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

Salva il proxy ed eseguine il deployment.

Chiama il proxy di cui è stato eseguito il deployment:

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

Mostra una risposta di esempio

Questa è una risposta di esempio senza alcun filtro applicato. Tieni presente che la risposta include un attributo 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"
  }

Questa è un'altra risposta di esempio con la policy JavaScript applicata. L'attributo modelVersion viene rimosso.

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

Questo esempio mostra come filtrare i dati di una risposta SSE prima di restituirli al client. In questo caso, filtriamo i dati sugli eventi dalla risposta utilizzando una policy JavaScript. La policy analizza la risposta dell'evento in JSON, modifica il JSON per rimuovere i dati sugli eventi e poi invia i dati della risposta modificati al client.

Come nell'esempio precedente, questo esempio recupera il valore della variabile di flusso response.event.current.content e lo analizza in JSON, quindi applica la logica per implementare il filtro previsto.

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

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

Aggiungi la policy JavaScript al EventFlow del proxy. EventFlow è collegato alla rete TargetEndpoint o ProxyEndpoint predefinita. Questo esempio utilizza l'API Gemini in Vertex AI per generare contenuti.

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

Salva il proxy ed eseguine il deployment.

Chiama il proxy di cui è stato eseguito il deployment:

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

Mostra una risposta di esempio

Ecco un esempio di come potrebbe apparire la risposta senza applicare alcun filtro. Nota che include i dati sugli errori:

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

Ecco un altro esempio di risposta dopo l'applicazione del filtro con il messaggio di errore rimosso.

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

Invia un evento SSE a un sistema esterno

In questo esempio, colleghiamo la policy PublishMessage di Apigee a EventFlow per inviare un evento SSE a un argomento Pub/Sub.

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy PublishMessage con la seguente definizione:

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

Aggiungi la policy PublishMessage come passaggio in EventFlow del proxy 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>

Esegui il deployment e testa il proxy API.
Dopo aver aggiunto i contenuti generati all'argomento Pub/Sub, puoi, ad esempio, creare una funzione Cloud Run per elaborare i messaggi dell'argomento.

Utilizza una policy Apigee Model Armor in un EventFlow

Puoi utilizzare la policy SanitizeModelResponse per eseguire la sanitizzazione degli eventi inviati dal server in un EventFlow. Imposta <ContinuousScan> su true per controllare l'intero stream, come nei passaggi seguenti, o lascialo su false (impostazione predefinita) per controllare ogni evento separatamente. Con true, la policy trattiene ogni evento finché Model Armor non conferma che il testo è pulito, quindi lo rilascia intero, invariato e in ordine per i passaggi successivi e per il client.

Per saperne di più su come la policy gestisce corrispondenze, errori e quote, consulta Sanitizza le risposte di Server-Sent Events (SSE) in streaming. Per saperne di più su Model Armor, consulta la panoramica di Model Armor.

Per configurare la policy, completa i seguenti passaggi:

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una policy SanitizeModelResponse con <ContinuousScan>true</ContinuousScan> e <LLMResponseSource> configurati per estrarre il delta del testo da response.event.current.data. L'esempio seguente è per gli stream SSE di 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>

Per gli stream SSE compatibili con OpenAI, imposta <LLMResponseSource> su {jsonPath('$.choices[-1].delta.content',response.event.current.data,true)}. Per gli stream SSE di Anthropic Claude, imposta <LLMResponseSource> su {jsonPath('$.delta.text',response.event.current.data,true)}.

Collega la policy SanitizeModelResponse all'elemento <Response> di EventFlow nel proxy. Se utilizzi anche LLMTokenQuota o DataCapture in EventFlow, posiziona SanitizeModelResponse per prima in modo che queste policy ricevano solo gli eventi controllati:
```
<EventFlow name="EventFlow" content-type="text/event-stream">
  <Request/>
  <Response>
    <Step>
      <Name>SMR-modelresponse</Name>
    </Step>
  </Response>
</EventFlow>
```
Esegui il deployment e testa il proxy API.

Gestione degli errori in EventFlow

Per impostazione predefinita, lo stream di eventi termina quando si verifica un errore. Tuttavia, se vuoi eseguire un debug aggiuntivo, puoi inviare informazioni sugli errori a Cloud Logging, come mostrato in questo esempio.

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy RaiseFault con la seguente definizione:

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

Collega la policy RaiseFault al EventFlow del 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>

Crea una policy MessageLogging per registrare gli errori. Ad esempio:

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

Aggiungi la policy MessageLogging alle regole FaultRules dell'endpoint di destinazione o 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>

Esegui il deployment e testa il proxy API.

Propaga gli errori di errore in un EventFlow

In questo esempio, mostriamo come utilizzare un EventFlow per propagare gli errori di errore al client. La procedura funziona in modo da informare immediatamente il client degli errori durante l'esecuzione della policy.

Crea un nuovo proxy con il modello di proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy JavaScript con la seguente definizione:

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

Questa policy è progettata per generare un errore quando viene soddisfatta una condizione specifica.

Collega la policy JavaScript a EventFlow del proxy SSE all'interno della configurazione TargetEndpoint o ProxyEndpoint. Questo passaggio garantisce che EventFlow elabori la policy durante la gestione della risposta:

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

Esegui il deployment del proxy API.

Esegui il test con il comportamento del proxy utilizzando il seguente comando curl:

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

Sostituisci ENVIRONMENT_GROUP_NAME con il nome del tuo gruppo di ambiente.

L'output dovrebbe avere un aspetto simile all'esempio riportato di seguito:

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

L'output mostra il stream di dati iniziale seguito da un messaggio fault. In caso di errore, Apigee acquisisce le informazioni sull'errore e le invia al client come evento.

Per saperne di più sulla gestione degli errori in Apigee, consulta Gestione degli errori.

Utilizza DataCapture per raccogliere i conteggi dei token in Analytics

Questo esempio mostra come utilizzare la policy DataCapture all'interno di un EventFlow per raccogliere i conteggi dei token dagli eventi SSE da utilizzare in Apigee Analytics.

Crea raccoglitori di dati per i conteggi dei token. Esegui i seguenti comandi per creare i raccoglitori di dati richiesti nella tua organizzazione:

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 nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy JavaScript per analizzare i conteggi dei token dai dati sugli eventi SSE. Nell'editor proxy Apigee, fai clic su + accanto a Policy, seleziona JavaScript come tipo di policy e inserisci un nome per la policy (ad esempio JS-parse-token). Il codice XML della policy risultante dovrebbe essere simile al seguente:

<?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 nuova policy DataCapture per acquisire i conteggi dei token. Nell'editor proxy Apigee, fai clic su + accanto a Policy, seleziona DataCapture come tipo di policy e inserisci un nome per la policy (ad esempio DC-capture-tokencount). Il codice XML della policy risultante dovrebbe essere simile al seguente:

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

Collega prima la policy JavaScript, seguita dalla policy DataCapture, all'interno di EventFlow del proxy SSE nella configurazione TargetEndpoint o ProxyEndpoint. Questo passaggio garantisce che EventFlow elabori le policy durante la gestione delle risposte. Il file XML risultante dovrebbe avere il seguente aspetto:

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

Salva ed esegui il deployment del proxy. Per maggiori dettagli, consulta Deployment dei proxy API.
Inviare richieste di test al proxy. Dopo che il proxy ha elaborato le risposte SSE, i conteggi dei token acquisiti sono disponibili in Analisi di Apigee tramite i raccoglitori di dati dc_tokenCount e dc_thoughtsTokenCount.

Utilizza una policy LLMTokenQuota per applicare le quote di token in un EventFlow

Questo esempio mostra come utilizzare la policy LLMTokenQuota all'interno di un EventFlow per conteggiare e applicare le quote di token per uno stream SSE. Durante l'elaborazione dello stream di eventi, il conteggio dei token viene eseguito solo quando nell'evento vengono trovati i metadati di utilizzo dei token della risposta LLM. Quando vengono rilevati i metadati di utilizzo dei token, questi vengono estratti e la policy viene eseguita. Per tutti gli altri eventi, la policy non esegue alcuna operazione.

Nota: Quando la politica LLMTokenQuota fa parte di unEventFlow, IL<LLMTokenUsageSource> E<LLMModelSource> gli elementi devono leggere dai dati dell'evento corrente utilizzando ilresponse.event.current.data variabile di flusso, con iljsonPath Il terzo argomento della funzione è impostato sufalse.

Crea un nuovo proxy con il template del proxy SSE. Consulta Crea un proxy API con eventi inviati dal server (SSE).
Apri il proxy nell'editor proxy Apigee e fai clic sulla scheda Sviluppa.

Crea una nuova policy LLMTokenQuota per conteggiare i token consumati dallo stream SSE. Nell'editor proxy Apigee, fai clic su + accanto a Policy, seleziona LLMTokenQuota come tipo di policy e inserisci un nome per la policy (ad esempio LTQ-Count-Only). Il codice XML della policy risultante dovrebbe essere simile al seguente:

<?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 seconda policy LLMTokenQuota per applicare la quota di token alle richieste in entrata. Nell'editor proxy Apigee, fai clic su + accanto a Policy, seleziona LLMTokenQuota come tipo di policy e inserisci un nome per la policy (ad esempio LTQ-Enforce-Only). Questa policy utilizza lo stesso <SharedName> della policy di conteggio, in modo che entrambe le policy condividano lo stesso contatore di quota. Il file XML della policy risultante dovrebbe avere il seguente aspetto:
```
<?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>
```
Aggiungi la policy LTQ-Enforce-Only come passaggio nel Request del PreFlow del ProxyEndpoint. Questo passaggio garantisce che la quota venga controllata per ogni richiesta in arrivo. Il codice XML risultante dovrebbe essere simile al seguente:
```
<ProxyEndpoint name="default">
  <PreFlow name="PreFlow">
    <Request>
      <Step>
        <Name>LTQ-Enforce-Only</Name>
      </Step>
    </Request>
    <Response/>
  </PreFlow>
</ProxyEndpoint>
```

Aggiungi la policy LTQ-Count-Only come passaggio all'interno di EventFlow del proxy SSE nella configurazione TargetEndpoint o ProxyEndpoint. Questo passaggio garantisce che EventFlow elabori la policy durante la gestione delle risposte e incrementi il contatore della quota condivisa. Il codice XML risultante dovrebbe avere il seguente aspetto:

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

Salva ed esegui il deployment del proxy. Per maggiori dettagli, consulta Deployment dei proxy API.
Inviare richieste di test al proxy. Per ogni richiesta in entrata, la policy LTQ-Enforce-Only controlla il contatore della quota condivisa e rifiuta la richiesta se il limite configurato è già stato raggiunto. Mentre il proxy elabora le risposte SSE, la policy LTQ-Count-Only conteggia i token riportati nei metadati di utilizzo dei token di ogni evento e incrementa lo stesso contatore della quota condivisa. Quando il contatore raggiunge il limite configurato, le richieste successive che consumano token vengono rifiutate fino al reset del contatore.

Per saperne di più sulla policy LLMTokenQuota, incluso come utilizzarla con le risposte SSE e come configurare i contatori di quote condivise, consulta la policy LLMTokenQuota.

Visualizzazione dei dati SSE in Apigee Analytics

I dati per i proxy SSE vengono visualizzati in Apigee Analytics come previsto per qualsiasi proxy API. Nella console Cloud, vai ad Analisi > Metriche API.

Debug dei proxy SSE

Utilizza lo strumento di debug di Apigee per eseguire il debug dei proxy SSE. I dati di debug vengono acquisiti per EventFlow esattamente come per gli altri tipi di flusso.

Risoluzione dei problemi

Per i problemi di traffico in tempo reale, controlla i log di accesso di Apigee per determinare la causa.

Limitazioni

Ai proxy SSE si applicano le seguenti limitazioni:

Se la policy DataCapture viene applicata a ogni evento, verrà registrato solo l'ultimo valore.
Sebbene una connessione SSE possa essere mantenuta aperta a tempo indeterminato, la sua durata è in ultima analisi limitata dall'impostazione di timeout del bilanciatore del carico a valle di Apigee. Per impostazione predefinita, il timeout del bilanciatore del carico Apigee è impostato su 30 secondi. Per le connessioni di lunga durata, ti consigliamo di aumentare questo timeout o di creare un servizio di backend separato con un valore di timeout più elevato per gestire il traffico SSE.
Poiché i dati di analisi vengono registrati dopo la chiusura della sessione SSE, potresti notare un certo ritardo nella generazione dei report dei dati di analisi.
I guasti all'interno di un EventFlow causano l'uscita immediata dallo stream e generano un errore. Per informazioni su come registrare manualmente questi tipi di errori o inviarli al client, consulta Casi d'uso ed esempi di EventFlow.
Un client che riceve risposte SSE in streaming riceverà le intestazioni HTTP, inclusi eventuali codici di stato, all'inizio dello stream di eventi. Di conseguenza, se lo stream di eventi entra in uno stato di errore, il codice di stato ricevuto inizialmente non rifletterà lo stato di errore.
Questa limitazione può essere visualizzata quando si visualizza una sessione di debug. Nella sessione, potresti notare che il codice di stato HTTP per gli stream che entrano nello stato di errore è diverso dai codici di stato inviati al client. Questo può accadere perché la voce della sessione di debug viene generata dopo l'elaborazione dell'intera richiesta, anziché all'inizio dello stream di eventi. La sessione di debug potrebbe riflettere il codice di errore generato dall'errore, mentre il client vede solo lo stato 2xx inizialmente ricevuto nelle intestazioni.

Eventi inviati dal server in streaming Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.

Crea un proxy API per SSE

Configura un EventFlow

Posizionamento di un EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Aggiungi policy a un EventFlow

<ProxyEndpoint>

<TargetEndpoint>

Variabili di flusso

Casi d'uso ed esempi di EventFlow

Modifica una risposta SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostra una risposta di esempio

Filtra una risposta SSE

<ProxyEndpoint>

<TargetEndpoint>

Mostra una risposta di esempio

Invia un evento SSE a un sistema esterno

<ProxyEndpoint>

<TargetEndpoint>

Utilizza una policy Apigee Model Armor in un EventFlow

Gestione degli errori in EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Propaga gli errori di errore in un EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Utilizza DataCapture per raccogliere i conteggi dei token in Analytics

<TargetEndpoint>

<ProxyEndpoint>

Utilizza una policy LLMTokenQuota per applicare le quote di token in un EventFlow

<TargetEndpoint>

<ProxyEndpoint>

Visualizzazione dei dati SSE in Apigee Analytics

Debug dei proxy SSE

Risoluzione dei problemi

Limitazioni

Eventi inviati dal server in streaming