[GECX 엔지니어링 1편] 이벤트 기반 Trigger Pattern 설계와 상태 보존(Early-Return) 메커니즘

LLM의 비결정성을 통제하는 ‘Trigger Pattern’과 음성 통화 특화

 

 

생성형 AI(Generative AI) 기반의 고객 경험(CX) 시스템, 특히 대규모 실시간 음성 통화(Voicebot/Callbot) 환경을 구축할 때 엔지니어가 마주하는 가장 큰 장벽은 LLM의 비결정성(Non-determinism), 음성 통화 특유의 침묵·끼어들기 등 비동기 오디오 제어, 그리고 텔레포니(Telephony/Genesys) 인프라 연동 시 발생하는 비용/품질 제약입니다.

챗봇이 텍스트로 대화할 때의 사소한 실수는 유저가 다시 질문하면 그만이지만, 교환기(PBX/CCaaS) 인프라와 결합된 음성 상담 환경에서는 단 한 번의 파라미터 누락, 침묵 오판단, 상태 증발이 곧바로 통화 강제 종료나 치명적인 상담 장애로 직결됩니다.

이번 글에서는 Google Cloud의 차세대 대화형 AI 플랫폼인 GECX(Gemini Enterprise for Customer Experience) 환경에서 세션 상태를 완벽히 보존하는 Early-Return Guard, 결정론(Determinism)을 100% 보장하는 Trigger Pattern(트리거 패턴), 음성 채널에 특화된 침묵 처리 및 이중 발화 방지 기법, 그리고 이를 Genesys Cloud와 비용 효율적으로 통합하는 하이브리드 연동 아키텍처를 심층 분석합니다.

 

1. 문제의식: LLM과 음성 런타임의 숨겨진 함정

GECX의 Agent Studio나 복합 다이얼로그 환경에서 복잡한 비즈니스 로직(본인 인증, 상담사 에스컬레이션, 침묵 처리, 세션 종료 등)을 다룰 때 크게 세 가지 기술적 난관에 부딪힙니다.

 

1.1 런타임 콜백의 매 턴 실행과 상태 초기화 (State Wipeout)

많은 개발자가 before_agent_callback이 세션 시작 시 단 한 번만 실행될 것이라고 오해합니다. 하지만 GECX 런타임에서 이 콜백은 모든 에이전트 턴(Turn)마다 반복 실행됩니다. 특히 서브 에이전트(AS 접수, 고장진단 등)에서 루트 에이전트로 제어권이 돌아올 때마다 고객 식별 API를 재호출하거나, 기존에 세팅해 둔 인증 상태(auth_status, customer_name)를 덮어써 초기화해 버리는 치명적인 상태 유실이 발생합니다.

 

1.2 LLM에게 직접 도구를 맡겼을 때의 비결정성

프롬프트 지시문(Instruction)만으로 상담사 연결이나 종료를 처리하게 하면 다음과 같은 안티 패턴이 나타납니다.

[고객] “상담원 연결해 주세요.”

[LLM 추론] 상담사 연결 도구(payload_update_tool) 호출 및 세션 종료(end_session) 필요 판단

-> 문제 1: payload_update_tool 호출 시 필수 인자(Argument)를 빈 딕셔너리({})로 누락

-> 문제 2: 통화 종료 전 안내 멘트(Farewell Message) 없이 곧바로 end_session 호출 (고객은 무음 후 통화 끊김 경험)

-> 문제 3: 복잡한 컨텍스트에서 엉뚱한 서브 에이전트로 전송하거나 세션 상태(State) 유실

 

1.3 음성 통화 특유의 묵음 및 멀티 모델 호출 오류

• 침묵(Silence)에 대한 환각: 고객이 전화를 받고 잠시 말을 하지 않을 때 텔레포니 플랫폼은 <context>no user activity detected…</context> 신호를 보냅니다. 이를 LLM 프롬프트에만 맡기면 엉뚱한 말을 지어내거나(Hallucination), 아무런 조치 없이 방치합니다.
• 이중 멘트 출력(Double-Text): 한 턴의 대화 안에서 LLM이 1차 텍스트 생성 → 2차 툴 호출 → 3차 종료 호출 등 여러 번의 모델 호출(Multi-Model Call)을 거칠 때, 매 호출마다 콜백이 트리거되어 작별 인사를 두 번 말하거나 엉키는 현상이 발생합니다.

