4,078
고객지원(Customer Support) 에이전트를 직접 만들면서 OpenAI Agents SDK의 세 가지 핵심 개념을 정리한다. 하나의 Triage Agent(분류 담당)가 사용자의 요청을 받아 → 주제에 맞는지 검사하고(Guardrails) → 적절한 전문 상담원에게 넘기는(Handoff) 구조다. 그 과정에서 사용자 정보는 Context를 통해 모든 도구로 흐른다.
💬 이 글은 실제 프로젝트의 소스코드 주석과 학습 노트를 통합한 것이다. 코드와 함께 "왜 이렇게 동작하는가"를 따라가며 읽으면 좋다.
Context는 AI 모델이 아니라 "내 코드"에 제공하는 데이터다. 이 구분이 가장 중요하다.
⚠️ AI 모델이 호출될 때 볼 수 있는 데이터는 대화 기록(conversation history)뿐이다. Context는 모델에게 보이지 않는다. 모델에게 Context의 내용을 보여주고 싶다면 프롬프트 안에 직접 넣어야 한다.
이 프로젝트에서 Context는 Pydantic BaseModel로 정의한다.
from pydantic import BaseModel
# 유저 계정 context를 위한 BaseModel
class UserAccountContext(BaseModel):
customer_id: int
name: str
tier: str = "basic"Context는 두 곳으로 자동 전달된다.
function_tool****의 첫 번째 인자로 들어온다.# Runner에서 사용된 context는 모든 function_tool의 첫번째 arg로 들어온다. 아래는 예시.
# @function_tool
# def get_user_tier(wrapper: RunContextWrapper[UserAccountContext]):
# return (
# f"The user {wrapper.context.customer_id} has a {wrapper.context.tier} account."
# )Context 인스턴스는 Runner에 넘겨주는 순간부터 흐르기 시작한다.
# context 인스턴스
user_account_ctx = UserAccountContext(
customer_id=1,
name="toma",
tier="basic",
)
stream = Runner.run_streamed(
triage_agent,
message,
session=session,
context=user_account_ctx, # context 인스턴스를 Runner에서 사용.
# 자동으로 에이전트(=모델)에게 전달되는 것은 아님.
# 이제 모든 function_tool들이 context를 다 받게 됨.
)💡
context=user_account_ctx를Runner에 한 번 넘기면, 이후 실행되는 모든function_tool과 동적 지침 함수가 같은 Context 인스턴스를 받는다. 일일이 넘겨줄 필요가 없다.
off_topic_guardrail, handle_handoff, dynamic_triage_agent_instructions 같은 콜백 함수들은 모두 첫 인자로 wrapper(context)를 받는다. 그런데 이 함수들은 우리 코드 어디에서도 직접 호출하지 않는다. 그렇다면 누가 호출하고, context는 어떻게 채워지는가? 여기엔 서로 다른 두 경로가 있다.
💬 흔한 오해 — "
input_guardrails에 등록했으니 그 옵션이 context도 전달해준다"고 생각하기 쉽다. 하지만 등록과 주입은 별개다.
| 구분 | 무엇을 하는가 | 어디서 |
|---|---|---|
| 등록 (Registration) | "이 함수를 언제 실행할지"를 정함 | Agent의 옵션 (input_guardrails=[...], handoffs=[...], instructions=...) |
| 주입 (Injection) | "함수의 첫 인자에 어떤 context를 꽂을지"를 정함 | Runner.run(context=...) |
동작 순서는 다음과 같다.
Runner.run(triage_agent, context=user_account_ctx)
│
├─ user_account_ctx 를 RunContextWrapper 로 감쌈
│
├─ triage_agent.input_guardrails 목록을 봄 ← "등록"이 여기서 읽힘
│ │
│ └─ off_topic_guardrail(wrapper, agent, input) 호출
│ ↑ wrapper = Runner가 감싼 그 context ← "주입"은 Runner가 직접
│
└─ (가드레일 통과 시) triage_agent 본 실행✅ 핵심: 호출 주체는 Runner다. 그리고 Runner는 자신이 호출하는 모든 사용자 콜백(가드레일·핸드오프 콜백·동적 지침·function_tool)의 첫 인자에 동일한
RunContextWrapper를 꽂아준다. 그래서handle_handoff처럼input_guardrails와 전혀 무관한 함수도 context를 받는다. 등록 경로(어떤 옵션에 넣었는지)는 "언제 부를지"만 정하고, context 주입은 Runner가 일괄 처리한다. 이것이 "context는 어디든 따라간다"의 실제 정체다.
Context가 모든 도구에 전달되기 때문에 다음과 같은 도구를 만들 수 있다.
✅ 가장 큰 장점: 민감한 정보를 AI 모델에게 넘기지 않아도 된다. 고객 ID, 등급 같은 데이터를 모델 프롬프트에 노출하지 않고도, 코드 레벨의 도구에서 안전하게 활용할 수 있다.
Guardrail은 에이전트가 주제를 벗어나지 않도록 막는 안전장치다. 두 가지 타입이 있다.
| 타입 | 시점 | 동작 |
|---|---|---|
| 입력 (Input) | 에이전트 실행 전 | guardrail 함수가 tripwire_triggered=True를 반환하면 SDK(Runner)가 예외를 던져 해당 에이전트 실행을 중단시킨다. |
| 출력 (Output) | 에이전트 답변 후 | 에이전트가 우리가 원하지 않는 방식으로 답변할 경우 tripwire를 발동시킨다. |
⚠️ 흔한 오해 — "에이전트가 스스로 멈춘다"가 아니다. 주체는 Runner/SDK다. guardrail 함수는 "발동 여부(
tripwire_triggered)"만 알려주고, 실제로 예외를 던져 실행을 중단시키는 것은 SDK다. 💬 이 프로젝트는 입력 가드레일만 구현한다. 출력 가드레일은 개념으로만 함께 정리한다.
1단계. 규칙을 검사하는 전용 에이전트를 만든다. 이 에이전트는 "주제에 맞는 요청인가?"를 판단해 구조화된 결과(InputGuardRailOutput)를 반환한다.
# 1. 입력 가드레일(안전장치) 에이전트
input_guardrail_agent = Agent(
name="Input Guardrail Agent",
# 규칙을 작성하여 유저 요청이 규칙에 맞는지 확인하고 그 외엔 tripwire 발동
instructions="""
Ensure the user's request specifically pertains to User Account details, Billing inquiries, Order information,
or Technical Support issues, and is not off-topic. If the request is off-topic,
return a reason for the tripwire. You can make small conversation with the user,
specially at the beginning of the conversation,
but don't help with requests that are not related to User Account details,
Billing inquiries, Order information, or Technical Support issues.
""",
output_type=InputGuardRailOutput
)반환 타입은 Pydantic 모델로 명확히 정의한다.
# 에이전트 -> 구조화된 답변
class InputGuardRailOutput(BaseModel):
is_off_topic: bool
reason: str2단계. 그 에이전트를 실제로 실행하는 함수를 @input_guardrail로 감싼다. 이 함수는 triage_agent가 호출되기 전에 실행되며, 유저의 context·에이전트·사용자 입력을 받는다.
# 입력 안전장치 에이전트 실행 함수
# triage_agent가 호출되기 전에 실행.
# 유저의 context와 에이전트, 사용자 입력을 가져옴.
@input_guardrail
async def off_topic_guardrail(
wrapper: RunContextWrapper[UserAccountContext],
agent: Agent[UserAccountContext],
input: str
):
result = await Runner.run(
input_guardrail_agent,
input,
context=wrapper.context
)
# input_guardrail과 output_guardrail 함수는 GuardrailFunctionOutput을 필수로 반환해야 하며,
# 출력과 tripwire 작동 여부를 전달해야 함.
return GuardrailFunctionOutput(
output_info=result.final_output,
tripwire_triggered=result.final_output.is_off_topic
)💡
input_guardrail과output_guardrail함수는 반드시GuardrailFunctionOutput을 반환해야 한다. 그 안에 (1) 출력 정보(output_info)와 (2) tripwire 작동 여부(tripwire_triggered)를 담아 전달한다.
triage_agent = Agent(
name="Triage_Agent",
instructions=dynamic_triage_agent_instructions,
# input guardrail은 첫번째(진입) 에이전트에서만 실행됨.
# handoff로 넘어간 다른 에이전트에는 적용되지 않음.
input_guardrails=[
off_topic_guardrail
], # 이렇게 해두면 triage_agent가 실행되기 전에 off_topic_guardrail 함수가 먼저 실행됨.
...
)⚠️ 입력 가드레일은 첫 번째(진입) 에이전트에서만 실행된다. Handoff로 넘어간 다른 전문 에이전트에는 적용되지 않는다. 이 프로젝트에서 모든 사용자 입력은
triage_agent를 거쳐 들어오므로, 여기에만 가드레일을 달면 충분하다. (그래서order_agent,billing_agent등 전문 에이전트에는 가드레일이 없다.)
주제를 벗어난 질문이 들어오면 입력 가드레일에서 InputGuardrailTripwireTriggered 예외가 발생한다. 이를 잡아서 처리하면 된다.
try:
stream = Runner.run_streamed(
triage_agent,
message,
session=session,
context=user_account_ctx,
)
async for event in stream.stream_events():
...
# 입력 가드레일 예외 발생 처리
except InputGuardrailTripwireTriggered:
st.write("I can't help you with that.")Handoff가 처리되는 방식에는 크게 두 가지 옵션이 있다.
대화 자체가 다른 에이전트에게 넘어감.
예) A 상담원과 대화 중 B 상담원으로 연결 전환. A 상담원과는 통신이 끊긴다.
다른 에이전트를 도구(tool)로 처리.
예) A 상담원과 대화 중, A가 다른 상담원에게서 정보를 얻어와 A가 나에게 답변하는 형태.
💡 이 프로젝트는 **1번 방식(대화 전환)**을
handoffs=[...]로 구현한다. 2번 방식은agent.as_tool(...)로 구현하며, 아래 "대안"에서 다룬다.
Handoff가 일어날 때 함께 넘길 메타데이터를 Pydantic 모델로 정의한다.
class HandoffData(BaseModel):
to_agent_name: str
issue_type: str
issue_description: str
reason: strdef make_handoff(agent):
return handoff(
agent=agent,
on_handoff=handle_handoff,
# AI 모델이 handoff를 호출할 때, HandoffData의 필드를 직접 채워서 tool 인자로 생성,
# 그 객체가 on_handoff 콜백의 input_data로 전달됨
input_type=HandoffData,
input_filter=handoff_filters.remove_all_tools, # 새로운 에이전트가 볼 데이터를 골라서 넘길 수 있게 해주는 필터. (extension에서 import)
# 이 경우 에이전트가 호출했던 tool 사용 기록을 지워주고 유저와 에이전트 간 메세지만 남김.
)handle_handoff 함수를 정의(등록)만 하고 실제 호출은 Runner가 수행합니다.
LLM이 handoff 도구를 호출하면서 input_type에 맞는 JSON을 생성하고 Runner(SDK)가 그 JSON을 HandoffData 객체로 변환하고 그 객체를 input_data 자리에 넣어주는 과정을 거칩니다.
각 파라미터의 역할:
| 파라미터 | 역할 |
|---|---|
agent | 인계받을 대상 에이전트 |
on_handoff | handoff가 일어나는 순간 호출되는 콜백 함수 |
input_type | 모델이 handoff를 호출할 때 채울 데이터의 타입(HandoffData). 모델이 필드를 채워 tool 인자로 만들고 → SDK가 검증 → on_handoff의 input_data로 전달 |
input_filter | 인계받는 에이전트가 볼 대화 데이터를 골라서 넘기는 필터 |
💡 **
input_type**의 데이터 흐름: AI 모델이HandoffData의 필드(어느 에이전트로, 어떤 이슈인지, 이유 등)를 직접 채워 handoff 도구의 인자로 생성한다. 그 객체가on_handoff콜백의input_data로 전달된다. 덕분에 "왜 이 상담원에게 넘겼는지"를 코드에서 받아 쓸 수 있다. ✅remove_all_tools(agents.extensions.handoff_filters에서 import): 인계 시 이전 에이전트의 도구 호출 기록을 모두 지우고, 유저와 에이전트 간 메시지만 남긴다. 새 에이전트가 불필요한 도구 사용 내역에 혼란받지 않게 한다.
on_handoff 콜백은 Context를 받을 수 있다. Context는 어디든 따라가기 때문이다.
# context는 어디든 따라가기 때문에 여기서도 context를 가져올 수 있음.
def handle_handoff(
wrapper: RunContextWrapper[UserAccountContext],
input_data: HandoffData,
):
# handoff가 발생하게 된 이유 서술
with st.sidebar:
st.write(
f"""
Handing off to {input_data.to_agent_name}
Reason: {input_data.reason}
Issue Type: {input_data.issue_type}
Description: {input_data.issue_description}
"""
)💡
input_data는 위에서 모델이 채운HandoffData객체다. 그래서 사이드바에 "어느 상담원에게, 왜, 어떤 이슈로 넘겼는지"를 그대로 표시할 수 있다.
triage_agent = Agent(
name="Triage_Agent",
# instructions 시그니처를 보면 문자열을 넘길 수도 있고
# RunContextWrapper와 에이전트를 매개변수로 받는 함수를 넘길 수도 있음.
# 에이전트까지 호출(전달)하는 이유는 여러 에이전트에서 해당 함수가 사용될 수 있기 때문.
instructions=dynamic_triage_agent_instructions,
input_guardrails=[off_topic_guardrail],
# handoff를 import 해서 handoff가 일어날 때 호출되는 함수 등등 여러가지를 할 수 있음.
handoffs=[
make_handoff(technical_agent),
make_handoff(billing_agent),
make_handoff(account_agent),
make_handoff(order_agent),
]
)💡
instructions시그니처: 단순 문자열을 넘길 수도 있고,RunContextWrapper와agent를 받는 함수를 넘길 수도 있다. 함수에agent까지 전달되는 이유는, 동일한 지침 함수를 여러 에이전트에서 재사용할 수 있기 때문이다.
handoff가 있는 에이전트의 동적 지침에는 SDK 권장 프리픽스를 최상단에 넣는다.
def dynamic_triage_agent_instructions(
wrapper: RunContextWrapper[UserAccountContext],
agent: Agent[UserAccountContext]
):
# RECOMMENDED_PROMPT_PREFIX는 OpenAI SDK에서 제공하는 프롬프트로,
# agent에 handoff가 있을 경우 agent 지침 최상단에 넣으라고 권장함.
return f"""
{RECOMMENDED_PROMPT_PREFIX}
You are a customer support agent. ...
The customer's name is {wrapper.context.name}.
The customer's tier is {wrapper.context.tier}.
...
"""✅
RECOMMENDED_PROMPT_PREFIX는 모델에게 handoff 도구의 존재와 사용법을 자연스럽게 이해시키기 위한 권장 프롬프트다. handoff를 쓰는 에이전트라면 지침 최상단에 넣는 것이 좋다.
위에서 언급한 2번 방식(다른 에이전트를 도구로 처리)은 handoffs 대신 tools에 as_tool()로 등록한다.
# 에이전트가 여러 개일 경우 일종의 tool로 사용 가능.
# tools=[
# technical_agent.as_tool(
# tool_name="Technical Help Tool",
# tool_description="Use this when the user tech support"
# )
# ]💬 Handoff vs. as_tool 정리 — Handoff는 대화의 _주도권 자체_가 넘어간다(전환). as_tool은 주도권은 그대로 둔 채, 다른 에이전트를 호출해 정보만 받아오는 도구로 쓴다.
사용자 입력
│
▼
[입력 Guardrail] ── off-topic이면 → InputGuardrailTripwireTriggered 예외 → "I can't help you with that."
│ (통과)
▼
[Triage Agent] ── Context(이름·등급)로 동적 지침 생성, 이슈 4분류
│
▼
[Handoff] ── HandoffData(모델이 채움) → handle_handoff 콜백(사이드바 표시) → remove_all_tools 필터
│
▼
[전문 에이전트] Technical / Billing / Order / Account| 개념 | 한 줄 요약 |
|---|---|
| Context | 모델이 아닌 _내 코드_에 흐르는 데이터. 모든 도구·동적 지침의 첫 인자. 민감정보를 모델에 노출 안 함. |
| Guardrails | 주제 이탈 차단. 입력/출력 2종. 발동 주체는 Runner/SDK. 입력 가드레일은 진입 에이전트에만 적용. |
| Handoff | 대화를 다른 에이전트로 전환. input_type으로 모델이 메타데이터를 채우고, 필터로 넘길 데이터를 정리. |
Handoff가 실제로 일어났을 때, 사용자에게 "지금 다른 상담원에게 연결됐다"는 것을 보여주려면 두 가지가 필요하다. (1) 전환을 감지하고, (2) 다음 메시지부터 전환된 에이전트로 이어서 실행하는 것이다.
메시지를 보낼 때마다 run_agent가 실행되고, 이 함수는 triage_agent가 들어 있는 Runner를 실행한다. 하지만 일단 전문 에이전트로 전환됐다면, 다음 메시지부터는 _전환된 그 에이전트_로 이어서 실행하고 싶다.
💡 그래서 현재 활성 에이전트를
st.session_state["agent"]에 캐시로 저장하고, 전환이 감지될 때마다 이 값을 갱신한다.Runner에는triage_agent가 아니라 이 캐시된 에이전트를 넘긴다.
# 진입 시 기본값은 triage_agent
if "agent" not in st.session_state:
st.session_state["agent"] = triage_agent
# Runner에는 항상 캐시된(=현재 활성) 에이전트를 넘긴다
stream = Runner.run_streamed(
st.session_state["agent"], # triage_agent 가 아니라 캐시된 에이전트
message,
session=session,
context=user_account_ctx,
)에이전트 전환은 스트림 이벤트 타입 중 agent_updated_stream_event로 확인할 수 있다. 새 에이전트 이름이 캐시된 이름과 다르면 전환이 일어난 것이다.
elif event.type == "agent_updated_stream_event":
if st.session_state["agent"].name != event.new_agent.name:
st.write(f"🤖 Transfered from {st.session_state['agent'].name} to {event.new_agent.name}")
# 캐시를 새 에이전트로 갱신 → 다음 메시지부터 이 에이전트로 이어감
st.session_state["agent"] = event.new_agent
# 새 에이전트의 응답을 받을 빈 placeholder 준비
text_placeholder = st.empty()
response = ""✅ 핵심 흐름: 감지(
agent_updated_stream_event) → 표시(st.write) → 캐시 갱신(session_state["agent"]). 캐시를 갱신해 두기 때문에, 사용자가 다음 메시지를 보내면 triage를 거치지 않고 전환된 전문 에이전트가 바로 이어받는다.
OpenAI Agents SDK의 Hook은 에이전트 실행 중 일어나는 이벤트를 감지하는 listener다. 이벤트를 "구독"하듯 연결해 두면, 해당 이벤트가 발생할 때마다 알림을 받아 원하는 동작(로깅·모니터링)을 실행할 수 있다.
💡 두 종류가 있다. **
AgentHooks**는 _특정 에이전트 하나_의 이벤트를, **RunHooks**는 _Runner 실행 전체_의 이벤트를 구독한다. 어느 쪽이든 서브클래스를 만들어 원하는 이벤트 메서드만 오버라이드하면 된다.
AgentHooks를 상속한 클래스에서 아래 메서드들을 오버라이드하면, 해당 시점마다 호출된다.
| 메서드 | 호출 시점 |
|---|---|
on_start | 에이전트가 활성화될 때 |
on_tool_start | 도구(tool) 호출 직전 |
on_tool_end | 도구 실행이 끝나고 결과가 나왔을 때 |
on_handoff | 다른 에이전트로 handoff가 일어날 때 |
on_end | 에이전트 실행이 완료될 때 |
각 메서드의 첫 인자가 self인 것은, 이 메서드들이 클래스의 인스턴스 메서드이기 때문이다(AgentHooks를 상속한 서브클래스의 인스턴스). SDK가 이벤트 발생 시 이 인스턴스의 메서드를 호출하면서 context, agent, tool 등을 인자로 넘겨준다.
from agents import AgentHooks, Agent, Tool, RunContextWrapper
class AgentToolUsageLoggingHooks(AgentHooks):
async def on_start(self, context, agent):
with st.sidebar:
st.write(f"🚀 **{agent.name}** activated")
async def on_tool_start(
self,
context: RunContextWrapper[UserAccountContext],
agent: Agent[UserAccountContext],
tool: Tool,
):
with st.sidebar:
st.write(f"🔧 **{agent.name}** starting tool: `{tool.name}`")
async def on_tool_end(self, context, agent, tool, result):
with st.sidebar:
st.write(f"🔧 **{agent.name}** used tool: `{tool.name}`")
st.code(result) # 도구 실행 결과를 사이드바에 그대로 표시
async def on_handoff(self, context, agent, source):
with st.sidebar:
st.write(f"🔄 Handoff: **{source.name}** → **{agent.name}**")
async def on_end(self, context, agent, output):
with st.sidebar:
st.write(f"🏁 **{agent.name}** completed")💬 위 콜백들은 모두 첫 인자로
context(RunContextWrapper)를 받는다. Guardrail·handoff 콜백과 동일하게, context는 어디든 따라간다는 원칙이 Hook에도 그대로 적용된다.
📝 정리 — **
self**와 서브클래스
self: 파이썬에서 인스턴스가 메서드를 호출할 때, 첨 번째 인자로 인스턴스 자기 자신을 파이썬이 자동으로 넘긴다. 그래서 메서드의 첫 인자 자리는self로 비워둡는다. (본문에서self.…를 안 써도 자리는 필요)주입 출처는 둘:
on_tool_start(self, context, agent, tool)에서self는 _파이썬_이, 나머지context·agent·tool은 _Runner_가 채운다.서브클래스(상속): 다른 객체지향 언어의 상속과 같은 개념.
class 내클래스(AgentHooks)형태로AgentHooks를 상속한다.왜 상속?:
AgentHooks는 이벤트 메서드들을 본문pass(빈 껍데기)로 미리 갖추고 있다. 상속하면 5개 메서드가 다 갖춰져 Runner가 불러도 에러가 안 나고, 그중 관심 있는 것만 오버라이드해 덮어쓰면 된다.
정의한 Hook 인스턴스를 각 에이전트의 hooks 파라미터에 넘기면 구독이 시작된다.
account_agent = Agent(
name="Account Management Agent",
instructions=dynamic_account_agent_instructions,
tools=[
reset_user_password,
enable_two_factor_auth,
update_account_email,
deactivate_account,
export_account_data,
],
hooks=AgentToolUsageLoggingHooks(), # ← Hook 인스턴스 등록
)✅ 이렇게 등록해 두면 해당 에이전트가 도구를 쓰거나 handoff 할 때마다 사이드바에 실시간 로그가 찍힌다. 디버깅과 모니터링에 유용하다.
Input Guardrail과 이름만 다를 뿐 역할은 동일하다. 단, 시점이 반대다 — 에이전트가 응답할 때 실행되어, 에이전트가 우리가 원하지 않는 방식으로 답할 경우 tripwire를 발동시킨다.
⚠️ Guardrail은 에이전트보다 먼저 실행된다기보다 거의 동시에 병렬로 실행된다. 그래서 에이전트가 응답을 생성하는 중에 가드레일이 함께 돌 수도 있다. 💡 Input Guardrail이 진입 에이전트에만 붙는 것과 달리, Output Guardrail은 에이전트별로 각 주제에 맞춰 붙여야 한다. 이 프로젝트에서는 Technical Agent가 결제·주문·계정 정보를 내놓지 않도록 전용 가드레일을 달았다.
에이전트 응답을 검사해 어떤 종류의 부적절한 정보가 섞였는지를 구조화된 결과로 반환한다. 필드들이 하나라도 True면 tripwire가 발동되도록 한다.
class TechnicalOutputGuardRailOutput(BaseModel):
contains_off_topic: bool
contains_billing_data: bool # 결제·환불·구독 정보 포함 여부
contains_account_data: bool # 비밀번호·이메일·계정 설정 정보 포함 여부
reason: strInput Guardrail과 똑같이, (1) 규칙을 검사하는 전용 에이전트를 만들고 (2) 그것을 실행하는 함수를 @output_guardrail로 감싼다.
# 출력 가드레일 (예시)
from agents import (
Agent,
output_guardrail,
Runner,
RunContextWrapper,
GuardrailFunctionOutput,
)
from models import TechnicalOutputGuardRailOutput, UserAccountContext
technical_output_guardrail_agent = Agent(
name="Technical Support Guardrail",
instructions="""
Analyze the technical support response to check if it inappropriately contains:
- Billing information (payments, refunds, charges, subscriptions)
- Order information (shipping, tracking, delivery, returns)
- Account management info (passwords, email changes, account settings)
Technical agents should ONLY provide technical troubleshooting, diagnostics, and product support.
Return true for any field that contains inappropriate content for a technical support response.
""",
output_type=TechnicalOutputGuardRailOutput,
)
# output_guardrail 함수, 에이전트의 output_guardrails에 전달 (ex. output_guardrails=[technical_output_guardrail])
@output_guardrail
async def technical_output_guardrail(
wrapper: RunContextWrapper[UserAccountContext],
agent: Agent,
# 에이전트가 낸 output
output: str,
):
# 에이전트의 output과 context를 받아 Runner 실행,
result = await Runner.run(
technical_output_guardrail_agent,
output,
context=wrapper.context,
)
# output type을 지정해놓았으니 하나라도 true면 trip wire 발동.
validation = result.final_output
triggered = (
validation.contains_off_topic
or validation.contains_billing_data
or validation.contains_account_data
)
# input guardrail과 마찬가지로 GuardrailFunctionOutput을 필수로 반환해야 함.
return GuardrailFunctionOutput(
output_info=validation,
tripwire_triggered=triggered,
)💡 구조는 Input Guardrail과 동일하다. 검사 에이전트의
output_type을TechnicalOutputGuardRailOutput으로 지정해 둔 덕분에, 필드 중 하나라도True면triggered가 되고 tripwire가 발동한다. 그리고 Input/Output 구분 없이 모두GuardrailFunctionOutput을 필수로 반환해야 한다.
Input Guardrail을 input_guardrails=[...]로 달았던 것처럼, Output Guardrail은 해당 전문 에이전트의 output_guardrails=[...]에 등록한다.
from output_guardrails import technical_output_guardrail
technical_agent = Agent(
name="Technical Support Agent",
instructions=dynamic_technical_agent_instructions,
tools=[
run_diagnostic_check,
provide_troubleshooting_steps,
escalate_to_engineering,
],
hooks=AgentToolUsageLoggingHooks(),
output_guardrails=[technical_output_guardrail], # ← 이 에이전트의 응답을 검사
)입력 가드레일이 InputGuardrailTripwireTriggered를 던졌던 것처럼, 출력 가드레일은 OutputGuardrailTripwireTriggered 예외를 던진다. 이를 잡아 처리한다.
from agents import OutputGuardrailTripwireTriggered
try:
stream = Runner.run_streamed(
st.session_state["agent"], message, session=session, context=user_account_ctx,
)
async for event in stream.stream_events():
...
# 입력 가드레일 예외
except InputGuardrailTripwireTriggered:
st.write("I can't help you with that.")
# 출력 가드레일 예외
except OutputGuardrailTripwireTriggered:
st.write("Cant show you that answer.")
st.session_state["text_placeholder"].empty() # 이미 스트림된 부적절 응답을 화면에서 지움✅ 출력 가드레일은 응답과 병렬로 돌기 때문에, 부적절한 텍스트가 이미 화면에 일부 스트리밍되었을 수 있다. 그래서 예외 처리에서
text_placeholder.empty()로 그 잘린 응답을 지워주는 것이 중요하다.
음성 에이전트 구현을 위해서는 VoicePipeline(음성 처리과정) 이 필요함.
이는 voice를 Text로 변환하고 이를 가지고 agentic workflow(에이전트 작업흐름)을 수행, 즉 text로 지금까지와 같이 Runner.run()으로 에이전트 호출.
그리고 반환받은 응답을 다시 Audio로 변환하는 것까지가 VoicePipeline 이 하는 역할임.
그런데 VoicePipeline를 사용하게 되면 단 하나의 voice 에이전트만 돌릴 수 있고 나만의 workflow 커스터마이징도 어렵다는 단점이 있는데,
최근 RealtimeSession 이라는 것을 통해 speech-to-speech voice agent를 구현할 수 있는 듯 하다.