2026年openlux api 怎么流式输出：Python 与 Node.js 流式调用示例及参数说明

2026年openlux api 怎么流式输出：Python 与 Node.js 流式调用示例及参数说明

调用大模型时，用户最难受的不是等待，而是没有任何反馈的等待。流式输出把结果一段段推回来，第一段内容出现的时间通常远早于完整回答生成的时间，体验差距非常明显。

下面从 Python 和 Node.js 两条路径，把 openlux api 怎么流式输出 的关键点拆开讲：请求参数怎么设、返回的数据长什么样、逐块解析的代码怎么写，以及最容易踩的几个坑。

流式输出在协议层发生了什么

绝大多数 OpenAI 兼容接口的流式输出走的是 SSE（Server-Sent Events）：服务端保持 HTTP 连接不关闭，按行推送 data: {...} 形式的数据块，最后一个块是 data: [DONE]，表示本次生成结束。每个块里通常只包含这次新增的内容增量，而不是完整文本。

data: {\"choices\":[{\"delta\":{\"content\":\"流式\"}}]}
data: {\"choices\":[{\"delta\":{\"content\":\"输出\"}}]}
data: [DONE]

所以客户端要做两件事：逐块读取并解析 JSON，然后把 delta 内容追加到已有文本上。一个常见错误是把原始字符串直接显示出来，结果界面上会混进 JSON 包装、空行和 data: 前缀——看起来“能跑”，但输出是脏的。

动手之前先确认三件事

• Base URL：以控制台给出的接口地址为准，注意末尾是否需要带 /v1，拼错通常直接 404。
• 模型名称：必须使用控制台中实际列出的名称，大小写和连字符都要一致。
• 是否支持流式：部分模型或接口形态对 stream 参数支持有限，建议先用一段短提示词小流量验证。

这三项信息可以在 千聚AI中转站 的控制台与接入文档中核对，确认接口地址和模型名称无误后再写业务代码，能省掉大量排查时间。

Python 流式调用示例

使用官方风格的 SDK 时，只需把 stream 设为 True，然后迭代返回对象，逐个取增量内容。

from openai import OpenAI

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="https://your-endpoint.example.com/v1",
)

stream = client.chat.completions.create(
    model="MODEL_NAME",
    messages=[{"role": "user", "content": "用三句话说明流式输出的价值"}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

如果不依赖 SDK，直接用 requests 也能实现，关键是把 stream=True 传给 requests，并用 iter_lines() 逐行处理响应。下面这段保留了跳过空行和处理结束标记的逻辑：

import json, requests

resp = requests.post(
    "https://your-endpoint.example.com/v1/chat/completions",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={
        "model": "MODEL_NAME",
        "messages": [{"role": "user", "content": "你好"}],
        "stream": True,
    },
    stream=True,
    timeout=(10, 120),
)

for line in resp.iter_lines(decode_unicode=True):
    if not line or not line.startswith("data:"):
        continue
    payload = line[5:].strip()
    if payload == "[DONE]":
        break
    delta = json.loads(payload)["choices"][0]["delta"].get("content")
    if delta:
        print(delta, end="", flush=True)

注意 timeout 写成了元组：第一个值是建立连接的超时，第二个值是读取数据的超时。流式连接存活时间比普通请求长，读超时设得太短会在生成中途被掐断。

Node.js 流式调用示例

Node.js 18 之后内置 fetch，可以直接读取响应体的可读流。难点在于分块到达的数据可能把一行 SSE 拆成两半，因此必须保留一个缓冲区。

const res = await fetch(`${BASE_URL}/chat/completions`, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: `Bearer ${API_KEY}`,
  },
  body: JSON.stringify({
    model: MODEL_NAME,
    messages: [{ role: "user", content: "你好" }],
    stream: true,
  }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder("utf-8");
let buffer = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n");
  buffer = lines.pop();

  for (const line of lines) {
    if (!line.startsWith("data:")) continue;
    const payload = line.slice(5).trim();
    if (payload === "[DONE]") continue;
    const delta = JSON.parse(payload).choices[0].delta?.content;
    if (delta) process.stdout.write(delta);
  }
}

这里有两个细节值得留意：decoder.decode(value, { stream: true }) 让多字节汉字在跨块时不会被截断；lines.pop() 把最后一段不完整的行留在缓冲区里，等下一次数据到达再拼接。

几个参数的实际影响

常见问题排查

首字迟迟不出现

如果接口本身是流式的，但前端一直空白，最先怀疑的是中间层缓冲。反向代理、网关或某些云函数默认会等响应结束才转发，需要关闭缓冲。其次检查代码里是否把整个响应读完才处理。

中文乱码或末尾缺字

这是典型的跨块解码问题。UTF-8 下一个汉字占三个字节，如果按块用普通方式解码，可能把汉字切成两半。Node.js 要用流式解码模式，Python 用 iter_lines(decode_unicode=True) 一般不会遇到这个问题。

生成到一半被断开

多半是超时或空闲检测。检查读取超时、代理的空闲连接时间、以及负载均衡器是否对长连接有额外限制。流式请求的持续时间通常比普通接口长得多，默认超时值往往不够用。

流式输出能不能用、好不好用，取决于你实际使用的接口地址、模型和网络链路。建议先用一段短提示词验证是否能稳定逐块返回，再接入正式业务，不要直接在生产环境上做第一次测试。

把上面几段代码跑通之后，可以进一步做工程化处理：为每个请求加唯一标识，记录首字延迟和总耗时；在客户端做打字机效果时把增量内容放进队列，避免频繁更新界面造成卡顿；失败时保留已输出内容，而不是整段重来。这些细节决定了流式体验是“看得见的快”还是“看起来更乱”。

如果要长期维护多模型调用，建议把接口地址和密钥放在配置层统一管理，而不是散落在各个项目里。像 千聚AI中转站 这类平台提供统一的接入方式和模型列表，切换模型时只需改配置中的模型名称，流式解析代码基本可以复用。

代码已经跑通，接下来要解决的是接入环境：注册账号后获取专属 API Key，在控制台确认自己的 Base URL 与可用模型名称，然后用一段短提示词完成第一次流式测试，确认逐块返回正常再接入业务。

注册千聚AI中转站，获取 API Key 开始流式测试