이를 해결하기 위해 Early-Return Guard, Trigger Pattern, 그리고 음성 특화 콜백 엔지니어링이라는 견고한 방어선이 필요합니다.

 

2. 엔드투엔드 아키텍처: 상태 보존과 음성 제어 파이프라인

전체 에이전트 턴의 라이프사이클은 상태 보존, 음성 침묵 감지, 결정론적 도구 실행을 위해 유기적으로 맞물려 동작합니다.

 

 

3. 실전 코드 분석

3.1 상태 증발 방지: before_agent_callback과 Early-Return Guard

고객 식별값(account_id, customer_id)은 전화 유입 시 IVR/SIP 헤더를 통해 전달됩니다. 이미 프로필 조회가 완료되었다면 이후 턴에서는 절대 재조회하지 않도록 가드를 쳐야 합니다.

 

# before_agent_callback 구현부
from typing import Optional

def before_agent_callback(callback_context: CallbackContext) -> Optional[Content]:
state = callback_context.state

# ————————————————————————-
# [핵심] EARLY RETURN GUARD:
# 이미 auth_status가 세팅되어 있다면 이전 턴이나 서브 에이전트 전환 과정에서
# 조회가 완료된 것이므로 즉시 리턴하여 상태 덮어쓰기(Wipeout)를 방지합니다.
# ————————————————————————-
if state.get(“auth_status”):
return None

# 세션 파라미터에서 식별자 추출 (SIP 헤더/telephony 파라미터 연동)
account_id = state.get(“account_id”, “”)
customer_id = state.get(“customer_id”, “”)

if not account_id or not customer_id:
state[“auth_status”] = “unauthenticated”
state[“user_role”] = “unknown”
state[“customer_name”] = “”
return None

try:
# 플랫폼 샌드박스 내부 툴을 통해 백엔드 고객 데이터스토어 조회
response = tools.Read_Customer_Datastore_readDatastore(
account_id=account_id,
customer_id=customer_id,
)
customer_data = response.get(“customer”, {})
state[“auth_status”] = “authenticated”
state[“user_role”] = customer_data.get(“role”, “standard”)
state[“customer_name”] = customer_data.get(“name”, “”)
state[“service_status”] = customer_data.get(“service_status”, “unknown”)

except Exception as e:
# API 장애 시 여기서 에러 텍스트를 바로 반환하면 오프닝 인사가 덮어써지는 문제가 발생함
# 플래그만 기록해 두고 before_model_callback에서 자연스럽게 에스컬레이션 유도
state[“auth_status”] = “unauthenticated”
state[“user_role”] = “unknown”
state[“customer_name”] = “”
state[“api_failed”] = “true”
state[“api_error_detail”] = str(e)

return None

 

3.2 상태 기록 도구: set_session_state

LLM은 세션 파라미터를 직접 변조할 수 없습니다. 따라서 플래그만 안전하게 전달할 수 있는 전용 도구를 배치합니다.

# tools/set_session_state/python_function/python_code.py
def set_session_state(
_action_trigger: str = “”,
_escalation_reason: str = “”,
_escalation_topic: str = “”
) -> dict:
“””LLM이 액션 트리거와 메타데이터를 세션 상태에 기록하는 도구”””
updated = {}
if _action_trigger:
context.state[“_action_trigger”] = _action_trigger
updated[“_action_trigger”] = _action_trigger
if _escalation_reason:
context.state[“_escalation_reason”] = _escalation_reason
updated[“_escalation_reason”] = _escalation_reason
if _escalation_topic:
context.state[“_escalation_topic”] = _escalation_topic
updated[“_escalation_topic”] = _escalation_topic

if not updated:
return {“agent_action”: “At least one parameter must be provided.”}

return {“status”: “success”, “updated_variables”: updated}

 

