openai-agents-python Responses WebSocket Session 完全指南：共享连接、多轮复用与源码级原理

openai-agents-python Responses WebSocket Session 完全指南：共享连接、多轮复用与源码级原理

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

ResponsesWebSocketSession是 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—— 固定使用 WebSocket；
  • OpenAIResponsesModel—— 固定使用 HTTP；
  • OpenAIChatCompletionsModel—— 固定停留在 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_websocket=True, # 可选；省略时优先使用 OPENAI_WEBSOCKET_BASE_URL 环境变量 websocket_base_url="wss://your-proxy.example/v1", # 可选的底层 keepalive 设置 responses_websocket_options={"ping_interval": 20.0, "ping_timeout": 60.0}, ) agent = Agent(name="Assistant") result = await Runner.run( agent, "Hello", run_config=RunConfig(model_provider=provider), )

启用 WebSocket 传输后，仍可使用常规RunnerAPI（Runner.run/Runner.run_streamed）。会话辅助器只是推荐用于连接复用，并非强制要求。

二、responses_websocket_session()：一次调用，构造完整的共享会话

responses_websocket_session()是模块暴露的异步上下文管理器（@asynccontextmanager），签名如下（依据 src/agents/responses_websocket_session.py）：

async 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]

全部参数为关键字参数，语义如下：

该函数在内部完成三件事（源码可见 src/agents/responses_websocket_session.py）：

1. 构造一个MultiProvider，并显式打开两个关键开关：

model_provider = MultiProvider( openai_api_key=api_key, openai_base_url=base_url, openai_websocket_base_url=websocket_base_url, openai_organization=organization, openai_project=project, openai_use_responses=True, openai_use_responses_websocket=True, openai_prefix_mode=openai_prefix_mode, unknown_prefix_mode=unknown_prefix_mode, openai_responses_websocket_options=responses_websocket_options, )

其中openai_use_responses=True确保走 Responses API 模型路径，openai_use_responses_websocket=True确保传输方式为 WebSocket。使用MultiProvider而非裸OpenAIProvider，是为了在保留前缀路由能力（例如openai/gpt-4.1）的同时持有同一个OpenAIProvider实例。

2. 从MultiProvider取出其内部openai_provider，构造ResponsesWebSocketSession(provider=..., run_config=RunConfig(model_provider=model_provider))，即一份绑定共享 provider 的共享RunConfig。

3. 通过try/finally保证上下文退出时调用await session.aclose()，关闭缓存的 provider 模型资源——其中就包括 WebSocket 连接。

三、ResponsesWebSocketSession类：共享 provider + 共享 RunConfig 的源码级原理

ResponsesWebSocketSession是一个@dataclass(frozen=True)（依据 src/agents/responses_websocket_session.py），只有两个字段：

@dataclass(frozen=True) 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( f"Do 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, match="run_config")）。

3.run()/run_streamed()转发到 Runner

async 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是异步的、返回RunResult；run_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.py）：

def __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.py）：

session = ResponsesWebSocketSession( provider=provider.openai_provider, run_config={ "model_provider": provider, "model_settings": {"temperature": 0.0, "retry": {"max_retries": 0}}, }, ) # 合法：字典被归一化为 RunConfig，model_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(name="Assistant", instructions="Be 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(name="Assistant", instructions="Be 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_id=first.last_response_id, ) async for _event in second.stream_events(): pass asyncio.run(main())

这是推荐模式：跨轮次（以及继承同一run_config的嵌套 agent-as-tool 调用）共享同一条 WebSocket 连接，使其保持「温热」状态。第二轮通过previous_response_id=first.last_response_id衔接上下文，RunResultStreaming.last_response_id提供了上一轮的响应 ID。

五、前缀路由：openai_prefix_mode与unknown_prefix_mode

MultiProvider保留了两个历史默认行为（依据 docs/models/index.md）：

• openai/...被视为 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_url="https://openrouter.ai/api/v1", openai_api_key="...", openai_use_responses_websocket=True, openai_prefix_mode="model_id", unknown_prefix_mode="model_id", ) agent = Agent( name="Assistant", instructions="Be concise.", model="openai/gpt-4.1", ) result = await Runner.run( agent, "Hello", run_config=RunConfig(model_provider=provider), )

选择规则：

• openai_prefix_mode="model_id"：后端期望字面的openai/...字符串时使用；
• unknown_prefix_mode="model_id"：后端期望其他命名空间模型 ID（如openrouter/openai/gpt-4.1-mini）时使用。

