סטרימינג של אירועים שנשלחים מהשרת

הדף הזה רלוונטי ל-Apigee ול-Apigee Hybrid.

לעיון במסמכי התיעוד של Apigee Edge

‫Apigee תומך בסטרימינג רציף של תשובות מנקודות קצה של אירועים שנשלחים מהשרת (SSE) ללקוחות בזמן אמת. התכונה Apigee SSE שימושית לטיפול בממשקי API של מודלים גדולים של שפה (LLM) שפועלים בצורה הכי יעילה על ידי סטרימינג של התשובות שלהם בחזרה ללקוח. הזרמת SSE מפחיתה את זמן האחזור, והלקוחות יכולים לקבל נתוני תגובה ברגע שהם נוצרים על ידי מודל LLM. התכונה הזו תומכת בשימוש בסוכני AI שפועלים בסביבות בזמן אמת, כמו צ'אט בוטים לשירות לקוחות או כלי תזמור של תהליכי עבודה.

כדי להשתמש ב-SSE עם Apigee, פשוט מפנים proxy ל-API ליעד עם SSE או לנקודת קצה של שרת proxy. כדי להשיג שליטה מדויקת יותר בתגובת ה-SSE, ‏ Apigee מספק זרימת נקודת קצה מיוחדת שנקראת EventFlow. במסגרת EventFlow, אפשר להוסיף קבוצה מוגבלת של כללי מדיניות כדי לבצע פעולות בתגובת ה-SSE, כמו סינון, שינוי או טיפול בשגיאות. מידע נוסף על תהליכי proxy זמין במאמר שליטה ב-API proxy באמצעות תהליכים.

יצירת proxy ל-API ל-SSE

ממשק המשתמש של Apigee מספק תבנית ליצירת שרת proxy חדש שכולל EventFlow.

הערה: השימוש בתבנית של שרת proxy של SSE הוא אופציונלי. היא מאפשרת לכם לגשת למשתני זרימה ספציפיים ל-SSE, שיכולים להיות שימושיים לביצוע פעולות מסוימות במדיניות. עם זאת, לא צריך להשתמש ב-EventFlow כדי להשתמש ב-SSE עם Apigee.

חשוב: צריך להשתמש בהגדרות נפרדות של נקודות קצה לטירגוט עבור יעדי SSE. שילוב של נקודות קצה של יעדים עם SSE ונקודות קצה של יעדים ללא SSE עלול לגרום להתנהגות לא עקבית, כמו response.content משתני זרימה ריקים. מידע נוסף אפשר לקרוא בקטע בעיות מוכרות.

בצע את השלבים הבאים כדי ליצור proxy ל-API עם תבנית EventFlow באמצעות ממשק המשתמש של Apigee:

במסוף Google Cloud , נכנסים לדף Apigee > Proxy Development > API Proxies.

מעבר אל API Proxies
בחלונית API Proxies (שרתי proxy של API), לוחצים על + Create (יצירה).
בחלונית Create a proxy (יצירת שרת proxy), בקטע Proxy template (תבנית שרת proxy), בוחרים באפשרות Proxy with Server-Sent Events (SSE) (שרת proxy עם אירועים שנשלחים מהשרת (SSE)).
בקטע פרטי שרת proxy, מזינים את הפרטים הבאים:
- שם ה-proxy: מזינים שם ל-proxy, למשל myproxy.
- נתיב בסיס: מוגדר אוטומטית לערך שהזנתם עבור שם שרת ה-proxy. נתיב הבסיס הוא חלק מכתובת ה-URL שמשמשת לשליחת בקשות ל-API. ‫Apigee משתמש בכתובת ה-URL כדי להתאים בקשות נכנסות ולנתב אותן אל proxy ה-API המתאים.
- תיאור (אופציונלי): מזינים תיאור ל-proxy ל-API החדש, למשל "בדיקת Apigee באמצעות proxy פשוט".
- יעד (API קיים): מזינים את כתובת ה-URL של יעד ה-SSE עבור ה-proxy ל-API. לדוגמה: https://mocktarget.apigee.net/sse-events/5
- לוחצים על הבא.
פריסה (אופציונלי):
- סביבות פריסה: אופציונלי. משתמשים בתיבות הסימון כדי לבחור סביבה אחת או יותר שבהן רוצים לפרוס את ה-proxy. אם אתם לא רוצים לפרוס את ה-proxy בשלב הזה, אתם יכולים להשאיר את השדה סביבות פריסה ריק. תמיד אפשר לפרוס את ה-proxy מאוחר יותר.
- חשבון שירות: אופציונלי. חשבון שירות לשרת ה-proxy. חשבון השירות מייצג את הזהות של ה-proxy שנפרס, וקובע אילו הרשאות יש לו. זו תכונה מתקדמת, ולצורך המדריך הזה אפשר להתעלם ממנה.
שרתי proxy ל-API שנפרסו עם הגדרת EventFlow יחויבו כשרתי proxy ניתנים להרחבה.
לוחצים על יצירה.

