流式传输服务器发送的事件

本页面适用于 Apigee 和 Apigee Hybrid。

查看 Apigee Edge 文档。

Apigee 支持从服务器发送的事件 (SSE) 端点向客户端实时进行连续响应流式传输。Apigee SSE 功能非常适合处理通过将响应流式传输回客户端来以最高效方式运行的大语言模型 (LLM) API。SSE 流式传输可缩短延迟时间，并且客户端可以在 LLM 生成响应数据后立即接收。此功能支持使用在实时环境中运行的 AI 代理，例如客户服务机器人或工作流编排程序。

注意：Apigee 和 Apigee Hybrid v1.15.0 及更高版本支持从 SSE 端点进行流式传输。该功能还支持与 Apigee Extension Processor 搭配使用。

如需将 SSE 与 Apigee 搭配使用，只需将 API 代理指向启用了 SSE 的目标端点或代理端点即可。为了更精细地控制 SSE 响应，Apigee 提供了一个名为 EventFlow 的特殊端点流。在 EventFlow 的上下文中，您可以添加一组有限的政策来对 SSE 响应执行操作，例如过滤、修改或处理错误。如需详细了解代理流，请参阅使用流控制 API 代理。

注意：Apigee 将每个响应事件的数据量限制为最多 10 MB。

为 SSE 创建 API 代理

Apigee 界面提供了一个用于创建新代理的模板，其中包含 EventFlow。

注意：使用 SSE 代理模板是可选的。它使您能够访问 SSE 特有的流变量，这些变量对于执行某些政策操作非常有用。不过，您无需使用 EventFlow 即可将 SSE 与 Apigee 搭配使用。

重要提示：请为 SSE 目标使用单独的目标端点定义。 将 SSE 和非 SSE 目标端点混用可能会导致行为不一致，例如 response.content 流变量为空。*查看已知问题

如需使用 Apigee 界面通过 EventFlow 模板创建 API 代理，请按照以下步骤操作：

1. 在 Google Cloud 控制台中，前往代理开发 > API 代理页面。

   前往 API 代理

2. 在 API 代理窗格，点击 + 创建。
3. 在创建代理窗格的代理模板下，选择包含服务器发送的事件 (SSE) 的代理。
4. 在代理详情下，输入以下内容：
   • 代理名称：输入代理的名称，例如 myproxy。
   • 基本路径：自动设置为您为代理名称输入的值。基本路径是用于向 API 发出请求的网址的一部分。Apigee 使用网址来匹配传入的请求并将其路由到相应的 API 代理。
   • 说明（可选）：输入新 API 代理的说明，例如“使用简单代理测试 Apigee”。
   • 目标（现有 API）：输入 API 代理的 SSE 目标网址。例如 https://mocktarget.apigee.net/sse-events/5
   • 点击下一步。
5. 部署（可选）：
   • 部署环境：可选。使用复选框选择要将代理部署到的一个或多个环境。如果您不想在此时部署代理，请将部署环境字段留空。您可以稍后部署该代理。
     注意：对于随用随付环境，唯一支持使用 SSE 的环境类型是全面。请参阅环境类型。
   • 服务账号：可选。代理的服务账号。服务账号代表已部署代理的身份，并决定了该代理拥有的权限。这是一项高级功能，在本教程中，您可以忽略它。

   使用 EventFlow 配置部署的 API 代理将按可扩展型计费。

6. 点击创建。

另请参阅构建简单的 API 代理。

配置 EventFlow

为了更精细地控制 SSE 响应，Apigee 提供了一个名为 EventFlow 的特殊端点流。在 EventFlow 的上下文中，您可以添加一组有限的政策，以在将 SSE 响应流式传输回客户端之前对其进行修改。如需详细了解代理流，请参阅使用流控制 API 代理。

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 节之前的所有节。

另请务必注意，虽然您可以向 TargetEndpoint、ProxyEndpoint 或两者都添加 EventFlow，但系统只会执行一个 EventFlow。

下表展示了基于端点放置位置的 EventFlow 节执行情况：

ProxyEndpoint	TargetEndpoint	使用的 EventFlow
ProxyEndpoint 中的 EventFlow	TargetEndpoint 中的 EventFlow	TargetEndpoint 中的 EventFlow
否 EventFlow	TargetEndpoint 中的 EventFlow	TargetEndpoint 中的 EventFlow
ProxyEndpoint 中的 EventFlow	否 EventFlow	ProxyEndpoint 中的 EventFlow

向 EventFlow 添加政策

您最多可以向 EventFlow 的 Response 元素添加 4 个政策。与所有流一样，政策的执行顺序与添加顺序一致，您可以添加条件步骤来控制政策的执行。请务必注意，您可以添加到 EventFlow 的政策类型仅限于以下几种。EventFlow 中不允许使用其他类型的政策：

