이 페이지는 Apigee 및 Apigee Hybrid에 적용됩니다.
Apigee Edge 문서 보기
Apigee는 서버 전송 이벤트 (SSE) 엔드포인트에서 클라이언트로의 연속 응답 스트리밍을 실시간으로 지원합니다. Apigee SSE 기능은 클라이언트에 응답을 스트리밍하여 가장 효과적으로 작동하는 대규모 언어 모델 (LLM) API를 처리하는 데 유용합니다. SSE 스트리밍은 지연 시간을 줄여주며 클라이언트는 LLM에서 생성되는 즉시 응답 데이터를 수신할 수 있습니다. 이 기능은 고객 서비스 봇이나 워크플로 오케스트레이터와 같은 실시간 환경에서 작동하는 AI 에이전트의 사용을 지원합니다.
Apigee에서 SSE를 사용하려면 API 프록시를 SSE 지원 대상 엔드포인트로 지정하면 됩니다. SSE 응답을 더 세부적으로 제어하기 위해 Apigee는 EventFlow
라는 특수 대상 엔드포인트 흐름을 제공합니다. EventFlow
컨텍스트 내에서 제한된 정책 집합을 추가하여 필터링, 수정, 오류 처리와 같은 SSE 응답 작업을 실행할 수 있습니다. 프록시 흐름에 대해 자세히 알아보려면 흐름을 통한 API 프록시 제어를 참고하세요.
SSE용 API 프록시 만들기
Apigee UI는 EventFlow
가 포함된 새 프록시를 만들기 위한 템플릿을 제공합니다.
다음 단계에 따라 Apigee UI를 사용하여 EventFlow
템플릿으로 API 프록시를 만듭니다.
- 브라우저에서 Cloud 콘솔의 Apigee UI를 엽니다.
- 왼쪽 탐색창에서 프록시 개발 > API 프록시를 클릭합니다.
- API 프록시 창에서 + 만들기를 클릭합니다.
- 프록시 만들기 창의 프록시 템플릿에서 서버 전송 이벤트 (SSE)가 있는 프록시를 선택합니다.
- 프록시 세부정보 아래에서 다음을 입력합니다.
- 프록시 이름:
myproxy
와 같은 프록시 이름을 입력합니다. - 기본 경로:
Proxy name
에 입력하는 값으로 자동 설정됩니다. 기본 경로는 API에 요청을 수행하는 데 사용되는 URL의 일부입니다. Apigee는 URL을 사용하여 수신 요청을 적절한 API 프록시로 일치시키며 라우팅합니다. - 설명(선택사항): '간단한 프록시로 Apigee 테스트'와 같이 새 API 프록시에 설명을 입력합니다.
- 대상 (기존 API): API 프록시의 SSE 대상 URL을 입력합니다. 예를 들면
https://mocktarget.apigee.net/sse-events/5
입니다. - 다음을 클릭합니다.
- 프록시 이름:
- 배포(선택사항):
- 배포 환경: 선택사항. 체크박스를 사용하여 프록시를 배포할 환경을 하나 이상 선택합니다. 이 시점에서 프록시를 배포하지 않으려면 배포 환경 필드를 비워 둡니다. 나중에 언제든지 프록시를 배포할 수 있습니다.
EventFlow
구성으로 배포된 API 프록시는 확장 가능으로 청구됩니다.
EventFlow 구성
SSE 응답을 더 세부적으로 제어하기 위해 Apigee는 EventFlow
라는 특수 대상 엔드포인트 흐름을 제공합니다. EventFlow
컨텍스트 내에서 아래에 나열된 제한된 정책 집합을 추가하여 클라이언트로 다시 스트리밍되기 전에 SSE 응답을 수정할 수 있습니다. 프록시 흐름에 대해 자세히 알아보려면 흐름을 통한 API 프록시 제어를 참고하세요.
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
요소에 정책을 최대 4개까지 추가할 수 있습니다. 모든 흐름과 마찬가지로 정책은 추가된 순서대로 실행되며 조건부 단계를 추가하여 실행을 제어할 수 있습니다.
EventFlow
에 추가할 수 있는 정책 유형은 다음으로 제한됩니다.
EventFlow
에는 다른 유형의 정책이 허용되지 않습니다.
UI에서 정책 연결 및 구성 및 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 정책을 사용하여 흐름 변수의 값을 검색하고 파싱하고 수정합니다. 흐름 변수도 참고하세요.
- SSE 프록시 템플릿으로 새 프록시를 만듭니다. 서버 전송 이벤트 (SSE)를 사용하여 API 프록시 만들기를 참고하세요.
- 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>
- 프록시의
EventFlow
에 JavaScript 정책을 추가합니다.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>
- 프록시를 저장하고 배포합니다.
- 배포된 프록시를 호출합니다.
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 프록시 템플릿으로 새 프록시를 만듭니다. 서버 전송 이벤트 (SSE)를 사용하여 API 프록시 만들기를 참고하세요.
- 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>
- 프록시의
EventFlow
에 JavaScript 정책을 추가합니다.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>
- 프록시를 저장하고 배포합니다.
- 배포된 프록시를 호출합니다.
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
에 연결하여 Pub/Sub 주제에 SSE 이벤트를 전송합니다.
- SSE 프록시 템플릿으로 새 프록시를 만듭니다. 서버 전송 이벤트 (SSE)를 사용하여 API 프록시 만들기를 참고하세요.
- 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 프록시의
EventFlow
에 PublishMessage 정책을 단계로 추가합니다.<TargetEndpoint name="default"> <EventFlow content-type="text/event-stream"> <Response> <Step> <Name>PM-record-event</Name> </Step> </Response> </EventFlow> <HTTPTargetConnection> </TargetEndpoint>
- API 프록시를 배포하고 테스트합니다.
- 생성된 콘텐츠를 Pub/Sub 주제에 추가하면 예를 들어 주제의 메시지를 처리하는 Cloud Run 함수를 만들 수 있습니다.
EventFlow에서 Apigee Model Armor 정책 사용
SanitizeModelResponse 정책을 사용하여 EventFlow
에서 수신되는 서버 전송 이벤트를 정리할 수 있습니다.
이 정책은 대규모 언어 모델 (LLM)의 대답을 정리하여 AI 애플리케이션을 보호합니다. Model Armor에 대한 자세한 내용은 Model Armor 개요를 참고하세요. Apigee Model Armor 정책에 대한 자세한 내용은 Apigee Model Armor 정책 시작하기를 참고하세요.
- SSE 프록시 템플릿으로 새 프록시를 만듭니다. 서버 전송 이벤트 (SSE)를 사용하여 API 프록시 만들기를 참고하세요.
- 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>
- (선택사항) 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>
- 프록시의
EventFlow
단계에 JavaScript 및 ModelArmor 정책을 추가합니다.<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 프록시를 배포하고 테스트합니다.
EventFlow의 오류 처리
기본적으로 오류가 발생하면 이벤트 스트림이 종료됩니다. 하지만 추가 디버깅을 하려면 이 예와 같이 오류 정보를 Cloud Logging에 전송하면 됩니다.
- SSE 프록시 템플릿으로 새 프록시를 만듭니다. 서버 전송 이벤트 (SSE)를 사용하여 API 프록시 만들기를 참고하세요.
- 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>
- SSE 프록시의
EventFlow
에 RaiseFault 정책을 연결합니다.<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>
- 타겟 엔드포인트의 FaultRules에 MessageLogging 정책을 추가합니다.
<?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 프록시를 배포하고 테스트합니다.
- 분석 데이터는 SSE 세션이 종료된 후에 기록되므로 분석 데이터 보고에 약간의 지연이 발생할 수 있습니다.
- EventFlow 내부의 오류로 인해 스트림이 즉시 종료되며 특정 오류 이벤트가 최종 클라이언트에 발생하지 않습니다. 이러한 종류의 오류를 수동으로 로깅하는 방법에 대한 자세한 내용은 EventFlow 사용 사례 및 예시를 참고하세요.
- 스트리밍된 SSE 응답을 수신하는 클라이언트는 이벤트 스트림 시작 시 상태 코드를 포함한
HTTP
헤더를 수신합니다. 따라서 이벤트 스트림이 오류 상태가 되면 처음에 수신된 상태 코드가 오류 상태를 반영하지 않습니다.이 제한은 디버그 세션을 볼 때 확인할 수 있습니다. 세션에서 오류 상태가 된 스트림의
HTTP
상태 코드가 클라이언트에 전송된 상태 코드와 다를 수 있습니다. 이는 디버그 세션 항목이 이벤트 스트림의 시작이 아닌 전체 요청이 처리된 후에 생성되기 때문에 발생할 수 있습니다. 디버그 세션에는 오류로 인해 생성된 결함 코드가 반영될 수 있지만 클라이언트에는 헤더에서 처음 수신된 2xx 상태만 표시됩니다.
Apigee 분석에서 SSE 데이터 보기
SSE 프록시의 데이터는 API 프록시와 마찬가지로 Apigee 분석에 표시됩니다. Cloud 콘솔에서 분석 > API 측정항목으로 이동합니다.
SSE 프록시 디버깅
Apigee 디버그 도구를 사용하여 SSE 프록시를 디버그합니다.
다른 흐름 유형과 마찬가지로 EventFlow
에 대한 디버그 데이터가 캡처됩니다.
문제 해결
실시간 트래픽 문제의 경우 Apigee 액세스 로그를 확인하여 원인을 파악합니다.
제한사항
SSE 프록시에는 다음과 같은 제한사항이 적용됩니다.