אפשר גם לעיין במאמר יצירת שרת proxy פשוט ל-API.

הגדרת 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 stanzas שמופיעים לפני האחרון.

חשוב גם לציין שאף על פי שאפשר להוסיף EventFlow ל-TargetEndpoint, ל-ProxyEndpoint או לשניהם, רק EventFlow אחד מבוצע.

בטבלה הבאה מוצגות הפסקאות של EventFlow על סמך מיקום נקודת הקצה:

ProxyEndpoint	TargetEndpoint	השימוש ב-EventFlow
‫EventFlow ב-`ProxyEndpoint`	‫EventFlow ב-`TargetEndpoint`	‫EventFlow ב-`TargetEndpoint`
לא `EventFlow`	‫EventFlow ב-`TargetEndpoint`	‫EventFlow ב-`TargetEndpoint`
‫EventFlow ב-`ProxyEndpoint`	לא `EventFlow`	‫EventFlow ב-`ProxyEndpoint`

הוספת מדיניות ל-EventFlow

אפשר להוסיף עד ארבע מדיניות לרכיב Response של EventFlow. כמו בכל התהליכים, כללי המדיניות מופעלים לפי הסדר שבו הם נוספו, ואפשר להוסיף שלבים מותנים כדי לשלוט בהפעלה שלהם. חשוב לציין שאפשר להוסיף ל-EventFlow רק את סוגי המדיניות הבאים. אסור להשתמש בסוגים אחרים של מדיניות ב-EventFlow:

כשמשתמשים ב-LLMTokenQuota עם זרמי SSE, המערכת לא אוכפת את המכסה על אירועים שחסרים בהם מטא-נתונים של שימוש באסימון. המיכסות מחושבות רק על סמך אירועים שמכילים ספירות טוקנים מפורשות.

אפשר לעיין גם במאמרים צירוף מדיניות והגדרת מדיניות בממשק המשתמש וצירוף מדיניות והגדרת מדיניות בקובצי XML.

בדוגמאות הבאות מוצג EventFlow עם שלב מדיניות RaiseFault מותנה שנוסף:

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

משתני Flow

הפעולה EventFlow מאכלסת שלושה משתני זרימת תשובות. שימו לב שאפשר להשתמש במשתנים האלה רק בהקשר של האירוע הנוכחי שעובר עיבוד ב-EventFlow. לגישה למשתנים האלה או להגדרתם מחוץ להיקף EventFlow אין השפעה. הם רלוונטיים רק בהקשר של EventFlow.

‫response.event.current.content: מחרוזת שמכילה את התגובה המלאה של האירוע הנוכחי. מערכת Apigee לא מנתחת את המחרוזת בשום צורה. הוא מכיל את כל התשובה ללא שינוי, כולל כל שדות הנתונים.
הערה: ב-Apigee, כל אירוע תגובה מוגבל לנתונים בגודל של עד 10MB.

הערה: אם מגדירים את הערך של המשתנה הזה למחרוזת ריקה (""), האירוע לא יישלח ללקוח. לדוגמה, אפשר לעיין במאמר בנושא סינון תגובה של SSE.
‫response.event.current.data: מחרוזת שמכילה את מטען הנתונים של האירוע הנוכחי. אפשר לשנות את המשתנה הזה ב-EventFlow כדי לשנות את מטען ייעודי (payload) הנתונים שנשלח ללקוח.
‫response.event.current.count: ספירה מצטברת של מספר אירועי התגובה שנשלחו. הערך הזה מתעדכן לכל אירוע שמתקבל. הערך יהיה 1 לאירוע הראשון, והוא יגדל לאירועים הבאים.
הערה: אם ניגשים לערך הזה, למשל במדיניות JavaScript, הערך של האירוע הראשון יהיה 1, אבל מספר האירועים שנשלחים ללקוח בשלב הזה יהיה 0.

אפשר גם לעיין במאמר בנושא הפניית משתני זרימה.

תרחישי שימוש ודוגמאות ל-EventFlow

בדוגמאות הבאות מוצגות דרכים להטמיע תרחישי שימוש נפוצים בשרתי proxy של SSE:

