本頁面由 Cloud Translation API 翻譯而成。

串流伺服器傳送的事件

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

查看 Apigee Edge 說明文件。

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

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

為 SSE 建立 API Proxy

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

注意：您可以自由選擇是否使用 SSE Proxy 範本。您可藉此存取 SSE 專屬的流程變數，有助於執行特定政策作業。不過，您不必使用 EventFlow，即可透過 Apigee 使用 SSE。

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

在瀏覽器中開啟 Cloud 控制台中的 Apigee 使用者介面。
在左側導覽窗格中，按一下「Proxy development」>「API proxies」。
在「API Proxies」窗格中，按一下「+ Create」。
在「建立 Proxy」窗格中，選取「Proxy 範本」下方的「Proxy with Server-Sent Events (SSE)」。
在「Proxy 詳細資料」下方，輸入下列資訊：
- Proxy name：輸入 Proxy 的名稱，例如 myproxy。
- 基本路徑：系統會自動將此值設為您在 Proxy name 中輸入的值。基本路徑是向 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。
  注意：對於隨用隨付環境，SSE 僅支援「全面」環境類型。請參閱「環境類型」。

服務帳戶：選用。Proxy 的服務帳戶。服務帳戶代表已部署 Proxy 的身分，並決定其擁有的權限。這是進階功能，在本教學課程中可以忽略。

使用 EventFlow 設定部署的 API Proxy 會以 Extensible 計費。

按一下 [Create] (建立)。另請參閱建構簡單的 API Proxy。

設定 EventFlow

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

EventFlow 必須放在 TargetEndpoint 定義內，如以下程式碼範例所示：

注意：如果將多個 EventFlow 詩節新增至 TargetEndpoint，系統只會執行 TargetEndpoint 中的最後一個詩節。系統會忽略最後一個 EventFlow 節之前的所有節。

<TargetEndpoint name="default">
  <Description/>
  <FaultRules/>
  <PreFlow name="PreFlow">
    <Request/>
    <Response/>
  </PreFlow>
  <PostFlow name="PostFlow">
    <Request/>
    <Response/>
  </PostFlow>
  <Flows/>
  <EventFlow name="EventFlow" content-type="text/event-stream">
    <Response/>
  </EventFlow>
  <HTTPTargetConnection>
    <Properties/>
    <URL>https://httpbun.org/sse</URL>
  </HTTPTargetConnection>
</TargetEndpoint>

EventFlow 具有兩項屬性：

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

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

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

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

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

<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.count：逐步計算傳送的回應事件數。系統會針對收到的每個事件更新這個值。第一個事件的計數為 1，後續事件的計數則會遞增。
注意：如果您在 JavaScript 政策中存取這個值，例如第一個事件的值會是 1，但此時傳送至用戶端的事件數為 0。

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

EventFlow 用途與範例

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

修改 SSE 回應
篩選 SSE 回應
將 SSE 事件傳送至外部系統
在 EventFlows 中使用 Apigee Model Armor 政策
EventFlow 中的錯誤處理

修改 SSE 回應

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

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee 代理程式編輯器中開啟代理程式，然後按一下「開發」分頁標籤。

使用下列定義建立新的 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。這個範例使用 Vertex AI 中的 Gemini API 生成內容。

<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 代理程式編輯器中開啟代理程式，然後按一下「開發」分頁標籤。

使用下列定義建立新的 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。這個範例會使用 Vertex AI 中的 Gemini API 生成內容。

<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 代理程式編輯器中開啟代理程式，然後按一下「開發」分頁標籤。

使用下列定義建立新的 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 政策做為步驟。

<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 的相關資訊，請參閱Model Armor 總覽。如要瞭解 Apigee Model Armor 政策，請參閱「開始使用 Apigee Model Armor 政策」。

使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」。
在 Apigee 代理程式編輯器中開啟代理程式，然後按一下「開發」分頁標籤。

使用下列定義建立新的 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>

(選用) 先將 JavaScript 政策新增至事件群組，再傳送至 Apigee Model Armor 政策。

<?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 代理程式編輯器中開啟代理程式，然後按一下「開發」分頁標籤。

建立新的 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 政策新增至目標端點的 FaultRules：

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

部署及測試 API Proxy。

在 Apigee Analytics 中查看 SSE 資料

SSE Proxy 的資料會如預期顯示在 Apigee Analytics 中，就像任何 API Proxy 一樣。前往 Cloud 控制台，依序點選「Analytics」>「API 指標」。

偵錯 SSE Proxy

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

疑難排解

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

限制

SSE Proxy 有下列限制：

由於系統會在 SSE 工作階段結束後記錄數據分析資料，因此您可能會發現數據分析資料的報表會延遲顯示。
EventFlow 內發生錯誤時，串流會立即結束，且不會向終端用戶端擲回任何特定錯誤事件。如要瞭解如何手動記錄這類錯誤，請參閱「EventFlow 用途和範例」。
接收串流 SSE 回應的用戶端會在事件串流的開頭收到 HTTP 標頭，包括任何狀態碼。因此，如果事件串流進入錯誤狀態，一開始收到的狀態碼不會反映錯誤狀態。
查看偵錯工作階段時，可能會看到這項限制。在工作階段中，您可能會發現進入錯誤狀態的串流 HTTP 狀態碼，與傳送給用戶端的狀態碼不同。這是因為系統是在處理完整要求後才產生偵錯工作階段項目，而不是在事件串流開始時產生。偵錯工作階段可能會反映錯誤產生的錯誤代碼，而用戶端最初只會在標頭中看到 2xx 狀態。

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

為 SSE 建立 API Proxy

設定 EventFlow

流程變數

EventFlow 用途與範例

修改 SSE 回應

顯示回覆範例

篩選 SSE 回應

顯示回覆範例

將 SSE 事件傳送至外部系統

在 EventFlow 中使用 Apigee Model Armor 政策

EventFlow 中的錯誤處理

在 Apigee Analytics 中查看 SSE 資料

偵錯 SSE Proxy

疑難排解

限制

串流伺服器傳送的事件