786 lines
31 KiB
TypeScript
786 lines
31 KiB
TypeScript
/**
|
||
* SSE 流式解析工具
|
||
*
|
||
* 解析 OpenAI 兼容的 Server-Sent Events (SSE) 流式响应，
|
||
* 产出 MetonaStreamEvent。DeepSeek、Agnes AI 和 MiMo 共享此工具。
|
||
*
|
||
* SSE 格式：data: {json}\n\n
|
||
* 结束标记：data: [DONE]
|
||
*/
|
||
|
||
import { nanoid } from 'nanoid';
|
||
import log from 'electron-log';
|
||
import type { MetonaStreamEvent, MetonaTokenUsage } from '../../types';
|
||
import { MetonaStreamEventType } from '../../types';
|
||
import { ContentFilterError } from '../base-adapter';
|
||
|
||
/**
|
||
* v0.6.4 根治「上游错误帧黑洞」:
|
||
*
|
||
* OpenAI 兼容网关常在中途发送 `data: {"error":{...}}` 数据帧（网关超时、限流、
|
||
* 配额耗尽、鉴权失效等）。旧解析器整条处理链只从 `chunk.choices?.[0]?.delta` 取数，
|
||
* 这类帧两个分支都不命中、零日志 —— 结果任何上游错误都伪装成"干净的空回复 +
|
||
* 正常 DONE"，且异常以普通事件而非异常形式出现，绕过了引擎 chatStreamWithRetry
|
||
* 的重试/故障转移通道。
|
||
*
|
||
* 现契约：上游错误帧在解析层**直接抛出**携带结构化 status/providerCode 的
|
||
* SseUpstreamError —— 异常沿 async generator 传播进 chatStreamWithRetry 的 catch，
|
||
* 使 429/5xx 自动走指数退避重试、401 等不可重试错误走 fallback 故障转移，
|
||
* 与 HTTP 状态码路径的行为完全对齐（错误分类单轨化的流式半边）。
|
||
*/
|
||
export class SseUpstreamError extends Error {
|
||
/** 归一化后的 HTTP status（当帧内无数值 status 时按 providerCode 推断） */
|
||
readonly status?: number;
|
||
/** 上游原始错误码（如 "rate_limit_exceeded" / "insufficient_quota"） */
|
||
readonly providerCode?: string;
|
||
|
||
constructor(message: string, options?: { status?: number; providerCode?: string }) {
|
||
super(message);
|
||
this.name = 'SseUpstreamError';
|
||
this.status = options?.status;
|
||
this.providerCode = options?.providerCode;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* v0.7.4 P1-2: 带空闲超时的流读取辅助 — 全协议读循环统一入口。
|
||
*
|
||
* 背景：fetch 流式响应在"服务器保活但不再推送数据"时，reader.read() 会无限挂起。
|
||
* 引擎 totalTimeoutMs 只在迭代之间检查，无法兜底流内挂死；用户只能手动 abort。
|
||
*
|
||
* 本函数在每次 read 前启动 IDLE_TIMEOUT_MS 计时器，数据到达即重置；
|
||
* 连续超时未收到数据则抛 SseUpstreamError(504) —— 沿 async generator 传播进
|
||
* 引擎 chatStreamWithRetry 的 catch，自动走既有重试/故障转移通道。
|
||
* SSE / Ollama NDJSON / Anthropic 事件机三处读循环共用，杜绝三份重复实现漂移。
|
||
*
|
||
* v0.8.2 P3-3 根治: 新增外部中断贯通 —— 此前 abort 监听只在 fetch 头阶段有效
|
||
* （BaseAdapter.fetchWithTimeout 在响应头返回后解除监听），流式消费阶段的
|
||
* reader.read() 对用户中断完全无感：配合保活/心跳型上游，"中断"按钮无法真正
|
||
* 终止挂起的 run（E2E 中断链路实测暴露）。现把外部 signal 传入本辅助：
|
||
* abort 触发时 cancel reader 并抛 AbortError —— 引擎 chatStreamWithRetry 由
|
||
* this.aborted 拦截原样抛出，executeRunStream 以 USER_INTERRUPT 收尾。
|
||
*
|
||
* @param reader 流的 reader
|
||
* @param idleTimeoutMs 空闲超时（默认 60s — 慢速思考模型正常 chunk 间隔可达数十秒）
|
||
* @param externalSignal 外部中断信号（引擎 abortController；可选）
|
||
* @returns { done, value }，done=true 表示流正常结束
|
||
*/
|
||
export async function readStreamChunkWithIdleTimeout(
|
||
reader: ReadableStreamDefaultReader<Uint8Array>,
|
||
idleTimeoutMs = 60_000,
|
||
externalSignal?: AbortSignal,
|
||
): Promise<{ done: boolean; value: Uint8Array | undefined }> {
|
||
let idleExpired = false;
|
||
let idleTimer: NodeJS.Timeout | undefined;
|
||
const idleController = new Promise<{ done: true; value: undefined }>((resolve) => {
|
||
idleTimer = setTimeout(() => {
|
||
idleExpired = true;
|
||
resolve({ done: true, value: undefined });
|
||
}, idleTimeoutMs);
|
||
});
|
||
|
||
// 外部中断竞速（流式消费阶段的中断贯通）
|
||
let onExternalAbort: (() => void) | null = null;
|
||
const abortController = externalSignal
|
||
? new Promise<never>((_, reject) => {
|
||
if (externalSignal.aborted) {
|
||
reject(new Error('Aborted'));
|
||
return;
|
||
}
|
||
onExternalAbort = () => reject(new Error('Aborted'));
|
||
externalSignal.addEventListener('abort', onExternalAbort, { once: true });
|
||
})
|
||
: null;
|
||
|
||
const abortError = (): Error => {
|
||
const err = new Error('Aborted');
|
||
err.name = 'AbortError';
|
||
void reader.cancel().catch(() => {
|
||
/* 连接销毁时 cancel 可能失败，忽略 */
|
||
});
|
||
return err;
|
||
};
|
||
|
||
try {
|
||
const result = await Promise.race([
|
||
reader.read(),
|
||
idleController,
|
||
...(abortController ? [abortController] : []),
|
||
]);
|
||
if (externalSignal?.aborted) {
|
||
throw abortError();
|
||
}
|
||
if (idleExpired) {
|
||
throw new SseUpstreamError(
|
||
`Stream idle timeout after ${idleTimeoutMs}ms (no data received)`,
|
||
{ status: 504 },
|
||
);
|
||
}
|
||
return result;
|
||
} catch (err) {
|
||
// 中断竞速赢时把底层读错误替换为标准 AbortError（reader.read 会因 cancel 拒绝）
|
||
if (externalSignal?.aborted) {
|
||
throw abortError();
|
||
}
|
||
throw err;
|
||
} finally {
|
||
if (idleTimer) clearTimeout(idleTimer);
|
||
if (externalSignal && onExternalAbort) {
|
||
externalSignal.removeEventListener('abort', onExternalAbort);
|
||
}
|
||
}
|
||
}
|
||
|
||
/** v0.6.4: OpenAI 兼容流帧的最小结构化类型（仅承载本解析器实际消费的字段） */
|
||
interface SseStreamFrame {
|
||
choices?: Array<{
|
||
delta?: {
|
||
content?: string;
|
||
reasoning_content?: string;
|
||
annotations?: unknown;
|
||
tool_calls?: Array<{
|
||
index?: number;
|
||
function?: { name?: string; arguments?: string };
|
||
}>;
|
||
};
|
||
/** 部分网关在最后一个 chunk 附带完整 message（含 annotations） */
|
||
message?: { annotations?: unknown };
|
||
finish_reason?: string;
|
||
}>;
|
||
/** v0.8.2 P2-6: MiMo 联网搜索引用注释（服务端 web_search 工具） */
|
||
annotations?: unknown;
|
||
usage?: {
|
||
prompt_tokens?: number;
|
||
completion_tokens?: number;
|
||
total_tokens?: number;
|
||
prompt_cache_hit_tokens?: number;
|
||
prompt_cache_miss_tokens?: number;
|
||
completion_tokens_details?: { reasoning_tokens?: number };
|
||
prompt_tokens_details?: { cached_tokens?: number };
|
||
};
|
||
}
|
||
|
||
/**
|
||
* v0.8.2 P2-6 根治: MiMo 联网搜索引用（annotations）全链路丢失的收口。
|
||
*
|
||
*MiMo enableWebSearch 服务端工具在响应中返回 annotations[].{title,url,site_name,...}
|
||
*（非流式在 choices[].message.annotations；流式按文档"其余字段与非流式相同"，
|
||
* 可能出现在最后一个 chunk 的顶层 / message / delta）。此前 sse-stream 完全不
|
||
* 读取该字段 —— mimo.adapter 注释宣称"由上层归并为文本内容展示"实为断头链路，
|
||
* 引用信息全链路丢失。
|
||
*
|
||
* 采集策略：按 url 去重累积；流结束时（[DONE] / 断流兜底）把引用格式化为
|
||
* Markdown 列表以 TEXT_DELTA 追加到正文 —— 引擎/渲染层按既有文本管线自然
|
||
* 消费，无需新事件类型。
|
||
*/
|
||
function collectAnnotations(
|
||
annotations: unknown,
|
||
sink: Map<string, { title: string; url: string; siteName?: string }>,
|
||
): void {
|
||
if (!Array.isArray(annotations)) return;
|
||
for (const item of annotations) {
|
||
if (!item || typeof item !== 'object') continue;
|
||
const rec = item as Record<string, unknown>;
|
||
const url = typeof rec.url === 'string' ? rec.url : '';
|
||
if (!url || sink.has(url)) continue;
|
||
sink.set(url, {
|
||
title: typeof rec.title === 'string' && rec.title ? rec.title : url,
|
||
url,
|
||
siteName: typeof rec.site_name === 'string' ? rec.site_name : undefined,
|
||
});
|
||
}
|
||
}
|
||
|
||
function formatAnnotationsBlock(
|
||
sink: Map<string, { title: string; url: string; siteName?: string }>,
|
||
): string | null {
|
||
if (sink.size === 0) return null;
|
||
const MAX_CITATIONS = 20;
|
||
const lines: string[] = ['', '', '**References**', ''];
|
||
let count = 0;
|
||
for (const { title, url, siteName } of sink.values()) {
|
||
if (count >= MAX_CITATIONS) break;
|
||
const safeUrl = /^https?:\/\//i.test(url) ? url : '';
|
||
if (!safeUrl) continue;
|
||
lines.push(`- [${title}](${safeUrl})${siteName ? ` — ${siteName}` : ''}`);
|
||
count++;
|
||
}
|
||
return count > 0 ? lines.join('\n') : null;
|
||
}
|
||
|
||
/**
|
||
* 从一条已 JSON.parse 的 SSE 数据帧中提取上游错误信息。
|
||
* 兼容三种形态：
|
||
* 1. 顶层 `{error:{message,status}}` — OpenAI 兼容网关最常见
|
||
* 2. `{choices:[{error:{...}}]}` — 少数代理的变体包装
|
||
* 3. `{error:"plain string"}` — 极简实现
|
||
* 返回 null 表示该帧不含错误（正常数据帧）。
|
||
*/
|
||
function extractUpstreamErrorFrame(
|
||
chunk: unknown,
|
||
): { message: string; status?: number; code?: string } | null {
|
||
if (!chunk || typeof chunk !== 'object') return null;
|
||
const c = chunk as Record<string, unknown>;
|
||
let errObj: unknown = c.error;
|
||
if ((!errObj || typeof errObj !== 'object') && Array.isArray(c.choices) && c.choices.length > 0) {
|
||
errObj = (c.choices[0] as Record<string, unknown> | undefined)?.error;
|
||
}
|
||
|
||
if (typeof errObj === 'string') {
|
||
return errObj.trim() ? { message: errObj } : null;
|
||
}
|
||
if (!errObj || typeof errObj !== 'object') return null;
|
||
|
||
const e = errObj as Record<string, unknown>;
|
||
const rawStatus =
|
||
typeof e.status === 'number'
|
||
? e.status
|
||
: typeof e.status_code === 'number'
|
||
? e.status_code
|
||
: undefined;
|
||
const rawCode = typeof e.code === 'string' ? e.code : typeof e.type === 'string' ? e.type : '';
|
||
// 无消息且无状态码的无害空对象不算错误（防御性）
|
||
// v0.8.0 P1-3.1 根治: 旧实现 message 兜底取 JSON.stringify(errObj) —— '{}' 恒为
|
||
// 真值，使本防御分支永不可达，`{"error":{}}` 被误判为错误帧抛 SseUpstreamError
|
||
// （原行为被 sse-stream.test.ts 锁定留档，本次随修复同步改约）。
|
||
// 现契约：仅当存在显式 message/msg 或 status/code 标识之一才构成错误帧；
|
||
// 描述文本只在有标识但无显式文本时由序列化/占位生成。
|
||
const explicitMessage =
|
||
typeof e.message === 'string' ? e.message : typeof e.msg === 'string' ? e.msg : '';
|
||
if (!explicitMessage && rawStatus === undefined && !rawCode) return null;
|
||
return {
|
||
message: explicitMessage || `upstream error (${rawCode || rawStatus})`,
|
||
status: rawStatus,
|
||
code: rawCode || undefined,
|
||
};
|
||
}
|
||
|
||
/** 上游字符串错误码 → 归一化 HTTP status（用于帧内缺失数值 status 时仍能驱动重试判定） */
|
||
function providerCodeToStatus(code: string): number | undefined {
|
||
const c = code.toLowerCase();
|
||
if (/rate_limit|too_many/.test(c)) return 429;
|
||
if (/quota|insufficient|billing|exceeded_balance/.test(c)) return 402;
|
||
if (/invalid_api_key|api_key_invalid|unauthorized|authentication/.test(c)) return 401;
|
||
if (/forbidden|permission/.test(c)) return 403;
|
||
if (/model_not_found|no_such_model/.test(c)) return 404;
|
||
if (/overloaded|capacity|unavailable/.test(c)) return 503;
|
||
if (/server_error|internal_error|internal/.test(c)) return 500;
|
||
// 请求级 400 家族（无效参数/上下文超限）— 不映射到可重试区间，保持非重试语义
|
||
return undefined;
|
||
}
|
||
|
||
/**
|
||
* 由提取出的错误信息构造待抛出的 SseUpstreamError /
|
||
* ContentFilterError（content_filter 类直接复用既有专用类型）。
|
||
*/
|
||
function makeUpstreamThrowable(info: { message: string; status?: number; code?: string }): Error {
|
||
if (info.code && info.code.toLowerCase().includes('content_filter')) {
|
||
return new ContentFilterError(info.message, 'SSE 流中收到上游安全审核错误');
|
||
}
|
||
const status = info.status ?? (info.code ? providerCodeToStatus(info.code) : undefined);
|
||
return new SseUpstreamError(
|
||
`upstream_error${info.code ? ` (${info.code})` : ''}: ${info.message}`,
|
||
{
|
||
status,
|
||
providerCode: info.code,
|
||
},
|
||
);
|
||
}
|
||
|
||
/**
|
||
* 截断参数自愈载荷的唯一构造点（v0.6.4: 非流式/Ollama NDJSON/Anthropic 全线共用）。
|
||
*
|
||
* 原 v0.6.3 只覆盖了 OpenAI 共享 SSE 层；此处抽出为共享函数后，所有协议路径的
|
||
* 截断工具调用统一转为显式错误参数 —— 工具执行失败 → 错误结果回传模型 → 模型
|
||
* 重试/分块写入（ReAct 自愈闭环），彻底消灭"静默丢弃 → 空回复 → 会话无声终止"。
|
||
*/
|
||
export function truncatedArgumentsPayload(
|
||
parseErrorMessage: string,
|
||
rawTailSample: string,
|
||
): Record<string, unknown> {
|
||
return {
|
||
_truncatedArguments: true,
|
||
_truncatedReason:
|
||
'The tool-call arguments JSON was truncated before completion ' +
|
||
'(likely max_tokens output limit reached while generating this tool call). ' +
|
||
'The original arguments are lost and cannot be recovered. Please retry with a ' +
|
||
'smaller output (e.g. write the file in smaller chunks) — do NOT reuse or repeat ' +
|
||
'the previous oversized arguments.' +
|
||
` [parser: ${parseErrorMessage}; tail sample: ...${rawTailSample}]`,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* L-4 修复: 提取 flushToolCallBuffer 辅助函数，消除 [DONE] 分支和 finish_reason='tool_calls' 分支的重复代码
|
||
*
|
||
* 遍历工具调用缓冲区，对每个缓冲的工具调用：
|
||
* 1. JSON.parse argsBuffer
|
||
* 2. yield 一个 TOOL_CALL_COMPLETE 事件
|
||
* 3. 清空缓冲区
|
||
*
|
||
* v0.6.3 会话停止修复: argsBuffer 解析失败（流截断致 JSON 半截 — 典型场景：
|
||
* 模型写大文件时输出 token 达上限 finish_reason=length）时，不再静默丢弃该
|
||
* 工具调用。丢弃会让引擎看到"零工具调用 + 零文本"→ 误判为模型已完成 →
|
||
* COMPLETED + 空回复 → 会话无声停止（main.log 20:31/20:32 两次实锤）。
|
||
* 现转为 yield 一个携带截断错误说明的 tool call：工具执行将因参数缺失失败，
|
||
* 错误结果回传模型 → 模型感知截断后重试/分块写入（ReAct 自愈路径）。
|
||
* 无限循环由引擎死循环检测器兜底。
|
||
*
|
||
* @param toolCallsBuffer - 工具调用缓冲区（index → { name, argsBuffer }）
|
||
* @param requestId - 请求 ID
|
||
* @param sessionId - 会话 ID
|
||
* @param iteration - 当前迭代轮次
|
||
* @param seqRef - seq 计数器引用（递增）
|
||
* @yields MetonaStreamEvent
|
||
*/
|
||
function* flushToolCallBuffer(
|
||
toolCallsBuffer: Map<number, { name: string; argsBuffer: string }>,
|
||
requestId: string,
|
||
sessionId: string,
|
||
iteration: number,
|
||
seqRef: { seq: number },
|
||
): Generator<MetonaStreamEvent> {
|
||
for (const [, buf] of toolCallsBuffer) {
|
||
if (buf.argsBuffer) {
|
||
try {
|
||
yield {
|
||
type: MetonaStreamEventType.TOOL_CALL_COMPLETE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
toolCall: {
|
||
id: `tc_${nanoid(8)}`,
|
||
name: buf.name,
|
||
args: JSON.parse(buf.argsBuffer),
|
||
iteration,
|
||
timestamp: Date.now(),
|
||
},
|
||
};
|
||
} catch (err) {
|
||
// v0.6.3 → v0.6.4: 截断的工具调用转显式错误参数（不丢弃）— 工具执行失败后
|
||
// 错误结果回传模型，触发重试/分块写入。载荷构造已收敛到共享的
|
||
// truncatedArgumentsPayload（Ollama NDJSON / Anthropic 事件机 / 非流式同步复用）。
|
||
const rawTail = buf.argsBuffer.slice(-120);
|
||
log.warn(
|
||
`[SSE] Tool call args truncated (unparseable JSON, ${(err as Error).message}). ` +
|
||
`Forwarding as error to model for self-healing. Tail: ...${rawTail}`,
|
||
);
|
||
yield {
|
||
type: MetonaStreamEventType.TOOL_CALL_COMPLETE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
toolCall: {
|
||
id: `tc_${nanoid(8)}`,
|
||
name: buf.name,
|
||
args: truncatedArgumentsPayload((err as Error).message, rawTail),
|
||
iteration,
|
||
timestamp: Date.now(),
|
||
},
|
||
};
|
||
}
|
||
} else {
|
||
// 空 argsBuffer：模型发了空 arguments（合法 — 无参工具）
|
||
yield {
|
||
type: MetonaStreamEventType.TOOL_CALL_COMPLETE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
toolCall: {
|
||
id: `tc_${nanoid(8)}`,
|
||
name: buf.name,
|
||
args: {},
|
||
iteration,
|
||
timestamp: Date.now(),
|
||
},
|
||
};
|
||
}
|
||
}
|
||
toolCallsBuffer.clear();
|
||
}
|
||
|
||
/**
|
||
* 解析 OpenAI 兼容 SSE 流式响应
|
||
*
|
||
* @param responseBody - fetch Response.body (ReadableStream<Uint8Array>)
|
||
* @param requestId - 对应的请求 ID
|
||
* @param sessionId - 会话 ID
|
||
* @param iteration - 当前迭代轮次
|
||
* @param externalSignal - 外部中断信号（v0.8.2 P3-3: 流式消费阶段的中断贯通，可选）
|
||
* @yields MetonaStreamEvent
|
||
*/
|
||
export async function* parseSSEStream(
|
||
responseBody: ReadableStream<Uint8Array>,
|
||
requestId: string,
|
||
sessionId: string,
|
||
iteration: number,
|
||
externalSignal?: AbortSignal,
|
||
): AsyncGenerator<MetonaStreamEvent> {
|
||
const reader = responseBody.getReader();
|
||
const decoder = new TextDecoder();
|
||
const seqRef = { seq: 0 };
|
||
let buffer = '';
|
||
// v0.6.3: 是否收到过 [DONE]（流断开兜底用）
|
||
let sawDone = false;
|
||
// v0.8.0 P0-1: 跟踪 Provider 原生 finish_reason，随 DONE 事件携带给引擎
|
||
// （'stop' | 'length' | 'tool_calls' | 'content_filter' 等；引擎据此区分
|
||
// 自然完成与输出上限截断，此前 length 仅落日志、引擎不可见）
|
||
let lastFinishReason: string | undefined;
|
||
|
||
// v0.7.4 P1-2: 流空闲超时 — 服务器保活但不再推送数据（连接挂死）时，
|
||
// reader.read() 会无限挂起，totalTimeoutMs 只在迭代之间检查，无法兜底。
|
||
// 连续 IDLE_TIMEOUT_MS 无任何数据则抛出 SseUpstreamError(504)，
|
||
// 异常沿 chatStreamWithRetry 的 catch 进入既有重试/故障转移通道。
|
||
// 选择 60s 而非 30s：慢速模型（思考模式）正常 chunk 间隔可达数十秒，
|
||
// 过短会误杀仍在思考的合法请求。实现收敛到共享 readStreamChunkWithIdleTimeout。
|
||
const IDLE_TIMEOUT_MS = 60_000;
|
||
|
||
// 工具调用缓冲区：index → { name, argsBuffer }
|
||
const toolCallsBuffer = new Map<number, { name: string; argsBuffer: string }>();
|
||
// v0.8.2 P2-6: MiMo 联网搜索引用采集（按 url 去重，流结束时回填正文）
|
||
const annotationsSink = new Map<string, { title: string; url: string; siteName?: string }>();
|
||
|
||
try {
|
||
while (true) {
|
||
// 数据到达即重置空闲窗口（辅助函数内部实现）
|
||
const { done, value } = await readStreamChunkWithIdleTimeout(
|
||
reader,
|
||
IDLE_TIMEOUT_MS,
|
||
externalSignal,
|
||
);
|
||
if (done) break;
|
||
|
||
buffer += decoder.decode(value, { stream: true });
|
||
const lines = buffer.split('\n');
|
||
buffer = lines.pop() ?? '';
|
||
|
||
for (const line of lines) {
|
||
const trimmed = line.trim();
|
||
// v0.6.4: 兼容 `data:{...}`（无空格）变体 — 部分代理网关不带空格，
|
||
// 原实现的 startsWith('data: ') 会将其整帧跳过
|
||
if (!trimmed || !trimmed.startsWith('data:')) continue;
|
||
const data = trimmed.slice(5).trim();
|
||
if (!data) continue;
|
||
|
||
// 流结束
|
||
if (data === '[DONE]') {
|
||
sawDone = true;
|
||
// L-4 修复: 使用 flushToolCallBuffer 替代重复的遍历代码
|
||
yield* flushToolCallBuffer(toolCallsBuffer, requestId, sessionId, iteration, seqRef);
|
||
|
||
// v0.8.2 P2-6: 引用注释回填正文（在 DONE 之前以 TEXT_DELTA 追加）
|
||
const citationBlock = formatAnnotationsBlock(annotationsSink);
|
||
if (citationBlock) {
|
||
yield {
|
||
type: MetonaStreamEventType.TEXT_DELTA,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
delta: citationBlock,
|
||
};
|
||
}
|
||
|
||
yield {
|
||
type: MetonaStreamEventType.DONE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
// v0.8.0 P0-1: 携带 Provider 原生 finish_reason（未见任何帧时 undefined）
|
||
...(lastFinishReason ? { finishReason: lastFinishReason } : {}),
|
||
};
|
||
return;
|
||
}
|
||
|
||
// v0.6.4: 结构化类型承载帧内容（替代 JSON.parse 的隐式 any，杜绝字段漂移）
|
||
let chunk: SseStreamFrame;
|
||
try {
|
||
chunk = JSON.parse(data) as SseStreamFrame;
|
||
} catch (parseErr) {
|
||
// P2-8 修复: 不再静默跳过，记录 warning 便于排查 SSE 数据损坏
|
||
log.warn(
|
||
`[SSE] Failed to parse stream line: ${(parseErr as Error).message}`,
|
||
line.slice(0, 200),
|
||
);
|
||
continue;
|
||
}
|
||
|
||
// v0.6.4: 上游错误帧检测（根治"错误帧黑洞"）。解析失败直接 throw，
|
||
// 异常沿 chatStreamWithRetry 的 catch 走重试/故障转移通道；
|
||
// 工具调用缓冲不 flush —— 重试会从零重建整个响应。
|
||
const upstreamError = extractUpstreamErrorFrame(chunk);
|
||
if (upstreamError) {
|
||
log.warn(
|
||
`[SSE] Upstream error frame received: code=${upstreamError.code ?? 'n/a'} status=${upstreamError.status ?? 'n/a'} message=${upstreamError.message.slice(0, 300)} — throwing for retry/failover handling`,
|
||
);
|
||
throw makeUpstreamThrowable(upstreamError);
|
||
}
|
||
|
||
const delta = chunk.choices?.[0]?.delta;
|
||
|
||
// v0.8.2 P2-6: 采集引用注释（顶层 / message / delta 三处兼容）
|
||
collectAnnotations(chunk.annotations, annotationsSink);
|
||
collectAnnotations(chunk.choices?.[0]?.message?.annotations, annotationsSink);
|
||
collectAnnotations(delta?.annotations, annotationsSink);
|
||
|
||
// 文本内容增量
|
||
if (delta?.content) {
|
||
yield {
|
||
type: MetonaStreamEventType.TEXT_DELTA,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
delta: delta.content,
|
||
};
|
||
}
|
||
|
||
// 推理内容增量（Thinking 模式）
|
||
if (delta?.reasoning_content) {
|
||
yield {
|
||
type: MetonaStreamEventType.REASONING_DELTA,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
delta: delta.reasoning_content,
|
||
};
|
||
}
|
||
|
||
// 工具调用增量 — 缓冲拼接
|
||
if (delta?.tool_calls) {
|
||
for (const tc of delta.tool_calls) {
|
||
const idx = tc.index ?? 0;
|
||
if (!toolCallsBuffer.has(idx)) {
|
||
toolCallsBuffer.set(idx, { name: tc.function?.name ?? '', argsBuffer: '' });
|
||
}
|
||
const buf = toolCallsBuffer.get(idx)!;
|
||
if (tc.function?.name) buf.name = tc.function.name;
|
||
if (tc.function?.arguments) buf.argsBuffer += tc.function.arguments;
|
||
|
||
yield {
|
||
type: MetonaStreamEventType.TOOL_CALL_DELTA,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
toolCallDelta: {
|
||
index: idx,
|
||
name: tc.function?.name,
|
||
argsDelta: tc.function?.arguments,
|
||
},
|
||
};
|
||
}
|
||
}
|
||
|
||
// Token 使用统计 / finish_reason
|
||
const usageRaw = chunk.usage;
|
||
if (usageRaw) {
|
||
const usage: MetonaTokenUsage = {
|
||
inputTokens: usageRaw.prompt_tokens ?? 0,
|
||
outputTokens: usageRaw.completion_tokens ?? 0,
|
||
totalTokens: usageRaw.total_tokens ?? 0,
|
||
reasoningTokens: usageRaw.completion_tokens_details?.reasoning_tokens,
|
||
// DeepSeek: prompt_cache_hit_tokens / prompt_cache_miss_tokens
|
||
// MiMo: prompt_tokens_details.cached_tokens
|
||
cacheHitTokens:
|
||
usageRaw.prompt_cache_hit_tokens ?? usageRaw.prompt_tokens_details?.cached_tokens,
|
||
cacheMissTokens: usageRaw.prompt_cache_miss_tokens,
|
||
};
|
||
|
||
yield {
|
||
type: MetonaStreamEventType.USAGE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
usage,
|
||
};
|
||
}
|
||
|
||
// 非 [DONE] 但 finish_reason 为 tool_calls 时提前 flush 缓冲区
|
||
const finishReason = chunk.choices?.[0]?.finish_reason as string | undefined;
|
||
// v0.8.0 P0-1: 记录最近一次 finish_reason（最终帧通常携带于最后一个
|
||
// 含 choices 的 chunk），随 DONE 事件上交通知引擎与前端
|
||
if (finishReason) {
|
||
lastFinishReason = finishReason;
|
||
}
|
||
if (finishReason === 'tool_calls') {
|
||
// L-4 修复: 使用 flushToolCallBuffer 替代重复的遍历代码
|
||
yield* flushToolCallBuffer(toolCallsBuffer, requestId, sessionId, iteration, seqRef);
|
||
}
|
||
// v0.6.3 归因: 输出 token 上限截断（长工具参数/长文本的常见根因）显式落日志
|
||
if (finishReason === 'length') {
|
||
log.warn(
|
||
`[SSE] finish_reason=length — output truncated by max_tokens limit ` +
|
||
`(accumulated argsBuffer: ${[...toolCallsBuffer.values()].reduce((n, b) => n + b.argsBuffer.length, 0)} chars, ` +
|
||
`model may retry with smaller output)`,
|
||
);
|
||
}
|
||
// v0.6.4: 流式 content_filter 终止映射 —— 非流式路径早已支持
|
||
// （throwHttpError → ContentFilterError），流式此前既不映射也不打日志，
|
||
// 引擎拿到普通结束、用户看不到拦截原因。抛专用类型使 finish() 映射为
|
||
// CONTENT_FILTERED 错误码 + 友好提示，且不会被重试逻辑反复重放。
|
||
if (finishReason === 'content_filter') {
|
||
log.warn(
|
||
'[SSE] finish_reason=content_filter — provider safety filter terminated the response',
|
||
);
|
||
throw new ContentFilterError(
|
||
'流式响应被 Provider 安全审核终止',
|
||
'SSE stream (finish_reason=content_filter)',
|
||
);
|
||
}
|
||
}
|
||
}
|
||
} finally {
|
||
// 空闲计时器已由 readStreamChunkWithIdleTimeout 内部清理，此处仅防御残留
|
||
//（无需额外处理——辅助函数 try/finally 保证清理）
|
||
}
|
||
|
||
// v0.6.3 流断开兜底: read() done 但从未收到 [DONE]（连接中断/服务端异常收尾）。
|
||
// 原实现直接结束 generator —— 工具缓冲不 flush、DONE 事件缺失（引擎侧等待
|
||
// 流收尾的路径行为未定义，且缓冲的工具调用整体丢失）。补 flush + DONE，
|
||
// 截断的参数由 flushToolCallBuffer 转为错误结果回传模型自愈。
|
||
if (!sawDone) {
|
||
log.warn(
|
||
'[SSE] Stream ended without [DONE] marker — flushing buffers (connection likely dropped)',
|
||
);
|
||
yield* flushToolCallBuffer(toolCallsBuffer, requestId, sessionId, iteration, seqRef);
|
||
// v0.8.2 P2-6: 断流兜底路径同样回填引用注释
|
||
const citationBlock = formatAnnotationsBlock(annotationsSink);
|
||
if (citationBlock) {
|
||
yield {
|
||
type: MetonaStreamEventType.TEXT_DELTA,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
delta: citationBlock,
|
||
};
|
||
}
|
||
yield {
|
||
type: MetonaStreamEventType.DONE,
|
||
requestId,
|
||
sessionId,
|
||
iteration,
|
||
seq: seqRef.seq++,
|
||
timestamp: Date.now(),
|
||
// v0.8.0 P0-1: 断流合成路径同样携带已观察到的 finish_reason
|
||
//（注意：断流时 finishReason 多为 undefined —— 引擎据此走空响应守卫/重试，
|
||
// 而不是误判为模型自然说完了）
|
||
...(lastFinishReason ? { finishReason: lastFinishReason } : {}),
|
||
};
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析 OpenAI 兼容的非流式 JSON 响应 → MetonaResponse
|
||
*/
|
||
export function parseOpenAICompatibleResponse(data: Record<string, unknown>): {
|
||
content: string;
|
||
reasoningContent?: string;
|
||
toolCalls?: Array<{
|
||
id: string;
|
||
name: string;
|
||
args: Record<string, unknown>;
|
||
iteration: number;
|
||
timestamp: number;
|
||
}>;
|
||
finishReason: string;
|
||
usage: MetonaTokenUsage;
|
||
} {
|
||
const choice = (data.choices as Array<Record<string, unknown>>)?.[0];
|
||
const message = choice?.message as Record<string, unknown> | undefined;
|
||
const usage = data.usage as Record<string, unknown> | undefined;
|
||
const rawToolCalls = message?.tool_calls as Array<Record<string, unknown>> | undefined;
|
||
|
||
// v0.8.2 P2-6: 非流式路径的引用注释回填（MiMo 联网搜索）
|
||
const annotationsSink = new Map<string, { title: string; url: string; siteName?: string }>();
|
||
collectAnnotations(message?.annotations, annotationsSink);
|
||
collectAnnotations(data.annotations, annotationsSink);
|
||
let content = (message?.content as string) ?? '';
|
||
const citationBlock = formatAnnotationsBlock(annotationsSink);
|
||
if (citationBlock) content += citationBlock;
|
||
|
||
return {
|
||
content,
|
||
reasoningContent: message?.reasoning_content as string | undefined,
|
||
toolCalls: rawToolCalls?.map((tc) => {
|
||
const fn = tc.function as Record<string, unknown>;
|
||
let args: Record<string, unknown> = {};
|
||
const rawArgs = fn?.arguments;
|
||
if (typeof rawArgs === 'string') {
|
||
try {
|
||
args = JSON.parse(rawArgs);
|
||
} catch (err) {
|
||
// v0.6.4: 非流式路径与流式截断自愈对齐 —— 此前坏参静默降级 {} 与流式
|
||
// 的显式自愈行为不一致（v0.6.3 只修了流式半边）。空参数会让工具以
|
||
// "缺少必要参数"泛化失败，模型无法得知发生了截断；现在统一转为
|
||
// _truncatedArguments 错误参数，触发模型分块重试。
|
||
const sample = rawArgs.slice(-120);
|
||
log.warn(
|
||
`[SSE] Non-stream tool call args truncated (unparseable JSON, ${(err as Error).message}). Tail: ...${sample}`,
|
||
);
|
||
args = truncatedArgumentsPayload((err as Error).message, sample);
|
||
}
|
||
} else if (rawArgs && typeof rawArgs === 'object') {
|
||
args = rawArgs as Record<string, unknown>;
|
||
}
|
||
return {
|
||
id: tc.id as string,
|
||
name: fn.name as string,
|
||
args,
|
||
iteration: 0,
|
||
timestamp: Date.now(),
|
||
};
|
||
}),
|
||
finishReason: mapOpenAIFinishReason(choice?.finish_reason as string),
|
||
usage: {
|
||
inputTokens: (usage?.prompt_tokens as number) ?? 0,
|
||
outputTokens: (usage?.completion_tokens as number) ?? 0,
|
||
totalTokens: (usage?.total_tokens as number) ?? 0,
|
||
reasoningTokens: (usage?.completion_tokens_details as Record<string, unknown>)
|
||
?.reasoning_tokens as number | undefined,
|
||
// DeepSeek: prompt_cache_hit_tokens / MiMo: prompt_tokens_details.cached_tokens
|
||
cacheHitTokens:
|
||
(usage?.prompt_cache_hit_tokens as number | undefined) ??
|
||
((usage?.prompt_tokens_details as Record<string, unknown> | undefined)?.cached_tokens as
|
||
| number
|
||
| undefined),
|
||
cacheMissTokens: usage?.prompt_cache_miss_tokens as number | undefined,
|
||
},
|
||
};
|
||
}
|
||
|
||
function mapOpenAIFinishReason(reason: string): string {
|
||
switch (reason) {
|
||
case 'stop':
|
||
return 'stop';
|
||
case 'length':
|
||
return 'length';
|
||
case 'tool_calls':
|
||
return 'tool_calls';
|
||
case 'content_filter':
|
||
return 'content_filter';
|
||
// MiMo 特有：检测到复读截断
|
||
case 'repetition_truncation':
|
||
return 'stop';
|
||
default:
|
||
return 'stop';
|
||
}
|
||
}
|