שינוי של תשובה בפורמט SSE
סינון תגובת SSE
שליחת אירוע SSE למערכת חיצונית
שימוש במדיניות הגנה מוגברת על המודל של Apigee ב-EventFlows
טיפול בשגיאות ב-EventFlow
הפצת הודעות על תקלות ב-EventFlow

שינוי תשובה של SSE

בדוגמה הזו מוצג איך להסיר נתונים מתגובת EventFlow של SSE לפני שמחזירים אותה ללקוח. התוכן של תגובת ה-SSE מאוחסן במשתנה של תהליך בשם response.event.current.content. במקרה הזה, אנחנו משתמשים במדיניות JavaScript כדי לאחזר את הערך של משתנה הזרימה, לנתח אותו ולשנות אותו. מידע נוסף זמין במאמר בנושא משתני Flow.

יוצרים פרוקסי חדש באמצעות תבנית הפרוקסי של SSE. מידע נוסף זמין במאמר יצירת proxy ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה 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 לתג EventFlow של ה-proxy. ה-EventFlow מצורף ל-TargetEndpoint או ל-ProxyEndpoint שמוגדרים כברירת מחדל. בדוגמה הזו נעשה שימוש ב-Gemini API ב-Vertex AI כדי ליצור תוכן.

<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 ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה Develop (פיתוח).

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

<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 למערכת חיצונית

בדוגמה הזו, אנחנו מצרפים את מדיניות PublishMessage של Apigee אל EventFlow כדי לשלוח אירוע SSE אל נושא ב-Pub/Sub.

יוצרים פרוקסי חדש באמצעות תבנית הפרוקסי של SSE. מידע נוסף זמין במאמר יצירת proxy ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה 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 כשלב ב-EventFlow של proxy ל-API.

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

פריסה ובדיקה של ה-proxy ל-API.
אחרי שמוסיפים את התוכן שנוצר לנושא Pub/Sub, אפשר למשל ליצור פונקציית Cloud Run לעיבוד הודעות מהנושא.

שימוש במדיניות Apigee הגנה מוגברת על המודל ב-EventFlow

אפשר להשתמש במדיניות SanitizeModelResponse כדי לבצע סניטציה של אירועים נכנסים שנשלחים מהשרת ב-EventFlow. המדיניות הזו מגנה על אפליקציות ה-AI שלכם על ידי ניקוי התשובות ממודלים גדולים של שפה (LLM). מידע על הגנה מוגברת על המודל זמין במאמר סקירה כללית על הגנה מוגברת על המודל. מידע על מדיניות Apigee Model Armor זמין במאמר תחילת העבודה עם מדיניות Apigee Model Armor.

יוצרים פרוקסי חדש באמצעות תבנית הפרוקסי של SSE. מידע נוסף זמין במאמר יצירת proxy ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה 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>

(אופציונלי) מוסיפים מדיניות JavaScript לאירועים של קבוצה לפני ששולחים אותם למדיניות Apigee הגנה מוגברת על המודל.

<?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 לשלב בEventFlow של ה-proxy:

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

פריסה ובדיקה של ה-proxy ל-API.

טיפול בשגיאות ב-EventFlow

כברירת מחדל, זרם האירועים מסתיים כשמתרחשת תקלה. עם זאת, אם רוצים לבצע ניפוי באגים נוסף, אפשר לשלוח מידע על תקלות אל Cloud Logging כמו בדוגמה הזו.

יוצרים פרוקסי חדש באמצעות תבנית הפרוקסי של SSE. מידע נוסף זמין במאמר יצירת proxy ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה 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 ל-EventFlow של שרת ה-proxy של SSE:

<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 של נקודת הקצה של היעד או של ה-proxy:

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

פריסה ובדיקה של ה-proxy ל-API.

הפצת שגיאות בתהליך EventFlow

בדוגמה הזו אנחנו מראים איך להשתמש ב-EventFlow כדי להעביר שגיאות של תקלות ללקוח. התהליך נועד להודיע ללקוח על שגיאות באופן מיידי במהלך ביצוע המדיניות.

יוצרים פרוקסי חדש באמצעות תבנית הפרוקסי של SSE. מידע נוסף זמין במאמר יצירת proxy ל-API עם אירועים שנשלחים מהשרת (SSE).
פותחים את ה-proxy בכלי לעריכת proxy של Apigee ולוחצים על הכרטיסייה 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>

המדיניות הזו נועדה להקפיץ הודעת שגיאה (throw) כשמתקיים תנאי מסוים.

מצרפים את מדיניות JavaScript ל-EventFlow של שרת ה-proxy של SSE בהגדרות של TargetEndpoint או ProxyEndpoint. בשלב הזה מוודאים ש-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>

