스트리밍

스트리밍을 사용하면 에이전트 실행이 진행되는 동안 업데이트를 구독할 수 있습니다. 최종 사용자에게 진행 상황 업데이트와 부분 응답을 표시할 때 유용합니다.

스트리밍하려면 Runner.run_streamed()을 호출하면 됩니다. 이 메서드는 RunResultStreaming을 반환합니다. result.stream_events()를 호출하면 아래에서 설명하는 StreamEvent 객체의 비동기 스트림을 가져올 수 있습니다.

비동기 이터레이터가 완료될 때까지 result.stream_events()를 계속 소비해야 합니다. 스트리밍 실행은 이터레이터가 끝날 때까지 완료되지 않으며, 세션 영속화, 승인 관련 기록 처리 또는 기록 압축과 같은 후처리는 마지막으로 표시되는 토큰이 도착한 후에도 계속될 수 있습니다. 루프가 종료되면 result.is_complete에 최종 실행 상태가 반영됩니다.

Raw 응답 이벤트

RawResponsesStreamEvent 객체는 LLM에서 직접 전달된 raw 이벤트를 래핑합니다. 각 객체의 data 필드에는 response.created 또는 response.output_text.delta와 같은 유형의 OpenAI Responses API 이벤트가 포함됩니다. 이러한 이벤트는 응답 메시지가 생성되는 즉시 사용자에게 스트리밍하려는 경우 유용합니다.

컴퓨터 도구의 raw 이벤트는 저장된 결과와 동일하게 프리뷰와 GA를 구분합니다. 프리뷰 흐름은 하나의 action이 포함된 computer_call 항목을 스트리밍하는 반면, 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로 변환하고, 인터럽션(중단 처리)을 승인하거나 거부한 다음 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

전체 일시 중지 및 재개 과정은 휴먼인더루프 가이드를 참조하세요.

현재 턴 완료 후 스트리밍 취소

스트리밍 실행을 도중에 중지해야 한다면 result.cancel()을 호출하세요. 기본적으로 실행이 즉시 중지됩니다. 중지하기 전에 현재 턴이 정상적으로 완료되도록 하려면 대신 result.cancel(mode="after_turn")를 호출하세요.

스트리밍 실행은 result.stream_events()가 완료될 때까지 완료된 것이 아닙니다. 마지막으로 표시되는 토큰 이후에도 SDK가 세션 항목을 영속화하거나, 승인 상태를 확정하거나, 기록을 압축하고 있을 수 있습니다.

result.to_input_list(mode="normalized")에서 수동으로 실행을 이어가고 있으며 cancel(mode="after_turn")가 도구 턴 이후 중지되는 경우, 즉시 새 사용자 턴을 추가하지 말고 정규화된 입력으로 result.last_agent을 다시 실행하여 완료되지 않은 기존 사용자 턴을 이어가세요.

완료되지 않은 실행이 재개되기 전에 새 사용자 입력이 도착하면 소진된 결과를 result.to_state()으로 변환하고 state.add_input(...)을 호출한 다음 해당 상태에서 재개하세요. 러너는 다음 모델 호출 직전에 준비된 입력을 반영합니다. 재개 전 입력 추가를 참조하세요.
스트리밍 실행이 도구 승인을 위해 중지된 경우 이를 새 턴으로 취급하지 마세요. 스트림을 끝까지 소진하고 result.interruptions를 확인한 다음 result.to_state()에서 재개하세요.
가져온 세션 기록과 새 사용자 입력이 다음 모델 호출 전에 병합되는 방식을 사용자 지정하려면 RunConfig.session_input_callback을 사용하세요. 여기에서 새 턴 항목을 다시 작성하면 다시 작성된 버전이 해당 턴에 영속화됩니다.

실행 항목 이벤트와 에이전트 이벤트

RunItemStreamEvent는 상위 수준 이벤트입니다. 항목이 완전히 생성되면 이를 알려줍니다. 따라서 각 토큰이 아니라 "메시지 생성됨", "도구 실행됨" 등의 수준에서 진행 상황 업데이트를 전달할 수 있습니다. 마찬가지로 AgentUpdatedStreamEvent는 현재 에이전트가 변경될 때 업데이트를 제공합니다(예: 핸드오프의 결과로 변경되는 경우).

실행 항목 이벤트 이름

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_requested으로만 내보내지며 tool_called로 중복해서 내보내지지 않습니다. 동일한 턴의 일반 함수 도구 호출은 계속 tool_called를 내보냅니다.

호스티드 툴 검색을 사용할 때 모델이 도구 검색 요청을 보내면 tool_search_called이 내보내지고, Responses API가 로드된 하위 집합을 반환하면 tool_search_output_created가 내보내집니다.

프로그래매틱 도구 호출(Programmatic Tool Calling)을 사용하면 생성된 program과 프로그램이 소유한 일반 하위 도구 호출에 대해 tool_called가 내보내집니다. 하위 도구 출력과 생성된 program에 대응하는 program_output에 대해서는 tool_output이 내보내집니다. 프로그램이 소유한 호스티드 MCP mcp_approval_request 및 mcp_list_tools 항목은 예외입니다. 각각 MCPApprovalRequestItem와 MCPListToolsItem을 래핑하는 mcp_approval_requested 및 mcp_list_tools으로 내보내집니다. 나머지 항목을 구분하려면 raw 항목의 type을 확인하세요. 프로그램이 소유한 하위 호출에는 유형이 program이고 호출자 ID가 상위 프로그램을 식별하는 caller도 포함됩니다.

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

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