logo
홈블로그소개
3,994

Built with Next.js, Bun, Tailwind CSS and Shadcn/UI

·개인정보처리방침
AIPython

OpenAI Agents SDK 기초 - Runner, Streaming, Session, Handoff, Tracing

Toma
2026년 6월 10일
약 32분
목차
🚀 OpenAI Agent SDK Intro
🧩 Agents and Runners
🌊 Stream Events
🧠 Session Memory
🤝 Hand Off
📊 Viz and Structured Outputs
🔍 Tracing
이전 포스트밑바닥부터 만드는 AI 에이전트 - OpenAI API 만으로 메모리, 도구 호출 구현하기
다음 포스트Streamlit로 만드는 에이전트 UI - 위젯과 rerun, session_state 데이터 흐름

목차

🚀 OpenAI Agent SDK Intro
🧩 Agents and Runners
🌊 Stream Events
🧠 Session Memory
🤝 Hand Off
📊 Viz and Structured Outputs
🔍 Tracing

🤖 이 글은 본격적인 에이전트 제작에 앞서 SDK의 기본 사용법(Agents·Runner·Streaming·Session·Handoff·Tracing)을 정리한 기록입니다.


🚀 OpenAI Agent SDK Intro

🧩 Agents and Runners

💡 SDK란? 개발자를 위한 도구 모음으로, 여러 개의 레고블럭(primitives, 기본요소)을 제공하면 사용자가 이를 조립해 사용한다. (예: Agents, Handoffs, Guardrails, Sessions 같은 기본요소)

💡 **Runner**란? While True Loop를 처리하는 존재다. 기본적으로 agent와 input을 받아 우리 대신 while true loop를 처리해준다.

즉, Runner가 input과 main agent를 받아

OpenAI 모델에 request를 보내면 OpenAI 모델이 응답하고

Runner는 그 response를 추출해 parsing한 뒤 호출해야 할 tool이 있는지 확인한다.

Runner가 그 tool들을 호출하고 그 결과를 다시 OpenAI 모델에 보낸다.

그리고 Runner가 final Response를 받아 우리에게 보낸다.

이것이 Runner의 역할 중 하나이자, Runner가 필요한 이유다.

그러므로Runner 가 while-loop를 돌면서 agent가 tool 호출이 필요하다고 판단할 때마다 API 요청을 반복하는데 만약 tool이 1개면 최소 2번, 여러 개일 경우엔 그 이상 요청이 발생하기 때문에 최종 응답을 받기까지 시간이 걸린다.

python
사용자 Input
    ↓
Runner → 사용자 Input을 가지고 OpenAI API 요청 (1번째)
    ↓
OpenAI → "get_weather tool 써야겠어"
    ↓
Runner → get_weather 실행
    ↓
Runner → tool 결과 포함해서 OpenAI API 요청 (2번째)
    ↓
OpenAI → "get_forecast tool도 써야겠어"
    ↓
Runner → get_forecast 실행
    ↓
Runner → tool 결과 포함해서 OpenAI API 요청 (3번째)
    ↓
OpenAI → Final Response 생성
    ↓
Runner → 사용자에게 반환

Runner.run()으로 Runner를 실행하며(Runner 인스턴스를 실행하는 클래스 메서드 run), 이때 에이전트와 input을 인자로 넘긴다.

그리고 에이전트에게 tool을 주고 싶다면 아래와 같이 @function_tool 어노테이션으로 함수를 정의하고 Agent에 tools로 전달하기만 하면 된다.

python
from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str):
    """Get weather by city"""
    print(city)
    return "30 degrees"

agent = Agent(
    name="Assistant Agent",
    instructions="You are a helpful assistant that can answer questions and help with tasks.",
    tools=[get_weather],
)

result = await Runner.run(agent, "Hello, how about the weather in Tokyo?")
print(result.final_output)

그러나 일반적으로 ChatGPT 같은 에이전트의 응답은 실시간으로 tool을 호출·실행하며, 사용자에게 보이는 응답도 실시간으로 작성된다. 이와 동일하게 하려면 Runner.run_streamed 메서드를 사용해야 한다.

🌊 Stream Events

**Streaming**을 활용하면 실제 GPT처럼 실시간으로 답변이 작성되는 모습을 볼 수 있다. **run_streamed**는 말 그대로 agent가 무엇을 하고 있는지 실시간으로 업데이트하는 메서드다.

run 메서드가 전체 완료 후 결과를 반환하는 데 반해, run_streamed 메서드는 이벤트 단위로 실시간 스트리밍한다.

위 코드에서 Runner 부분을 아래와 같이 수정하고 실행하면 이벤트 목록이 나오고, 이 중 type만 필터링해 보면 아래와 같다.

python
stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")

async for event in stream.stream_events():
    print(event.type)
    print("=" * 20)
python
agent_updated_stream_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
raw_response_event
====================
run_item_stream_event
====================
raw_response_event
...
raw_response_event
====================
run_item_stream_event
====================

