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），仅供参考