ストリーミング
ストリーミングを使用すると、エージェント の実行が進むにつれて更新を購読できます。これは、エンドユーザーに進捗更新や部分的な応答を表示するのに役立ちます。
ストリーミングするには、Runner.run_streamed() を呼び出します。これによりRunResultStreaming が返されます。result.stream_events() を呼び出すと、以下で説明するStreamEvent オブジェクトの非同期ストリームが得られます。
raw response イベント
RawResponsesStreamEvent は、LLM から直接渡される raw なイベントです。これは 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())Run item イベントと エージェント イベント
RunItemStreamEvent は、より高レベルなイベントです。これは、アイテムが完全に生成されたタイミングを通知します。これにより、各トークンではなく「メッセージが生成された」「ツールが実行された」などのレベルで進捗更新を行えます。同様に、AgentUpdatedStreamEvent は、現在のエージェント が変更されたとき(例: ハンドオフの結果として)の更新を提供します。
たとえば、次の例では raw イベントを無視し、ユーザーに更新をストリーミングします。
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())