AI Agent : OpenAI Agents SDK

 

지난 포스팅에서 while 루프와 Tool Calling으로 AI Agent의 밑바닥을 직접 구현했다.
이번엔 OpenAI Agents SDK를 써서 그 구조가 어떻게 추상화되는지, 그리고 실시간 스트리밍이 뭔지 알아본다.

---

1. OpenAI Agents SDK : Swarm (이었던것)

지난 포스팅 마지막에 언급했던 OpenAI Agents SDK, 원래 이름은 Swarm이었다. 가볍고 개념이 적고 추상화가 거의 없어서 입문용으로 딱 좋다.

SDK를 쓰면 지난번에 직접 짰던 while 루프, messages 관리, tool 매핑 같은 것들을 대신 처리해준다. 핵심 구성 요소는 두 가지다.

• Agent : 에이전트 정의. 이름, 역할(instructions), 쓸 수 있는 도구(tools)를 설정
• Runner : 에이전트를 실행하는 주체. 지난번에 직접 짰던 while 루프를 대신 돌려준다

from agents import Agent, Runner, function_tool

@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. Use tools when needed to answer questions",
    tools=[get_weather],
)

눈에 띄는 변화가 있다. 지난번엔 TOOLS 딕셔너리를 직접 작성하고 FUNCTION_MAP 으로 매핑했는데, 이제 @function_tool 데코레이터 하나로 끝난다. docstring이 곧 description 역할을 한다.

---

2. Runner 실행 방법 세 가지

에이전트를 실행하는 방법은 세 가지다.

• Runner.run() : async/await 환경에서 사용. 응답이 완성될 때까지 기다렸다가 한 번에 반환
• Runner.run_streamed() : async/await 환경에서 사용. 응답을 실시간으로 받음
• Runner.run_sync() : async 환경이 아닐 때 사용. 일반 .py 파일처럼 이벤트 루프가 없는 환경에서 await 없이 호출 가능

#참고
Jupyter notebook에서 async for가 바로 돌아가는 이유는 Jupyter가 자체 이벤트 루프를 갖고 있어서다. 일반 스크립트에서 돌리면 이벤트 루프가 없으니 run_sync()를 써야 하는 상황이 온다.

---

3. 스트리밍 : 실시간성

보통의 채팅형 서비스에서는 AI가 타이핑하듯 한 글자씩 응답해주는걸 실시간으로 볼 수 있다. 모든 응답이 완성될 때까지 로딩 화면만 보여주면 사용자 경험이 좋지 않을것이다. 그래서인지 실제 서비스에서 AI Agent는 거의 다 이 방식으로 동작한다.

stream = Runner.run_streamed(
    agent, "Hello how are you? What is the weather in the capital of Spain?"
)

async for event in stream.stream_events():
    ...

stream_events()로 실시간 이벤트를 받아서 처리하도록 한다. 핸들링할 이벤트는 세 종류다.

---

raw_response_event

모델이 생성하는 토큰을 가장 날것으로 받는 이벤트다. 여기서 토큰이란 LLM이 텍스트를 생성하는 최소 단위다. 단어일 수도 있고 단어의 일부일 수도 있다. 스트리밍 출력에서 글자가 조각조각 붙어나오는 이유는 바로 이 토큰 단위로 언어 출력이 생성되기 때문이다.

 

현재 토큰 생성 상태를 event.data.type으로 세부 타입을 구분한다.

• response.output_text.delta : 텍스트가 토큰 단위로 생성되는 중
• response.function_call_arguments.delta : tool 인자가 생성되는 중
• response.completed : 응답 완료. 버퍼 초기화할 타이밍

message = ""
args = ""

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":
            args += event.data.delta
            print(args)
        elif event_type == "response.completed":
            message = ""
            args = ""

실제 출력을 보면 이렇게 찍힌다.

