fix(bom-costing): stop completed model streams promptly
This commit is contained in:
parent
e4554d86cf
commit
dbf48330bd
@ -44,6 +44,48 @@ describe('BOM 多模态网关', () => {
|
||||
expect((cancelReason as Error | undefined)?.name).toBe('TimeoutError')
|
||||
})
|
||||
|
||||
it('收到 DONE 标记后立即结束,不等待上游主动关闭长连接', async () => {
|
||||
let controller: ReadableStreamDefaultController<Uint8Array> | null = null
|
||||
let cancelReason: unknown
|
||||
const encoder = new TextEncoder()
|
||||
const keepAliveStream = new ReadableStream<Uint8Array>({
|
||||
start(streamController) {
|
||||
controller = streamController
|
||||
streamController.enqueue(encoder.encode(
|
||||
'data: {"id":"call-1","model":"gpt-5.6-terra","choices":[{"delta":{"content":"{\\"ok\\":true}"},"finish_reason":"stop"}]}\n\n' +
|
||||
'data: [DONE]\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('{"ok":true}')
|
||||
expect(result.finishReason).toBe('stop')
|
||||
expect(cancelReason).toBe('completed')
|
||||
expect(controller).not.toBeNull()
|
||||
})
|
||||
|
||||
it('规格汇总遇到网络超时时立即交由上层拆分,不白白重试整批', async () => {
|
||||
const networkTimeout = Object.assign(
|
||||
new Error('socket timed out'),
|
||||
|
||||
@ -249,12 +249,17 @@ async function readStreamingChatCompletion(
|
||||
let promptTokens: number | null = null
|
||||
let completionTokens: number | null = null
|
||||
let finishReason: string | null = null
|
||||
let receivedDoneMarker = false
|
||||
|
||||
const consumeLine = (line: string) => {
|
||||
const trimmed = line.trim()
|
||||
if (!trimmed.startsWith('data:')) return
|
||||
const data = trimmed.slice(5).trim()
|
||||
if (!data || data === '[DONE]') return
|
||||
if (!data) return
|
||||
if (data === '[DONE]') {
|
||||
receivedDoneMarker = true
|
||||
return
|
||||
}
|
||||
|
||||
let chunk: {
|
||||
id?: string
|
||||
@ -294,21 +299,32 @@ async function readStreamingChatCompletion(
|
||||
}
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
while (!receivedDoneMarker) {
|
||||
const { done, value } = await reader.read()
|
||||
throwIfAborted()
|
||||
if (done) break
|
||||
buffer += decoder.decode(value, { stream: true })
|
||||
const lines = buffer.split(/\r?\n/)
|
||||
buffer = lines.pop() ?? ''
|
||||
lines.forEach(consumeLine)
|
||||
for (const line of lines) {
|
||||
consumeLine(line)
|
||||
if (receivedDoneMarker) break
|
||||
}
|
||||
}
|
||||
buffer += decoder.decode()
|
||||
if (buffer.trim()) {
|
||||
if (!receivedDoneMarker) buffer += decoder.decode()
|
||||
if (!receivedDoneMarker && buffer.trim()) {
|
||||
buffer.split(/\r?\n/).forEach(consumeLine)
|
||||
}
|
||||
} finally {
|
||||
signal?.removeEventListener('abort', cancelReaderOnAbort)
|
||||
if (receivedDoneMarker) {
|
||||
// Some OpenAI-compatible gateways keep the SSE connection alive after
|
||||
// the protocol-level completion marker. Waiting for the TCP socket to
|
||||
// close turns a completed response into a false timeout and retains the
|
||||
// response buffers until the deadline. The marker is authoritative, so
|
||||
// release the stream immediately.
|
||||
await reader.cancel('completed').catch(() => undefined)
|
||||
}
|
||||
reader.releaseLock()
|
||||
}
|
||||
|
||||
|
||||
@ -197,7 +197,7 @@ describe('BOM v6 结构蓝图与 SKU 展开', () => {
|
||||
onCheckpoint: async (checkpoint) => { checkpoints.push(checkpoint) },
|
||||
})
|
||||
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(3)
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(4)
|
||||
expect(output.result.schemaVersion).toBe('competitor-analysis/v2')
|
||||
expect(output.result.skus).toHaveLength(7)
|
||||
expect(output.result.skus.map((sku) => sku.externalSkuId)).toEqual(
|
||||
@ -210,7 +210,7 @@ describe('BOM v6 结构蓝图与 SKU 展开', () => {
|
||||
expect(output.result.skus.flatMap((sku) => sku.bomLines).every((line) => (
|
||||
line.estimatedUnitPrice === undefined && line.suggestedLossFactor === undefined
|
||||
))).toBe(true)
|
||||
expect(output.trace.instantiationCallCount).toBe(2)
|
||||
expect(output.trace.instantiationCallCount).toBe(3)
|
||||
expect(stages.map((stage) => stage.stage)).toEqual(expect.arrayContaining([
|
||||
'BLUEPRINT',
|
||||
'INSTANTIATE',
|
||||
|
||||
@ -33,7 +33,7 @@ const MAX_BLUEPRINT_CONTRACT_ATTEMPTS = 2
|
||||
const MAX_VERIFICATION_ROUNDS_PER_GROUP = 3
|
||||
const MAX_QUESTIONS_PER_VERIFICATION_ROUND = 8
|
||||
const MAX_IMAGES_PER_VERIFICATION_ROUND = 4
|
||||
const MAX_SKUS_PER_INSTANTIATION_CALL = 6
|
||||
const MAX_SKUS_PER_INSTANTIATION_CALL = 3
|
||||
|
||||
const BLUEPRINT_SYSTEM_PROMPT = `你是家具产品结构工程师。根据系统保存的商品事实、真实 SKU 和版本化结构知识,先建立产品级结构蓝图,不输出价格,不输出每个 SKU 的最终 BOM。
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user