AI Agent : 사용자와 더 똑똑하게 대화하기 (Context, Guardrails, Handoffs, Hooks)

지난 포스팅에서 OpenAI Agents SDK의 기본 구성요소(Agent, Runner, SQLiteSession, handoffs)를 훑었다. 이번엔 거기서 한 발 더 들어간 개념들을 정리한다.

 

Context → Dynamic Instructions → Handoffs → Input/Output Guardrails → Hooks

 

이들은 SDK 사용법 그 자체보단, 에이전트가 여러 개일 때 코드가 어디서 정보를 주고받고, 누가 무엇을 검사하고, 흐름이 어디서 어떻게 끊기는지에 대한 관점이다. 이게 잡혀야 어떤 프레임워크를 써도 흐름을 따라갈 수 있다.

예시로 고객 지원 에이전트라는 느낌으로, Triage 에이전트가 사용자 질문을 받아서 Billing/Order/Technical/Account 4명의 전문 에이전트 중 하나로 넘기는 구조다. 일종의 콜센터 라우팅이라고 보면 된다.

---

0. 왜 이런 개념들이 필요할까

위에서 말한 콜센터 라우팅 구조를 좀 더 구체적으로 그려보면 이렇다.

• 사용자가 질문을 던지면 먼저 Triage Agent가 받는다 (분류 담당)
• 질문 카테고리에 따라 Billing / Order / Technical / Account 4명의 전문 에이전트 중 하나로 handoff
• 전문 에이전트는 각자의 tool을 써서 실제 작업을 처리

이 구조를 만들다 보면 자연스럽게 부딪히는 문제들이 있다.

• 고객의 이름, 등급(basic / premium) 같은 정보를 모든 에이전트와 tool이 알아야 한다. 어떻게 공유하지?
• 같은 에이전트라도 고객 등급에 따라 응대 방식이 달라야 한다. instructions를 어떻게 동적으로?
• 주제와 무관한 질문(예: 오늘 점심 뭐 먹지)은 아예 받지 말아야한다. 어디서 막지?
• Technical Agent가 갑자기 환불 얘기를 꺼내면 곤란하다. 응답을 내보내기 전에 검증할 방법은?
• handoff가 일어나고, tool이 호출되고, 흐름이 복잡한데 어디서 무슨 일이 일어났는지 추적하고 싶다.

때문에 각 분야의 전문가를 모시는거다.

---

1. Context : 에이전트들이 공유하는 사전 지식

지금까지 우리는 에이전트에게 정보를 줄 때 messages(대화 기록)로 넘겼다. 하지만 "이 사용자는 customer_id가 1번이고, premium 등급이고, 이메일은 어쩌고..." 같은 메타데이터까지 매번 user 메시지로 박아넣는 건 비효율적이다. 모델이 봐야 할 정보도 아니다.

이럴 때 쓰는 게 Context다. 코드가 공유하는 백그라운드 지식 역할이다.

# models.py
from pydantic import BaseModel
from typing import Optional


class UserAccountContext(BaseModel):

    customer_id: int
    name: str
    tier: str = "basic"
    email: Optional[str] = None

 

그냥 pydantic에서 정의된 BaseModel이다. (타입 힌트를 위해 얘로 썼다) 일반적인 데이터 구조체 정의와 다를 게 없다.

이걸로 인스턴스 하나 따서 Runner에 넘기면 된다.

# main.py
user_account_ctx = UserAccountContext(
    customer_id=1,
    name="larva",
    tier="basic",
)

result = Runner.run_streamed(
    triage_agent,
    transcription,
    session=session,
    context=user_account_ctx,  # 여기!
)

 

여기서 중요한 포인트 하나를 짚고 가야 한다.

 

Context는 모델에 직접 전달되는 게 아니다. 모델이 보는 건 여전히 messages뿐이다. Context는 코드(워크플로우, tool, dynamic instructions 등)에 주입되는 "의존성"이다.
즉, Context에 담긴 정보를 모델이 알게하려면, 코드에서 그 값을 꺼내 프롬프트나 메시지로 변환해서 모델에게 넘겨줘야 한다.
(다음에 나올 dynamic instructions가 정확히 그 역할이다.)

 

실제로 OpenAI SDK에선 @function_tool로 정의한 함수들을 보면 첫 번째 인자로 context를 받는다. (위치는 자유지만 첫번째가관례다)