에이전트가 다른 에이전트에게 작업을 전달할 때(handoff) agent_updated_stream_event가 발생한다. 즉, 대화의 주도권이 A 에이전트에서 B 에이전트로 넘어갔을 때, 또는 스트림 시작 시 현재 활성 에이전트가 설정될 때 발생하는 이벤트다.

run_item_stream_event는 agent가 취한 액션을 의미한다. 즉, Runner의 While-Loop에서 일어나는 각 단계를 이벤트로 노출한 것이다.

python
stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")

async for event in stream.stream_events():
    if event.type == "raw_response_event":
        continue
    elif event.type == "agent_updated_stream_event":
        print("Agent updated to", event.new_agent.name)
    elif event.type == "run_item_stream_event":
        print(event.item.type)
    print("=" * 20)
python
Agent updated to Assistant Agent  # <- Agent()로 만들었던 에이전트가 업데이트
====================
tool_call_item  # <- agent가 tool을 호출했다는 알림  
====================
tool_call_output_item  # <- tool이 output을 주었다는 알림
====================
message_output_item  # <- agent가 우리에게 message를 주었다는 알림
====================

ItemHelpers는 response에서 text를 추출·포맷팅한다.

python
from agents import Agent, Runner, function_tool, ItemHelpers

@function_tool
def get_weather(city: str):
    """Get weather by city"""
    # print(city)
    return "30 degrees"

agent = Agent(
    name="Assistant Agent",
    instructions="You are a helpful assistant that can answer questions and help with tasks.",
    tools=[get_weather],
)

stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")

async for event in stream.stream_events():
    if event.type == "raw_response_event":
        continue
    elif event.type == "agent_updated_stream_event":
        print("Agent updated to", event.new_agent.name)
    elif event.type == "run_item_stream_event":
        if event.item.type == "tool_call_item":
            print(event.item.raw_item.to_dict())
        elif event.item.type == "tool_call_output_item":
            print(event.item.output)
        elif event.item.type == "message_output_item":
            print(ItemHelpers.text_message_output(event.item))
    print("=" * 20)
python
# 에이전트 업데이트
Agent updated to Assistant Agent
====================
# function 호출 시 argument 등
{'arguments': '{"city":"Tokyo"}', 'call_id': 'call_4teEMh7LMEUPjwZw49f0Z26u', 'name': 'get_weather', 'type': 'function_call', 'id': 'fc_0259facd41af4d67006a2ab613d1c4819aa5f8c5f7d2417299', 'status': 'completed'}
====================
# tool call output item -> tool의 response
30 degrees
====================
# final response
Tokyo: 30 degrees.
====================

위의 방식으로 실행한 에이전트도 꽤나 실시간이지만 좀 더 실시간으로 agent가 어떤 step을 밟는지 정확히 볼 수 있다.


이를 위해서는 raw_response_event를 사용하면 된다. raw_response_event는 agent에게 무슨 일이 일어나고 있는지를 더 자세히 실시간으로 listen할 수 있는, run_item_stream_event보다 더 낮은 수준의 이벤트다. (예: tool 호출 인자를 글자 하나씩 생성하고 있다는 수준)

python
async for event in stream.stream_events():
    if event.type == "raw_response_event":
        print(event.data.type)
    print("=" * 20)
python
====================
response.created
====================
response.in_progress
====================
response.output_item.added
====================
response.function_call_arguments.delta
====================
response.function_call_arguments.delta
====================
response.function_call_arguments.delta
====================
response.function_call_arguments.delta
====================
response.function_call_arguments.delta
====================
response.function_call_arguments.done
====================
response.output_item.done
====================
====================
response.completed
====================
====================
response.created
====================
response.in_progress
====================
response.output_item.added
====================
response.content_part.added
====================
response.output_text.delta
====================
response.output_text.delta
====================
response.output_text.delta
====================
response.output_text.delta
====================
response.output_text.delta
====================
response.output_text.delta
====================
response.output_text.done
====================
response.content_part.done
====================
response.output_item.done
====================
response.completed
====================
====================

여기서 delta는 에이전트가 텍스트를 조금씩 작성하고 있는 것을 표현한 것이다.

raw_response_event 안의 response.output_text.delta를 사용하면 실제 GPT와 같이 텍스트가 한 글자씩 출력되는 효과를 구현할 수 있다.

python
stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")

async for event in stream.stream_events():
    if event.type == "raw_response_event":
        event_type = event.data.type
        if event_type == "response.output_text.delta":
            print(event.data.delta)
    print("=" * 20)

최종 코드

python
stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")

message = ""
arg = ""

async for event in stream.stream_events():
    if event.type == "raw_response_event":
        event_type = event.data.type
        if event_type == "response.output_text.delta":
            message += event.data.delta
            print(message)
        elif event_type == "response.function_call_arguments.delta":
            arg += event.data.delta
            print(arg)
        elif event_type == "response.completed":
            message = ""
            arg = ""