{"
{"city
{"city":"
{"city":"Madrid
{"city":"Madrid"}
Hello
Hello!
Hello! I'm
Hello! I'm doing
...
Hello! I'm doing well, thank you. The weather in Madrid is currently 30 degrees Celsius.

 

tool 인자가 토큰 단위로 조각조각 쌓이다가, 완성되면 텍스트 응답이 한 토큰씩 붙어나가는 흐름을 그대로 볼 수 있따.

 

agent_updated_stream_event

현재 어떤 에이전트가 동작 중인지 알려준다. 에이전트가 여러 개일 때 다른 에이전트로 넘어가는 handoff가 일어나면 이 이벤트가 발생한다. 

elif event.type == "agent_updated_stream_event":
    print("Agent updated to", event.new_agent.name)

run_item_stream_event

그래서 어떤 에이전트가 뭘 했는지 행동 단위로 알려준다. event.item.type으로 구분한다.

• tool_call_item : 어떤 tool을 호출했는지
• tool_call_output_item : tool 실행 결과
• message_output_item : 최종 메시지 출력

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))

#참고
raw_response_event가 토큰 단위의 실시간 스트림이라면, run_item_stream_event는 행동 단위의 요약이다. 용도에 따라 골라 쓰면 된다.

 

---

4. Session Memory : 대화를 디스크에 저장하기

지난 포스팅에서 가장 기초적인 Agent 를 구현해보며, 간단한 메모리를 messages 리스트로 직접 관리했다. 하지만 알다시피 그 방식은 프로세스가 종료되면 날아간다. 디스크에 저장하기 위해 SDK에선 SQLiteSession으로 해결한다.

from agents import Agent, Runner, SQLiteSession

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

result = await Runner.run(
    agent,
    "What was my name again?",
    session=session,
)
print(result.final_output)

session_id 와 DB 경로(저장할 대상 파일 경로) 만 넘기면 된다. 해당 session_id 를 기준으로 컨텍스트가 분리된다는 점을 유념하도록 하자. (서로 다른 세션간 데이터 교환하지 않음)

이후 경로를 지정하면 해당 경로 디스크에 저장되고, 지정하지 않으면 메모리에만 올라간다. 삭제, 수정, 메모리 관리도 자동으로 처리해준다. 다른 DB를 쓰고 싶다해도 인터페이스만 맞추면 된다. (다른 서버에 있는 디비나 회사 API를 통하고 싶거나 등등...)

 

세션에 누적된 대화 이력을 session.get_items()로 꺼내보면

[
  {'content': 'Hello how are you? My name is Larva', 'role': 'user'},
  {'role': 'assistant', 'content': 'Hello Larva! How can I assist you today?'},
  {'content': 'I live in Spain', 'role': 'user'},
  {'role': 'assistant', 'content': "That's great! Spain is a beautiful country."},
  {'content': 'What is the weather in the third biggest city of the country i live on', 'role': 'user'},
  {'name': 'get_weather', 'arguments': '{"city":"Valencia"}', 'type': 'function_call'},
  {'output': '30 degrees', 'type': 'function_call_output'},
  {'role': 'assistant', 'content': 'The weather in Valencia is currently 30 degrees.'},
  {'content': 'What was my name again?', 'role': 'user'},
  {'role': 'assistant', 'content': 'Your name is Larva.'}
]

 

지난 포스팅에서 직접 쌓던 그 messages 리스트 구조가 그대로 나온다. 그냥 테이블 처럼 바로 볼수도 있다.

아무튼 좋은건, 귀찮은 메모리 관리를 SDK가 대신 관리해 준다는거다.

---

5. Handoffs : 다른 에이전트에게 넘기기

에이전트가 하나일 때는 모든 걸 혼자 처리했다. 하지만 규모가 커지고 필요한 역할이 많아지면 전문 에이전트들을 선언하여 적합한 일감을  나눠주는 게 낫다. 그럴때 바톤터치 시키는게 handoff다.

geography_agent = Agent(
    name="Geo Expert Agent",
    instructions="You are an 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 an expert in economics, you answer questions related to them.",
    handoff_description="Use this to answer economics related questions.",
)

main_agent = Agent(
    name="Main Agent",
    instructions="You are a helpful assistant.",
    handoffs=[
        economics_agent,
        geography_agent,
    ],
)

#참고
tools는 에이전트가 알아서 실행하지만, handoffs는 대화 자체를 다른 에이전트에게 넘긴다.
handoff_description은 main agent가 "언제 이 에이전트한테 넘길지"를 판단하는 근거가 된다. description을 잘 써야 하는 이유가 여기서도 나온다.
handoff가 일어났는지는 앞서 배운 agent_updated_stream_event로 확인할 수 있다. (어떤 에이전트가 동작중인지 확인)

---

6. Structured Outputs & draw_graph

이건 소소한 팁 같은 것인데, 에이전트가 자유롭게 텍스트로 답하는 게 기본이지만, 특정 포맷으로 응답받고 싶을 때가 있다. pydantic BaseModel로 원하는 구조를 정의하고 output_type으로 넘기면 된다.

from pydantic import BaseModel

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

geography_agent = Agent(
    name="Geo Expert Agent",
    instructions="You are an expert in geography.",
    handoff_description="Use this to answer geography related questions.",
    tools=[get_weather],
    output_type=Answer,
)

 

실제 응답은 아래와 같다. 

Geo Expert Agent
answer="The capital of Thailand's northern province, Chiang Mai, is Chiang Mai City."
background_explanation="Chiang Mai is both a city and a province in northern Thailand..."

 

추가로 에이전트 구조가 복잡해지면 한눈에 보고 싶을 때가 있을 때는 draw_graph로 시각화할 수 있다.

from agents.extensions.visualization import draw_graph

draw_graph(main_agent)

 

main agent → economics agent, geo agent → get_weather tool 흐름이 그래프로 그려진다.

---

7. Tracing : OpenAI 콘솔에서 워크플로우 보기

에이전트가 어떤 흐름으로 동작했는지 OpenAI 홈페이지 콘솔의 트레이싱 페이지에서 확인할 수 있다. 별도 설정 없이도 자동으로 기록된다. 특정 사용자의 트레이스만 모아서 보고 싶다면 trace()로 묶어주면 된다:

from agents import trace

with trace("user_1_session"):
    result = await Runner.run(agent, message, session=session)

 

이 또한 취향껏 쓰면 되는 기능이다.

---

 

직접 다 짰던 것들이 OpenAI Agents SDK에선 이렇게 기능으로 제공하여 추상화될 수 있고, 그또한 까보면 본질은 달라진 게 없다.

 

그래도 처음보다도 쌩짜로 안짜도 되는게 많아졌다. 그러나 이친구는 추상화 수준이 많이 낮은 편이고,
앞으로 공부해볼 도구들은 꽤나 복잡하고 심도있는 작업도 프레임워크에서 해줄 예정이다.