스트리밍
스트리밍을 사용하면 에이전트 실행이 진행되는 동안 업데이트를 구독할 수 있습니다. 이는 최종 사용자에게 진행 상황 업데이트와 부분 응답을 보여줄 때 유용합니다.
스트리밍하려면Runner.run_streamed()를 호출하여RunResultStreaming을 받을 수 있습니다.result.stream_events()를 호출하면 아래에 설명된StreamEvent 객체의 비동기 스트림을 제공합니다.
원문 응답 이벤트
RawResponsesStreamEvent는 LLM에서 직접 전달되는 원문 이벤트입니다. 이들은 OpenAI Responses API 형식이며, 각 이벤트는 유형(예:response.created,response.output_text.delta 등)과 데이터를 가집니다. 이 이벤트는 생성되는 즉시 사용자에게 응답 메시지를 스트리밍하려는 경우에 유용합니다.
예를 들어, 다음은 LLM이 생성한 텍스트를 토큰 단위로 출력합니다.
importasynciofromopenai.types.responsesimportResponseTextDeltaEventfromagentsimportAgent,Runnerasyncdefmain():agent=Agent(name="Joker",instructions="You are a helpful assistant.",)result=Runner.run_streamed(agent,input="Please tell me 5 jokes.")asyncforeventinresult.stream_events():ifevent.type=="raw_response_event"andisinstance(event.data,ResponseTextDeltaEvent):print(event.data.delta,end="",flush=True)if__name__=="__main__":asyncio.run(main())실행 항목 이벤트 및 에이전트 이벤트
RunItemStreamEvent는 상위 수준 이벤트입니다. 항목이 완전히 생성되었을 때 알려줍니다. 이를 통해 각 토큰 대신 "메시지 생성됨", "도구 실행됨" 등의 수준에서 진행 상황 업데이트를 보낼 수 있습니다. 마찬가지로,AgentUpdatedStreamEvent는 현재 에이전트가 변경될 때(예: 핸드오프의 결과로) 업데이트를 제공합니다.
예를 들어, 다음은 원문 이벤트를 무시하고 사용자에게 업데이트를 스트리밍합니다.
importasyncioimportrandomfromagentsimportAgent,ItemHelpers,Runner,function_tool@function_tooldefhow_many_jokes()->int:returnrandom.randint(1,10)asyncdefmain():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 ===")asyncforeventinresult.stream_events():# We'll ignore the raw responses event deltasifevent.type=="raw_response_event":continue# When the agent updates, print thatelifevent.type=="agent_updated_stream_event":print(f"Agent updated:{event.new_agent.name}")continue# When items are generated, print themelifevent.type=="run_item_stream_event":ifevent.item.type=="tool_call_item":print("-- Tool was called")elifevent.item.type=="tool_call_output_item":print(f"-- Tool output:{event.item.output}")elifevent.item.type=="message_output_item":print(f"-- Message output:\n{ItemHelpers.text_message_output(event.item)}")else:pass# Ignore other event typesprint("=== Run complete ===")if__name__=="__main__":asyncio.run(main())