Streaming

Streaming (스트리밍)

스트리밍을 쓰면 에이전트 실행이 진행되는 동안 그 업데이트를 구독할 수 있어요. 최종 사용자에게 진행 상황 업데이트나 부분 응답을 보여주는 데 유용하죠. 스트리밍하려면 Runner.run_streamed()을 호출하면 되고, 그러면 RunResultStreaming을 받아요. result.stream_events()를 호출하면 아래에서 설명할 StreamEvent 객체의 async 스트림이 나와요.

출처: 문서

본문

async 반복자가 끝날 때까지 result.stream_events()를 계속 소비하세요. 스트리밍 실행은 반복자가 끝나야 완료돼요. 세션 영속화·승인 기록·기록 압축 같은 후처리는 마지막 보이는 토큰이 도착한 뒤에 끝날 수 있어요. 루프가 끝나면 result.is_complete가 최종 실행 상태를 반영해요.

원시 응답 이벤트

RawResponsesStreamEvent 객체는 LLM에서 직접 전달된 원시 이벤트를 감싸요. 각 객체의 data 필드에는 response.created나 response.output_text.delta 같은 타입의 OpenAI Responses API 이벤트가 들어 있어요. 이 이벤트들은 응답 메시지를 생성되는 즉시 사용자에게 스트리밍하고 싶을 때 유용해요.

컴퓨터 도구 원시 이벤트는 저장된 결과와 같은 preview-vs-GA 구분을 유지해요. Preview 흐름은 computer_call 항목을 하나의 action으로 스트리밍하고, gpt-5.5는 배치된 actions[]로 computer_call 항목을 스트리밍할 수 있어요. 더 높은 수준의 RunItemStreamEvent 표면은 이것을 위한 컴퓨터 전용 이벤트 이름을 추가하지 않아요. 두 형태 모두 tool_called로 표시되고, 스크린샷 결과는 computer_call_output 항목을 감싼 tool_output으로 돌아와요.

예를 들어 다음 코드는 LLM이 생성한 텍스트를 토큰 단위로 출력해요.

import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

async def main():
    agent = Agent(
        name="Joker",
        instructions="You are a helpful assistant.",
    )

    result = Runner.run_streamed(agent, input="Please tell me 5 jokes.")
    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)

if __name__ == "__main__":
    asyncio.run(main())

스트리밍과 승인

스트리밍은 도구 승인을 위해 일시 중지되는 실행과 호환돼요. 도구가 승인을 요구하면 result.stream_events()가 끝나고, 대기 중인 승인은 RunResultStreaming.interruptions에 노출돼요. 결과를 result.to_state()로 RunState로 바꾸고, interruption을 승인하거나 거부한 다음 Runner.run_streamed(...)로 재개하면 돼요.

result = Runner.run_streamed(agent, "Delete temporary files if they are no longer needed.")
async for _event in result.stream_events():
    pass

if result.interruptions:
    state = result.to_state()
    for interruption in result.interruptions:
        state.approve(interruption)
    result = Runner.run_streamed(agent, state)
    async for _event in result.stream_events():
        pass

전체 일시 중지/재개 워크스루는 human-in-the-loop 가이드를 참고하세요.

현재 턴이 끝난 뒤 스트리밍 취소

실행 중에 스트리밍 실행을 멈춰야 한다면 result.cancel()을 호출하세요. 기본적으로 실행을 즉시 중지해요. 현재 턴을 깔끔하게 끝내고 멈추려면 result.cancel(mode="after_turn")을 대신 쓰세요.

스트리밍 실행은 result.stream_events()가 끝나야 완료돼요. 마지막 보이는 토큰이 온 뒤에도 SDK가 세션 항목을 영속화하거나 승인 상태를 마무리하거나 기록을 압축하고 있을 수 있어요.

result.to_input_list(mode="normalized")로 수동으로 이어가고 있고, cancel(mode="after_turn")이 도구 턴 뒤에 멈췄다면, 그 정규화된 입력으로 result.last_agent를 다시 실행해 미완료 기존 사용자 턴을 이어가세요. 새 사용자 턴을 바로 추가하지 말고요.