@function_tool
def process_refund_request(
    context: UserAccountContext, refund_amount: float, reason: str
) -> str:
    """Process a refund request for the customer."""
    processing_days = 3 if context.is_premium_customer() else 5
    refund_id = f"REF-{random.randint(100000, 999999)}"
    # ...

 

모델이 함수를 호출할 때 LLM이 채워주는 인자는 refund_amount와 reason뿐이다. context는 우리가 코드 레벨에서 자동으로 주입해준 것. 의존성 주입(DI)과 정확히 같은 패턴이다.

 

덕분에 tool 코드 안에서 context.is_premium_customer()를 호출해서 premium 사용자에겐 3일 처리, 일반 사용자에겐 5일 처리로 분기할 수 있다. 이걸 모델한테 매번 시키려고 하면 LLM이 헷갈리거나 거짓말을 할 텐데, 그냥 코드로 처리하니 깔끔하다.

---

2. Dynamic Instructions : 컨텍스트를 프롬프트에 녹이기

앞서 컨텍스트는 코드용이라고 했다. 그런데 모델도 사용자 이름이나 등급을 알아야 응대를 할 수 있지 않나? 즉, 모델에게도 컨텍스트를 알려줘야 한다.

def dynamic_triage_agent_instructions(
    wrapper: RunContextWrapper[UserAccountContext],
    agent: Agent[UserAccountContext],  # 같은 함수를 여러 에이전트가 쓸 수 있으니
):
    return f"""
    You are a customer support agent.
    The customer's name is {wrapper.context.name}.
    The customer's email is {wrapper.context.email}.
    The customer's tier is {wrapper.context.tier}.
    
    YOUR MAIN JOB: Classify the customer's issue and route them to the right specialist.
    ...
    """


triage_agent = Agent(
    name="Triage Agent",
    instructions=dynamic_triage_agent_instructions,  # 문자열 대신 함수!
    handoffs=[...],
)

 

지난 포스팅에서는 instructions에 문자열을 넘겼다. 여기선 함수를 넘기고 있다.

왜냐면 문자열은 에이전트 정의 시점에 고정되지만, 함수는 매 실행마다 호출되어 그때그때 instructions를 새로 만든다. 때문에 같은 코드로도 고객별로 다른 instructions가 만들어지는것.

즉, 그때그때 다른 컨텍스트를 instructions에 녹이려면 함수로 정의해야 한다. 에이전트가 실행될 때마다 SDK가 이 함수를 호출해서 그 결과 문자열을 instructions로 쓴다. 덕분에 같은 코드로도 고객별로 다른 instructions가 만들어진다.

# billing_agent의 dynamic instructions 발췌
return f"""
    You are a Billing Support specialist helping {wrapper.context.name}.
    Customer tier: {wrapper.context.tier} {"(Premium Billing Support)" if wrapper.context.tier != "basic" else ""}
    ...
    {"PREMIUM BENEFITS: Fast-track refund processing and flexible payment options available." if wrapper.context.tier != "basic" else ""}
    """

 

basic 고객한테는 PREMIUM BENEFITS 문장 자체가 instructions에 안 들어간다. 모델은 자기가 모르는 혜택을 안내할 수가 없다. 코드로 차단하는 셈.

#참고
Context는 tool에 자동 주입되고, dynamic instructions 함수에도 wrapper로 전달된다. 같은 데이터를 두 군데서 활용할 수 있다는 의미. "코드에 심어두고, 어떻게 활용할지(프롬프트에 녹이기 / tool에서 꺼내쓰기)는 자유" 라고 이해하면 된다.

---

3. Handoffs : 단순 위임에서 한 단계 더

지난 포스팅에서 handoff의 기본은 다뤘다. handoffs=[agent1, agent2]처럼 에이전트 리스트를 넘기는 방식. 그런데 이번엔 좀 더 세밀한 제어가 필요하다.

• handoff가 일어났을 때 어떤 부가 작업을 하고 싶다 (예: 로깅, 알림 발송)
• handoff하면서 다음 에이전트에게 전달할 정보를 구조화하고 싶다
• handoff 이후 다음 에이전트가 보는 대화 이력에서 일부를 걸러내고 싶다

이걸 위해 handoff() 함수를 직접 호출해서 커스터마이즈한다.

from agents import handoff
from agents.extensions import handoff_filters
from models import HandoffData