这两个选项在 WebSocket 传输之外同样适用于MultiProvider；示例中保持openai_use_responses_websocket=True只是为了贴合本节「WebSocket 传输」的上下文。测试分别验证了三种路由行为（tests/models/test_responses_websocket_session.py）：

• 默认"alias"：get_model("openai/gpt-4.1")捕获到"gpt-4.1"；
• openai_prefix_mode="model_id"：捕获到"openai/gpt-4.1"；
• unknown_prefix_mode="model_id"：get_model("openrouter/openai/gpt-4.1")捕获到"openrouter/openai/gpt-4.1"。

六、底层连接行为：keepalive、消息上限与 60 分钟约束

通过responses_websocket_options可以定制底层 WebSocket 行为（依据 docs/running_agents.md 与 docs/models/index.md）：

• ping_interval/ping_timeout：控制心跳保活。长推理轮次或网络延迟抖动导致 keepalive 超时时，增大ping_timeout以容忍延迟的 pong 帧；或设置ping_timeout=None关闭心跳超时（同时保持 ping 启用）。官方示例使用{"ping_interval": 20.0, "ping_timeout": 60.0}。
• max_size：SDK 默认关闭传入消息大小限制（max_size=None）。对运行在代理之后或内存受限容器中的长生命周期 Agent 进程，可设responses_websocket_options={"max_size": 8 * 1024 * 1024}来限制单条消息的内存占用。

服务端约束（连接复用不会消除这些限制）：

• 每条 WebSocket 连接一次只处理一个响应，且连接限制为60 分钟；超过限制需新建连接，需要并行运行时应使用多条连接；
• 服务端只在连接本地内存中保留最近一个响应；某轮4xx/5xx失败会逐出previous_response_id所引用的响应。重连后，store=False与 ZDR（zero data retention）流程无法恢复未缓存的previous_response_id——此时应开启新链条（previous_response_id=None并发送完整输入上下文），或从本地管理的 session 状态重建上下文；
• 可靠性优先于延迟的场景，建议退回 HTTP/SSE 传输。

两个工程注意事项：

1. 退出上下文前必须排空/关闭流式迭代器。若在 WebSocket 请求仍在飞行时退出async with上下文，可能强制关闭共享连接（docs/running_agents.md）。
2. 若环境中未安装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_MODE=auto（脚本化运行自动批准 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_approval=True) def submit_refund(order_id: str, amount: float, reason: str) -> dict[str, Any]: """创建退款申请。该工具需要人工审批。""" async def run_streamed_turn(ws, agent, prompt, *, previous_response_id=None): result = ws.run_streamed(agent, prompt, previous_response_id=previous_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="", flush=True) elif raw.type == "response.output_text.delta": print(raw.delta, end="", flush=True) 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(f"Approve {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(name="RefundPolicySpecialist", model=model_name, ...) support_agent = Agent( name="SupportAgent", tools=[lookup_order, policy_agent.as_tool(...), submit_refund], model=model_name, model_settings=ModelSettings(max_tokens=200, reasoning=Reasoning(effort="medium", summary="detailed")), ) 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_id=first_response_id, )

示例注释明确指出：完全可以跳过该辅助器直接调用Runner.run_streamed(...)，那样也能工作，但每次运行都会重新建连/连接，除非手动复用同一份RunConfig/provider；而该辅助器让跨轮次（以及嵌套 agent-as-tool 运行）的复用变得简单，从而保持 WebSocket 连接温热。

值得注意的是中断（interruption）处理流程：needs_approval=True的工具触发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 完整覆盖了本模块的行为契约：

ResponsesWebSocketSession与responses_websocket_session均通过 src/agents/init.py 作为公开 API 导出，并纳入 tests/fixtures/released_api_contract.json 的发布 API 契约校验范围。

九、使用要点速查

1. 会话生命周期：始终用async with responses_websocket_session(...) as ws:包裹，退出时自动aclose()释放连接；不要在请求飞行时退出上下文。
2. 连接复用：所有轮次共享同一provider与RunConfig，嵌套 agent-as-tool 调用也会继承同一run_config；不要把run_config再次传给ws.run(...)/ws.run_streamed(...)。
3. 多轮衔接：流式结果用result.last_response_id配合previous_response_id衔接上下文；若发生重连或失败逐出，需用完整输入上下文开新链条。
4. 长推理场景：调大ping_timeout或设ping_timeout=None；内存受限环境设max_size上限。
5. 兼容端点：自定义 OpenAI 兼容端点需支持 WebSocket/responses端点，必要时显式设置websocket_base_url；期望字面命名空间模型 ID 时使用openai_prefix_mode="model_id"/unknown_prefix_mode="model_id"。
6. 概念区分：这是 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