openai-agents-python Responses WebSocket Session 完全指南共享连接、多轮复用与源码级原理【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonResponsesWebSocketSession是 openai-agents-python 中为 OpenAI Responses API 的 WebSocket 传输模式提供的会话辅助层它把「一个共享的 WebSocket 能力的 provider」与「一份共享的 RunConfig」绑定在一起让多轮对话、agent-as-tool 嵌套调用可以复用同一条 WebSocket 连接。本文围绕 docs/ref/responses_websocket_session.md 指向的模块 src/agents/responses_websocket_session.py结合 docs/running_agents.md、docs/models/index.md、examples/basic/stream_ws.py 与测试用例完整讲解会话的创建方式、全部参数、底层机制与实战注意事项。读完你将掌握如何用responses_websocket_session()在多次Runner调用间复用 WebSocket 连接、如何配置前缀路由与 keepalive、如何在流式多轮对话中处理工具调用与人工审批以及每条连接的服务端约束。一、定位这是 Responses API 的 WebSocket 传输不是 Realtime API默认情况下OpenAI Responses API 请求走 HTTP 传输。SDK 允许在 OpenAI Responses provider 路径下切换到 WebSocket 传输以减少多轮对话的建连开销、降低端到端延迟。需要首先澄清的概念边界依据 docs/models/index.md这是Responses API over WebSocket 传输不是 Realtime API也不适用于 Chat Completions传输选择发生在 SDK 把模型名解析成模型实例的时刻OpenAIResponsesWSModel—— 固定使用 WebSocketOpenAIResponsesModel—— 固定使用 HTTPOpenAIChatCompletionsModel—— 固定停留在 Chat Completions如果向Runner传入RunConfig(model_provider...)则由该 provider 决定传输方式全局默认值set_default_openai_responses_transport不再生效对非 OpenAI 的第三方 provider只有在其支持 Responses WebSocket/responses端点时才适用。开启 WebSocket 传输有两种粒度from agents import set_default_openai_responses_transport # 全局默认影响由默认 OpenAI provider 解析出的所有 Responses 模型 set_default_openai_responses_transport(websocket)以及按 provider / 按 run 配置from agents import Agent, OpenAIProvider, RunConfig, Runner provider OpenAIProvider( use_responses_websocketTrue, # 可选省略时优先使用 OPENAI_WEBSOCKET_BASE_URL 环境变量 websocket_base_urlwss://your-proxy.example/v1, # 可选的底层 keepalive 设置 responses_websocket_options{ping_interval: 20.0, ping_timeout: 60.0}, ) agent Agent(nameAssistant) result await Runner.run( agent, Hello, run_configRunConfig(model_providerprovider), )启用 WebSocket 传输后仍可使用常规RunnerAPIRunner.run/Runner.run_streamed。会话辅助器只是推荐用于连接复用并非强制要求。二、responses_websocket_session()一次调用构造完整的共享会话responses_websocket_session()是模块暴露的异步上下文管理器asynccontextmanager签名如下依据 src/agents/responses_websocket_session.pyasync def responses_websocket_session( *, api_key: str | None None, base_url: str | None None, websocket_base_url: str | None None, organization: str | None None, project: str | None None, openai_prefix_mode: MultiProviderOpenAIPrefixMode alias, unknown_prefix_mode: MultiProviderUnknownPrefixMode error, responses_websocket_options: OpenAIResponsesWebSocketOptions | None None, ) - AsyncIterator[ResponsesWebSocketSession]全部参数为关键字参数语义如下参数类型默认值说明api_keystr \| NoneNoneOpenAI API Key透传给底层MultiProvider(openai_api_key...)base_urlstr \| NoneNoneHTTP 基地址openai_base_url一般无需设置websocket_base_urlstr \| NoneNoneWebSocket 基地址openai_websocket_base_url。使用自定义代理或 OpenAI 兼容端点且其/responses走 WebSocket 时通常需要显式指定organizationstr \| NoneNoneOrganization 标识openai_organizationprojectstr \| NoneNoneProject 标识openai_projectopenai_prefix_modealias \| model_idalias控制openai/...前缀的处理方式alias剥离前缀路由到 OpenAI provideropenai/gpt-4.1→ 模型gpt-4.1model_id保留字面前缀模型openai/gpt-4.1unknown_prefix_modeerror \| model_iderror控制未知前缀如openrouter/...的处理方式error抛UserErrormodel_id原样透传responses_websocket_optionsOpenAIResponsesWebSocketOptions \| NoneNone底层 WebSocket keepalive 等低层行为的定制项如ping_interval、ping_timeout、max_size该函数在内部完成三件事源码可见 src/agents/responses_websocket_session.py构造一个MultiProvider并显式打开两个关键开关model_provider MultiProvider( openai_api_keyapi_key, openai_base_urlbase_url, openai_websocket_base_urlwebsocket_base_url, openai_organizationorganization, openai_projectproject, openai_use_responsesTrue, openai_use_responses_websocketTrue, openai_prefix_modeopenai_prefix_mode, unknown_prefix_modeunknown_prefix_mode, openai_responses_websocket_optionsresponses_websocket_options, )其中openai_use_responsesTrue确保走 Responses API 模型路径openai_use_responses_websocketTrue确保传输方式为 WebSocket。使用MultiProvider而非裸OpenAIProvider是为了在保留前缀路由能力例如openai/gpt-4.1的同时持有同一个OpenAIProvider实例。从MultiProvider取出其内部openai_provider构造ResponsesWebSocketSession(provider..., run_configRunConfig(model_providermodel_provider))即一份绑定共享 provider 的共享RunConfig。通过try/finally保证上下文退出时调用await session.aclose()关闭缓存的 provider 模型资源——其中就包括 WebSocket 连接。三、ResponsesWebSocketSession类共享 provider 共享 RunConfig 的源码级原理ResponsesWebSocketSession是一个dataclass(frozenTrue)依据 src/agents/responses_websocket_session.py只有两个字段dataclass(frozenTrue) class ResponsesWebSocketSession: provider: OpenAIProvider run_config: RunConfig其核心职责是「把多次 Runner 调用钉在同一个共享 provider 上」。理解它需要看四个关键方法1._validate_provider_alignment()对齐校验def _validate_provider_alignment(self) - MultiProvider: model_provider self.run_config.model_provider if not isinstance(model_provider, MultiProvider): raise TypeError( ResponsesWebSocketSession.run_config.model_provider must be a MultiProvider. ) if model_provider.openai_provider is not self.provider: raise ValueError( ResponsesWebSocketSession provider and run_config.model_provider are not aligned. ) return model_provider两个硬性约束run_config.model_provider必须是MultiProvider且该MultiProvider内部持有的openai_provider必须与 session 的provider是同一个对象身份比较is。这保证了「共享连接」不会在运行时被悄悄替换。2._prepare_runner_kwargs()注入共享配置def _prepare_runner_kwargs(self, method_name: str, kwargs: Mapping[str, Any]) - dict[str, Any]: self._validate_provider_alignment() if run_config in kwargs: raise ValueError( fDo not pass run_config to ResponsesWebSocketSession.{method_name}(). ) runner_kwargs dict(kwargs) runner_kwargs[run_config] self.run_config return runner_kwargs也就是说调用ws.run(...)/ws.run_streamed(...)时不允许再传run_config否则直接抛ValueError——这正是「所有轮次共享同一份RunConfig」的强制性保证。测试 tests/models/test_responses_websocket_session.py 专门验证了这一行为pytest.raises(ValueError, matchrun_config)。3.run()/run_streamed()转发到 Runnerasync def run(self, starting_agent, input, **kwargs) - RunResult: runner_kwargs self._prepare_runner_kwargs(run, kwargs) return await Runner.run(starting_agent, input, **runner_kwargs) def run_streamed(self, starting_agent, input, **kwargs) - RunResultStreaming: runner_kwargs self._prepare_runner_kwargs(run_streamed, kwargs) return Runner.run_streamed(starting_agent, input, **runner_kwargs)run是异步的、返回RunResultrun_streamed是同步返回RunResultStreaming的流式接口通过stream_events()消费事件。注意session 刻意不暴露run_sync测试 tests/models/test_responses_websocket_session.py 验证了not hasattr(ws, run_sync)因为 WebSocket 会话天然面向异步事件循环。4.aclose()释放连接async def aclose(self) - None: Close cached provider model resources (including websocket connections). await self._validate_provider_alignment().aclose()aclose委托给MultiProvider.aclose()依次关闭其内部 OpenAI provider 缓存的所有模型资源——包括活跃的 WebSocket 连接。上下文管理器退出时自动调用因此推荐用async with保证释放。测试 tests/models/test_responses_websocket_session.py 验证了退出async with块后 provider 的aclose被恰好调用一次。3.1 字典形式 RunConfig 的自动归一化构造ResponsesWebSocketSession时run_config也接受字典在__post_init__中通过_coerce_run_config归一化为RunConfig实例源码 src/agents/responses_websocket_session.pydef __post_init__(self) - None: object.__setattr__(self, run_config, _coerce_run_config(self.run_config)) self._validate_provider_alignment()测试用例展示了字典形态的合法与非法用法tests/models/test_responses_websocket_session.pysession ResponsesWebSocketSession( providerprovider.openai_provider, run_config{ model_provider: provider, model_settings: {temperature: 0.0, retry: {max_retries: 0}}, }, ) # 合法字典被归一化为 RunConfigmodel_settings.temperature 0.0而包含未知字段如tracin_disabled的字典会抛出TypeError: Unknown run_config settings: ...即_coerce_run_config对字典键有严格校验。四、两种使用模式对比裸 Runner 与共享会话docs/running_agents.md 给出了两种模式模式一不用会话辅助器可用但多轮会重连import asyncio from agents import Agent, Runner, set_default_openai_responses_transport async def main(): set_default_openai_responses_transport(websocket) agent Agent(nameAssistant, instructionsBe concise.) result Runner.run_streamed(agent, Summarize recursion in one sentence.) async for event in result.stream_events(): if event.type raw_response_event: continue print(event.type) asyncio.run(main())这种模式适合单次运行。如果反复调用Runner.run()/Runner.run_streamed()每次运行都可能重新建连除非手动复用同一份RunConfig/ provider 实例。模式二responses_websocket_session()多轮复用推荐import asyncio from agents import Agent, responses_websocket_session async def main(): agent Agent(nameAssistant, instructionsBe concise.) async with responses_websocket_session( responses_websocket_options{ping_interval: 20.0, ping_timeout: 60.0}, ) as ws: first ws.run_streamed(agent, Say hello in one short sentence.) async for _event in first.stream_events(): pass second ws.run_streamed( agent, Now say goodbye., previous_response_idfirst.last_response_id, ) async for _event in second.stream_events(): pass asyncio.run(main())这是推荐模式跨轮次以及继承同一run_config的嵌套 agent-as-tool 调用共享同一条 WebSocket 连接使其保持「温热」状态。第二轮通过previous_response_idfirst.last_response_id衔接上下文RunResultStreaming.last_response_id提供了上一轮的响应 ID。五、前缀路由openai_prefix_mode与unknown_prefix_modeMultiProvider保留了两个历史默认行为依据 docs/models/index.mdopenai/...被视为 OpenAI provider 的别名openai/gpt-4.1路由为模型gpt-4.1未知前缀默认抛UserError不透明透传。当 OpenAI provider 指向一个「期望字面命名空间模型 ID」的 OpenAI 兼容端点时需要显式开启透传。responses_websocket_session()上同样提供这两个开关用法与MultiProvider一致from agents import Agent, MultiProvider, RunConfig, Runner provider MultiProvider( openai_base_urlhttps://openrouter.ai/api/v1, openai_api_key..., openai_use_responses_websocketTrue, openai_prefix_modemodel_id, unknown_prefix_modemodel_id, ) agent Agent( nameAssistant, instructionsBe concise., modelopenai/gpt-4.1, ) result await Runner.run( agent, Hello, run_configRunConfig(model_providerprovider), )选择规则openai_prefix_modemodel_id后端期望字面的openai/...字符串时使用unknown_prefix_modemodel_id后端期望其他命名空间模型 ID如openrouter/openai/gpt-4.1-mini时使用。这两个选项在 WebSocket 传输之外同样适用于MultiProvider示例中保持openai_use_responses_websocketTrue只是为了贴合本节「WebSocket 传输」的上下文。测试分别验证了三种路由行为tests/models/test_responses_websocket_session.py默认aliasget_model(openai/gpt-4.1)捕获到gpt-4.1openai_prefix_modemodel_id捕获到openai/gpt-4.1unknown_prefix_modemodel_idget_model(openrouter/openai/gpt-4.1)捕获到openrouter/openai/gpt-4.1。六、底层连接行为keepalive、消息上限与 60 分钟约束通过responses_websocket_options可以定制底层 WebSocket 行为依据 docs/running_agents.md 与 docs/models/index.mdping_interval/ping_timeout控制心跳保活。长推理轮次或网络延迟抖动导致 keepalive 超时时增大ping_timeout以容忍延迟的 pong 帧或设置ping_timeoutNone关闭心跳超时同时保持 ping 启用。官方示例使用{ping_interval: 20.0, ping_timeout: 60.0}。max_sizeSDK 默认关闭传入消息大小限制max_sizeNone。对运行在代理之后或内存受限容器中的长生命周期 Agent 进程可设responses_websocket_options{max_size: 8 * 1024 * 1024}来限制单条消息的内存占用。服务端约束连接复用不会消除这些限制每条 WebSocket 连接一次只处理一个响应且连接限制为60 分钟超过限制需新建连接需要并行运行时应使用多条连接服务端只在连接本地内存中保留最近一个响应某轮4xx/5xx失败会逐出previous_response_id所引用的响应。重连后storeFalse与 ZDRzero data retention流程无法恢复未缓存的previous_response_id——此时应开启新链条previous_response_idNone并发送完整输入上下文或从本地管理的 session 状态重建上下文可靠性优先于延迟的场景建议退回 HTTP/SSE 传输。两个工程注意事项退出上下文前必须排空/关闭流式迭代器。若在 WebSocket 请求仍在飞行时退出async with上下文可能强制关闭共享连接docs/running_agents.md。若环境中未安装websockets包需要先安装。七、实战多轮流式输出 工具调用 人工审批完整示例examples/basic/stream_ws.py 给出了一个完整的用户侧 WebSocket 工作流覆盖流式输出含 reasoning summary delta、普通函数工具、Agent.as_tool(...)专家 Agent、敏感工具调用的 HITL 审批以及基于previous_response_id的后续轮次。所需环境变量OPENAI_API_KEY必需OPENAI_MODEL默认gpt-5.6-sol、OPENAI_BASE_URL、OPENAI_WEBSOCKET_BASE_URL、EXAMPLES_INTERACTIVE_MODEauto脚本化运行自动批准 HITL 提示均为可选。核心骨架如下from agents import ( Agent, ModelSettings, ResponsesWebSocketSession, responses_websocket_session, trace, ) from agents.decorators import tool tool def lookup_order(order_id: str) - dict[str, Any]: 返回演示用的确定性订单数据。 tool(needs_approvalTrue) def submit_refund(order_id: str, amount: float, reason: str) - dict[str, Any]: 创建退款申请。该工具需要人工审批。 async def run_streamed_turn(ws, agent, prompt, *, previous_response_idNone): result ws.run_streamed(agent, prompt, previous_response_idprevious_response_id) while True: async for event in result.stream_events(): if event.type raw_response_event: raw event.data if raw.type response.reasoning_summary_text.delta: print(raw.delta, end, flushTrue) elif raw.type response.output_text.delta: print(raw.delta, end, flushTrue) continue if event.type ! run_item_stream_event: continue item event.item if item.type tool_call_item: print(f\n[tool call] {getattr(item.raw_item, name, unknown)}) elif item.type tool_call_output_item: print(f[tool result] {item.output}) if not result.interruptions: break # HITL逐项审批然后携带审批状态继续同一会话 state result.to_state() for interruption in result.interruptions: if ask_approval(fApprove {interruption.name} with args {interruption.arguments}?): state.approve(interruption) else: state.reject(interruption) result ws.run_streamed(agent, state) return result.last_response_id, str(result.final_output) async def main(): model_name os.getenv(OPENAI_MODEL, gpt-5.6-sol) policy_agent Agent(nameRefundPolicySpecialist, modelmodel_name, ...) support_agent Agent( nameSupportAgent, tools[lookup_order, policy_agent.as_tool(...), submit_refund], modelmodel_name, model_settingsModelSettings(max_tokens200, reasoningReasoning(effortmedium, summarydetailed)), ) async with responses_websocket_session() as ws: with trace(Responses WS support example) as current_trace: first_response_id, _ await run_streamed_turn(ws, support_agent, 退款请求...) await run_streamed_turn( ws, support_agent, What refund ticket did you just create? Reply with only the ticket., previous_response_idfirst_response_id, )示例注释明确指出完全可以跳过该辅助器直接调用Runner.run_streamed(...)那样也能工作但每次运行都会重新建连/连接除非手动复用同一份RunConfig/provider而该辅助器让跨轮次以及嵌套 agent-as-tool 运行的复用变得简单从而保持 WebSocket 连接温热。值得注意的是中断interruption处理流程needs_approvalTrue的工具触发result.interruptions通过result.to_state()拿到运行状态逐项approve/reject后把状态对象再次传给ws.run_streamed(agent, state)——审批状态与共享连接在同一会话内延续无需重建 provider。示例还演示了错误处理捕获RuntimeError且消息包含closed before any response events时通常意味着该账号/模型尚未启用 WebSocket 模式可提示用户后优雅退出。八、测试验证行为契约一览tests/models/test_responses_websocket_session.py 完整覆盖了本模块的行为契约测试验证点test_responses_websocket_session_builds_shared_run_configws.provider是OpenAIProvider_use_responses与_use_responses_websocket均为Truerun_config.model_provider是MultiProvider且其openai_provider is ws.providertest_responses_websocket_session_normalizes_dictionary_run_config字典形态的run_config被归一化为RunConfigtemperature、retry.max_retries正确解析test_responses_websocket_session_rejects_unknown_dictionary_run_config_fields未知字段抛TypeErrortest_responses_websocket_session_preserves_openai_prefix_routing默认alias模式下openai/gpt-4.1路由为gpt-4.1test_responses_websocket_session_can_preserve_openai_prefix_model_idsopenai_prefix_modemodel_id保留完整openai/gpt-4.1test_responses_websocket_session_can_preserve_unknown_prefix_model_idsunknown_prefix_modemodel_id透传openrouter/openai/gpt-4.1test_responses_websocket_session_run_injects_run_config/..._run_streamed_injects_run_configrun/run_streamed均注入共享run_configtest_responses_websocket_session_rejects_run_config_override显式传run_config抛ValueErrortest_responses_websocket_session_context_manager_closes_provider上下文退出时 provider 的aclose被调用test_responses_websocket_session_does_not_expose_run_sync不暴露同步run_syncResponsesWebSocketSession与responses_websocket_session均通过 src/agents/init.py 作为公开 API 导出并纳入 tests/fixtures/released_api_contract.json 的发布 API 契约校验范围。九、使用要点速查会话生命周期始终用async with responses_websocket_session(...) as ws:包裹退出时自动aclose()释放连接不要在请求飞行时退出上下文。连接复用所有轮次共享同一provider与RunConfig嵌套 agent-as-tool 调用也会继承同一run_config不要把run_config再次传给ws.run(...)/ws.run_streamed(...)。多轮衔接流式结果用result.last_response_id配合previous_response_id衔接上下文若发生重连或失败逐出需用完整输入上下文开新链条。长推理场景调大ping_timeout或设ping_timeoutNone内存受限环境设max_size上限。兼容端点自定义 OpenAI 兼容端点需支持 WebSocket/responses端点必要时显式设置websocket_base_url期望字面命名空间模型 ID 时使用openai_prefix_modemodel_id/unknown_prefix_modemodel_id。概念区分这是 Responses API 的 WebSocket 传输不是 Realtime API也不适用于 Chat Completions需要安装websockets包单连接一次处理一个响应、限时 60 分钟。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考