通过 OpenAI 兼容的 Responses API 调用 GPT-5.6 Sol 时,只要向 /v1/responses 提交请求并把 stream 设为 true,服务端就会用服务器发送事件(Server-Sent Events,SSE)持续返回响应事件。客户端不应把每一行都当成最终文本,而要解析事件类型,只累加 response.output_text.delta 的 delta 字段,并在完成事件或流结束后收尾。
OpenAI 兼容描述的是请求和响应的协议形状,而不是说两个接口的数据结构完全相同。Chat Completions 通常从 choices[0].delta.content 取增量;Responses API 则发送带 type 的语义事件,文本增量常见于 response.output_text.delta。如果把旧接口的解析循环原样搬过来,即使连接成功,也可能一直打印空字符串。
Responses API 的请求主体使用 input,而不是必须构造 messages。简单文本可直接传字符串;需要区分角色或提交多段内容时,再使用结构化输入数组。GPT-5.6 Sol 支持 Responses 端点和流式输出,因此关键不在模型名称的特殊写法,而在兼容服务是否声明该模型具备 Responses 能力,以及客户端是否按 Responses 事件协议消费数据。
示例把密钥和兼容服务地址放进环境变量,避免硬编码到源码。服务地址应是兼容服务提供的 API 根地址,是否包含 /v1 要以服务商说明为准;拼接时应确保最终请求路径只有一段 /v1/responses。还要确认当前账户可以使用 gpt-5.6-sol,因为“接口兼容”并不自动代表每个模型都已开通。
export COMPAT_API_KEY="替换为实际密钥"
export COMPAT_BASE_URL="替换为兼容服务的 API 根地址"
生产环境应通过密钥管理服务或部署平台注入变量,不要把真实密钥提交到 Git、日志或前端代码中。Responses API 必须从可信后端调用;浏览器直连会把长期密钥暴露给访问者。
OpenAI Python SDK 可以代为处理 SSE 分帧和 JSON 解码。创建客户端时设置兼容服务的 base_url,调用 responses.create 并开启流式模式,然后按事件类型提取文本。下面的代码只输出文本增量,同时保留错误事件和完成状态的处理位置。
import os
import sys
from openai import OpenAI
client = OpenAI(
api_key=os.environ["COMPAT_API_KEY"],
base_url=os.environ["COMPAT_BASE_URL"].rstrip("/"),
timeout=60.0,
)
try:
stream = client.responses.create(
model="gpt-5.6-sol",
input="用三点说明 SSE 流式输出适合哪些场景。",
stream=True,
)
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
elif event.type == "response.failed":
print("n响应生成失败", file=sys.stderr)
elif event.type == "error":
print(f"n流错误: {event}", file=sys.stderr)
print()
except Exception as exc:
print(f"请求失败: {exc}", file=sys.stderr)
raise
flush=True 很重要。终端、容器日志和部分 Web 框架会缓冲标准输出,不主动刷新时,服务端虽然逐段发送,用户却可能在最后一刻才看到全部内容。不要依赖某个 SDK 版本必然暴露完全相同的事件类属性;升级依赖后,应先打印测试事件的 type,确认兼容服务与 SDK 的组合仍符合预期。
Node.js SDK 返回可异步迭代的流。循环主体同样按 event.type 分派,而不是读取 Chat Completions 的 choices。服务端程序还应下游连接是否关闭,避免用户离开页面后仍持续消耗模型额度。
import OpenAI from "openai";
const client = new OpenAI({
apiKey: process.env.COMPAT_API_KEY,
baseURL: process.env.COMPAT_BASE_URL.replace(//$/, ""),
timeout: 60_000,
});
const stream = await client.responses.create({
model: "gpt-5.6-sol",
input: "给出一个简短的 SSE 健康检查清单。",
stream: true,
});
for await (const event of stream) {
if (event.type === "response.output_text.delta") {
process.stdout.write(event.delta);
} else if (event.type === "response.failed") {
console.error("响应生成失败", event.response?.error);
} else if (event.type === "error") {
console.error("流错误", event);
}
}
process.stdout.write("n");
若使用 TypeScript,可让 SDK 提供的事件联合类型完成类型收窄。不要为图省事而把所有事件强制转换成任意类型,否则事件名称拼错、字段层级变化等问题会延迟到运行时才暴露。
需要调试代理、核对原始帧或在不支持官方 SDK 的运行环境中接入时,可以直接发送 HTTP 请求。请求头至少包含 Bearer 鉴权和 JSON 内容类型,并建议声明接受 text/event-stream。响应成功后,客户端必须按空行切分 SSE 消息,再逐行处理 event: 与 data: 字段。
网络数据块与 SSE 消息没有一一对应关系。一次读取可能只得到半行,也可能包含多个事件。因此不能对每个网络数据块直接调用 JSON 解析;必须保留未完成的尾部缓冲区,等遇到消息分隔符后再解析完整帧。还要兼容换行符差异,并忽略 SSE 注释行。
const endpoint = `${process.env.COMPAT_BASE_URL.replace(//$/, "")}/responses`;
const response = await fetch(endpoint, {
method: "POST",
headers: {
Authorization: `Bearer ${process.env.COMPAT_API_KEY}`,
"Content-Type": "application/json",
Accept: "text/event-stream",
},
body: JSON.stringify({
model: "gpt-5.6-sol",
input: "解释什么是事件增量。",
stream: true,
}),
});
if (!response.ok) {
const detail = await response.text();
throw new Error(`HTTP ${response.status}: ${detail}`);
}
if (!response.body) throw new Error("响应体为空");
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { value, done } = await reader.read();
buffer += decoder.decode(value || new Uint8Array(), { stream: !done });
const messages = buffer.split(/r?nr?n/);
buffer = messages.pop() || "";
for (const message of messages) {
const data = message
.split(/r?n/)
.filter((line) => line.startsWith("data:"))
.map((line) => line.slice(5).trimStart())
.join("n");
if (!data || data === "[DONE]") continue;
const event = JSON.parse(data);
if (event.type === "response.output_text.delta") {
process.stdout.write(event.delta);
}
if (event.type === "error" || event.type === "response.failed") {
console.error("流中出现错误", event);
}
}
if (done) break;
}
示例中的根地址应按服务商约定配置。如果根地址已经以 /v1 结尾,拼接 /responses 即可;如果配置的是域名根路径,则应在统一的客户端配置层补齐版本段,避免业务代码到处猜测地址形式。
常见架构是浏览器请求自己的应用后端,再由后端调用兼容 Responses API。此时后端既可以把上游 SSE 原样透传,也可以解析上游事件后,只向前端发送业务需要的文本事件。原样透传保留信息最完整;转换后再发送则更容易屏蔽供应商差异,但必须定义稳定的内部事件协议。
无论采用哪种方式,都要避免中间层缓冲。反向代理、压缩中间件、Serverless 网关和框架响应封装都可能攒够一定字节才下发。后端应设置正确的 SSE 内容类型,关闭不必要的响应缓存,并在每次写入后及时刷新。前端若用 fetch 发 POST 请求,需要读取 ReadableStream;浏览器原生 EventSource 更适合 GET,不能直接满足带 JSON 请求体和自定义鉴权头的常见调用。
流式连接建立成功只代表服务器开始处理,不代表整个生成成功。应用至少要区分正常完成、模型生成失败、协议错误、HTTP 错误和客户端主动取消。文本已经输出一部分后仍可能出现失败事件,因此业务层不能仅以“收到过 delta”判定成功。
用户点击停止时,应通过取消信号终止下游读取,并尽可能把取消传递给上游请求。若应用需要审计完整回答,应在服务端同步累积 delta,并且只在收到完成事件后把记录标记为完成;连接意外断开时可保留为中断状态,不能把不完整文本冒充最终答案。
连接前错误与流中断的重试策略不同。对于尚未收到任何输出的临时网络错误或限流,可以采用带随机抖动的指数退避;一旦已经把部分文本展示给用户,自动重试可能产生重复段落、重复计费或两条回答拼接,因此通常应提示中断,让用户明确选择重新生成。
鉴权失败、模型不可用、请求字段不合法等确定性错误不应盲目重试。遇到 401 或 403,应检查密钥与权限;遇到 400,应核对模型是否支持 Responses、参数名是否正确;遇到 429,应遵循服务端的重试提示并限制并发;遇到网关超时,应同时检查客户端超时、代理读取超时和供应商请求期限。
首先记录请求开始、收到响应头、首个文本 delta、完成事件四个时间点。首个 delta 明显早于完整响应完成,才能说明用户确实获得了流式体验。其次,在开发环境记录事件类型和序号,但不要记录密钥或敏感提示词。最后,用包含多段输出的提示进行测试,观察终端或页面是否连续更新,而不是结束时一次性出现。
若 HTTP 状态为 200 却没有文字,优先检查是否误读了 choices、是否过滤错事件类型,以及返回的是否只有推理或生命周期事件。若所有内容最后一起出现,重点检查输出刷新和代理缓冲。若 JSON 偶发解析失败,通常是把网络数据块误当成了完整 SSE 帧,应修复缓冲和分帧逻辑。
上线前应确认模型列表或服务说明明确包含 GPT-5.6 Sol 的 Responses 能力;最终路径指向 /v1/responses;请求使用 input 和 stream: true;消费者按 Responses 事件类型解析;密钥只保存在后端;代理不会缓存事件流;应用能处理失败、取消和超时;日志能够定位首字节延迟与完成状态,又不会泄露敏感数据。
把这些环节逐一验证后,SSE 实现就不只是“能看到逐字输出”,而是具备清晰的协议边界、可靠的错误状态和可观测性。真正需要长期维护的是事件解析与连接生命周期,而不是把某个非流式示例简单加上 stream: true。
Work Agent 与 Workflow 在产品架构和应用场景上有什么区别?
AI自动化业务工作流的搭建与落地实践
Work Agent、Workflow 与高代码应用应如何按自主性和可控性选型?
Workspace Agent 如何通过团队知识、审批和跨工具协作形成差异化?
从手写代码到 Vibe Coding:程序员真正不能丢掉的能力
Code Agent 应从集成、编排、治理、记忆和部署等哪些维度比较?