• Javascript
• MessageLogging
• PublishMessage
• RaiseFault
• SanitizeModelResponse

另请参阅在界面中附加和配置政策以及在 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.count：以增量方式统计发送的响应事件数量。 此值会针对收到的每个事件进行更新。 第一个事件的计数为 1，后续事件的计数会递增。
  注意：如果您在 JavaScript 政策中访问此值，例如，第一个事件的值将为 1；不过，此时发送到客户端的事件数量将为 0。

另请参阅流变量参考文档。

EventFlow 应用场景和示例

以下示例展示了如何实现 SSE 代理的常见应用场景：

• 修改 SSE 响应
• 过滤 SSE 响应
• 向外部系统发送 SSE 事件
• 在 EventFlow 中使用 Apigee Model Armor 政策
• 在 EventFlow 中进行错误处理
• 在 EventFlow 中传播故障消息

修改 SSE 响应

以下示例展示了如何在将 SSE EventFlow 响应返回客户端之前从中移除数据。SSE 响应的内容存储在名为 response.event.current.content 的流变量中。在本例中，我们使用 JavaScript 政策来检索流变量的值，并对其进行解析和修改。另请参阅流变量。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
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 政策添加到代理的 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>
   

5. 保存并部署代理。
6. 调用已部署的代理：

   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，然后应用逻辑来实现预期的过滤。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
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 政策添加到代理的 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>
   

5. 保存并部署代理。
6. 调用已部署的代理：

   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 主题。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
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 代理的 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>

5. 部署并测试 API 代理。
6. 将生成的内容添加到 Pub/Sub 主题后，您可以执行创建 Cloud Run 函数来处理主题中的消息等操作。

在 EventFlow 中使用 Apigee Model Armor 政策

注意：Apigee Hybrid 不支持 Model Armor 政策。

您可以使用 SanitizeModelResponse 政策来清理 EventFlow 中传入的服务器发送的事件。此政策通过清理大语言模型 (LLM) 的响应来保护您的 AI 应用。如需了解 Model Armor，请参阅 Model Armor 概览。如需了解 Apigee Model Armor 政策，请参阅 Apigee Model Armor 政策使用入门。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
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 政策添加到代理的 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 代理。

在 EventFlow 中进行错误处理

默认情况下，当发生故障时，事件流会结束。不过，如果您想进行额外的调试，可以按照此示例所示将故障信息发送到 Cloud Logging。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
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 代理的 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 政策添加到目标端点或代理端点的 FaultRule 中：

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

7. 部署并测试 API 代理。

在 EventFlow 中传播故障错误

在以下示例中，我们展示了如何使用 EventFlow 将故障错误传播到客户端。该流程旨在在执行政策期间立即将错误通知客户端。

1. 使用 SSE 代理模板创建新代理。请参阅创建包含服务器发送的事件 (SSE) 的 API 代理。
2. 在 Apigee 代理编辑器中打开代理，然后点击开发标签页。
3. 创建具有以下定义的新 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>

   此政策旨在在满足特定条件时抛出错误。

4. 将 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>

5. 部署 API 代理。
6. 使用以下 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."}]}]}'

   将 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 中的故障处理，请参阅处理故障。

在 Apigee 分析中查看 SSE 数据

SSE 代理的数据会像任何 API 代理一样，按预期显示在 Apigee 分析中。在 Cloud 控制台中，前往分析 > API 指标。

调试 SSE 代理

使用 Apigee 调试工具调试 SSE 代理。系统会捕获 EventFlow 的调试数据，就像捕获其他流类型的调试数据一样。

问题排查

对于实时流量问题，请查看 Apigee 访问日志以确定原因。

限制

SSE 代理存在以下限制：

• 由于分析数据是在 SSE 会话关闭后记录的，因此您可能会注意到分析数据报告存在一些延迟。
• EventFlow 内部的故障会导致流立即退出并抛出错误。如需了解如何手动记录这些类型的错误或将其发送到客户端，请参阅 EventFlow 应用场景和示例。
• 接收流式 SSE 响应的客户端会在事件流开始时收到 HTTP 标头（包括任何状态代码）。因此，如果事件流进入错误状态，最初收到的状态代码将无法反映该错误状态。

  查看调试会话时，您可以看到此限制。在此会话中，您可能会注意到，进入错误状态的数据流的 HTTP 状态代码与发送到客户端的状态代码不同。之所以会发生这种情况，是因为调试会话条目是在处理整个请求后生成的，而不是在事件流开始时生成的。调试会话可能会反映由错误生成的故障代码，而客户端只会看到标头中最初收到的 2xx 状态。