3.3 프롬프트(Instruction) 지침 규격

프롬프트에서는 복잡한 도구를 여러 개 나열하지 않고, 상황에 맞춰 set_session_state만 호출하도록 유도합니다.

<subtask name=”Escalation_And_Errors”>
<step name=”Escalate_To_Human”>
<trigger>고객이 상담원 연결을 요청하거나 AI 해결이 불가능한 경우</trigger>
<action>
1. 고객에게 상담사 연결을 진행함을 정중히 알립니다.
2. {@TOOL: set_session_state}를 다음 인자와 함께 호출합니다:
– _action_trigger = “escalate”
– _escalation_reason = 고객의 핵심 문의 요약
– _escalation_topic = 해당 상담 분류 카테고리
</action>
</step>
</subtask>

 

3.4 콜백 인터셉터와 음성 침묵 제어: before_model_callback

실제 비즈니스 로직, 고정 오프닝 인사, 침묵 3진 아웃 제어는 모델 호출 직전 가로채는 before_model_callback에서 수행됩니다.

import re
from typing import Iterator, Optional

# 침묵 감지 정규식
SILENCE_PATTERN = r”<context>no user activity detected for \d+ seconds\.</context>”

def is_user_inactive(contents: list) -> bool:
“””최신 유저 메시지가 텔레포니 침묵 이벤트인지 확인”””
return len(contents) > 1 and any(
re.search(SILENCE_PATTERN, p.text, re.IGNORECASE)
for p in contents[-1].parts if p.text
)

def get_reversed_agent_messages(contents: list) -> Iterator[str]:
“””이전 에이전트 멘트를 역순으로 탐색하여 추출”””
for content in reversed(contents):
texts = [p.text for p in content.parts if content.role == “model” and p.text]
if texts:
yield “”.join(texts)

# 트리거별 실행 매핑 테이블
ESCALATION_MAP = {
“escalate”: {
“text”: “상담사 연결을 도와드리겠습니다. 잠시만 기다려 주세요.”,
“payload”: {
“escalation_reason”: “_escalation_reason”, # 세션 state 참조
“main_topic”: “_escalation_topic”,
“summary”: “고객 요청에 의한 상담사 호전환”,
},
},
“api_failure_escalate”: {
“text”: “시스템 조회 중 지연이 발생하고 있습니다. 전문 상담원에게 연결해 드리겠습니다.”,
“payload”: {
“escalation_reason”: “System API failure”,
“main_topic”: “technical”,
“summary”: “백엔드 API 응답 실패에 따른 자동 에스컬레이션”,
},
},
}

def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]:
state = callback_context.state

# 1. 고정 인사말(Deterministic Greeting): 시작 이벤트 즉시 반환
for part in callback_context.get_last_user_input():
if part.text == “<event>session start</event>”:
greeting = “안녕하세요, AI 고객센터입니다. 무엇을 도와드릴까요?”
return LlmResponse.from_parts(parts=[Part.from_text(text=greeting)])

# 2. 음성 침묵 3진 아웃(Silence Handling) 제어
try:
if is_user_inactive(llm_request.contents):
no_input_counter = int(state.get(“no_input_counter”, 0)) + 1
state[“no_input_counter”] = str(no_input_counter)

if no_input_counter < 3:
reversed_msgs = get_reversed_agent_messages(llm_request.contents)
if no_input_counter == 1:
last_msg = next(reversed_msgs, “무엇을 도와드릴까요?”)
return LlmResponse.from_parts(parts=[
Part.from_text(text=f”말씀이 없으셔서 다시 안내해 드립니다. {last_msg}”)
])
else:
next(reversed_msgs, None) # 이전 반복 구문 건너뛰기
orig_msg = next(reversed_msgs, “말씀을 듣지 못했습니다.”)
return LlmResponse.from_parts(parts=[
Part.from_text(text=f”고객님, 목소리가 들리지 않습니다. {orig_msg}”)
])
else:
# 3회 연속 침묵 시 정중한 종료 안내 후 세션 종료
return LlmResponse.from_parts(parts=[
Part.from_text(text=”고객님의 응답이 없어 통화를 종료합니다. 필요 시 다시 연락해 주세요.”),
Part.from_function_call(
name=”end_session”,
args={“session_escalated”: False, “reason”: “no_input_limit”},
),
])
else:
state[“no_input_counter”] = “0” # 정상 발화 시 카운터 리셋
except Exception as e:
print(f”Silence handling error: {e}”)

