Node/Express集成ChatGPT：混合分块响应的客户端接收异常问题

在Node/Express中集成ChatGPT时，期望先向客户端返回JSON格式的元数据，再流式传输ChatGPT的响应内容。目前程序无报错，但客户端的onprogress处理器需等待服务器开始发送GPT的分块响应后，才会开始接收之前的元数据分块。

服务端代码 #

// Send response message array to client
res.setHeader('Transfer-Encoding', 'chunked')
let chunkedJSON = chunkString(JSON.stringify(myJSON), 20)
for(let i = 0; i < chunkedJSON.length; i++){
  res.write(chunkedJSON[i])
}
res.write('END-JSON')

// Get GPT Response
const stream = await openai.chat.completions.create(
  {
    messages: messages,
    model: thread.model,
    stream: true
  },
  {
    responseType: "stream"
  }
)

// Stream ChatGPT response
for await (const part of stream) {
  tokens++ // each chunk from the completion API is a token
  if (part.choices[0].finish_reason === "stop") {
    res.end();
  } else {
    res.write(part.choices[0]?.delta?.content || "")
  }
}

客户端代码 #

let xhr = new XMLHttpRequest();
xhr.open("POST", environment.serverBaseUrl + '/chat/message', true);
xhr.setRequestHeader("Content-Type", "application/json");
xhr.setRequestHeader("x-access-token", this.userAuth!);
xhr.onprogress = function () {
  var curr_index = xhr.responseText.length;
  if (last_index == curr_index) return; 
  var s = xhr.responseText.substring(last_index, curr_index);
  last_index = curr_index;
  console.log("PROGRESS:", s);
}
xhr.onload = function () {
  console.log("COMPLETE:", xhr.responseText);
}
xhr.send(JSON.stringify({
  test: true,
  message: message
}));

问题根源在于Express默认会缓冲响应数据，直到积累到一定大小才会发送给客户端，导致先写入的JSON元数据被暂存在服务器缓冲区，直到GPT流的分块进来才一起推送。

修改服务端代码，强制立即发送分块 #

1. 添加X-Accel-Buffering: no头部禁用反向代理（如Nginx）的缓冲，同时调用res.flushHeaders()确保响应头部立即发送。
2. 每次调用res.write()后，调用res.flush()强制将当前缓冲区数据发送给客户端。

修改后的服务端代码：

// Send response message array to client
res.setHeader('Transfer-Encoding', 'chunked')
res.setHeader('X-Accel-Buffering', 'no') // 禁用代理层缓冲
res.flushHeaders() // 立即发送响应头部

let chunkedJSON = chunkString(JSON.stringify(myJSON), 20)
for(let i = 0; i < chunkedJSON.length; i++){
  res.write(chunkedJSON[i])
  await res.flush() // 强制发送当前元数据分块
}
res.write('END-JSON')
await res.flush() // 发送元数据结束标记

// Get GPT Response
const stream = await openai.chat.completions.create(
  {
    messages: messages,
    model: thread.model,
    stream: true
  },
  {
    responseType: "stream"
  }
)

// Stream ChatGPT response
for await (const part of stream) {
  tokens++
  if (part.choices[0].finish_reason === "stop") {
    res.end();
  } else {
    const content = part.choices[0]?.delta?.content || ""
    res.write(content)
    await res.flush() // 实时发送GPT响应分块
  }
}

客户端优化（可选） #

可以在onprogress中识别END-JSON标记，分离元数据和GPT响应内容：

let last_index = 0;
let metadataReceived = false;
xhr.onprogress = function () {
  var curr_index = xhr.responseText.length;
  if (last_index == curr_index) return; 
  var s = xhr.responseText.substring(last_index, curr_index);
  last_index = curr_index;

  if (!metadataReceived) {
    const endMarkerIndex = s.indexOf('END-JSON');
    if (endMarkerIndex !== -1) {
      // 解析元数据
      const metadataStr = s.substring(0, endMarkerIndex);
      try {
        const metadata = JSON.parse(metadataStr);
        console.log("元数据:", metadata);
      } catch (e) {
        console.error("解析元数据失败:", e);
      }
      metadataReceived = true;
      // 处理标记后的GPT响应片段
      const gptContent = s.substring(endMarkerIndex + 8);
      if (gptContent) console.log("GPT响应片段:", gptContent);
    } else {
      console.log("元数据片段:", s);
    }
  } else {
    console.log("GPT响应片段:", s);
  }
}

内容的提问来源于stack exchange，提问作者E-Madd