python
{"
{"city
{"city":"
{"city":"Tokyo
{"city":"Tokyo"}
Tokyo
Tokyo:
Tokyo: 
Tokyo: 30
Tokyo: 30 degrees
Tokyo: 30 degrees.

🧠 Session Memory

💡 SQLiteSession — conversation history를 SQLite database에 저장하는 메모리 역할을 한다.

세션 메모리를 직접 만든다면 배열을 만들고 거기에 사용자 message를 넣고 다시 그 배열을 모델에 전달하는 등의 과정이 필요하지만, OpenAI가 제공하는 session memory를 사용하면 이게 자동으로 수행된다.

인자로는 session_id와 db_path 값을 받는다.

⚠️ session_id가 다르면 다른 세션으로 간주되며, 같은 DB라 할지라도 세션 메모리를 공유하지 않는다.

python
from agents import Agent, Runner, function_tool, SQLiteSession

session = SQLiteSession("user_1", "ai-memory.db")

@function_tool
def get_weather(city: str):
    """Get weather by city"""
    return "30 degrees"

agent = Agent(
    name="Assistant Agent",
    instructions="You are a helpful assistant that can answer questions and help with tasks.",
    tools=[get_weather],
)

result = await Runner.run(
    agent,
    "where is my location?",
    session=session
)

print(result.final_output)

# await session.clear_session() user_1 세션 초기화
# await session.add_items([{"role" : "user", "content": "Hello"}]) 세션에 내용 추가
# await session.pop_item() 마지막 item 삭제

기본 제공 Session 외에 기존 DB나 다른 것을 사용하고 싶다면, Session Class 규약에 맞춰 Class를 구현해 사용하면 된다. (공식문서 제공)

🤝 Hand Off

💡 Hand Off란? agent가 다른 agent로 작업을 넘기는 과정을 말한다.

메인 역할을 하는 에이전트에게 다른 서브 에이전트가 있다고 알려주어야, 메인 에이전트가 이들이 어떤 역할을 하는지 확인하고 적절하게 task를 분배할 수 있다. 그리고 서브 에이전트들은 main 에이전트에게 자신이 무엇을 할 수 있는지에 대한 description을 가지고 있어야 한다.

python
main_agent = Agent(
    name="Main_Agent",
    instructions="You are user facing agent. Transfer to the agent most capable of answering the user's question.",
    handoffs=[
        economics_agent,
        geography_agent
    ]
)
python
geography_agent = Agent(
    name="Geo_Expert_Agent",
    instructions="You are a expert in geography, you answer questions related to them.",
    handoff_description="Use this to answer geography related questions."
)

economics_agent = Agent(
    name="Economics_Expert_Agent",
    instructions="You are a expert in economics, you answer questions related to them.",
    handoff_description="Use this to answer economics related questions."
)

📊 Viz and Structured Outputs

python
from agents.extensions.visualization import draw_graph

...

draw_graph(main_agent)

위 기본 도구를 활용하면 아래와 같이 에이전트를 도식화해 보여준다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-06-14_10.11.30.png

⚠️ 이 기능은 mac에 graphviz가 설치되어 있어야 한다. (brew install graphviz)

그리고 특정 format을 활용해 응답 형태를 강제할 수도 있다.

💡 **pydantic**은 python의 데이터 검증 라이브러리다. **BaseModel**을 상속받아 클래스를 정의하면, 그 클래스가 “이 데이터는 반드시 이 형태여야 한다”는 스키마가 된다. (pydantic에서 스키마 클래스를 만들 때 상속받는 기반 클래스)

그래서 아래 Answer와 같이 클래스를 정의하면, Answer 객체는 반드시 answer와 background_explanation을 가져야 하며, 타입이 맞지 않으면 Pydantic이 자동으로 에러를 발생시킨다.

python
from pydantic import BaseModel

class Answer(BaseModel):
    answer: str
    background_explanation: str


geography_agent = Agent(
    name="Geo Expert Agent",
    instructions="You are a expert in geography, you answer questions related to them.",
    handoff_description="Use this to answer geography related questions.",
    tools=[get_weather],
    output_type=Answer
)
python
answer='Yokohama is generally considered the second biggest city in Japan, after Tokyo.' background_explanation='By population, Yokohama is Japan’s second-largest city and part of the Greater Tokyo Area.'

🔍 Tracing

💡 Tracing(추적) — OpenAI Agents SDK는 agent로 하는 모든 것을 대시보드에서 확인할 수 있다. (예: Log, tool이 얼마나 걸리는지, 어떤 tool이 호출되는지 등)

OpenAI Platform > 내 프로젝트 > Logs > Agent Traces에서 에이전트 워크플로우를 확인할 수 있다. Hand offs가 몇 번 이루어졌는지, Tools를 몇 개 사용했는지 등을 볼 수 있다. Agent Workflow는 run을 실행할 때마다, 그리고 Agent를 생성할 때마다 생성된다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-06-14_10.29.43.png

특정 사용자의 여러 run을 하나의 Trace로 추적하고 싶을 수도 있다. 이런 경우 Runner.run()을 특정 Trace로 호출하면, 모든 run()이 같은 Trace에 들어가도록 할 수 있다.

python
from agents import Agent, Runner, trace

with trace("user_1"):
    result = await Runner.run(
        main_agent,
        "What is the second biggest city in Japan",
        session=session
    )

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-06-14_10.40.33.png