# 3. 백엔드 API 실패 플래그 확인
if str(state.get(“api_failed”, “false”)).lower() == “true”:
state[“api_failed”] = “false”
trigger_value = “api_failure_escalate”
else:
# 4. 트리거 확인 및 즉시 초기화 (Trigger Read + Clear 패턴)
trigger_value = state.get(“_action_trigger”, “”)
if trigger_value:
state[“_action_trigger”] = “” # 다음 턴 재발행 방지

if not trigger_value or trigger_value not in ESCALATION_MAP:
return None

escalation = ESCALATION_MAP[trigger_value]

# 세션 상태값 매핑 및 파라미터 빌드
payload_args = {}
for key, value in escalation[“payload”].items():
if isinstance(value, str) and value.startswith(“_”):
payload_args[key] = state.get(value, value)
else:
payload_args[key] = value

# 핵심: LLM에게 노출하지 않고 콜백 내부에서 도구를 직접 실행(Direct Call)
tools.payload_update_tool(payload_args)

# 안내 멘트와 세션 종료(호전환) 신호를 단일 LlmResponse로 확정적 반환
return LlmResponse.from_parts(parts=[
Part.from_text(text=escalation[“text”]),
Part.from_function_call(
name=”end_session”,
args={“session_escalated”: True},
),
])

 

4. 음성 통화 특화 엔지니어링: 침묵 처리 및 멀티 모델 호출 방어

음성 상담은 텍스트 챗봇과 달리 발화 주체(Turn)의 교차, 음성 버퍼(Streaming Chunk), 오디오 전사(Transcript)라는 독특한 제약이 따릅니다.

 

4.1 멀티 모델 호출 환경의 Double-Text 방지 (after_model_callback)

한 턴 내에서 LLM이 텍스트 생성 후 추가 도구(Tool)를 부르고 종료하는 등 모델 호출이 2~3회 연속 일어날 때, 단순한 널 체크만 적용하면 작별 인사가 두 번 발화되는 대참사가 일어납니다.

# after_model_callback 구현부
from typing import Optional

FAREWELL_TEXT = “고객센터를 이용해 주셔서 감사합니다. 좋은 하루 되세요!”

def after_model_callback(callback_context: CallbackContext, llm_response: LlmResponse) -> Optional[LlmResponse]:
has_end_session = False
has_text_this_call = False

# 1. 현재 모델 응답에 end_session 및 발화 텍스트가 있는지 확인
for part in llm_response.content.parts:
if part.has_function_call(“end_session”):
has_end_session = True
else:
# [핵심] 음성 모드에서는 part.text 대신 오디오 트랜스크립트 메타데이터가 들어오므로
# text_or_transcript()를 사용하여 오디오 세이프하게 검증
content = part.text_or_transcript()
if content and len(content.strip()) > 0:
has_text_this_call = True

# end_session이 없거나 이미 현재 호출에서 발화했다면 조기 반환
if not has_end_session or has_text_this_call:
return None

# 2. 멀티 모델 호출 이슈 방어: 이전 모델 호출에서 이미 발화했는지 이벤트 히스토리 역추적
for event in reversed(callback_context.events):
if event.is_user():
break # 최근 유저 발화까지 도달하면 탐색 종료
if event.is_agent():
for p in event.parts():
content = p.text_or_transcript()
if content and len(content.strip()) > 0:
# 이전 모델 호출에서 이미 발화한 내역이 있으므로 작별인사 중복 주입 방지 (Double-text 해결)
return None

# 3. 턴 전체에서 발화가 전무한 채 end_session만 호출된 경우: 종료 직전 작별 인사 Prepend
new_parts = [Part.from_text(text=FAREWELL_TEXT)]
new_parts.extend(llm_response.content.parts)
return LlmResponse.from_parts(parts=new_parts)

 