פורסים את ה-proxy ל-API.

כדי לבדוק את התנהגות ה-proxy, משתמשים בפקודה 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 זמין במאמר בנושא טיפול בשגיאות.

צפייה בנתוני SSE בניתוח הנתונים של Apigee

הנתונים של שרתי proxy של SSE מופיעים בניתוח הנתונים של Apigee כמו שקורה בדרך כלל עם proxy ל-API. במסוף Cloud, עוברים אל Analytics > API metrics.

ניפוי באגים בשרתי proxy של SSE

אפשר להשתמש בכלי לניפוי באגים ב-Apigee כדי לנפות באגים בשרתי proxy של SSE. נתוני ניפוי הבאגים נאספים ב-EventFlow בדיוק כמו בסוגי התהליכים האחרים.

פתרון בעיות

כדי לפתור בעיות שקשורות לתנועה בזמן אמת, בודקים את יומני הגישה של Apigee כדי לזהות את הגורם לבעיה.

מגבלות

המגבלות הבאות חלות על שרתי proxy של SSE:

אפשר להשאיר חיבור SSE פתוח ללא הגבלת זמן, אבל משך החיבור מוגבל בסופו של דבר על ידי הגדרת הזמן הקצוב לתפוגה של מאזן העומסים במורד הזרם של Apigee. כברירת מחדל, הגדרת הזמן הקצוב לתפוגה של מאזן העומסים של Apigee היא 30 שניות. לחיבורים ארוכים יותר, מומלץ להגדיל את הזמן הקצוב לתפוגה או ליצור שירות קצה עורפי נפרד עם ערך גבוה יותר של זמן קצוב לתפוגה כדי לטפל בתנועת הנתונים של SSE.
נתוני Analytics מתועדים אחרי סגירת סשן ה-SSE, ולכן יכול להיות שתבחינו בעיכוב בדיווח של נתוני Analytics.
תקלות בתוך EventFlow גורמות ליציאה מיידית מהזרם ולזריקת שגיאה. מידע על רישום ידני של סוגי השגיאות האלה או על שליחתן ללקוח זמין במאמר תרחישים לדוגמה לשימוש ב-EventFlow.
לקוח שמקבל תגובות SSE בסטרימינג יקבל את הכותרות HTTP, כולל קודי סטטוס, בתחילת סטרימינג האירועים. כתוצאה מכך, אם זרם האירועים נכנס למצב שגיאה, קוד הסטטוס שמתקבל בהתחלה לא ישקף את מצב השגיאה.
אפשר לראות את ההגבלה הזו כשצופים בסשן ניפוי באגים. במהלך הסשן, יכול להיות שתבחינו שקוד הסטטוס HTTP של הסטרימינג שנכנס למצב שגיאה שונה מקודי הסטטוס שנשלחים ללקוח. זה יכול לקרות כי רשומת סשן הניפוי באגים נוצרת אחרי שכל הבקשה עברה עיבוד, ולא בתחילת זרם האירועים. יכול להיות שבסשן הניפוי באגים יוצג קוד השגיאה שנוצר על ידי השגיאה, בזמן שהלקוח רואה בהתחלה רק את הסטטוס 2xx שהתקבל בכותרות.

סטרימינג של אירועים שנשלחים מהשרת קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

יצירת proxy ל-API ל-SSE

הגדרת EventFlow

מיקום של EventFlow

<ProxyEndpoint>

<TargetEndpoint>

הוספת מדיניות ל-EventFlow

<ProxyEndpoint>

<TargetEndpoint>

משתני Flow

תרחישי שימוש ודוגמאות ל-EventFlow

שינוי תשובה של SSE

<ProxyEndpoint>

<TargetEndpoint>

הצגת דוגמה לתשובה

סינון של תגובת SSE

<ProxyEndpoint>

<TargetEndpoint>

הצגת דוגמה לתשובה

שליחת אירוע SSE למערכת חיצונית

<ProxyEndpoint>

<TargetEndpoint>

שימוש במדיניות Apigee הגנה מוגברת על המודל ב-EventFlow

טיפול בשגיאות ב-EventFlow

<TargetEndpoint>

<ProxyEndpoint>

הפצת שגיאות בתהליך EventFlow

<TargetEndpoint>

<ProxyEndpoint>

צפייה בנתוני SSE בניתוח הנתונים של Apigee

ניפוי באגים בשרתי proxy של SSE

פתרון בעיות

מגבלות

סטרימינג של אירועים שנשלחים מהשרת