From 75180418299539da5e976bf26fe6d77ddafc4291 Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Tue, 11 Aug 2026 02:49:47 +0800 Subject: [PATCH] fix(bom-costing): avoid stalled synthesis streams --- .../bom-costing-multimodal-gateway.test.ts | 43 ++++++++++++ .../bom-costing-multimodal-gateway.ts | 67 +++++++++++++++++-- .../bom-costing-structure-blueprint.ts | 19 +++--- 3 files changed, 114 insertions(+), 15 deletions(-) diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts index 7915af60..d7f58cd5 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts @@ -11,6 +11,49 @@ describe('BOM 多模态网关', () => { jest.restoreAllMocks() }) + it('结构汇总默认请求一次性 JSON，并读取标准非流式响应', async () => { + const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue( + new Response(JSON.stringify({ + id: 'call-json-response', + model: 'gpt-5.6-terra', + choices: [{ + message: { content: '{"ok":true}' }, + finish_reason: 'stop', + }], + usage: { prompt_tokens: 12, completion_tokens: 5 }, + }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ) + + 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, + }) + + const requestBody = JSON.parse(String(fetchSpy.mock.calls[0]?.[1]?.body)) as { + stream?: boolean + } + expect(requestBody.stream).toBe(false) + expect(result).toMatchObject({ + id: 'call-json-response', + model: 'gpt-5.6-terra', + content: '{"ok":true}', + finishReason: 'stop', + promptTokens: 12, + completionTokens: 5, + }) + }) + it('流式响应空转超过时限时取消读取并返回可恢复的超时错误', async () => { let cancelReason: unknown const stalledStream = new ReadableStream({ diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts index 112490c6..5c37ea90 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts @@ -30,6 +30,16 @@ export interface BomCostingMultimodalCallResult { completedFromJsonPayload?: boolean } +interface BomCostingChatCompletionResponse { + id: string | null + model: string | null + content: string + promptTokens: number | null + completionTokens: number | null + finishReason: string | null + completedFromJsonPayload: boolean +} + export type BomCostingMultimodalOperation = 'IMAGE_FACTS' | 'SYNTHESIS' const BOM_COSTING_REASONING_EFFORTS = new Set([ @@ -285,7 +295,7 @@ function isCompleteJsonObjectPayload(content: string) { async function readStreamingChatCompletion( response: Response, options: { signal?: AbortSignal } = {}, -) { +): Promise { if (!response.body) { throw new AppError( ErrorCode.BOM_COSTING_INVALID_PROVIDER, @@ -466,6 +476,41 @@ async function readStreamingChatCompletion( } } +async function readNonStreamingChatCompletion( + response: Response, +): Promise { + const payload = await response.json() as { + id?: string + model?: string + choices?: Array<{ + message?: { content?: string } + finish_reason?: string | null + }> + usage?: { + prompt_tokens?: number + completion_tokens?: number + } + } + const choice = payload.choices?.[0] + return { + id: payload.id?.trim() || null, + model: payload.model?.trim() || null, + content: typeof choice?.message?.content === 'string' + ? choice.message.content.trim() + : '', + promptTokens: Number.isFinite(payload.usage?.prompt_tokens) + ? payload.usage!.prompt_tokens! + : null, + completionTokens: Number.isFinite(payload.usage?.completion_tokens) + ? payload.usage!.completion_tokens! + : null, + finishReason: typeof choice?.finish_reason === 'string' && choice.finish_reason.trim() + ? choice.finish_reason.trim() + : null, + completedFromJsonPayload: false, + } +} + function readTransportErrorCode(error: unknown): string | null { let current = error for (let depth = 0; depth < 4; depth += 1) { @@ -577,6 +622,12 @@ export async function callBomCostingMultimodalModel(input: { for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { let response: Response + // The UI streams durable stage events, not raw model tokens. Synthesis only + // needs one strict JSON object, so asking the provider for a regular JSON + // response avoids OpenAI-compatible gateways that emit a partial SSE body + // and then keep the socket open for minutes. Image analysis keeps streaming + // because those responses can be substantially larger. + const requestStreamingResponse = input.operation === 'IMAGE_FACTS' const timeoutSignal = AbortSignal.timeout(input.timeoutMs) const requestSignal = input.signal ? AbortSignal.any([timeoutSignal, input.signal]) @@ -597,7 +648,7 @@ export async function callBomCostingMultimodalModel(input: { response_format: { type: 'json_object' }, temperature: 0.1, max_tokens: input.maxTokens, - stream: true, + stream: requestStreamingResponse, ...(reasoningEffort ? { reasoning_effort: reasoningEffort } : {}), }), cache: 'no-store', @@ -673,11 +724,15 @@ export async function callBomCostingMultimodalModel(input: { ) } - let result: Awaited> + let result: BomCostingChatCompletionResponse try { - result = await readStreamingChatCompletion(response, { - signal: requestSignal, - }) + const responseContentType = response.headers.get('content-type')?.toLowerCase() ?? '' + const responseUsesEventStream = responseContentType.includes('text/event-stream') || ( + requestStreamingResponse && !responseContentType.includes('application/json') + ) + result = responseUsesEventStream + ? await readStreamingChatCompletion(response, { signal: requestSignal }) + : await readNonStreamingChatCompletion(response) } catch (error) { const taskDeadlineError = readTaskDeadlineError(requestSignal) if (taskDeadlineError) throw taskDeadlineError diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-structure-blueprint.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-structure-blueprint.ts index 217f7d46..00461508 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-structure-blueprint.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-structure-blueprint.ts @@ -360,13 +360,14 @@ function sanitizeBlueprintFactIds( } } -function getTimeoutMs() { - return readBomCostingBoundedInteger( +function getTimeoutMs(maximumMs = 60_000) { + const configured = readBomCostingBoundedInteger( process.env.BOM_COSTING_STRUCTURE_TIMEOUT_MS, - 100_000, + 60_000, 30_000, 180_000, ) + return Math.min(configured, maximumMs) } function recordResponse(trace: BomCostingStructureTrace, response: BomCostingMultimodalCallResult) { @@ -463,8 +464,8 @@ async function generateBlueprint(input: { reviewInstruction: input.analysisInstruction?.trim() || undefined, }), maxTokens: 2_400, - timeoutMs: getTimeoutMs(), - retrySynthesisTimeout: true, + timeoutMs: getTimeoutMs(60_000), + retrySynthesisTimeout: false, signal: input.signal, }) recordResponse(input.trace, response) @@ -557,8 +558,8 @@ async function verifyBlueprintGroup(input: { systemPrompt: VERIFICATION_SYSTEM_PROMPT, userContent: content, maxTokens: 4_000, - timeoutMs: getTimeoutMs(), - retrySynthesisTimeout: true, + timeoutMs: getTimeoutMs(60_000), + retrySynthesisTimeout: false, signal: input.signal, }) recordResponse(input.trace, response) @@ -933,8 +934,8 @@ async function instantiateBlueprint(input: { reviewInstruction: input.analysisInstruction?.trim() || undefined, }), maxTokens: 2_200, - timeoutMs: getTimeoutMs(), - retrySynthesisTimeout: true, + timeoutMs: getTimeoutMs(45_000), + retrySynthesisTimeout: false, signal: input.signal, }) recordResponse(input.trace, response)