class HandoffData(BaseModel):
    to_agent_name: str
    issue_type: str
    issue_description: str
    reason: str


def handle_handoff(
    wrapper: RunContextWrapper[UserAccountContext],
    input_data: HandoffData,
):
    # 사이드바에 핸드오프 정보 표시
    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}
        """)


def make_handoff(agent):
    return handoff(
        agent=agent,
        on_handoff=handle_handoff,
        input_type=HandoffData,
        input_filter=handoff_filters.remove_all_tools,
    )


triage_agent = Agent(
    name="Triage Agent",
    instructions=dynamic_triage_agent_instructions,
    handoffs=[
        make_handoff(technical_agent),
        make_handoff(billing_agent),
        make_handoff(account_agent),
        make_handoff(order_agent),
    ],
)

 

한 줄씩 보자.

input_type=HandoffData

handoff가 일어날 때 LLM이 채워주는 인자의 형식을 정한 것. tool calling에서 LLM이 함수 인자를 채우던 것과 똑같은 메커니즘이다. handoff 자체가 결국 일종의 tool call이라고 봐도 되겠다.

덕분에 모델은 그냥 "Technical Agent에게 넘길게"가 아니라 "이유는 X, 이슈 타입은 Y, 설명은 Z"까지 구조화해서 넘기게 된다.

on_handoff=handle_handoff

handoff가 실제로 일어나는 순간 호출되는 콜백 함수다. 여기선 사이드바에 표시했지만, 슬랙 알림을 보내거나 DB에 기록하는 등 뭐든 가능하다.

input_filter=handoff_filters.remove_all_tools

이게 미묘하지만 중요한 옵션이다. 다음 에이전트가 받게 될 대화 이력에서 이전 tool 호출 기록을 모두 제거한다.

왜 필요할까? Triage Agent가 분류만 하다가 Billing Agent로 넘겼다고 치자. Billing Agent 입장에서 Triage Agent가 어떤 tool을 어떻게 호출했는지(만약 있었다면)는 알 필요가 없다. 오히려 노이즈만 된다. 깔끔하게 잘라내는 것.

handoff_filters에는 빌트인 필터가 몇 개 있고, 필요하면 직접 정의할 수도 있다.

 

#참고 : handoff vs as_tool 차이
- handoffs=[agent] : 대화 자체가 다음 에이전트로 넘어간다. 이후 응답은 그 에이전트가 책임짐.
- tools=[agent.as_tool(...)] : 다른 에이전트를 서브루틴처럼 호출한다. 호출이 끝나면 다시 원래 에이전트로 돌아옴.

대화가 길게 이어질 일이면 handoff, 단발성 위임이면 as_tool. 그래서 9장 코드에 주석으로 "이전 연결을 안 끊고 싶으면 tool로 쓰면 됨"이라고 적어둔 거다.

 

전환 후 상태 유지하기

한 가지 더. handoff가 일어났는데, 다음 메시지에서 다시 처음의 Triage Agent로 돌아가버리면 곤란하다. 마지막에 활성화된 에이전트를 기억해야 한다.

async def run(user_input):
    result = Runner.run_streamed(
        st.session_state["agent"],  # 마지막으로 활성화됐던 에이전트로 실행
        user_input,
        session=st.session_state["session"],
        context=user_account_ctx,
    )

    # ... 스트림 처리 ...

    st.session_state["agent"] = result.last_agent  # 다음 호출을 위해 저장

result.last_agent에 마지막 활성 에이전트가 들어 있다. 이걸 세션 상태에 저장해두고 다음 턴에 그 에이전트로 다시 시작하는 패턴이다.

---

4. Input Guardrails : 들어오는 질문 검사

이번 시스템은 고객 지원이 목적이다. 사용자가 "오늘 점심 뭐 먹지?"라고 물어도 답하는 건 비용 낭비고 시스템 목적에 안 맞는다. 들어오는 질문 자체를 거를 수단이 필요하다.

 

OpenAI Agents SDK의 guardrail은 또 다른 에이전트로 구현하고있다. LLM이 LLM을 검사하는 셈

from agents import (
    Agent,
    RunContextWrapper,
    input_guardrail,
    Runner,
    GuardrailFunctionOutput,
)


class InputGuardRailOutput(BaseModel):
    is_off_topic: bool
    reason: str


# 1. 검사 담당 에이전트
input_guardrail_agent = Agent(
    name="Input Guardrail Agent",
    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.
    """,
    output_type=InputGuardRailOutput,  # 구조화된 출력
)


