服务端怎么把大模型的流式响应转发给前端？

event: delta
data: {"text":"你好"}

event: done
data: {"messageId":"8f3c…","usage":{"input":1200,"output":356}}

import { randomUUID } from 'node:crypto'

app.post('/api/chat', async (req, res) => {
  const controller = new AbortController()
  res.on('close', () => {
    if (!res.writableFinished) controller.abort() // 客户端提前断开，中止上游
  })

  const upstream = await fetch(`${process.env.LLM_BASE_URL}/chat/completions`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${process.env.LLM_API_KEY}` },
    body: JSON.stringify({
      model: process.env.LLM_MODEL,
      messages: req.body.messages,
      stream: true,
      stream_options: { include_usage: true },
    }),
    signal: controller.signal,
  }).catch(() => null)
  if (!upstream?.ok) {
    if (upstream) console.error('upstream', upstream.status, await upstream.text()) // 详情只记日志
    return res.status(502).json({ message: 'AI 服务暂时不可用，请稍后再试' })
  }

  res.writeHead(200, { 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no' })
  const send = (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
  const heartbeat = setInterval(() => res.write(': ping\n\n'), 15000)

  const messageId = randomUUID()
  let text = ''
  let usage = null
  let status = 'completed'
  try {
    for await (const chunk of readUpstream(upstream.body)) {
      const delta = chunk.choices?.[0]?.delta?.content
      if (delta) {
        text += delta
        send('delta', { text: delta })
      }
      if (chunk.usage) usage = chunk.usage
    }
    send('done', { messageId, usage: usage && { input: usage.prompt_tokens, output: usage.completion_tokens } })
  } catch {
    status = controller.signal.aborted ? 'stopped' : 'failed'
    if (status === 'failed') send('error', { message: '生成中断，请重试' })
  } finally {
    clearInterval(heartbeat) // 先停心跳，再结束响应
    res.end()
    if (text) await saveMessage({ id: messageId, text, usage, status }).catch(console.error)
  }
})

// 逐行解析上游的 SSE。按 OpenAI 兼容格式简化处理：每个事件只有一行 data
async function* readUpstream(body) {
  const decoder = new TextDecoder()
  let buffer = ''
  for await (const bytes of body) {
    buffer += decoder.decode(bytes, { stream: true })
    const lines = buffer.split('\n')
    buffer = lines.pop() // 最后一行可能不完整，留到下次
    for (const line of lines) {
      if (!line.startsWith('data:')) continue
      const data = line.slice(5).trim()
      if (data === '[DONE]') return
      yield JSON.parse(data)
    }
  }
}