Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

串流伺服器傳送的事件

本頁內容適用於 Apigee 和 Apigee Hybrid。

查看 Apigee Edge 說明文件。

Apigee 支援從伺服器推送事件 (SSE) 端點，將回應即時串流至用戶端。Apigee SSE 功能適用於處理大型語言模型 (LLM) API，這類 API 會將回應串流回用戶端，以發揮最大效用。SSE 串流可減少延遲，且用戶端可以在 LLM 生成回應資料後立即接收。這項功能支援使用在即時環境中運作的 AI 代理，例如客戶服務機器人或工作流程協調器。

如要搭配 Apigee 使用 SSE，只要將 API Proxy 指向已啟用 SSE 的目標或 Proxy 端點即可。如要更精細地控管 SSE 回應，Apigee 提供名為 EventFlow 的特殊端點流程。在 EventFlow 的環境中，您可以新增一組有限的政策，對 SSE 回應執行作業，例如篩選、修改或處理錯誤。如要進一步瞭解 Proxy 流程，請參閱「使用流程控管 API Proxy」。

為 SSE 建立 API Proxy

Apigee 使用者介面提供範本，可建立包含 EventFlow 的新 Proxy。

注意：您可以選擇是否使用 SSE Proxy 範本。這個範本可讓您存取 SSE 專屬的流程變數，方便執行特定政策作業。不過，您不一定要使用 EventFlow，才能透過 Apigee 使用 SSE。

重要事項：請為 SSE 目標使用不同的目標端點定義。如果將 SSE 和非 SSE 目標端點混用，可能會導致不一致的行為，例如 response.content 流程變數為空白。請參閱「已知問題」。

請按照下列步驟，使用 Apigee UI 透過 EventFlow 範本建立 API Proxy：

在 Google Cloud 控制台中，前往「Apigee」>「Proxy 開發」>「API Proxy」頁面。

前往 API Proxy
在「API Proxy」窗格中，按一下「+ 建立」。
在「建立 Proxy」窗格的「Proxy template」下方，選取「Proxy with Server-Sent Events (SSE)」。
在「Proxy details」下方，輸入下列資訊：
- Proxy name：輸入 Proxy 名稱，例如 myproxy。
- 基本路徑：系統會自動將這個值設為您輸入的「Proxy 名稱」。基本路徑是向 API 發出要求時所用網址的一部分。Apigee 會使用這個網址比對及將傳入要求路由至適當的 API Proxy。
- 說明 (選用)：輸入新 API Proxy 的說明，例如「Testing Apigee with a simple proxy」。
- 目標 (現有 API)：輸入 API Proxy 的 SSE 目標網址。例如：https://mocktarget.apigee.net/sse-events/5
- 點選「下一步」。
部署 (選用)：
- 部署環境：選用。使用核取方塊選取要部署 Proxy 的一或多個環境。如果不想在這個時間點部署 Proxy，請將「部署環境」欄位留空。您隨時可以稍後部署 Proxy。
- 服務帳戶：選用。Proxy 的服務帳戶。服務帳戶代表已部署 Proxy 的身分，並決定 Proxy 擁有的權限。這是進階功能，在本教學課程中可以忽略。
使用 EventFlow 設定部署的 API Proxy 會以 Extensible 計費。
點選「建立」。

另請參閱建構簡單的 API Proxy。

設定 EventFlow

如要更精細地控制 SSE 回應，Apigee 提供名為 EventFlow 的特殊端點流程。在 EventFlow 的脈絡中，您可以新增下列一組有限的政策，在 SSE 回應串流回用戶端前進行修改。如要進一步瞭解 Proxy 流程，請參閱「使用流程控管 API Proxy」。

EventFlow 的位置

EventFlow 具有下列兩個屬性：

name：用於識別流程的名稱。
content-type：這項屬性的值必須為 text/event-stream。

另請參閱「流程設定參考資料」。

EventFlow 可放在 TargetEndpoint 或 ProxyEndpoint 定義中，如下列程式碼範例所示：

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

注意：如果為端點新增多個 EventFlow 節，系統只會執行端點定義中的最後一個節。系統會忽略最後一個 EventFlow 節之前的所有節。

此外，請務必注意，雖然您可以將 EventFlow 新增至 TargetEndpoint、ProxyEndpoint 或兩者，但只會執行一個 EventFlow。

下表顯示根據端點位置執行的 EventFlow 節：