# 2. guardrail 함수: 위 에이전트를 돌려서 tripwire 결정
@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,
    )

    return GuardrailFunctionOutput(
        output_info=result.final_output,                 # 선택적: 부가 정보
        tripwire_triggered=result.final_output.is_off_topic,  # 필수: 차단 여부
    )


# 3. 적용
triage_agent = Agent(
    name="Triage Agent",
    instructions=dynamic_triage_agent_instructions,
    input_guardrails=[off_topic_guardrail],
    handoffs=[...],
)

 

흐름을 정리하면

1. 사용자 입력이 Triage Agent로 들어가기 직전, off_topic_guardrail이 먼저 실행된다.
2. 이 함수가 내부적으로 input_guardrail_agent를 돌려서 "주제에 맞는 요청인지" 판단한다.
3. is_off_topic=True면 tripwire_triggered=True로 신호를 보낸다.
4. SDK는 즉시 InputGuardrailTripwireTriggered 예외를 발생시키고, 본 에이전트는 아예 실행되지 않는다.

이걸 main에서 잡으면 된다.

try:
    result = await pipeline.run(audio)
    # ...
except InputGuardrailTripwireTriggered:
    st.write("I can't help you with that.")

 

본 에이전트가 돌기 전에 차단되는 거니까, 모델 호출이 일어나지 않아 비용도 안 든다. 정책 위반 메시지나 명백히 무관한 질문을 일찍 쳐낼 때 유용하다.  기억하자. 토큰은 생각보다 비싸다!

---

5. Output Guardrails : 나가는 답변 검사

Input과 거의 대칭이다. 이번엔 에이전트가 답변을 생성한 후, 사용자에게 보여주기 전에 검사한다.

만약, Technical Agent가 기술 지원 중에 갑자기 환불 이야기를 꺼낸다면? 각 에이전트는 자기 영역에만 머물러야 한다. 그래서 에이전트별로 특화된 output guardrail을 만들어 두는게 좋다.

