openai-agents-python 流式（Streaming）编程全指南：从 `Runner.run_streamed()` 到事件流处理实战

openai-agents-python 流式Streaming编程全指南从Runner.run_streamed()到事件流处理实战【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python流式Streaming是 openai-agents-python 中让 Agent 运行过程可见的核心能力通过Runner.run_streamed()启动异步流式运行你可以在模型逐 token 生成、工具调用、切换 Agent、等待人工审批的每一个环节实时接收事件从而为终端用户渲染进度、实现打字机式输出并在暂停-审批-恢复的人机协同流程中保持对运行状态的完全掌控。读完本文你将掌握 raw 响应事件与高层级 RunItem 事件的完整模型、11 个语义事件名的触发语义、与工具审批/取消/断点续跑的组合用法以及这些行为背后的源码实现依据。流式运行的基本模型run_streamed与stream_events与一次性返回完整结果的Runner.run()不同流式运行通过Runner.run_streamed()启动返回一个RunResultStreaming对象。该对象的核心入口是result.stream_events()它返回一个类型为StreamEvent的异步迭代器async iterator。StreamEvent是三种事件的联合类型union type定义在 src/agents/stream_events.py事件类型事件类含义raw_response_eventRawResponsesStreamEventLLM 直接透传的 raw 事件data字段为 OpenAI Responses API 事件run_item_stream_eventRunItemStreamEvent高层级语义事件某个 RunItem消息、工具调用等已完整生成时触发agent_updated_stream_eventAgentUpdatedStreamEvent当前 Agent 发生变化如发生 handoff时触发携带new_agent使用方式遵循以下三个关键原则必须消费到迭代器结束流式运行在异步迭代器结束之前不算完成。stream_events()内部的实现src/agents/result.py通过后台run_loop_task向_event_queue写入事件直到放入QueueCompleteSentinel哨兵才结束迭代——会话持久化、审批记账approval bookkeeping、历史压缩等后处理可能发生在最后一个可见 token 之后因此即使你只关心最终结果也应完整遍历事件流。循环退出后检查is_complete迭代器结束后result.is_complete反映最终的运行状态src/agents/result.py。异常通过迭代器抛出stream_events()会在运行超时MaxTurnsExceeded、护栏绊线如InputGuardrailTripwireTriggered/OutputGuardrailTripwireTriggered时抛出对应异常如果运行在产生任何事件前就失败例如沙箱初始化阶段可检查result.run_loop_exception属性捕获静默失败src/agents/result.py。import asyncio 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: print((raw), event.data.type) print(Run complete:, result.is_complete) if __name__ __main__: asyncio.run(main())Raw 响应事件逐 token 输出 LLM 生成内容RawResponsesStreamEvent包装从 LLM 直接透传的 raw 事件。每个对象的data字段是 OpenAI Responses API 事件类型包括response.created、response.output_text.delta等。当你希望生成即展示——例如在聊天界面里实现打字机效果——就应该监听这一类事件。data的类型别名TResponseStreamEvent直接来自 OpenAI Python SDK见 src/agents/items.py 的导入因此可以直接使用openai.types.responses下的具体事件类做类型判断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())关于计算机工具computer tool的 raw 事件有一点需要特别留意raw 事件流与已存储的结果保持一致区分预览版preview与GA 版两种形态。预览版流程流式产出只含单个action的computer_call项而gpt-5.5等模型可以流式产出带批量actions[]的computer_call项。不过在上层RunItemStreamEvent接口中并不会为计算机工具单独增加事件名两种形态统一以tool_called对外发布截图结果则以包装computer_call_output项的tool_output返回。流式与工具审批Human-in-the-Loop的配合流式运行天然兼容等待工具审批而暂停的流程。当某个工具需要人工批准时result.stream_events()正常结束审批项不会被当作流式事件发出——见 src/agents/run_internal/streaming.pyToolApprovalItem代表的是中断而非流式事件挂起的审批会暴露在RunResultStreaming.interruptions中调用result.to_state()将结果转换为RunState用state.approve(interruption)/state.reject(interruption)作出判断再以Runner.run_streamed(agent, state)从断点恢复运行。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(): passto_state()的实现src/agents/result.py会携带当前轮次、已处理的模型响应、工具使用跟踪快照、未决输入等现场信息保证恢复后的运行从精确的中断点继续。RunState层面的approve/reject通过RunContextWrapper.approve_tool(...)落库判断且支持always_approveTrue/always_rejectTrue使同一判断在后续调用中持续生效src/agents/run_state.py。完整的暂停/恢复演练请参考 Human-in-the-loop 指南。在当前轮次结束后取消流式运行需要中途停止流式运行时调用result.cancel()默认modeimmediate立即停止取消所有后台任务、清空事件队列并把is_complete置为True以终止事件流modeafter_turn优雅停止——让当前 LLM 响应完整生成、执行完挂起的工具调用、正确保存会话状态并记录 usage然后在下一轮开始前停下。从源码可以看到after_turn模式本质是设置一个_cancel_mode标志流式循环会在合适的轮次边界检查该标志后自行收敛src/agents/result.py。另外调用cancel()之后应当继续消费stream_events()让取消流程完整落地见cancel()的 docstring 注释。记住流式运行在result.stream_events()结束之前不算完成。SDK 在最后一个可见 token 之后仍可能在后台持久化会话项、确定审批状态或压缩历史。关于after_turn停止后的续跑文档给出三条重要准则手动续跑与规范化输入如果你正使用result.to_input_list(modenormalized)手动续跑且cancel(modeafter_turn)在某个工具轮之后停止那么应当用这份规范化输入重新运行result.last_agent以继续那个未完成的既有用户轮次而不是立刻追加一个新的用户轮次。modenormalized会在手写过滤改写模型历史时优先返回规范化的续跑输入src/agents/result.py。新输入到达时如果在未完成运行恢复前有新的用户输入到达请把已排空drained的结果用result.to_state()转换调用state.add_input(...)暂存输入然后从该状态恢复。Runner 会在下一次模型调用之前立即接纳这份暂存输入字符串会被规范化为用户消息多次调用保持插入顺序暂存输入还会随RunState的to_json()/from_json()、to_string()/from_string()序列化往返而保留。详见 results.md 的恢复前添加输入小节。因审批而暂停时流式运行因工具审批暂停不要把它当作新的一轮。应完整消费流、检查result.interruptions并改用result.to_state()恢复。自定义会话历史合并可用RunConfig.session_input_callback自定义检索到的会话历史 新用户输入在下次模型调用前的合并方式。如果在该回调中重写了新一轮的 items重写后的版本会被持久化到该轮次。RunItem 事件与 Agent 更新事件RunItemStreamEvent是更高层级的语义事件它不关心 token 级增量而是在某个 item完整生成后通知你适合按消息已生成工具已执行的粒度推送进度。AgentUpdatedStreamEvent则在当前 Agent 切换时例如 handoff 之后给出更新new_agent字段携带新的 Agent 实例。固定的语义事件名RunItemStreamEvent.name使用一套固定的语义事件名src/agents/stream_events.py其触发映射可在 src/agents/run_internal/streaming.py 中一一对应事件名触发条件源码映射message_output_createdMessageOutputItem生成handoff_requestedHandoffCallItem生成handoff 调用发出handoff_occuredHandoffOutputItem生成handoff 完成tool_calledToolCallItem生成tool_search_calledToolSearchCallItem生成模型发出工具检索请求tool_search_output_createdToolSearchOutputItem生成Responses API 返回加载的子集tool_outputToolCallOutputItem生成reasoning_item_createdReasoningItem生成mcp_approval_requestedMCPApprovalRequestItem生成mcp_approval_responseMCPApprovalResponseItem生成mcp_list_toolsMCPListToolsItem生成几点值得注意的语义细节handoff_occured是故意拼错的——这是为了向后兼容而保留的拼写src/agents/stream_events.py 的注释明确说明无法修改否则会破坏兼容性。handoff 调用只以handoff_requested发出不会重复以tool_called发出同一轮内的普通函数工具调用仍然发出tool_called。托管式工具检索hosted tool search模型发出工具检索请求时发出tool_search_calledResponses API 返回加载的工具子集时发出tool_search_output_created。Programmatic Tool Calling生成的program及其拥有的普通子工具调用都会发出tool_called子工具的输出以及与生成program对应的program_output发出tool_output。例外是程序拥有的托管 MCP 的mcp_approval_request项与mcp_list_tools项它们分别以mcp_approval_requested和mcp_list_tools发出包装对应的MCPApprovalRequestItem与MCPListToolsItem。若要区分其余项目请检查 raw item 的type程序拥有的子调用还会携带caller字段其type为programcaller的 ID 标识父程序。完整示例忽略 raw 事件按语义粒度流式推送下面的示例演示了三种事件的分流处理跳过raw_response_event在 Agent 切换时打印新 Agent 名在 item 生成时按类型打印工具调用、工具输出与消息输出。ItemHelpers.text_message_output(...)用于从消息输出项中提取纯文本src/agents/items.py 中的ItemHelpers。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持有run_loop_task后台 asyncio 任务它执行实际的 Agent 运行循环并把事件写入_event_queuesrc/agents/result.py。Agent 切换时运行循环会向队列放入AgentUpdatedStreamEventsrc/agents/run_internal/run_loop.py。item 到事件的转换每一轮步骤产生的 items 由stream_step_items_to_queue转换并按上述表格映射为RunItemStreamEvent审批占位项ToolApprovalItem被跳过因为它代表的是中断而不是流式事件。消费与收尾stream_events()从_event_queue中取出事件并yield遇到QueueCompleteSentinel时会先安全等待输入护栏任务结束、再检查错误最后清理后台任务、排空队列src/agents/result.py。错误路径运行循环抛出的异常如超时、护栏绊线被捕获进_stored_exception在迭代过程中通过_check_errors()抛出保证异常信息不会在流式消费中丢失src/agents/result.py。流式运行中建议留意的实践要点事件粒度选型追求打字机效果用raw_response_event追求步骤化进度条用run_item_stream_event多 Agent 场景追踪当前执行者用agent_updated_stream_event。三者可以并存于同一个async for循环中用event.type分流。始终排空事件流即使只关心interruptions或最终输出也应完整消费stream_events()否则会话持久化与后处理无法正常收敛。取消后继续消费调用cancel()后继续遍历事件流让取消流程含沙箱、模型 provider 的资源清理见 src/agents/result.py完整执行。审批与续跑的状态一致性审批暂停时不要追加新输入、不要另起新轮统一走to_state()→approve/reject→run_streamed(agent, state)的闭环若确有新输入使用state.add_input(...)暂存。通过Runner.run_streamed()、三类StreamEvent与 11 个语义事件名的组合你可以在 openai-agents-python 中构建出从实时对话到审批驱动的完整流式应用相关的更多运行细节可继续阅读 运行 Agent、结果与中断处理 与 Human-in-the-loop 指南。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考