ProxyEndpoint	TargetEndpoint	使用的 EventFlow
`ProxyEndpoint` 中的 EventFlow	`TargetEndpoint` 中的 EventFlow	`TargetEndpoint` 中的 EventFlow
找不到「`EventFlow`」	`TargetEndpoint` 中的 EventFlow	`TargetEndpoint` 中的 EventFlow
`ProxyEndpoint` 中的 EventFlow	找不到「`EventFlow`」	`ProxyEndpoint` 中的 EventFlow

在 EventFlow 中新增政策

您最多可以在 EventFlow 的 Response 元素中新增四項政策。與所有流程一樣，政策會按照新增順序執行，您可以新增條件式步驟來控管政策的執行方式。請注意，您只能在 EventFlow 中新增下列類型的政策，其他類型的政策則不允許：EventFlow

另請參閱「透過使用者介面附加和設定政策」和「在 XML 檔案中附加及設定政策」。

以下範例顯示新增條件式 RaiseFault 政策步驟的 EventFlow：

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

如需更多 EventFlow 程式碼範例，請參閱「EventFlow 使用案例和範例」一節。

流程變數

EventFlow 會填入三個回應流程變數。請注意，這些變數只能在 EventFlow 中處理的目前事件範圍內使用。在 EventFlow 範圍外存取或設定這些變數不會產生任何影響。這些值只有在 EventFlow 的脈絡中才有意義。

response.event.current.content：包含目前事件完整回應的字串。Apigee 不會以任何方式剖析字串。字串包含完整的回應，包括所有資料欄位，且不會經過任何變更。
注意：Apigee 會將每個回應事件的資料量限制為最多 10 MB。

注意：將這個變數的值設為空白字串 ("")，可防止事件傳送至用戶端。如需範例，請參閱「篩選 SSE 回應」。
response.event.current.data：包含目前事件資料酬載的字串。您可以在 EventFlow 中修改這個變數，變更傳送至用戶端的資料酬載。
response.event.current.count：逐步計算傳送的回應事件數。系統會針對收到的每個事件更新這個值。第一個事件的計數為 1，後續事件的計數會遞增。
注意：如果您在 JavaScript 政策中存取這個值，例如第一個事件的值會是 1，但此時傳送至用戶端的事件數為 0。

另請參閱流程變數參考資料。

EventFlow 用途和範例

下列範例說明如何實作 SSE 代理程式的常見用途：

修改 SSE 回應
篩選 SSE 回應
將 SSE 事件傳送至外部系統
在 EventFlows 中使用 Apigee Model Armor 政策
EventFlow 中的錯誤處理
在 EventFlow 中傳播錯誤訊息
使用 DataCapture 收集 Analytics 中的權杖計數

修改 SSE 回應

這個範例說明如何從 SSE EventFlow 回應中移除資料，再傳回給用戶端。SSE 回應的內容會儲存在名為 response.event.current.content 的流程變數中。在本範例中，我們會使用 JavaScript 政策擷取流程變數的值、剖析並修改該值。另請參閱「流程變數」。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

建立新的 JavaScript 政策，並加入以下定義。在本範例中，JavaScript 程式碼直接納入政策。您也可以將 JavaScript 程式碼放在資源檔案中，設定政策。

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

將 JavaScript 政策新增至 Proxy 的 EventFlow。EventFlow 會附加至預設的 TargetEndpoint 或 ProxyEndpoint。這個範例會使用 Vertex AI 的 Gemini API 生成內容。

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

儲存並部署 Proxy。

呼叫部署的 Proxy：

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

顯示回覆範例

這是未套用任何篩選條件的回應範例。請注意，回應包含 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"
  }

這是套用 JavaScript 政策的另一個範例回應，其中已移除 modelVersion 屬性。

data: {
    "candidates": [
      {
        "content": {
          "parts": [
            {
              "text": " the fantastical creatures of her imagination.  The quiet beauty of a simple life was a magic all its own.\n"
            }
          ],
          "role": "model"
        },
        "finishReason": "STOP"
      }
    ],
    "usageMetadata": {
      "promptTokenCount": 8,
      "candidatesTokenCount": 601,
      "totalTokenCount": 609,
      "promptTokensDetails": [
        {
          "modality": "TEXT",
          "tokenCount": 8
        }
      ],
      "candidatesTokensDetails": [
        {
          "modality": "TEXT",
          "tokenCount": 601
        }
      ]
    }
  }

篩選 SSE 回應