4.2 음성 세이프 전사 추출 (text_or_transcript)

텍스트 채널과 달리 양방향 음성 스트리밍 환경(Bidi Streaming API)에서는 LLM 응답 페이로드가 단순 문자열(part.text)이 아니라 오디오 인라인 데이터(inline_data) 및 트랜스크립트 메타데이터 형태로 전달됩니다. 콜백에서 part.text만 검사하면 실제 AI 음성이 출력되고 있음에도 텍스트가 없다고 오판단하여 엉뚱한 로직이 실행될 수 있습니다. 반드시 text_or_transcript() 함수로 추상화하여 검사해야 합니다.

 

4.3 오디오 버퍼링과 끼어들기(Barge-in) 충돌 제어

AI 봇이 멘트를 읽고 있는 중간에 고객이 말을 끊고 들어오는 경우(Barge-in), 교환기 단에서 음성 패킷이 동시에 유입되면서 gRPC 세션 에러가 터질 수 있습니다.

• 콜봇 구간: 50ms 단위(1초당 20개 청크)로 스트리밍 처리하되, 고객 음성이 인입되는 즉시 봇의 오디오 큐를 플러시(Flush)하고 발화권을 고객에게 넘기는 게이트웨이 레벨 제어 필요.
• 안내 멘트 구간: 법적 필수 고지 사항이나 계좌번호 안내 등 끊기면 안 되는 주요 시나리오에서는 시스템 옵션으로 Barge-in을 임시 비활성화하여 멘트 청취 완결성 보장.

 

5. 실전 인프라 연동: Genesys Cloud와 Vertex AI AICC의 하이브리드 아키텍처

소프트웨어 레벨에서 Trigger Pattern과 음성 제어를 완성했다면, 다음 과제는 기존 엔터프라이즈 컨택센터 인프라(Genesys Cloud)와의 물리적 연동입니다.

일 2만~3만 건 이상의 대규모 콜 트래픽 환경에서는 전체 호에 동일한 고비용 양방향 음성 스트리밍을 연결할 경우 막대한 커넥터 라이선스 비용이 발생합니다. 실제 프로덕션에서는 비용과 목적에 맞춘 아키텍처 이원화가 필수적입니다.

 

 

5.1 Genesys Cloud 연동 방식 비교 및 비용 최적화

 

연동 방식	기술 특징	비용 수준	실무 적용 영역 및 설계 가이드
웹소켓 (Audio Connector)	Genesys Cloud에서 실시간 양방향 오디오 스트리밍을 직접 연결하여 레이턴시 최소화 (50ms 청킹, Barge-in 대응)	높음 (채널당 커넥터 라이선스 부과)	콜봇(Voicebot) 영역: 전체 콜이 아닌 IVR 분기(정수기 가입, 취소 등)를 거쳐 특정 대기열(Queue)로 진입한 자율 응대 세션에만 오디오 커넥터를 활성화하여 비용 최적화
단방향 오디오 포킹 (Symmetric/심맥)	통화 음성을 단방향으로 복제(Forking)하여 GECX/STT로 스트리밍	낮음 (라이선스 저렴)	상담원 조력(Agent Assist) 영역: 실시간 추천 답변 생성에는 양방향 제어가 불필요하므로 단방향 심맥을 적용하여 인프라 비용 대폭 절감
자체 STT API 직접 연동	사내 구축된 STT(예: 효성 STT) 전사 텍스트를 중간 미들웨어 경유하여 GECX로 직접 전송	매우 낮음 (기존 인프라 재활용)	상담사 화면(Zendesk/CRM) 팝업: 음성 스트림 전송 없이 순수 텍스트 결과만 워크스페이스에 즉시 반영하여 구간 지연 시간 대폭 개선

 

 

5.2 SBC 경유 연동과 통신망 제약 극복

