feat: AI 对话支持 A2UI 表单/审查/图表与 Excel 读取

apps/server/src/ai-chat/ai-model-stream.service.ts

@@ -4,6 +4,7 @@ import * as http from 'node:http';
 import * as https from 'node:https';
 import { isIP } from 'node:net';
 import type { AiRuntimeConfig } from '../ai-config/dto/ai-config.dto';
+import { AiProvider } from '../ai-config/ai-config.entity';
 import type { ModelMessage, ModelStreamEvent } from './ai-chat.types';
 
 interface ChatTool {
@@ -26,6 +27,17 @@ interface StreamChoiceDelta {
 }
 
 const MAX_UPSTREAM_EVENT_BYTES = 1024 * 1024;
+const MAX_UPSTREAM_RETRIES = 3;
+const UPSTREAM_RETRY_DELAYS_MS = [500, 1000, 2000];
+const RETRYABLE_STATUS_CODES = new Set([408, 425, 429, 500, 502, 503, 504]);
+const RETRYABLE_TRANSPORT_CODES = new Set([
+  'ECONNRESET',
+  'ECONNREFUSED',
+  'ETIMEDOUT',
+  'ENOTFOUND',
+  'EAI_AGAIN',
+  'EPIPE',
+]);
 
 // Known public provider hosts — trusted even if CDN resolves to private-range IPs
 const DNS_TRUSTED_HOSTS = new Set(['api.openai.com', 'api.deepseek.com']);
@@ -54,31 +66,76 @@ export class AiModelStreamService {
     tools: ChatTool[],
     signal: AbortSignal,
   ): AsyncGenerator<ModelStreamEvent> {
-    const timeout = AbortSignal.timeout(config.timeoutMs);
-    const combinedSignal = AbortSignal.any([signal, timeout]);
-    let response: PinnedResponse;
+    const requestBody = JSON.stringify({
+      model: config.defaultModel,
+      messages,
+      stream: true,
+      ...(tools.length ? { tools, tool_choice: 'auto' } : {}),
+      // reasoning_effort 仅对支持该参数的 OpenAI 兼容服务生效；
+      // DeepSeek 官方接口不接受该参数，避免请求被拒。
+      ...(config.reasoningEffort &&
+      config.reasoningEffort !== 'none' &&
+      config.provider !== AiProvider.DEEPSEEK
+        ? { reasoning_effort: config.reasoningEffort }
+        : {}),
+    });
+    const url = `${config.baseUrl.replace(/\/$/, '')}/chat/completions`;
+    const headers = {
+      Authorization: `Bearer ${config.apiKey}`,
+      'Content-Type': 'application/json',
+      Accept: 'text/event-stream',
+    };
+    let response: PinnedResponse | null = null;
+    let activeTimeout: AbortSignal | undefined;
 
-    try {
-      response = await this.pinnedPost(
-        `${config.baseUrl.replace(/\/$/, '')}/chat/completions`,
-        {
-          Authorization: `Bearer ${config.apiKey}`,
-          'Content-Type': 'application/json',
-          Accept: 'text/event-stream',
-        },
-        JSON.stringify({
-          model: config.defaultModel,
-          messages,
-          stream: true,
-          ...(tools.length ? { tools, tool_choice: 'auto' } : {}),
-        }),
-        combinedSignal,
-      );
-    } catch (error) {
-      if (timeout.aborted && !signal.aborted) throw new RequestTimeoutException('AI 服务响应超时');
-      throw error;
+    for (let attempt = 0; attempt <= MAX_UPSTREAM_RETRIES; attempt += 1) {
+      activeTimeout = AbortSignal.timeout(config.timeoutMs);
+      const combinedSignal = AbortSignal.any([signal, activeTimeout]);
+      try {
+        response = await this.pinnedPost(url, headers, requestBody, combinedSignal);
+      } catch (error) {
+        if (activeTimeout.aborted && !signal.aborted) {
+          throw new RequestTimeoutException('AI 服务响应超时');
+        }
+        if (
+          attempt < MAX_UPSTREAM_RETRIES &&
+          !signal.aborted &&
+          this.isRetryableTransportError(error)
+        ) {
+          const delayMs = UPSTREAM_RETRY_DELAYS_MS[attempt];
+          yield {
+            type: 'retrying',
+            attempt: attempt + 1,
+            maxRetries: MAX_UPSTREAM_RETRIES,
+            delayMs,
+            reason: error instanceof Error ? error.message : '网络连接失败',
+          };
+          await this.sleep(delayMs);
+          continue;
+        }
+        throw error;
+      }
+
+      if (response.status >= 200 && response.status < 300) break;
+      if (attempt < MAX_UPSTREAM_RETRIES && RETRYABLE_STATUS_CODES.has(response.status)) {
+        response.body.resume?.();
+        const delayMs = UPSTREAM_RETRY_DELAYS_MS[attempt];
+        yield {
+          type: 'retrying',
+          attempt: attempt + 1,
+          maxRetries: MAX_UPSTREAM_RETRIES,
+          delayMs,
+          reason: `上游返回 ${response.status}`,
+        };
+        await this.sleep(delayMs);
+        continue;
+      }
+      const body = await this.readLimitedBody(response.body);
+      throw new BadGatewayException(this.safeUpstreamMessage(response.status, body));
     }
 
+    if (!response) throw new BadGatewayException('AI 服务暂时不可用');
+
     if (response.status < 200 || response.status >= 300) {
       const body = await this.readLimitedBody(response.body);
       throw new BadGatewayException(this.safeUpstreamMessage(response.status, body));
@@ -128,7 +185,9 @@ export class AiModelStreamService {
       buffer += decoder.decode();
       if (buffer.trim()) for (const parsed of consumeEvent(buffer)) yield parsed;
     } catch (error) {
-      if (timeout.aborted && !signal.aborted) throw new RequestTimeoutException('AI 服务响应超时');
+      if (activeTimeout?.aborted && !signal.aborted) {
+        throw new RequestTimeoutException('AI 服务响应超时');
+      }
       throw error;
     }
 
@@ -154,9 +213,21 @@ export class AiModelStreamService {
     }
   }
 
+  private isRetryableTransportError(error: unknown): boolean {
+    if (!(error instanceof Error)) return false;
+    const code = (error as NodeJS.ErrnoException).code;
+    if (code && RETRYABLE_TRANSPORT_CODES.has(code)) return true;
+    return /socket hang up|ECONNRESET|ETIMEDOUT|EAI_AGAIN/i.test(error.message);
+  }
+
+  private sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
+
   private safeUpstreamMessage(status: number, body: string): string {
     if (status === 401 || status === 403) return 'AI 服务认证失败';
     if (status === 429) return 'AI 服务请求过于频繁';
+    if (status === 503) return 'AI 服务繁忙，请稍后重试';
     if (status >= 500) return 'AI 服务暂时不可用';
     const message = this.extractErrorMessage(body);
     return message ? `AI 服务请求失败：${message}` : `AI 服务请求失败（${status}）`;