스트리밍
스트리밍을 사용하면 에이전트 실행이 진행되는 동안 업데이트를 구독할 수 있습니다. 이는 최종 사용자에게 진행 상태 업데이트와 부분 응답을 보여주는 데 유용합니다.
스트리밍을 사용하려면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())