串流伺服器傳送的事件

本頁適用於 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 UI 提供範本，可用來建立包含 EventFlow 的新 Proxy。

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

  1. 在瀏覽器中開啟 Cloud 控制台中的 Apigee UI。
  2. 在左側導覽窗格中，依序按一下「Proxy development」>「API proxies」。
  3. 在「API Proxy」窗格中，按一下「+ 建立」。
  4. 在「Create a proxy」窗格中的「Proxy template」下方，選取「Proxy with Server-Sent Events (SSE)」。
  5. 在「Proxy details」下方輸入以下內容：
    • Proxy name：輸入 Proxy 的名稱，例如 myproxy。
    • Base Path：自動設為您為 Proxy name 輸入的值。「Base Path」是用於向 API 提出要求的網址的一部分。Apigee 會使用網址比對並將傳入的要求轉送至適當的 API Proxy。
    • 說明 (選填)：輸入新 API Proxy 的說明，例如「使用簡易 Proxy 測試 Apigee」。
    • 目標 (現有 API)：輸入 API 代理程式的 SSE 目標網址。例如：https://mocktarget.apigee.net/sse-events/5
    • 點選「下一步」。
  6. 部署 (選用)：
    • 部署環境：選用。使用核取方塊選取一或多個要部署 Proxy 的環境。如果您不想在此時部署 Proxy，請將「部署環境」欄位留空。您之後隨時可以部署 Proxy。
  • 服務帳戶：選填。用於 Proxy 的服務帳戶。服務帳戶代表部署的 Proxy 身分，並決定其擁有的權限。這是進階功能，您可以忽略本教學課程中的這項功能。

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

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

    設定 EventFlow

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

    EventFlow 必須放在 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://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 不會以任何方式剖析字串。它包含整個回應，包括所有資料欄，且不會有所變動。
    • response.event.current.count：逐漸計算傳送的回應事件數量。系統會針對每個收到的事件更新這個值。第一個事件的計數為 1，後續事件的計數則會遞增。

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

    EventFlow 用途和範例

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

    修改 SSE 回應

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

    1. 使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」一文。
    2. 在 Apigee Proxy 編輯器中開啟 Proxy，然後按一下「Develop」分頁標籤。
    3. 使用以下定義建立新的 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>
    4. 將 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-1.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse</URL>
  </HTTPTargetConnection>
</TargetEndpoint>
    5. 儲存 Proxy 並部署。
    6. 呼叫已部署的 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-1.5-flash" 屬性。

      data: {
    "candidates": [
      {
        "content": {
          "parts": [
            {
              "text": "ara found the pen tucked away in a dusty antique shop, nestled amongst chipped tea"
            }
          ],
          "role": "model"
        }
      }
    ],
    "usageMetadata": {
      "promptTokenCount": 8,
      "totalTokenCount": 8
    },
    "modelVersion": "gemini-1.5-flash"
  }

      這是另一個已套用 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，然後套用邏輯來實作所需的篩選。

    1. 使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」一文。
    2. 在 Apigee Proxy 編輯器中開啟 Proxy，然後按一下「Develop」分頁標籤。
    3. 使用以下定義建立新的 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>
    4. 將 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-1.5-flash:streamGenerateContent?key=GEMINI_API_KEY&alt=sse	</URL>
  </HTTPTargetConnection>
</TargetEndpoint>
    5. 儲存 Proxy 並部署。
    6. 呼叫已部署的 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-1.5-flash"
  }
    data: {
    "error": "Service temporarily unavailable. We are experiencing high traffic.",
    "modelVersion": "gemini-1.5-flash"
    }

      以下是套用篩選條件後的另一個範例回應，其中已移除錯誤訊息。

      data: {
  "candidates": [
    {
      "content": {
        "parts": [
          {
            "text": "El"
          }
        ],
        "role": "model"
      }
    }
  ],
  "usageMetadata": {
    "promptTokenCount": 8,
    "totalTokenCount": 8
  },
  "modelVersion": "gemini-1.5-flash"
}
data: {
  "candidates": [
    {
      "content": {
        "parts": [
          {
            "text": "ara found the pen tucked away in a dusty antique shop, nestled amongst chipped tea"
          }
        ],
        "role": "model"
      }
    }
  ],
  "usageMetadata": {
    "promptTokenCount": 8,
    "totalTokenCount": 8
  },
  "modelVersion": "gemini-1.5-flash"
}

    將 SSE 事件傳送至外部系統

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

    1. 使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」一文。
    2. 在 Apigee Proxy 編輯器中開啟 Proxy，然後按一下「Develop」分頁標籤。
    3. 使用下列定義建立新的 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>
    4. 將 PublishMessage 政策新增為 API Proxy 的 EventFlow 中的步驟。
      <TargetEndpoint name="default">
  <EventFlow content-type="text/event-stream">
    <Response>
      <Step>
        <Name>PM-record-event</Name>
      </Step>
    </Response>
  </EventFlow>
  <HTTPTargetConnection>
</TargetEndpoint>
    5. 部署及測試 API Proxy。
    6. 將產生的內容新增至 Pub/Sub 主題後，您可以建立 Cloud Run 函式，以便處理主題中的訊息。

    在 EventFlow 中使用 Apigee Model Armor 政策

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

    1. 使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」一文。
    2. 在 Apigee Proxy 編輯器中開啟 Proxy，然後按一下「Develop」分頁標籤。
    3. 使用以下定義建立新的 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>
    4. (選用) 將 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>
    5. 將 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>
    6. 部署及測試 API Proxy。

    EventFlow 中的錯誤處理

    根據預設，事件串流會在發生錯誤時結束。不過，如果您想進行額外的偵錯作業，可以將錯誤資訊傳送至 Cloud Logging，如本範例所示。

    1. 使用 SSE Proxy 範本建立新的 Proxy。請參閱「使用伺服器傳送事件 (SSE) 建立 API Proxy」一文。
    2. 在 Apigee Proxy 編輯器中開啟 Proxy，然後按一下「Develop」分頁標籤。
    3. 使用以下定義建立新的 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>
    4. 將 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>
    5. 建立「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>
    6. 將 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>
    7. 部署及測試 API Proxy。

      8. 在 Apigee Analytics 中查看 SSE 資料

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

      對 SSE Proxy 進行偵錯

      使用 Apigee 偵錯工具偵錯 SSE 代理程式。EventFlow 的偵錯資料會與其他流程類型一樣擷取。

      疑難排解

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

      限制

      以下限制適用於 SSE 代理程式：

      • 由於數據分析資料是在 SSE 工作階段關閉後才記錄，因此您可能會發現報表中的數據分析資料有延遲。
      • EventFlow 中的錯誤會導致串流立即結束，且不會向終端用戶端擲回特定錯誤事件。如要瞭解如何手動記錄這類錯誤，請參閱「EventFlow 用途和範例」。
      • 接收串流 SSE 回應的用戶端會在事件串流開始時收到 HTTP 標頭，包括任何狀態碼。因此，如果事件串流進入錯誤狀態，最初收到的狀態碼不會反映錯誤狀態。

        查看偵錯工作階段時，您可以看到這項限制。在工作階段中，您可能會注意到，進入錯誤狀態的串流的 HTTP 狀態碼與傳送至用戶端的狀態碼不同。這可能是因為偵錯工作階段項目是在處理整個要求後產生，而不是在事件串流開始時產生。偵錯工作階段可能會反映錯誤產生的錯誤代碼，而用戶端只會看到標頭中最初收到的 2xx 狀態。

    •