# output_guardrails.py
class TechnicalOutputGuardRailOutput(BaseModel):
    contains_off_topic: bool
    contains_billing_data: bool
    contains_account_data: bool
    reason: str


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.
    """,
    output_type=TechnicalOutputGuardRailOutput,
)


@output_guardrail
async def technical_output_guardrail(
    wrapper: RunContextWrapper[UserAccountContext],
    agent: Agent,
    output: str,
):
    result = await Runner.run(
        technical_output_guardrail_agent,
        output,
        context=wrapper.context,
    )
    validation = result.final_output

    triggered = (
        validation.contains_off_topic
        or validation.contains_billing_data
        or validation.contains_account_data
    )

    return GuardrailFunctionOutput(
        output_info=validation,
        tripwire_triggered=triggered,
    )


# technical_agent.py에서 적용
technical_agent = Agent(
    name="Technical Support Agent",
    instructions=dynamic_technical_agent_instructions,
    tools=[...],
    output_guardrails=[technical_output_guardrail],
)

 

예외 잡는 것도 동일하게 진행하면 된다.

except OutputGuardrailTripwireTriggered:
    st.write("Cant show you that answer.")

 

이쪽은 모델이 응답을 다 생성한 뒤에 검사하니까 어쩔수없이 필수 비용이 발생한다. (생성된 토큰 + guardrail 에이전트의 검사 토큰)

그래도 잘못된 답변이 사용자에게 그대로 노출되는 것보단 낫다. 이미 답변이 거의 완성된 상태에서 막는 것이라 "거의 다 만들어 놓고 안 보여줄 수 있다"는게 역할이자 포인트

#참고
Input/Output guardrail 모두 본질은 같다.
또 다른 에이전트로 "이거 통과시켜도 돼?"를 물어보고, 그 결과로 tripwire를 발동시킬지 결정한다.
tripwire가 발동되면 즉시 예외가 던져지고 흐름이 끊긴다. 막을 거면 일찍, 어쩔 수 없이 늦게 막아야 한다면 늦게!

---

6. Hooks : 흐름 전체에 콜백 끼워넣기

여기까지 만들면 흐름이 꽤 복잡해진다.

[Triage가 받았다 → guardrail 통과 → Billing Agent로 handoff → tool 호출 → 응답 생성 → output guardrail 통과 → 사용자에게 표시]

그럼 우리같은 개발자는 슬슬 불안해진다. 중간에 뭐가 어떻게 됐는지 추적하고 싶어진다.

이럴 때 쓰는 게 Hooks다. 에이전트 생애주기의 각 시점에 콜백을 걸 수 있다.

# tools.py
from agents import AgentHooks


class AgentToolUsageLoggingHooks(AgentHooks):

    async def on_start(self, context, agent):
        with st.sidebar:
            st.write(f"🚀 **{agent.name}** activated")

    async def on_end(self, context, agent, output):
        with st.sidebar:
            st.write(f"🏁 **{agent.name}** completed")

    async def on_tool_start(self, context, agent, 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}**")

 

적용은 hooks= 한 줄

technical_agent = Agent(
    name="Technical Support Agent",
    instructions=dynamic_technical_agent_instructions,
    tools=[...],
    hooks=AgentToolUsageLoggingHooks(),   # 여기!
    output_guardrails=[technical_output_guardrail],
)

 

이렇게 해두면 사이드바에 활성화 → tool 호출 → handoff 같은 흐름이 실시간으로 찍힌다.

디버깅 도구로도 쓰지만, 사용자에게 보여주는 진행 상태로도 쓸 수 있다.

 

참고로, 앞서 handoff에서 on_handoff=handle_handoff로 콜백을 따로 등록한 적이 있다. 그런데 Hooks 클래스에도 on_handoff 메서드가 있다. 즉 같은 이벤트를 두 가지 방법으로 잡을 수 있는 셈. 콜백 하나만 박을 거면 전자, 생애주기 전반을 다룰 거면 후자 — 보통은 한 클래스로 묶어서 통째로 등록하는 게 깔끔하다.

---

7. 전체 그림 다시 보기

주요 개념을 다 봤으니 전체 흐름을 다시 정리해보자. (클로드야 고마워)

[사용자 입력]
       │
       ▼
┌─────────────────────────┐
│ Input Guardrails        │ ← LLM으로 "주제 맞는 질문인지" 검사
│  (off_topic_guardrail)  │   tripwire 시 즉시 차단
└─────────┬───────────────┘
          │ pass
          ▼
┌─────────────────────────┐
│ Triage Agent            │ ← dynamic instructions로
│ (분류 담당)             │   고객 이름·등급 반영
└─────────┬───────────────┘
          │ handoff (HandoffData로 구조화 + input_filter)
          ▼
┌─────────────────────────┐
│ Specialist Agent        │ ← Context는 tool에도 자동 주입됨
│ (Billing/Order/Tech/Acc)│   premium 여부로 동작 분기
│  - tools                │
│  - hooks                │ ← on_start, on_tool_start/end 등 콜백
│  - output_guardrails    │ ← 응답 생성 후 검사, tripwire 시 차단
└─────────┬───────────────┘
          │ pass
          ▼
   [사용자에게 응답]
   + last_agent를 세션에 저장해두어
     다음 턴에 같은 specialist 유지

 

겉으로 보면 SDK 기능이 많아 보이지만, 본질은 단순하다.

• Context는 코드 레벨 의존성 주입
• Dynamic Instructions는 그 Context를 프롬프트로 변환하는 함수
• Guardrails는 또 다른 에이전트로 입/출력을 검사하고 tripwire 발동
• Handoffs는 구조화된 데이터(input_type)와 필터(input_filter)를 곁들인 대화 위임
• Hooks는 생애주기 각 지점의 콜백 묶음

---

첫 포스팅에서 정리했던것 처럼 AI Agent의 본질은 결국 "입력 받기 → 메모리 추가 → AI 호출 → 응답 추가"의 while 루프다. 이번 9장에서 본 것들은 그 루프의 각 지점에 검사대를 세우고(guardrails), 분기를 만들고(handoffs), 사전 지식을 주입하고(context), 관측 포인트를 박는(hooks) 작업에 가깝다.

 

프레임워크는 결국 이런 패턴들을 깔끔하게 추상화한 레이어다. 본질이 안 바뀌었다는 것도, 그 위에 얹힌 추상화가 꽤 유용하다는 것도 둘 다 사실인 것 같다.

어떤 프레임워크를 쓰더라도 결국 "Context가 어디로 흐르나, 누가 누구에게 위임하나, 검사는 어디서 일어나나"를 묻게 될 거다.