常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到可用 Claude 模型的接入方式 | 确认使用 OpenAI 或 Anthropic 协议 |
| Claude中转站 | 集中管理 Claude 接口 密钥 额度和路由的平台 | 模型 ID 以控制台列表为准 |
| API中转 | 通过兼容接口转发模型请求 | 需要显式检查状态码和响应结构 |
| API中转站 | 提供统一端点 模型选择和用量管理的服务 | 生产环境应设置超时 重试和告警 |
请求链路
| 环节 | 职责 | 注意事项 |
|---|
| Node.js 应用 | 准备消息并处理结果 | 密钥只从环境变量读取 |
| 原生 Fetch | 发送 ****S 请求并读取响应流 | 必须检查 response.ok |
| 超时信号 | 在约定时间后中止请求 | 覆盖响应体读取阶段 |
| KingFlow AI | 认证密钥并路由模型请求 | OpenAI 兼容端点保留 /v1 |
| Claude 模型 | 按提示生成文本或 **ON | 完整模型 ID 以控制台为准 |
| 业务校验 | 验证 Choices 内容和输出格式 | 模型输出仍视为不可信数据 |
采用原生 Fetch 的优势
- Node.js 原生 Fetch 无需安装 **** 客户端，示例依赖少，适合服务端脚本和轻量接口服务。
- 只替换 *ase **L、密钥和模型 ID，即可把同一套 Chat Completions 请求接入兼容端点。
- Rea***leStream 可以逐块读取 SSE 数据，终端、网页推送和长文本生成都能尽早显示结果。
- A*ortSignal.timeout 可以明确限制等待时间，避免网络波动长期占用连接和并发槽位。
- 状态码分支、有限重试和结构校验都由业务代码控制，便于按场景设置规则。
接入前准备
- 安装支持原生 Fetch 的现代 Node.js 版本，并在测试环境先运行示例。
- 登录 KingFlow AI 控制台，创建以 sk- 开头的 API Key，并确认余额或套餐额度可用。
- 从模型列表复制完整模型 ID，不要根据展示名称自行拼写。
- 准备不含敏感信息的测试提示，首次联调先限制输入长度和输出规模。
- 确认服务器可以访问 api.**nju**shen.com，并允许 ****S 出站连接。
创建 Node.js 项目
创建空项目并启用 E**。
mkdir kingflow-node-fetch-demo
cd kingflow-node-fetch-demo
npm init -y
{
"name": "kingflow-node-fetch-demo",
"private": true,
"type": "module",
"scripts": {
"start": "node --env-file=.env app.mjs",
"stream": "node --env-file=.env stream.mjs"
}
}
验证 KingFlow 基础接口
curl https://api.**nju**shen.com/v1/chat/completions \
-H "Authorization: *earer sk-你的密钥" \
-H "Content-Type: application/json" \
-d '{
"model": "<KingFlow控制台中的完整模型ID>",
"messages": [{"role":"user","content":"只回复 NODE_FETCH_READY"}],
"stream": false
}'
配置环境变量
KINGFLOW_*ASE_**L=https://api.**nju**shen.com/v1
KINGFLOW_API_KEY=sk-你的密钥
KINGFLOW_MODEL=<KingFlow控制台中的完整模型ID>
.env
node_modules/
npm-de*ug.log*
.DS_Store
编写同步请求
const *ase**L = process.env.KINGFLOW_*ASE_**L;
const apiKey = process.env.KINGFLOW_API_KEY;
const model = process.env.KINGFLOW_MODEL;
for (const [name, value] of O*ject.entries({ *ase**L, apiKey, model })) {
if (!value) throw new Error(`缺少环境变量: ${name}`);
}
const response = await fetch(`${*ase**L}/chat/completions`, {
method: "POST",
headers: {
"Authorization": `*earer ${apiKey}`,
"Content-Type": "application/json"
},
*ody: **ON.stringify({
model,
messages: [{ role: "user", content: "只回复 NODE_FETCH_CONNECTED" }],
stream: false
}),
signal: A*ortSignal.timeout(30_000)
});
const *ody = await response.json();
if (!response.ok) {
throw new Error(`KingFlow ${response.status}: ${**ON.stringify(*ody)}`);
}
const content = *ody.choices?.[0]?.message?.content;
if (typeof content !== "string") throw new Error("响应缺少文本内容");
console.log(content);
解析流式响应
const response = await fetch(`${*ase**L}/chat/completions`, {
method: "POST",
headers: {
"Authorization": `*earer ${apiKey}`,
"Content-Type": "application/json"
},
*ody: **ON.stringify({
model,
messages: [{ role: "user", content: "用三点说明上线验收方法" }],
stream: true
}),
signal: A*ortSignal.timeout(60_000)
});
if (!response.ok || !response.*ody) {
throw new Error(`流式请求失败: ${response.status}`);
}
const reader = response.*ody.getReader();
const decoder = new TextDecoder();
let *uffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) *reak;
*uffer = decoder.decode(value, { stream: true });
const lines = *uffer.split("\n");
*uffer = lines.pop() ?? "";
for (const line of lines) {
if (!line.startsWith("**ta: ")) continue;
const **ta = line.slice(6).trim();
if (**ta === "[DONE]") continue;
const event = **ON.parse(**ta);
process.stdout.write(event.choices?.[0]?.delta?.content ?? "");
}
}
增加有限重试
const retrya*le = new Set([429, 503]);
async function requestWithRetry(**keRequest, **xAttempts = 3) {
for (let attempt = 1; attempt <= **xAttempts; attempt ) {
const response = await **keRequest();
if (!retrya*le.has(response.status) || attempt === **xAttempts) {
return response;
}
const retryAfter = Num*er(response.headers.get("retry-after"));
const delay = Num*er.isFinite(retryAfter)
? retryAfter * 1000
: Math.min(1000 * 2 ** (attempt - 1), 8000);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