• mTLS 암호화 및 트렁킹 제약: Genesys Cloud와 Google Cloud GECX 간 직접 SIP 연결 시 상호 인증서(mTLS) 암호화 규격 차이 및 트렁킹 기능 제약이 발생합니다. 따라서 중간에 가이드 SBC(Session Border Controller)를 배치하여 시그널링과 미디어를 중계하는 토폴로지가 필수로 요구됩니다.
• 통신망 레이턴시 최적화: 국내 통신사 망에서 미국 멀티 리전(US)으로 직접 호가 전달되면 홉(Hop) 수 증가로 지연 시간과 음질 저하가 발생합니다. 인접 리전(도쿄/오사카)에 오픈소스 SBC(Asterisk 등) 및 프록시 환경을 구축하여 SIP 트렁킹을 주(Primary) 경로로, Audio Connector를 장애 대비(Fallback) 경로로 이원화 구성합니다.

 

5.3 텔레포니 메타데이터 바인딩 파이프라인

전화 유입 시 전달되는 시스템 변수(telephony_ani, telephony_dnis)의 국가번호 형태(+8210…)를 콜백 함수를 통해 010…으로 정제한 뒤 고객 DB를 조회합니다. 이후 고객 등급(VIP 여부), 인입 경로, 에스컬레이션 사유는 앞서 다룬 Trigger Pattern의 payload_update_tool을 통해 Genesys의 SIP 헤더 및 상담사 CRM(Zendesk ZAF, Salesforce) 화면 팝업 데이터로 무결하게 전달됩니다.

 

5.4 실시간 침묵 시나리오의 확정적 제어

고객이 말을 하지 않을 때 LLM이 엉뚱하게 헛소리하지 못하도록 차단하고, 1차/2차 재안내 후 3차에 정중히 세션을 마감하는 표준화된 전화 응대 프로토콜을 백엔드에서 100% 보장합니다.

 

5.5 인프라 비용과 라이선스의 획기적 절감

모든 인바운드 콜에 고가의 양방향 오디오 커넥터를 연결하지 않고, 콜봇 구간(양방향 WebSocket)과 상담사 지원 구간(단방향 포킹/STT 텍스트)을 아키텍처적으로 분리하여 일 수만 건의 대규모 트래픽에서도 TCO를 통제할 수 있습니다.

 

7. 결론 및 요약

 

구분	As-Is (프롬프트 및 단일 연동 의존)	To-Be (Early-Return + Trigger Pattern + 음성 특화 엔지니어링)
상태 관리	멀티턴/에이전트 전환 시 인증 정보 유실 빈발	Early-Return Guard로 상태 보존 및 API 재호출 방지
상담사 호전환 성공률	확률적 (인자 누락 및 문맥 손실 빈발)	100% 확정적 (Deterministic 콜백 인터셉트)
통화 종료 UX	간헐적 무음 끊김 또는 인사 중복(Double-text)	이벤트 역추적 기반 멘트 송출 보장 (LlmResponse 통합)
침묵 대응	LLM 환각 발생 또는 방치	1~2차 재안내 및 3차 확정 종료 (Silence 3진 아웃)
CCaaS 연동 비용	전체 호에 고비용 Audio Connector 일괄 적용	콜봇(양방향) vs 상담원 지원(단방향/STT) 분리로 라이선스 절감
네트워크 안정성	글로벌 리전 직접 연결 시 레이턴시 이슈	중간 SBC 경유 및 SIP/WebSocket Fallback 이중화

• LLM의 비결정성 통제: API 런타임에서 발생하는 예기치 못한 매 턴 콜백 상태의 누수를 차단하기 위해 Early-Return Guard 논리 구조를 정립했습니다.
• 트리거 패턴(Trigger Pattern): 다중 에이전트 간 에스컬레이션 시 발생할 수 있는 시점 불일치를 해결하는 Trigger Read + Clear 설계로 논리적 결정론을 확보했습니다.
• 구현체 안정성: 단순 개념에 그치지 않고 context.state 제어 및 pytest 단위 테스트 코드를 직접 수록하여 신뢰할 수 있는 개발 가이드를 제시합니다.

 

자세한 내용이 궁금하시다면, 메가존소프트 문의포탈을 통해 궁금한 부분을 남겨주세요.