Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

串流伺服器傳送的事件

本頁面適用於 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 UI 提供範本，可用於建立包含 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 Proxies」(API Proxy) 頁面
在「API Proxies」窗格中，點選「+ Create」。
在「Create a proxy」窗格的「Proxy template」下方，選取「Proxy with Server-Sent Events (SSE)」。
在「Proxy details」下方，輸入下列資訊：
- Proxy 名稱：輸入 Proxy 名稱，例如 myproxy。
- Base Path：系統會自動設為您在「Proxy name」中輸入的值。「Base Path」(基本路徑) 是用來向 API 提出要求的部分網址。Apigee 會使用網址比對傳入的要求，並將要求轉送至適當的 API Proxy。
- 說明 (選填)：輸入新 API Proxy 的說明，例如「使用簡易 Proxy 測試 Apigee」。
- Target (Existing API) (目標 (現有 API))：輸入 API Proxy 的 SSE 目標網址。例如：https://mocktarget.apigee.net/sse-events/5
- 點選「下一步」。
部署 (選用)：
- 部署環境：選用。使用核取方塊選取一或多個要部署 Proxy 的環境。如果目前不想部署 Proxy，請將「Deployment environments」(部署環境) 欄位留空。您之後隨時可以部署 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 中不得包含其他類型的政策：

另請參閱「在 UI 中附加及設定政策」和「在 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 Proxy 的常見用途：

修改 SSE 回應
篩選 SSE 回應
將 SSE 事件傳送至外部系統
在 EventFlow 中使用 Apigee Model Armor 政策
EventFlow 中的錯誤處理
在 EventFlow 中傳播錯誤訊息
使用 DataCapture 收集數據分析中的權杖數量
使用 LLMTokenQuota 政策在 EventFlow 中強制執行詞元配額

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

將 PublishMessage 政策新增為 API Proxy EventFlow 中的步驟。

<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 中的伺服器傳送事件。將 <ContinuousScan> 設為 true，即可檢查整個串流 (如以下步驟所示)，或保留 false (預設值)，分別檢查每個事件。如果設為 true，政策會保留每個事件，直到 Model Armor 確認文字不含不當內容，然後再將事件完整、不變且依序釋出，傳送至後續步驟和用戶端。

如要進一步瞭解政策如何處理相符項目、錯誤和配額，請參閱「清理串流伺服器傳送事件 (SSE) 回應」。如要進一步瞭解 Model Armor，請參閱「Model Armor 總覽」。

如要設定政策，請完成下列步驟：

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

使用 <ContinuousScan>true</ContinuousScan> 和 <LLMResponseSource> 建立 SanitizeModelResponse 政策，並設定從 response.event.current.data 擷取文字差異。以下是 Gemini SSE 串流的範例：

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

如果是與 OpenAI 相容的 SSE 串流，請將 <LLMResponseSource> 設為 {jsonPath('$.choices[-1].delta.content',response.event.current.data,true)}。如果是 Anthropic Claude SSE 串流，請將 <LLMResponseSource> 設為 {jsonPath('$.delta.text',response.event.current.data,true)}。

