🤖 이 글은 본격적인 에이전트 제작에 앞서 SDK의 기본 사용법(Agents·Runner·Streaming·Session·Handoff·Tracing)을 정리한 기록입니다.
💡 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번, 여러 개일 경우엔 그 이상 요청이 발생하기 때문에 최종 응답을 받기까지 시간이 걸린다.
사용자 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로 전달하기만 하면 된다.
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 메서드를 사용해야 한다.
**
Streaming**을 활용하면 실제 GPT처럼 실시간으로 답변이 작성되는 모습을 볼 수 있다. **run_streamed**는 말 그대로 agent가 무엇을 하고 있는지 실시간으로 업데이트하는 메서드다.
run 메서드가 전체 완료 후 결과를 반환하는 데 반해, run_streamed 메서드는 이벤트 단위로 실시간 스트리밍한다.
위 코드에서 Runner 부분을 아래와 같이 수정하고 실행하면 이벤트 목록이 나오고, 이 중 type만 필터링해 보면 아래와 같다.
stream = Runner.run_streamed(agent, "Hello, how about the weather in Tokyo?")
async for event in stream.stream_events():
print(event.type)
print("=" * 20)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에서 일어나는 각 단계를 이벤트로 노출한 것이다.
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)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를 추출·포맷팅한다.
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)# 에이전트 업데이트
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 호출 인자를 글자 하나씩 생성하고 있다는 수준)
async for event in stream.stream_events():
if event.type == "raw_response_event":
print(event.data.type)
print("=" * 20)====================
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와 같이 텍스트가 한 글자씩 출력되는 효과를 구현할 수 있다.
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)최종 코드
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 = ""{"
{"city
{"city":"
{"city":"Tokyo
{"city":"Tokyo"}
Tokyo
Tokyo:
Tokyo:
Tokyo: 30
Tokyo: 30 degrees
Tokyo: 30 degrees.💡
SQLiteSession— conversation history를 SQLite database에 저장하는 메모리 역할을 한다.
세션 메모리를 직접 만든다면 배열을 만들고 거기에 사용자 message를 넣고 다시 그 배열을 모델에 전달하는 등의 과정이 필요하지만, OpenAI가 제공하는 session memory를 사용하면 이게 자동으로 수행된다.
인자로는 session_id와 db_path 값을 받는다.
⚠️ session_id가 다르면 다른 세션으로 간주되며, 같은 DB라 할지라도 세션 메모리를 공유하지 않는다.
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란? agent가 다른 agent로 작업을 넘기는 과정을 말한다.
메인 역할을 하는 에이전트에게 다른 서브 에이전트가 있다고 알려주어야, 메인 에이전트가 이들이 어떤 역할을 하는지 확인하고 적절하게 task를 분배할 수 있다. 그리고 서브 에이전트들은 main 에이전트에게 자신이 무엇을 할 수 있는지에 대한 description을 가지고 있어야 한다.
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
]
)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."
)from agents.extensions.visualization import draw_graph
...
draw_graph(main_agent)위 기본 도구를 활용하면 아래와 같이 에이전트를 도식화해 보여준다.

⚠️ 이 기능은 mac에
graphviz가 설치되어 있어야 한다. (brew install graphviz)
그리고 특정 format을 활용해 응답 형태를 강제할 수도 있다.
💡 **
pydantic**은 python의 데이터 검증 라이브러리다. **BaseModel**을 상속받아 클래스를 정의하면, 그 클래스가 “이 데이터는 반드시 이 형태여야 한다”는 스키마가 된다. (pydantic에서 스키마 클래스를 만들 때 상속받는 기반 클래스)
그래서 아래 Answer와 같이 클래스를 정의하면, Answer 객체는 반드시 answer와 background_explanation을 가져야 하며, 타입이 맞지 않으면 Pydantic이 자동으로 에러를 발생시킨다.
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
)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(추적) — OpenAI Agents SDK는 agent로 하는 모든 것을 대시보드에서 확인할 수 있다. (예: Log, tool이 얼마나 걸리는지, 어떤 tool이 호출되는지 등)
OpenAI Platform > 내 프로젝트 > Logs > Agent Traces에서 에이전트 워크플로우를 확인할 수 있다. Hand offs가 몇 번 이루어졌는지, Tools를 몇 개 사용했는지 등을 볼 수 있다. Agent Workflow는 run을 실행할 때마다, 그리고 Agent를 생성할 때마다 생성된다.

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