這個範例說明如何篩選 SSE 回應中的資料，再傳回給用戶端。在本例中，我們使用 JavaScript 政策篩選回應中的事件資料。這項政策會將事件回應剖析為 JSON，修改 JSON 以移除事件資料，然後將修改過的回應資料傳回給用戶端。

與上一個範例相同，這個範例會擷取 response.event.current.content 流程變數的值並剖析為 JSON，然後套用邏輯來實作預期的篩選作業。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

建立新的 JavaScript 政策，並使用以下定義。在本例中，JavaScript 程式碼直接納入政策。您也可以將 JavaScript 程式碼放在資源檔案中，設定政策。

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

將 JavaScript 政策新增至 Proxy 的 EventFlow。EventFlow 會附加至預設的 TargetEndpoint 或 ProxyEndpoint。這個範例會使用 Vertex AI 中的 Gemini API 生成內容。

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

儲存並部署 Proxy。

呼叫部署的 Proxy：

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

顯示回覆範例

以下範例顯示未套用任何篩選條件的回應。請注意，這項資料包含錯誤資料：

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

以下是套用篩選器後，並清除錯誤訊息的另一個範例回應。

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

將 SSE 事件傳送至外部系統

在本例中，我們會將 Apigee PublishMessage 政策附加至 EventFlow，將 SSE 事件傳送至 Pub/Sub 主題。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

使用下列定義建立新的 PublishMessage 政策：

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

在 API Proxy 的 EventFlow 中，新增 PublishMessage 政策做為步驟。

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

部署及測試 API Proxy。
將產生的內容新增至 Pub/Sub 主題後，您就可以建立 Cloud Run 函式來處理主題中的訊息。

在 EventFlow 中使用 Apigee Model Armor 政策

您可以使用 SanitizeModelResponse 政策，在 EventFlow 中清除傳入的伺服器推送事件。這項政策會清除大型語言模型 (LLM) 的回覆內容，保護您的 AI 應用程式。如要瞭解 Model Armor，請參閱這篇文章。如要瞭解 Apigee Model Armor 政策，請參閱「開始使用 Apigee Model Armor 政策」。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

使用下列定義建立新的 SanitizeModelResponse 政策：

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

(選用) 在將事件傳送至 Apigee Model Armor 政策之前，先新增 JavaScript 政策來分組事件。

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

將 JavaScript 和 ModelArmor 政策新增至 Proxy 的 EventFlow 中的步驟：

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

部署及測試 API Proxy。

EventFlow 中的錯誤處理機制

預設情況下，發生錯誤時，事件串流會終止。不過，如要進行額外偵錯，可以將錯誤資訊傳送至 Cloud Logging，如這個範例所示。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

建立新的 RaiseFault 政策，並加入以下定義：

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

將 RaiseFault 政策附加至 SSE Proxy 的 EventFlow：

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

建立 MessageLogging 政策，記錄錯誤。例如：

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

將 MessageLogging 政策新增至目標或 Proxy 端點的 FaultRules：

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

部署及測試 API Proxy。

在 EventFlow 中傳播錯誤

在本範例中，我們將說明如何使用 EventFlow 將錯誤傳播至用戶端。這個程序會在政策執行期間立即通知用戶端發生錯誤。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

建立新的 JavaScript 政策，並使用以下定義：

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

這項政策的目的是在符合特定條件時擲回錯誤。

將 JavaScript 政策附加至 TargetEndpoint 或 ProxyEndpoint 設定中的 SSE 代理項目 EventFlow。這個步驟可確保 EventFlow 在處理回應時處理政策：

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

部署 API Proxy。

使用下列 curl 指令測試 Proxy 行為：

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

將 ENVIRONMENT_GROUP_NAME 替換為環境群組的名稱。

輸出內容應如下列範例所示：

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

輸出內容會顯示初始資料串流，然後是 fault 訊息。發生錯誤時，Apigee 會擷取錯誤資訊，並以事件形式傳送給用戶端。

如要進一步瞭解 Apigee 中的錯誤處理，請參閱「處理錯誤」。

使用 DataCapture 收集 Analytics 中的權杖計數

這個範例說明如何在 EventFlow 內使用 DataCapture 政策，從 SSE 事件收集權杖計數，以用於 Apigee Analytics。

建立權杖計數的資料收集器。執行下列指令，在貴機構中建立必要資料收集器：

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

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee Proxy 編輯器中開啟 Proxy，然後點選「Develop」分頁標籤。

