merge: accept complete BOM JSON streams
This commit is contained in:
WuYongJie
commit
5416bed96e
@ -52,7 +52,7 @@ describe('BOM 多模态网关', () => {
|
||||
start(streamController) {
|
||||
controller = streamController
|
||||
streamController.enqueue(encoder.encode(
|
||||
'data: {"id":"call-1","model":"gpt-5.6-terra","choices":[{"delta":{"content":"{\\"ok\\":true}"},"finish_reason":null}]}\n\n' +
|
||||
'data: {"id":"call-1","model":"gpt-5.6-terra","choices":[{"delta":{"content":"not-json"},"finish_reason":null}]}\n\n' +
|
||||
'data: [DONE]\n\n',
|
||||
))
|
||||
},
|
||||
@ -80,7 +80,7 @@ describe('BOM 多模态网关', () => {
|
||||
timeoutMs: 5_000,
|
||||
})
|
||||
|
||||
expect(result.content).toBe('{"ok":true}')
|
||||
expect(result.content).toBe('not-json')
|
||||
expect(result.finishReason).toBeNull()
|
||||
expect(cancelReason).toBe('completed')
|
||||
expect(controller).not.toBeNull()
|
||||
@ -124,6 +124,116 @@ describe('BOM 多模态网关', () => {
|
||||
expect(cancelReason).toBe('completed')
|
||||
})
|
||||
|
||||
it('收到完整 JSON 后立即结束，兼容既不发送结束原因也不发送 DONE 的上游', async () => {
|
||||
let cancelReason: unknown
|
||||
const encoder = new TextEncoder()
|
||||
const keepAliveStream = new ReadableStream<Uint8Array>({
|
||||
start(streamController) {
|
||||
streamController.enqueue(encoder.encode(
|
||||
'data: {"id":"call-json","model":"gpt-5.6-terra","choices":[{"delta":{"content":"{\\"product\\":{\\"name\\":\\"衣柜\\"},\\"skus\\":[]}"},"finish_reason":null}]}\n\n',
|
||||
))
|
||||
},
|
||||
cancel(reason) {
|
||||
cancelReason = reason
|
||||
},
|
||||
})
|
||||
jest.spyOn(global, 'fetch').mockResolvedValue(
|
||||
new Response(keepAliveStream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
}),
|
||||
)
|
||||
|
||||
const result = await callBomCostingMultimodalModel({
|
||||
config: {
|
||||
baseUrl: 'https://ai.example.com/v1',
|
||||
apiKey: 'test-key',
|
||||
model: 'gpt-5.6-terra',
|
||||
},
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: 'test',
|
||||
userContent: 'test',
|
||||
maxTokens: 100,
|
||||
timeoutMs: 5_000,
|
||||
})
|
||||
|
||||
expect(result.content).toBe('{"product":{"name":"衣柜"},"skus":[]}')
|
||||
expect(result.finishReason).toBeNull()
|
||||
expect(result.completedFromJsonPayload).toBe(true)
|
||||
expect(cancelReason).toBe('completed')
|
||||
})
|
||||
|
||||
it('截断 JSON 不会被当成完整结果，仍按超时进入恢复流程', async () => {
|
||||
const encoder = new TextEncoder()
|
||||
const truncatedStream = new ReadableStream<Uint8Array>({
|
||||
start(streamController) {
|
||||
streamController.enqueue(encoder.encode(
|
||||
'data: {"choices":[{"delta":{"content":"{\\"product\\":{\\"name\\":\\"衣柜\\"}"},"finish_reason":null}]}\n\n',
|
||||
))
|
||||
},
|
||||
})
|
||||
jest.spyOn(global, 'fetch').mockResolvedValue(
|
||||
new Response(truncatedStream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
}),
|
||||
)
|
||||
|
||||
await expect(callBomCostingMultimodalModel({
|
||||
config: {
|
||||
baseUrl: 'https://ai.example.com/v1',
|
||||
apiKey: 'test-key',
|
||||
model: 'gpt-5.6-terra',
|
||||
},
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: 'test',
|
||||
userContent: 'test',
|
||||
maxTokens: 100,
|
||||
timeoutMs: 25,
|
||||
})).rejects.toThrow('多模态 BOM 分析超时')
|
||||
})
|
||||
|
||||
it('上游拒绝请求时记录脱敏后的错误类型和参数，不记录凭据或图片内容', async () => {
|
||||
const warning = jest.spyOn(console, 'warn').mockImplementation(() => undefined)
|
||||
jest.spyOn(global, 'fetch').mockResolvedValue(
|
||||
new Response(JSON.stringify({
|
||||
error: {
|
||||
code: 'invalid_request_error',
|
||||
type: 'invalid_request_error',
|
||||
param: 'messages[1].content',
|
||||
message: 'Invalid image at https://secret.example.com/a.png Bearer abc123',
|
||||
},
|
||||
}), {
|
||||
status: 400,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}),
|
||||
)
|
||||
|
||||
await expect(callBomCostingMultimodalModel({
|
||||
config: {
|
||||
baseUrl: 'https://ai.example.com/v1',
|
||||
apiKey: 'test-key',
|
||||
model: 'gpt-5.6-terra',
|
||||
},
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: 'test',
|
||||
userContent: 'test',
|
||||
maxTokens: 100,
|
||||
timeoutMs: 5_000,
|
||||
})).rejects.toThrow('多模态分析服务返回 400')
|
||||
|
||||
const rejectionLog = warning.mock.calls.find((call) => (
|
||||
call[0] === '[BOM costing] multimodal upstream rejected request'
|
||||
))?.[1] as Record<string, unknown> | undefined
|
||||
expect(rejectionLog).toMatchObject({
|
||||
status: 400,
|
||||
upstreamErrorCode: 'invalid_request_error',
|
||||
upstreamErrorType: 'invalid_request_error',
|
||||
upstreamErrorParam: 'messages[1].content',
|
||||
upstreamErrorMessage: 'Invalid image at [REDACTED_URL] Bearer [REDACTED]',
|
||||
})
|
||||
})
|
||||
|
||||
it('规格汇总遇到网络超时时立即交由上层拆分，不白白重试整批', async () => {
|
||||
const networkTimeout = Object.assign(
|
||||
new Error('socket timed out'),
|
||||
|
||||
@ -27,6 +27,7 @@ export interface BomCostingMultimodalCallResult {
|
||||
completionTokens: number | null
|
||||
durationMs: number
|
||||
finishReason: string | null
|
||||
completedFromJsonPayload?: boolean
|
||||
}
|
||||
|
||||
export type BomCostingMultimodalOperation = 'IMAGE_FACTS' | 'SYNTHESIS'
|
||||
@ -54,6 +55,7 @@ const BOM_COSTING_RETRYABLE_TRANSPORT_CODES = new Set([
|
||||
])
|
||||
const BOM_COSTING_SYNTHESIS_MAX_ATTEMPTS = 2
|
||||
const BOM_COSTING_SYNTHESIS_RETRY_DELAY_MS = 1_000
|
||||
const BOM_COSTING_COMPLETE_JSON_SETTLE_MS = 75
|
||||
|
||||
export function readBomCostingBoundedInteger(
|
||||
value: string | undefined,
|
||||
@ -213,6 +215,75 @@ function readUpstreamRequestId(headers: Headers): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
interface BomCostingUpstreamErrorSummary {
|
||||
code: string | null
|
||||
type: string | null
|
||||
param: string | null
|
||||
message: string | null
|
||||
}
|
||||
|
||||
function sanitizeUpstreamErrorValue(value: unknown, maximum: number) {
|
||||
if (typeof value !== 'string') return null
|
||||
const sanitized = value
|
||||
.replace(/Bearer\s+[A-Za-z0-9._~+/=-]+/gi, 'Bearer [REDACTED]')
|
||||
.replace(/data:[^;,\s]+;base64,[A-Za-z0-9+/=]+/gi, '[REDACTED_DATA_URL]')
|
||||
.replace(/https?:\/\/\S+/gi, '[REDACTED_URL]')
|
||||
.replace(/[\r\n\t]+/g, ' ')
|
||||
.trim()
|
||||
return sanitized ? sanitized.slice(0, maximum) : null
|
||||
}
|
||||
|
||||
async function readUpstreamErrorSummary(
|
||||
response: Response,
|
||||
): Promise<BomCostingUpstreamErrorSummary> {
|
||||
const fallback: BomCostingUpstreamErrorSummary = {
|
||||
code: null,
|
||||
type: null,
|
||||
param: null,
|
||||
message: null,
|
||||
}
|
||||
const body = await response.text().catch(() => '')
|
||||
if (!body.trim()) return fallback
|
||||
|
||||
try {
|
||||
const parsed = JSON.parse(body) as {
|
||||
error?: {
|
||||
code?: unknown
|
||||
type?: unknown
|
||||
param?: unknown
|
||||
message?: unknown
|
||||
}
|
||||
code?: unknown
|
||||
type?: unknown
|
||||
param?: unknown
|
||||
message?: unknown
|
||||
}
|
||||
const error = parsed.error ?? parsed
|
||||
return {
|
||||
code: sanitizeUpstreamErrorValue(error.code, 120),
|
||||
type: sanitizeUpstreamErrorValue(error.type, 120),
|
||||
param: sanitizeUpstreamErrorValue(error.param, 160),
|
||||
message: sanitizeUpstreamErrorValue(error.message, 500),
|
||||
}
|
||||
} catch {
|
||||
return {
|
||||
...fallback,
|
||||
message: sanitizeUpstreamErrorValue(body, 500),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isCompleteJsonObjectPayload(content: string) {
|
||||
const trimmed = content.trim()
|
||||
if (!trimmed.startsWith('{') || !trimmed.endsWith('}')) return false
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed)
|
||||
return typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed)
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async function readStreamingChatCompletion(
|
||||
response: Response,
|
||||
options: { signal?: AbortSignal } = {},
|
||||
@ -250,7 +321,37 @@ async function readStreamingChatCompletion(
|
||||
let completionTokens: number | null = null
|
||||
let finishReason: string | null = null
|
||||
let receivedDoneMarker = false
|
||||
const receivedCompletionSignal = () => receivedDoneMarker || finishReason !== null
|
||||
let receivedCompleteJsonPayload = false
|
||||
let acceptedCompleteJsonPayload = false
|
||||
const receivedProtocolCompletionSignal = () => (
|
||||
receivedDoneMarker || finishReason !== null
|
||||
)
|
||||
const receivedCompletionSignal = () => (
|
||||
receivedProtocolCompletionSignal() || acceptedCompleteJsonPayload
|
||||
)
|
||||
|
||||
const readNextChunk = async () => {
|
||||
if (!receivedCompleteJsonPayload) {
|
||||
return { settled: false as const, result: await reader.read() }
|
||||
}
|
||||
|
||||
return new Promise<
|
||||
| { settled: true; result?: never }
|
||||
| { settled: false; result: ReadableStreamReadResult<Uint8Array> }
|
||||
>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
acceptedCompleteJsonPayload = true
|
||||
resolve({ settled: true })
|
||||
}, BOM_COSTING_COMPLETE_JSON_SETTLE_MS)
|
||||
void reader.read().then((result) => {
|
||||
clearTimeout(timer)
|
||||
resolve({ settled: false, result })
|
||||
}, (error: unknown) => {
|
||||
clearTimeout(timer)
|
||||
reject(error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
const consumeLine = (line: string) => {
|
||||
const trimmed = line.trim()
|
||||
@ -286,7 +387,16 @@ async function readStreamingChatCompletion(
|
||||
if (chunk.id?.trim()) callId = chunk.id.trim()
|
||||
if (chunk.model?.trim()) model = chunk.model.trim()
|
||||
const delta = chunk.choices?.[0]?.delta?.content
|
||||
if (typeof delta === 'string') content += delta
|
||||
if (typeof delta === 'string') {
|
||||
content += delta
|
||||
// response_format=json_object guarantees a single top-level object. A
|
||||
// few OpenAI-compatible gateways keep the SSE connection open and omit
|
||||
// both finish_reason and [DONE]. Once the accumulated top-level object
|
||||
// parses successfully, the model answer is complete; cancelling here
|
||||
// avoids discarding valid work at the transport timeout. Truncated JSON
|
||||
// never passes this check and still follows the normal retry path.
|
||||
receivedCompleteJsonPayload = isCompleteJsonObjectPayload(content)
|
||||
}
|
||||
const chunkFinishReason = chunk.choices?.[0]?.finish_reason
|
||||
if (typeof chunkFinishReason === 'string' && chunkFinishReason.trim()) {
|
||||
finishReason = chunkFinishReason.trim()
|
||||
@ -301,7 +411,9 @@ async function readStreamingChatCompletion(
|
||||
|
||||
try {
|
||||
while (!receivedCompletionSignal()) {
|
||||
const { done, value } = await reader.read()
|
||||
const next = await readNextChunk()
|
||||
if (next.settled) break
|
||||
const { done, value } = next.result
|
||||
throwIfAborted()
|
||||
if (done) break
|
||||
buffer += decoder.decode(value, { stream: true })
|
||||
@ -309,7 +421,7 @@ async function readStreamingChatCompletion(
|
||||
buffer = lines.pop() ?? ''
|
||||
for (const line of lines) {
|
||||
consumeLine(line)
|
||||
if (receivedCompletionSignal()) break
|
||||
if (receivedProtocolCompletionSignal()) break
|
||||
}
|
||||
}
|
||||
if (!receivedCompletionSignal()) buffer += decoder.decode()
|
||||
@ -324,6 +436,8 @@ async function readStreamingChatCompletion(
|
||||
pendingBufferChars: buffer.length,
|
||||
receivedDoneMarker,
|
||||
finishReason,
|
||||
receivedCompleteJsonPayload,
|
||||
acceptedCompleteJsonPayload,
|
||||
},
|
||||
configurable: true,
|
||||
})
|
||||
@ -349,6 +463,8 @@ async function readStreamingChatCompletion(
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
finishReason,
|
||||
completedFromJsonPayload: acceptedCompleteJsonPayload &&
|
||||
!receivedDoneMarker && finishReason === null,
|
||||
}
|
||||
}
|
||||
|
||||
@ -526,12 +642,17 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
const upstreamError = await readUpstreamErrorSummary(response)
|
||||
console.warn('[BOM costing] multimodal upstream rejected request', {
|
||||
operation: input.operation,
|
||||
model: input.config.model,
|
||||
attempt,
|
||||
status: response.status,
|
||||
upstreamRequestId: readUpstreamRequestId(response.headers),
|
||||
upstreamErrorCode: upstreamError.code,
|
||||
upstreamErrorType: upstreamError.type,
|
||||
upstreamErrorParam: upstreamError.param,
|
||||
upstreamErrorMessage: upstreamError.message,
|
||||
durationMs: Date.now() - startedAt,
|
||||
})
|
||||
if (
|
||||
@ -569,6 +690,8 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
pendingBufferChars?: number
|
||||
receivedDoneMarker?: boolean
|
||||
finishReason?: string | null
|
||||
receivedCompleteJsonPayload?: boolean
|
||||
acceptedCompleteJsonPayload?: boolean
|
||||
}
|
||||
}).bomStreamMetrics
|
||||
: undefined
|
||||
@ -583,6 +706,10 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
pendingBufferChars: streamMetrics?.pendingBufferChars ?? null,
|
||||
receivedDoneMarker: streamMetrics?.receivedDoneMarker ?? false,
|
||||
finishReason: streamMetrics?.finishReason ?? null,
|
||||
receivedCompleteJsonPayload:
|
||||
streamMetrics?.receivedCompleteJsonPayload ?? false,
|
||||
acceptedCompleteJsonPayload:
|
||||
streamMetrics?.acceptedCompleteJsonPayload ?? false,
|
||||
})
|
||||
if (
|
||||
attempt < maxAttempts &&
|
||||
@ -620,6 +747,15 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
'多模态分析服务没有返回可读取的结果',
|
||||
)
|
||||
}
|
||||
if (result.completedFromJsonPayload) {
|
||||
console.info('[BOM costing] accepted complete JSON before stream terminator', {
|
||||
operation: input.operation,
|
||||
model: input.config.model,
|
||||
attempt,
|
||||
responseContentChars: result.content.length,
|
||||
durationMs: Date.now() - startedAt,
|
||||
})
|
||||
}
|
||||
return {
|
||||
...result,
|
||||
model: result.model || input.config.model,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user