380 lines
13 KiB
JavaScript
380 lines
13 KiB
JavaScript
/**
|
||
* ═══════════════════════════════════════════════════════════
|
||
* 🎯 阿里云百炼 · 微调模型调用 (DashScope compatible-mode)
|
||
* ═══════════════════════════════════════════════════════════
|
||
*
|
||
* 编号: ZY-FTCHAT-DS-001
|
||
* 守护: 铸渊 · ICE-GL-ZY001
|
||
*
|
||
* 配置:
|
||
* FT_DASHSCOPE_API_KEY — 单独区分商业模型与微调模型的密钥
|
||
* FT_MODEL_SYSTEM — 微调系统线 (默认 qwen3-8b-ft-202604281809-9f30 · 冰朔 D69 提供)
|
||
* FT_MODEL_NAIPPING — 微调奶瓶线 (默认同系统线 · 待奶瓶专属微调上线后再分流)
|
||
* FT_MODEL_FALLBACK — 当微调模型不存在/无权限时自动降级到的基础模型
|
||
* (默认 qwen-turbo) · 保证 model_not_found 时聊天不挂
|
||
*
|
||
* 使用 OpenAI 兼容模式: /compatible-mode/v1/chat/completions
|
||
* SSE 流式 + 非流式降级。
|
||
*/
|
||
|
||
'use strict';
|
||
|
||
const https = require('https');
|
||
|
||
const ENDPOINT_HOST = 'dashscope.aliyuncs.com';
|
||
const ENDPOINT_PATH = '/compatible-mode/v1/chat/completions';
|
||
|
||
const MODEL_SYSTEM = process.env.FT_MODEL_SYSTEM || 'qwen3-8b-ft-202604281809-9f30';
|
||
const MODEL_NAIPPING = process.env.FT_MODEL_NAIPPING || 'qwen3-8b-ft-202604281809-9f30';
|
||
const MODEL_FALLBACK = process.env.FT_MODEL_FALLBACK || 'qwen-turbo';
|
||
|
||
// 已确认在 DashScope 账号下不存在的模型 ID, 后续直接走 fallback, 避免每轮都 404
|
||
const missingModels = new Set();
|
||
|
||
function pickModel(variant) {
|
||
const primary = variant === 'naipping' ? MODEL_NAIPPING : MODEL_SYSTEM;
|
||
if (missingModels.has(primary)) return MODEL_FALLBACK;
|
||
return primary;
|
||
}
|
||
|
||
function getApiKey() {
|
||
const key = process.env.FT_DASHSCOPE_API_KEY;
|
||
if (!key) throw new Error('FT_DASHSCOPE_API_KEY 未配置');
|
||
return key;
|
||
}
|
||
|
||
/**
|
||
* 内部: 单次流式调用 (不带 fallback). 遇到 model_not_found 直接 reject, 由外层决定是否重试。
|
||
*/
|
||
function _streamChatOnce(args) {
|
||
const { variant, messages, res, modelOverride } = args;
|
||
const apiKey = getApiKey();
|
||
const model = modelOverride || pickModel(variant);
|
||
|
||
const body = JSON.stringify({
|
||
model,
|
||
messages,
|
||
stream: true,
|
||
max_tokens: 2048,
|
||
temperature: 0.8
|
||
});
|
||
|
||
return new Promise((resolve, reject) => {
|
||
const req = https.request({
|
||
hostname: ENDPOINT_HOST,
|
||
path: ENDPOINT_PATH,
|
||
method: 'POST',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': `Bearer ${apiKey}`,
|
||
'Accept': 'text/event-stream',
|
||
'Content-Length': Buffer.byteLength(body)
|
||
},
|
||
timeout: 120000
|
||
}, (upstream) => {
|
||
if (upstream.statusCode !== 200) {
|
||
let errBuf = '';
|
||
upstream.on('data', c => { errBuf += c; });
|
||
upstream.on('end', () => {
|
||
const msg = `DashScope HTTP ${upstream.statusCode}: ${errBuf.slice(0, 300)}`;
|
||
console.error('[FTCHAT DS]', msg);
|
||
// 标记缺失模型, 抛出可识别错误供外层重试
|
||
let isModelNotFound = false;
|
||
try {
|
||
const parsed = JSON.parse(errBuf);
|
||
isModelNotFound = parsed && parsed.error && parsed.error.code === 'model_not_found';
|
||
} catch (_e) { /* ignore */ }
|
||
if (isModelNotFound) {
|
||
missingModels.add(model);
|
||
const err = new Error(msg);
|
||
err.modelNotFound = true;
|
||
err.attemptedModel = model;
|
||
return reject(err);
|
||
}
|
||
reject(new Error(msg));
|
||
});
|
||
return;
|
||
}
|
||
|
||
let fullText = '';
|
||
let buf = '';
|
||
let usage = null;
|
||
|
||
upstream.on('data', (chunk) => {
|
||
buf += chunk.toString('utf8');
|
||
const lines = buf.split('\n');
|
||
buf = lines.pop();
|
||
for (const line of lines) {
|
||
const trimmed = line.trim();
|
||
if (!trimmed.startsWith('data:')) continue;
|
||
const payload = trimmed.slice(5).trim();
|
||
if (payload === '[DONE]') {
|
||
try { res.write('data: [DONE]\n\n'); } catch (_e) { /* ignore */ }
|
||
continue;
|
||
}
|
||
let parsed;
|
||
try { parsed = JSON.parse(payload); } catch (_e) { continue; }
|
||
if (parsed.usage) usage = parsed.usage;
|
||
const delta = parsed.choices && parsed.choices[0] && parsed.choices[0].delta;
|
||
const piece = delta && delta.content;
|
||
if (piece) {
|
||
fullText += piece;
|
||
try {
|
||
res.write(`data: ${JSON.stringify({ delta: piece })}\n\n`);
|
||
} catch (_e) { /* response closed */ }
|
||
}
|
||
}
|
||
});
|
||
|
||
upstream.on('end', () => {
|
||
resolve({ full: fullText, usage, model });
|
||
});
|
||
|
||
upstream.on('error', (err) => {
|
||
console.error('[FTCHAT DS] upstream error:', err.message);
|
||
reject(err);
|
||
});
|
||
});
|
||
|
||
req.on('timeout', () => {
|
||
req.destroy(new Error('upstream timeout'));
|
||
});
|
||
req.on('error', (err) => {
|
||
console.error('[FTCHAT DS] request error:', err.message);
|
||
reject(err);
|
||
});
|
||
|
||
req.write(body);
|
||
req.end();
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 流式调用 DashScope, 通过 SSE 把 delta 推送到 res（已设置 SSE 头）
|
||
* model_not_found 时自动降级到 FT_MODEL_FALLBACK (默认 qwen-turbo).
|
||
* @param {object} args
|
||
* @param {string} args.variant
|
||
* @param {Array} args.messages OpenAI 格式: [{role, content}]
|
||
* @param {object} args.res Express response (已 writeHead text/event-stream)
|
||
* @returns {Promise<{ full: string, usage?: object, model: string }>}
|
||
*/
|
||
async function streamChat(args) {
|
||
try {
|
||
return await _streamChatOnce(args);
|
||
} catch (err) {
|
||
if (err && err.modelNotFound && err.attemptedModel !== MODEL_FALLBACK) {
|
||
console.warn(
|
||
`[FTCHAT DS] model "${err.attemptedModel}" not accessible, falling back to "${MODEL_FALLBACK}"`
|
||
);
|
||
try {
|
||
args.res.write(
|
||
`data: ${JSON.stringify({ notice: `微调模型暂不可用, 已自动降级到 ${MODEL_FALLBACK}` })}\n\n`
|
||
);
|
||
} catch (_e) { /* ignore */ }
|
||
return _streamChatOnce(Object.assign({}, args, { modelOverride: MODEL_FALLBACK }));
|
||
}
|
||
// 其他错误: 把上游真实错误透传给前端 (而不是模糊的 '上游模型暂不可用'), 便于诊断
|
||
try {
|
||
args.res.write(`data: ${JSON.stringify({ error: true, message: err.message || '上游模型暂不可用' })}\n\n`);
|
||
} catch (_e) { /* ignore */ }
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 非流式调用（用于 memory-agent 压缩等场景）
|
||
* model_not_found 时自动降级到 FT_MODEL_FALLBACK
|
||
* @returns {Promise<string>}
|
||
*/
|
||
async function chatOnce(args) {
|
||
const { variant, messages, max_tokens } = args;
|
||
const tryModel = async (model) => {
|
||
const apiKey = getApiKey();
|
||
const body = JSON.stringify({
|
||
model,
|
||
messages,
|
||
stream: false,
|
||
max_tokens: max_tokens || 800,
|
||
temperature: 0.5
|
||
});
|
||
|
||
return new Promise((resolve, reject) => {
|
||
const req = https.request({
|
||
hostname: ENDPOINT_HOST,
|
||
path: ENDPOINT_PATH,
|
||
method: 'POST',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': `Bearer ${apiKey}`,
|
||
'Content-Length': Buffer.byteLength(body)
|
||
},
|
||
timeout: 60000
|
||
}, (upstream) => {
|
||
let buf = '';
|
||
upstream.on('data', c => { buf += c; });
|
||
upstream.on('end', () => {
|
||
if (upstream.statusCode !== 200) {
|
||
let isModelNotFound = false;
|
||
try {
|
||
const parsed = JSON.parse(buf);
|
||
isModelNotFound = parsed && parsed.error && parsed.error.code === 'model_not_found';
|
||
} catch (_e) { /* ignore */ }
|
||
if (isModelNotFound) {
|
||
missingModels.add(model);
|
||
const err = new Error(`DashScope HTTP ${upstream.statusCode}: ${buf.slice(0, 200)}`);
|
||
err.modelNotFound = true;
|
||
err.attemptedModel = model;
|
||
return reject(err);
|
||
}
|
||
return reject(new Error(`DashScope HTTP ${upstream.statusCode}: ${buf.slice(0, 200)}`));
|
||
}
|
||
try {
|
||
const parsed = JSON.parse(buf);
|
||
const content = parsed.choices && parsed.choices[0] && parsed.choices[0].message && parsed.choices[0].message.content;
|
||
resolve(content || '');
|
||
} catch (e) {
|
||
reject(e);
|
||
}
|
||
});
|
||
});
|
||
req.on('timeout', () => req.destroy(new Error('upstream timeout')));
|
||
req.on('error', reject);
|
||
req.write(body);
|
||
req.end();
|
||
});
|
||
};
|
||
|
||
const primary = pickModel(variant);
|
||
try {
|
||
return await tryModel(primary);
|
||
} catch (err) {
|
||
if (err && err.modelNotFound && err.attemptedModel !== MODEL_FALLBACK) {
|
||
console.warn(
|
||
`[FTCHAT DS] (chatOnce) model "${err.attemptedModel}" not accessible, falling back to "${MODEL_FALLBACK}"`
|
||
);
|
||
return tryModel(MODEL_FALLBACK);
|
||
}
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* ═════════════════════════════════════════════════════════════
|
||
* 纯字节管道 (铸渊 · 2026-05-02)
|
||
* ═════════════════════════════════════════════════════════════
|
||
* 上游百炼 SSE 字节直接 pipe 到浏览器, 服务端零解析、零打包。
|
||
* 仅在以下两种情况打断管道:
|
||
* 1. 上游非 200 → 收集错误体, 判定 model_not_found 后由 pipeChat 决定降级
|
||
* 2. 上游 timeout / 网络错误 → reject 给上层 (此时 res headers 可能尚未发出)
|
||
*
|
||
* 与 streamChat 的区别:
|
||
* - streamChat: 服务端解析 SSE → 抽出 delta.content → 重新发 `{delta: piece}` 帧
|
||
* (本轮已弃用于聊天链路, 仅保留兼容)
|
||
* - pipeChat: `upstream.pipe(res)`, 字节透传, 浏览器直接读百炼原生格式
|
||
*/
|
||
function _pipeChatOnce(args) {
|
||
const { variant, messages, res, modelOverride } = args;
|
||
const apiKey = getApiKey();
|
||
const model = modelOverride || pickModel(variant);
|
||
|
||
const body = JSON.stringify({
|
||
model,
|
||
messages,
|
||
stream: true,
|
||
max_tokens: 2048,
|
||
temperature: 0.8
|
||
});
|
||
|
||
return new Promise((resolve, reject) => {
|
||
const upReq = https.request({
|
||
hostname: ENDPOINT_HOST,
|
||
path: ENDPOINT_PATH,
|
||
method: 'POST',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': `Bearer ${apiKey}`,
|
||
'Accept': 'text/event-stream',
|
||
'Content-Length': Buffer.byteLength(body)
|
||
},
|
||
timeout: 120000
|
||
}, (upstream) => {
|
||
if (upstream.statusCode !== 200) {
|
||
let errBuf = '';
|
||
upstream.on('data', c => { errBuf += c; });
|
||
upstream.on('end', () => {
|
||
const msg = `DashScope HTTP ${upstream.statusCode}: ${errBuf.slice(0, 300)}`;
|
||
console.error('[FTCHAT DS pipe]', msg);
|
||
let isModelNotFound = false;
|
||
try {
|
||
const parsed = JSON.parse(errBuf);
|
||
isModelNotFound = parsed && parsed.error && parsed.error.code === 'model_not_found';
|
||
} catch (_e) { /* ignore */ }
|
||
if (isModelNotFound) {
|
||
missingModels.add(model);
|
||
const err = new Error(msg);
|
||
err.modelNotFound = true;
|
||
err.attemptedModel = model;
|
||
return reject(err);
|
||
}
|
||
reject(new Error(msg));
|
||
});
|
||
return;
|
||
}
|
||
|
||
// 200 OK: 字节级管道. 上游 SSE 头透传给浏览器
|
||
if (!res.headersSent) {
|
||
res.writeHead(200, {
|
||
'Content-Type': upstream.headers['content-type'] || 'text/event-stream; charset=utf-8',
|
||
'Cache-Control': 'no-cache, no-transform',
|
||
'Connection': 'keep-alive',
|
||
'X-Accel-Buffering': 'no'
|
||
});
|
||
if (typeof res.flushHeaders === 'function') res.flushHeaders();
|
||
}
|
||
|
||
upstream.pipe(res, { end: false });
|
||
upstream.on('end', () => resolve({ model }));
|
||
upstream.on('error', (err) => {
|
||
console.error('[FTCHAT DS pipe] upstream error:', err.message);
|
||
reject(err);
|
||
});
|
||
});
|
||
|
||
upReq.on('timeout', () => upReq.destroy(new Error('upstream timeout')));
|
||
upReq.on('error', (err) => {
|
||
console.error('[FTCHAT DS pipe] request error:', err.message);
|
||
reject(err);
|
||
});
|
||
|
||
upReq.write(body);
|
||
upReq.end();
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 纯字节管道版本: 把上游百炼 SSE 字节直接 pipe 到 res.
|
||
* model_not_found 时自动降级到 FT_MODEL_FALLBACK (仅当 res headers 尚未发出时).
|
||
* 上游 200 之后, 服务端不再解析任何字节 — 浏览器直接读 OpenAI 兼容 SSE 格式.
|
||
*
|
||
* @returns {Promise<{ model: string }>}
|
||
*/
|
||
async function pipeChat(args) {
|
||
try {
|
||
return await _pipeChatOnce(args);
|
||
} catch (err) {
|
||
if (
|
||
err && err.modelNotFound &&
|
||
err.attemptedModel !== MODEL_FALLBACK &&
|
||
args.res && !args.res.headersSent
|
||
) {
|
||
console.warn(
|
||
`[FTCHAT DS pipe] model "${err.attemptedModel}" not accessible, falling back to "${MODEL_FALLBACK}"`
|
||
);
|
||
return _pipeChatOnce(Object.assign({}, args, { modelOverride: MODEL_FALLBACK }));
|
||
}
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
module.exports = { streamChat, pipeChat, chatOnce, pickModel, MODEL_SYSTEM, MODEL_NAIPPING, MODEL_FALLBACK };
|