• 미완료 실행이 재개되기 전에 새 사용자 입력이 도착하면, 배출된(drained) 결과를 result.to_state()로 바꾸고 state.add_input(...)을 호출한 뒤 상태에서 재개하세요. 러너는 다음 모델 호출 직전에 스테이징된 입력을 즉시 받아들여요. 자세한 내용은 Add input before resuming을 참고하세요.
• 스트리밍 실행이 도구 승인 때문에 멈췄다면 그걸 새 턴으로 취급하지 마세요. 스트림을 끝까지 배출하고 result.interruptions를 조사한 뒤 result.to_state()에서 재개하세요.
• RunConfig.session_input_callback을 쓰면 다음 모델 호출 전에 검색된 세션 기록과 새 사용자 입력이 어떻게 병합되는지 커스터마이즈할 수 있어요. 거기서 새 턴 항목을 다시 쓰면, 그 다시 쓴 버전이 해당 턴에 영속돼요.

Run item 이벤트와 에이전트 이벤트

RunItemStreamEvent는 더 높은 수준의 이벤트예요. 항목이 완전히 생성됐을 때 알려주죠. 이것으로 토큰 단위가 아니라 "메시지 생성됨", "도구 실행됨" 수준에서 진행 업데이트를 밀어낼 수 있어요. 마찬가지로 AgentUpdatedStreamEvent는 현재 에이전트가 바뀔 때(예: handoff 결과) 업데이트를 줘요.

Run item 이벤트 이름

RunItemStreamEvent.name은 고정된 의미론적 이벤트 이름 집합을 써요.

• message_output_created
• handoff_requested
• handoff_occured
• tool_called
• tool_search_called
• tool_search_output_created
• tool_output
• reasoning_item_created
• mcp_approval_requested
• mcp_approval_response
• mcp_list_tools

handoff_occured는 역호환성을 위해 일부러 오타로 남겨둔 거예요.

handoff 호출은 handoff_requested로만 방출되고, 동시에 tool_called로는 방출되지 않아요. 같은 턴의 일반 function tool 호출은 여전히 tool_called를 방출해요.

호스팅 tool search를 쓸 때, 모델이 tool-search 요청을 내면 tool_search_called가 방출되고 Responses API가 로드된 부분집합을 돌려주면 tool_search_output_created가 방출돼요.

Programmatic Tool Calling에서 tool_called는 생성된 프로그램과 프로그램이 소유한 일반 자식 도구 호출에 대해 방출돼요. tool_output은 자식 도구 출력과 생성된 프로그램과 일치하는 program_output에 대해 방출돼요. 프로그램이 소유한 호스팅 MCP mcp_approval_request·mcp_list_tools 항목은 예외인데, 각각 MCPApprovalRequestItem·MCPListToolsItem을 감싼 mcp_approval_requested·mcp_list_tools로 방출돼요. 나머지 항목을 구분하려면 원시 항목의 type을 조사하세요. 프로그램이 소유한 자식 호출은 type이 program인 caller를 지니고, 그 caller ID가 부모 프로그램을 식별해요.

예를 들어 다음 코드는 원시 이벤트를 무시하고 사용자에게 업데이트를 스트리밍해요.

import asyncio
import random
from agents import Agent, ItemHelpers, Runner
from agents.decorators import tool

@tool
def how_many_jokes() -> int:
    return random.randint(1, 10)

async def main():
    agent = Agent(
        name="Joker",
        instructions="First call the `how_many_jokes` tool, then tell that many jokes.",
        tools=[how_many_jokes],
    )

    result = Runner.run_streamed(
        agent,
        input="Hello",
    )
    print("=== Run starting ===")

    async for event in result.stream_events():
        # We'll ignore the raw responses event deltas
        if event.type == "raw_response_event":
            continue
        # When the agent updates, print that
        elif event.type == "agent_updated_stream_event":
            print(f"Agent updated: {event.new_agent.name}")
            continue
        # When items are generated, print them
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("-- Tool was called")
            elif event.item.type == "tool_call_output_item":
                print(f"-- Tool output: {event.item.output}")
            elif event.item.type == "message_output_item":
                print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}")
            else:
                pass  # Ignore other event types

    print("=== Run complete ===")

if __name__ == "__main__":
    asyncio.run(main())

더 알아보기 (Learn more)

• OpenAI Agents SDK 문서에서 더 많은 가이드를 확인하세요.