openai-agents-python 流式输出实战指南StreamEvent 事件模型、审批暂停与运行取消【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python流式输出Streaming是 openai-agents-python 中向最终用户实时展示 Agent 运行进度与部分响应的核心能力。本文以 docs/streaming.md 为主线结合 src/agents/stream_events.py、src/agents/result.py、src/agents/run.py 等源码实现系统讲解Runner.run_streamed()的用法、三类StreamEvent事件、工具审批暂停与恢复、运行中途取消以及事件名称的完整语义帮助你写出可直接复制运行、可应对多智能体交接与人工审批场景的流式应用。流式运行的基础从 run_streamed 到 stream_events在 openai-agents-python 中普通运行使用Runner.run()而流式运行则调用Runner.run_streamed()。该方法返回一个RunResultStreaming对象再通过result.stream_events()获得一个StreamEvent对象的异步迭代器供async for逐条消费。从 src/agents/run.py 的签名可以看到run_streamed与run的参数基本一致核心参数包括参数类型说明starting_agentAgent[TContext]起始智能体运行可能因 handoff 切换到其他智能体inputstr \| list[TResponseInputItem] \| RunState[TContext]初始输入可以是字符串、输入项列表也可以是用于恢复运行的RunStatecontextTContext \| None运行上下文max_turnsint \| None最大轮数None表示不限制超过会抛出MaxTurnsExceededhooksRunHooks[TContext] \| None生命周期钩子run_configRunConfig \| dict[str, Any] \| None全局运行配置如session_input_callbackprevious_response_id/auto_previous_response_idstr \| None/boolResponses API 响应链相关conversation_id/sessionstr \| None/Session \| None会话持久化相关error_handlersRunErrorHandlers[TContext] \| None按错误类型注册的错误处理器必须消费到迭代器结束流式运行有一个容易踩坑的关键点异步迭代器未结束时运行并未真正完成。即使最后一个可见 token 已经到达SDK 仍可能在后台执行会话项持久化session persistence、审批状态收尾approval bookkeeping、历史压缩history compaction等后处理。因此必须持续消费result.stream_events()直到迭代器自然结束循环退出后result.is_complete才反映最终的运行状态。从 src/agents/result.py 的实现可以看到stream_events()内部通过asyncio.Queue消费后台运行循环产出的事件遇到QueueCompleteSentinel会先等待输入护栏任务完成并检查异常最后再等待run_loop_task彻底落定_await_task_safely(self.run_loop_task)才返回这从源码层面印证了迭代器结束才代表运行收尾完成的约定。三类流式事件StreamEvent是一个类型别名在 src/agents/stream_events.py 中定义为三种事件类型的联合StreamEvent: TypeAlias RawResponsesStreamEvent | RunItemStreamEvent | AgentUpdatedStreamEvent事件类型event.type取值含义RawResponsesStreamEventraw_response_eventLLM 直接透传的原始 Responses API 事件RunItemStreamEventrun_item_stream_event高层语义事件消息生成、工具调用、handoff 等AgentUpdatedStreamEventagent_updated_stream_event当前智能体发生变化如 handoff 之后原始响应事件逐 token 输出 LLM 文本RawResponsesStreamEvent包装的是 LLM 直接传递的原始事件其data字段是 OpenAI Responses API 的事件对象类型如response.created、response.output_text.delta等。当你希望响应消息一生成就立刻推送给用户时这类事件最合适。例如下面的代码可以把 LLM 生成的文本逐 token 打印出来完整代码见 examples/basic/stream_text.pyimport 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())计算机工具的原始事件preview 与 GA 的区别原文档特别指出计算机工具computer tool的原始事件与存储结果一样保留了 preview 与 GA 的差异。preview 流程流式产出包含单个action的computer_call项而gpt-5.5可以流式产出带批量actions[]的computer_call项。需要留意的是高层级接口RunItemStreamEvent并没有为此单独增加计算机专用的事件名——两种形态都以tool_called呈现截图结果则作为包装了computer_call_output项的tool_output返回。这意味着如果你要区分这两种形态不能只看事件名而需要检查原始项的字段结构。流式与审批工具审批暂停与恢复流式运行与因工具审批而暂停的流程完全兼容这对应 docs/human_in_the_loop.md 中的人工介入场景。当某个工具需要审批时result.stream_events()会正常结束待审批项暴露在RunResultStreaming.interruptions中src/agents/result.py 中定义为list[ToolApprovalItem]用result.to_state()把结果转换成RunState在RunState上对中断项调用approve()或拒绝把RunState作为输入再次调用Runner.run_streamed(...)恢复运行。to_state()的实现位于 src/agents/result.py它会把上下文、原始输入、起始智能体、当前轮数、最后一次处理的响应、会话 ID 等运行现场完整打包进新的RunState确保恢复时上下文不丢失。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取消流式运行立即停止与当前轮结束后停止如果需要在运行中途终止流式执行调用result.cancel()。默认modeimmediate立即停止运行若希望当前轮正常走完再停则调用result.cancel(modeafter_turn)。从 src/agents/result.py 的实现可以看清两种模式的差异immediate默认直接调用_cleanup_tasks()取消所有后台任务、把is_complete置为True以终止事件流、清空输入护栏队列并写入结束哨兵属于硬性中断after_turn只设置_cancel_mode标志不做任何清理由后台流式循环检查该标志后优雅停止。根据 docstring这种模式会允许 LLM 响应自然结束、执行完待处理的工具调用、正确保存会话状态并准确统计 usage然后在下一轮开始前停下。另外两个值得记住的约定取消后仍需继续消费调用cancel()之后应当继续消费stream_events()让取消流程完整收尾src/agents/result.py 的注释也明确说明了这一点。运行未完成前取消同样不算完成流式运行要等stream_events()结束才算完成SDK 可能在最后一个可见 token 之后仍在持久化会话项、确定审批状态或压缩历史。after_turn 取消后如何衔接如果你正通过result.to_input_list(modenormalized)手动继续推进而cancel(modeafter_turn)恰好在一个工具轮之后停止此时不要立即追加一个新的用户轮次而应该用规范化后的输入重新运行result.last_agent以继续那个尚未完成的既有用户轮次。同时注意三种特殊情形恢复前来了新用户输入把已消费完的结果result.to_state()转成状态调用state.add_input(...)暂存新输入再从该状态恢复。add_input的实现见 src/agents/run_state.py它会将输入暂存并在下一次恢复的模型调用之前正式纳入字符串输入会规范化为一条用户消息多次调用保持插入顺序。详见 docs/results.md。因工具审批而停下的运行不要把它当作新轮次。应当先完整消费流、检查result.interruptions然后从result.to_state()恢复。自定义历史与新输入的合并策略通过RunConfig.session_input_callback定制检索到的会话历史 新用户输入在下一次模型调用前的合并方式src/agents/run_config.py。默认行为None是把新输入追加到会话历史传入回调函数后回调接收历史与新输入并返回合并后的项列表。若在回调中重写了新轮次的项持久化的将是重写后的版本。运行项事件与智能体事件面向 UI 的高层进度RunItemStreamEvent是高层级事件它在一个项完整生成后才触发因此适合以消息已生成工具已执行的粒度推送进度而不是逐 token 更新。与之互补的AgentUpdatedStreamEvent在当前智能体切换例如 handoff 发生时触发其new_agent字段携带切换后的新智能体对象。RunItemStreamEvent 的固定事件名集合RunItemStreamEvent.name使用一套固定的语义事件名完整列表如下与 src/agents/stream_events.py 中的Literal定义一一对应message_output_created消息输出项生成handoff_requested发起 handoffhandoff_occuredhandoff 实际发生注意该拼写是刻意保留的occured为错误拼写但为了向后兼容无法修改tool_called工具被调用tool_search_called模型发起托管工具搜索请求tool_search_output_createdResponses API 返回加载的工具子集tool_output工具输出项生成reasoning_item_created推理项生成mcp_approval_requested请求 MCP 审批mcp_approval_responseMCP 审批响应mcp_list_toolsMCP 工具列表项几点语义细节值得注意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定义见 src/agents/items.py。要区分其余项需要检查原始项的type字段程序所属的子调用还会携带caller字段其type为programcaller的 ID 标识父程序。示例忽略原始事件向用户推送高层进度下面的完整示例演示了如何忽略原始响应事件只用高层事件驱动 UI 更新完整代码见 examples/basic/stream_items.py其文件末尾还附带了真实运行输出可直接对照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())对照 examples/basic/stream_items.py 底部注释中的真实输出运行流程依次为 Run starting →Agent updated: Joker→-- Tool was called: how_many_jokes→-- Tool output: 4→-- Message output:四则笑话正文→ Run complete 。这与RunItemStreamEvent项完整生成后触发的设计完全吻合工具调用项、工具输出项、消息输出项分别在各自完成时被一次性推送。小结openai-agents-python 的流式体系围绕三个层次展开原始层RawResponsesStreamEvent逐 token 透传 LLM 输出适合打字机式文本流语义层RunItemStreamEvent以项为粒度推送消息、工具、handoff、MCP 审批等高层进度切换层AgentUpdatedStreamEvent感知多智能体场景下的当前智能体切换。配合Runner.run_streamed()的完整参数体系、审批暂停与RunState恢复机制、cancel()的两种取消模式以及session_input_callback/add_input对会话衔接的精细控制你可以构建出生产可用的实时流式交互界面。更多相关主题可继续阅读 docs/results.md、docs/human_in_the_loop.md 与 docs/handoffs.md。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考