建立新的 JavaScript 政策，從 SSE 事件資料剖析權杖計數。在 Apigee 代理伺服器編輯器中，按一下「Policies」旁的「+」，選取「JavaScript」做為政策類型，然後輸入政策名稱 (例如 JS-parse-token)。產生的政策 XML 應如下所示：

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

建立新的 DataCapture 政策，擷取權杖計數。在 Apigee 代理程式編輯器中，按一下「政策」旁的「+」，選取「DataCapture」做為政策類型，然後輸入政策名稱 (例如 DC-capture-tokencount)。產生的政策 XML 應如下所示：

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

請先附加 JavaScript 政策，再附加 DataCapture 政策，位置在 TargetEndpoint 或 ProxyEndpoint 設定中的 SSE 代理程式 EventFlow 內。這個步驟可確保 EventFlow 在處理回應時會處理政策。產生的 XML 應如下所示：

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

儲存並部署 Proxy。詳情請參閱部署 API Proxy。
將測試要求傳送至 Proxy。Proxy 處理 SSE 回應後，您就能透過 dc_tokenCount 和 dc_thoughtsTokenCount 資料收集器，在 Apigee Analytics 中查看擷取的權杖計數。

在 Apigee Analytics 中查看 SSE 資料

SSE 代理伺服器的資料會如預期顯示在 Apigee Analytics 中，就像任何 API Proxy 一樣。在 Cloud 控制台中，前往「Analytics」>「API metrics」。

偵錯 SSE Proxy

使用 Apigee 偵錯工具為 SSE 代理項目偵錯。系統會擷取 EventFlow 的偵錯資料，就像擷取其他流程類型的偵錯資料一樣。

疑難排解

如要解決即時流量問題，請查看 Apigee 存取記錄，找出問題原因。

限制

SSE Proxy 有以下限制：

如果每個事件都套用 DataCapture 政策，系統只會記錄最新值。
雖然 SSE 連線可以無限期保持開啟，但最終還是會受到 Apigee 下游負載平衡器的逾時設定限制。根據預設，Apigee Load Balancer 逾時時間設為 30 秒。如果是執行時間較長的連線，建議增加這個逾時時間，或是建立逾時值較高的獨立後端服務，處理 SSE 流量。
由於系統會在 SSE 工作階段結束後記錄數據分析資料，因此您可能會發現數據分析資料的報表會延遲顯示。
EventFlow 內部的錯誤會導致串流立即結束並擲回錯誤。如要瞭解如何手動記錄這類錯誤，或將錯誤傳送至用戶端，請參閱「EventFlow 用途和範例」。
接收串流 SSE 回應的用戶端會在事件串流的開頭收到 HTTP 標頭，包括任何狀態碼。因此，如果事件串流進入錯誤狀態，一開始收到的狀態碼不會反映錯誤狀態。
查看偵錯工作階段時，可能會看到這項限制。在工作階段中，您可能會發現進入錯誤狀態的串流 HTTP 狀態碼，與傳送給用戶端的狀態碼不同。這是因為偵錯工作階段項目是在處理完整要求後產生，而不是在事件串流開始時產生。偵錯工作階段可能會反映錯誤產生的錯誤代碼，而用戶端最初只會在標頭中看到 2xx 狀態。

串流伺服器傳送的事件 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

為 SSE 建立 API Proxy

設定 EventFlow

EventFlow 的位置

<ProxyEndpoint>

<TargetEndpoint>

在 EventFlow 中新增政策

<ProxyEndpoint>

<TargetEndpoint>

流程變數

EventFlow 用途和範例

修改 SSE 回應

<ProxyEndpoint>

<TargetEndpoint>

顯示回覆範例

篩選 SSE 回應

<ProxyEndpoint>

<TargetEndpoint>

顯示回覆範例

將 SSE 事件傳送至外部系統

<ProxyEndpoint>

<TargetEndpoint>

在 EventFlow 中使用 Apigee Model Armor 政策

EventFlow 中的錯誤處理機制

<TargetEndpoint>

<ProxyEndpoint>

在 EventFlow 中傳播錯誤

<TargetEndpoint>

<ProxyEndpoint>

使用 DataCapture 收集 Analytics 中的權杖計數

<TargetEndpoint>

<ProxyEndpoint>

在 Apigee Analytics 中查看 SSE 資料

偵錯 SSE Proxy

疑難排解

限制

串流伺服器傳送的事件