將 SanitizeModelResponse 政策附加至 Proxy 中 EventFlow 的 <Response> 元素。如果您也在 EventFlow 中使用 LLMTokenQuota 或 DataCapture，請先放置 SanitizeModelResponse，讓這些政策只接收已檢查的事件：
```
<EventFlow name="EventFlow" content-type="text/event-stream">
  <Request/>
  <Response>
    <Step>
      <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 Proxy 的 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 收集數據分析中的權杖數量

這個範例說明如何在 EventFlow 中使用資料擷取政策，從 SSE 事件收集權杖數量，用於 Apigee 數據分析。

建立資料收集器，用於收集權杖數量。執行下列指令，在組織中建立必要的資料收集器：

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 Proxy 編輯器中，按一下「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>

建立新的資料擷取政策，擷取權杖數量。在 Apigee Proxy 編輯器中，按一下「Policies」旁邊的「+」，選取「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>

在 TargetEndpoint 或 ProxyEndpoint 設定中，先將 JavaScript 政策附加到 SSE Proxy 的 EventFlow，再附加 DataCapture 政策。這個步驟可確保 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 數據分析中。

使用 LLMTokenQuota 政策，在 EventFlow 中強制執行詞元配額

這個範例說明如何在 EventFlow 中使用 LLMTokenQuota 政策，計算並強制執行 SSE 串流的詞元配額。處理事件串流時，只有在事件中找到 LLM 回覆的詞元用量中繼資料時，才會執行詞元計數。系統發現權杖使用中繼資料時，會擷取該中繼資料並執行政策。對於所有其他事件，政策不會執行任何作業。

注意：如果 LLMTokenQuota 政策是 EventFlow 的一部分，<LLMTokenUsageSource> 和 <LLMModelSource> 元素必須使用 response.event.current.data 流程變數，從目前事件的資料讀取，並將 jsonPath 函式的第三個引數設為 false。

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

建立新的 LLMTokenQuota 政策，計算 SSE 串流使用的詞元。在 Apigee Proxy 編輯器中，按一下「Policies」旁的「+」，選取「LLMTokenQuota」做為政策類型，然後輸入政策名稱 (例如 LTQ-Count-Only)。產生的政策 XML 應如下所示：

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

建立第二個 LLMTokenQuota 政策，對傳入的要求強制執行詞元配額。在 Apigee Proxy 編輯器中，按一下「Policies」旁邊的「+」，選取「LLMTokenQuota」做為政策類型，然後輸入政策名稱 (例如 LTQ-Enforce-Only)。這項政策會使用與計數政策相同的 <SharedName>，因此兩項政策會共用同一個配額計數器。產生的政策 XML 應如下所示：
```
<?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>
```

將 LTQ-Enforce-Only 政策新增為 ProxyEndpoint 的 PreFlow 的 Request 中的步驟。這個步驟可確保系統會檢查每項傳入要求的配額。產生的 XML 應如下所示：

<ProxyEndpoint name="default">
  <PreFlow name="PreFlow">
    <Request>
      <Step>
        <Name>LTQ-Enforce-Only</Name>
      </Step>
    </Request>
    <Response/>
  </PreFlow>
</ProxyEndpoint>

在 TargetEndpoint 或 ProxyEndpoint 設定中，將 LTQ-Count-Only 政策新增為 SSE Proxy 的 EventFlow 內步驟。這個步驟可確保 EventFlow 在處理回應時處理政策，並增加共用配額計數器。產生的 XML 應如下所示：

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

儲存並部署 Proxy。詳情請參閱「部署 API Proxy」。
傳送測試要求至 Proxy。針對每個傳入要求，LTQ-Enforce-Only 政策會檢查共用配額計數器，並在達到設定的限制時拒絕要求。當 Proxy 處理 SSE 回應時，LTQ-Count-Only 政策會計算每個事件的權杖用量中繼資料所回報的權杖，並遞增相同的共用配額計數器。當計數器達到設定的限制時，系統會拒絕後續的權杖消耗要求，直到計數器重設為止。

如要進一步瞭解 LLMTokenQuota 政策，包括如何搭配 SSE 回應使用，以及如何設定共用配額計數器，請參閱「LLMTokenQuota 政策」。

在 Apigee Analytics 中查看 SSE 資料

SSE Proxy 的資料會顯示在 Apigee Analytics 中，與任何 API Proxy 的預期行為相同。在 Cloud 控制台中，依序前往「Analytics」>「API metrics」。

偵錯 SSE Proxy

使用 Apigee 偵錯工具對 SSE Proxy 進行偵錯。系統會擷取 EventFlow 的偵錯資料，就像擷取其他流程類型的資料一樣。

疑難排解

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

限制

SSE Proxy 有以下限制：

如果每項事件都套用 DataCapture 政策，系統只會記錄最新值。
雖然 SSE 連線可以無限期保持開啟，但其持續時間最終會受到 Apigee 下游負載平衡器的逾時設定限制。根據預設，Apigee 負載平衡器逾時時間設為 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 收集數據分析中的權杖數量

<TargetEndpoint>

<ProxyEndpoint>

使用 LLMTokenQuota 政策，在 EventFlow 中強制執行詞元配額

<TargetEndpoint>

<ProxyEndpoint>

在 Apigee Analytics 中查看 SSE 資料

偵錯 SSE Proxy

疑難排解

限制

串流伺服器傳送的事件