openai-agents-python 流式傳輸完全指南從原始事件到智能體更新的訂閱機制【免費下載鏈接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows項目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python流式傳輸Streaming允許你訂閱智能體運行過程中的實時更新是把智能體的部分響應與進度推送給最終用戶的核心手段。本文基于openai-agents-python的Runner.run_streamed()與RunResultStreaming系統講解三類流式事件、審批暫停恢復、運行取消以及事件消費的完整生命周期并給出可直接運行的代碼示例與源碼級原理分析。概述流式運行的三步流程流式傳輸的使用可以歸納為三個固定步驟調用Runner.run_streamed()啟動一次流式運行它返回一個RunResultStreaming對象調用result.stream_events()獲得由StreamEvent對象組成的異步流持續消費result.stream_events()直到異步迭代器結束運行才算真正完成。從源碼看StreamEvent是一個類型別名由三種具體事件聯合而成StreamEvent: TypeAlias RawResponsesStreamEvent | RunItemStreamEvent | AgentUpdatedStreamEvent因此你在消費事件流時實際上可能收到三種類型的事件后續章節會逐一展開。關于“運行完成”有一個容易誤解的細節只有迭代器結束后流式運行才算完成。會話持久化、審批記錄維護、歷史壓縮等后處理可能會在最后一個可見 token 到達之后才完成。當循環退出時result.is_complete會反映最終的運行狀態。這一點在stream_events()的實現中也有體現——迭代器在退出前會等待后臺run_loop_task完全落定再執行_check_errors()復查異常見 src/agents/result.py。原始響應事件RawResponsesStreamEventRawResponsesStreamEvent封裝了直接從 LLM 傳遞的原始事件。每個對象的data字段包含一個 OpenAI Responses API 事件類型可能是response.created、response.output_text.delta等。如果你希望響應消息一經生成就立即以流式方式發送給用戶這類事件是最直接的途徑。關于計算機工具computer tool原始事件與已存儲結果保持相同的預覽版與正式發布版之分預覽版流程會流式傳輸帶有一個action的computer_call條目gpt-5.5可以流式傳輸帶有批量actions[]的computer_call條目更高層級的RunItemStreamEvent接口不會為此添加計算機工具專用的特殊事件名稱兩種結構仍然都以tool_called的形式呈現而截圖結果會以封裝computer_call_output條目的tool_output形式返回。下面的示例會逐 token 輸出 LLM 生成的文本import asyncio from openai.types.responses import ResponseTextDeltaEvent from agents import Agent, Runner async def main(): agent Agent( nameJoker, instructionsYou are a helpful assistant., ) result Runner.run_streamed(agent, inputPlease 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, flushTrue) if __name__ __main__: asyncio.run(main())這里通過event.type raw_response_event過濾出原始事件再用isinstance(event.data, ResponseTextDeltaEvent)只保留文本增量從而實現逐 token 輸出。流式傳輸與審批Streaming and Approvals流式傳輸與因工具審批而暫停的運行是兼容的。如果某個工具需要審批result.stream_events()會結束迭代器正常退出而不是拋出異常待處理的審批會暴露在RunResultStreaming.interruptions中元素類型為ToolApprovalItem使用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從源碼看RunState.approve()支持always_approve參數RunState.reject()支持always_reject與rejection_message參數并且兩者都會自動處理嵌套智能體工具運行nested agent-tool runs的審批路由。有關完整的暫停和恢復操作流程請參閱 人在回路指南。當前輪次結束后的流式傳輸取消如果需要中途停止流式運行請調用result.cancel()。其mode參數決定取消策略模式行為immediate默認立即停止運行取消所有任務并清空事件隊列after_turn讓當前輪次正常完成后再停止允許 LLM 響應收尾、執行待處理的工具調用、正確保存會話狀態、準確記錄用量然后在下一輪次開始前停止再次強調只有result.stream_events()結束后流式運行才算完成。在最后一個可見 token 到達后SDK 可能仍在持久化會話條目、確定最終審批狀態或壓縮歷史記錄。因此調用cancel()之后應當繼續消費stream_events()讓取消過程正確收尾。如果使用cancel(modeafter_turn)在某個工具輪次后停止并且你正通過result.to_input_list(modenormalized)手動繼續那么應當使用規范化輸入重新運行result.last_agent以繼續尚未完成的現有用戶輪次而不是立即追加一個新的用戶輪次。以下三種情況需要特別注意如果在該未完成的運行恢復前收到了新的用戶輸入使用result.to_state()轉換已消費完畢的結果調用state.add_input(...)暫存輸入然后從該狀態恢復運行。運行器會在下一次模型調用前立即接納暫存的輸入字符串輸入會被規范化為用戶消息多次調用保持插入順序。參見 恢復前添加輸入。如果流式運行因工具審批而停止請勿將其視為新輪次。應先將流消費完畢檢查result.interruptions然后改為從result.to_state()恢復運行。自定義會話歷史合并使用RunConfig.session_input_callback自定義如何在下一次模型調用前合并檢索到的會話歷史與新的用戶輸入。默認行為None是將新輸入追加到會話歷史傳入SessionInputCallback自定義函數后函數接收歷史與新的輸入并返回合并后的條目列表。如果你在此處重寫新輪次條目則重寫后的版本會作為該輪次的持久化內容。運行條目事件與智能體事件RunItemStreamEvent是更高層級的事件它們在條目完全生成后通知你因此你可以按“消息已生成”“工具已運行”等粒度推送進度更新而不是按每個 token 推送。它包含name語義事件名稱與item被創建的RunItem兩個字段。同樣AgentUpdatedStreamEvent會在當前智能體發生變化時提供更新例如因任務轉移 handoff 而切換智能體通過new_agent字段暴露新的智能體對象。運行條目事件名稱RunItemStreamEvent.name使用一組固定的語義事件名稱message_output_createdhandoff_requestedhandoff_occuredtool_calledtool_search_calledtool_search_output_createdtool_outputreasoning_item_createdmcp_approval_requestedmcp_approval_responsemcp_list_tools兩點需要特別說明handoff_occured是特意保留的拼寫錯誤目的是保持向后兼容。源碼注釋中明確寫道“This is misspelled, but we cant change it because that would be a breaking change”見 src/agents/stream_events.py任務轉移調用只會以handoff_requested的形式發出不會同時以tool_called的形式發出同一輪次中的普通函數工具調用仍會發出tool_called。關于托管工具檢索hosted tool search當模型發出工具檢索請求時會發出tool_search_called當 Responses API 返回已加載的子集時會發出tool_search_output_created。關于程序化工具調用Programmatic Tool Calling系統會為生成的program以及由程序擁有的普通子工具調用發出tool_called系統會為子工具輸出以及與生成的program相匹配的program_output發出tool_output由程序擁有的托管 MCPmcp_approval_request和mcp_list_tools條目屬于例外它們會分別以mcp_approval_requested和mcp_list_tools的形式發出并分別封裝MCPApprovalRequestItem和MCPListToolsItem檢查原始條目的type以區分其余條目如tool_call_item、tool_call_output_item、message_output_item定義見 src/agents/items.py 與 src/agents/items.py由程序擁有的子調用還帶有一個類型為program的caller其調用方 IDcaller 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( nameJoker, instructionsFirst call the how_many_jokes tool, then tell that many jokes., tools[how_many_jokes], ) result Runner.run_streamed( agent, inputHello, ) print( Run starting ) async for event in result.stream_events(): # Well 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(fAgent 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())異常行為與后續排查從RunResultStreaming的文檔字符串與stream_events()的實現可以看出流式消費過程中可能拋出以下異常若智能體超過max_turns上限會拋出MaxTurnsExceeded若守衛guardrail被觸發會拋出 tripwire 異常例如InputGuardrailTripwireTriggered或OutputGuardrailTripwireTriggered。此外如果運行循環在產生任何流式事件之前就失敗例如沙箱初始化早期失敗異常可能不會通過stream_events()重新拋出。此時可以通過RunResultStreaming.run_loop_exception屬性可靠地檢查靜默失敗該屬性在運行循環無錯誤完成、尚未完成或被取消時返回None否則返回后臺運行循環的異常對象。result Runner.run_streamed(agent, hello) async for event in result.stream_events(): pass if result.run_loop_exception: raise result.run_loop_exception事件消費的底層機制從實現層面看流式事件的傳遞依賴一個后臺運行循環與事件隊列的協作見 src/agents/result.pyRunResultStreaming內部維護_event_queueasyncio.Queue后臺的run_loop_task把事件寫入隊列stream_events()則作為異步迭代器從隊列中取出事件并yield給調用方。隊列以QueueCompleteSentinel哨兵標記結束消費方在收到哨兵后會等待輸入守衛任務收尾、復查錯誤然后退出迭代器src/agents/result.py。這種“后臺生產、前臺消費”的設計保證了即使后處理如會話持久化、歷史壓縮晚于最后一個可見 token 完成你也能在流結束后通過is_complete與run_loop_exception獲取到真實的最終狀態。小結掌握openai-agents-python的流式傳輸核心在于理解三個層次原始事件層RawResponsesStreamEvent逐 token 級、運行條目層RunItemStreamEvent條目級語義事件與智能體切換層AgentUpdatedStreamEvent。在此基礎上將審批暫停/恢復interruptionsto_state()approve()、輪次級取消cancel(modeafter_turn)、待恢復輸入的暫存state.add_input()以及會話歷史合并回調RunConfig.session_input_callback組合使用即可構建出面向真實產品場景的流式交互體驗。事件類型的完整定義可繼續查閱 src/agents/stream_events.py運行入口見 src/agents/run.py運行結果與取消邏輯見 src/agents/result.py。【免費下載鏈接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows項目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考