feat: 增加fetch请求，优化axios对流式数据处理的不友好 (#640)

King5-12 · weizhongjindragonplus · web-flow · commit 12a3d969361f · 2025-07-11T16:16:54.000+08:00
fix: 修复消息刷新后的更新问题

Co-authored-by: kim &lt;weizhong.jin@dragonplus.com&gt;
diff --git a/src/api/index.ts b/src/api/index.ts
@@ -1,7 +1,8 @@
 import type { AnnounceConfig, AuditConfig, ConfigState, GiftCard, KeyConfig, MailConfig, SearchConfig, SiteConfig, Status, UserInfo, UserPassword, UserPrompt } from '@/components/common/Setting/model'
 import type { SettingsState } from '@/store/modules/user/helper'
-import { useAuthStore, useUserStore } from '@/store'
+import { useUserStore } from '@/store'
 import { get, post } from '@/utils/request'
+import fetchService from '@/utils/request/fetchService'
 
 export function fetchAnnouncement<T = any>() {
   return post<T>({
@@ -26,7 +27,7 @@ interface SSEEventHandlers {
   onEnd?: () => void
 }
 
-// SSE chat processing function
+// SSE chat processing function using custom fetch service
 export function fetchChatAPIProcessSSE(
   params: {
     roomId: number
@@ -40,7 +41,6 @@ export function fetchChatAPIProcessSSE(
   handlers: SSEEventHandlers,
 ): Promise<void> {
   const userStore = useUserStore()
-  const authStore = useAuthStore()
 
   const data: Record<string, any> = {
     roomId: params.roomId,
@@ -55,92 +55,66 @@ export function fetchChatAPIProcessSSE(
   }
 
   return new Promise((resolve, reject) => {
-    const baseURL = import.meta.env.VITE_GLOB_API_URL || ''
-    const url = `${baseURL}/api/chat-process`
-
-    fetch(url, {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        'Authorization': authStore.token ? `Bearer ${authStore.token}` : '',
+    fetchService.postStream(
+      {
+        url: '/chat-process',
+        body: data,
+        signal: params.signal,
       },
-      body: JSON.stringify(data),
-      signal: params.signal,
-    }).then((response) => {
-      if (!response.ok) {
-        throw new Error(`HTTP error! status: ${response.status}`)
-      }
-
-      const reader = response.body?.getReader()
-      if (!reader) {
-        throw new Error('No reader available')
-      }
-
-      const decoder = new TextDecoder()
-      let buffer = ''
-
-      function readStream(): void {
-        reader!.read().then(({ done, value }) => {
-          if (done) {
-            handlers.onEnd?.()
-            resolve()
+      {
+        onChunk: (line: string) => {
+          if (line.trim() === '')
             return
-          }
 
-          buffer += decoder.decode(value, { stream: true })
-          const lines = buffer.split('\n')
-          buffer = lines.pop() || '' // Keep the incomplete line in buffer
+          if (line.startsWith('event: ')) {
+            // const _eventType = line.substring(7).trim()
+            return
+          }
 
-          for (const line of lines) {
-            if (line.trim() === '')
-              continue
+          if (line.startsWith('data: ')) {
+            const data = line.substring(6).trim()
 
-            if (line.startsWith('event: ')) {
-              // const _eventType = line.substring(7).trim()
-              continue
+            if (data === '[DONE]') {
+              handlers.onEnd?.()
+              resolve()
+              return
             }
 
-            if (line.startsWith('data: ')) {
-              const data = line.substring(6).trim()
+            try {
+              const jsonData = JSON.parse(data)
 
-              if (data === '[DONE]') {
-                handlers.onEnd?.()
-                resolve()
-                return
+              // Dispatch to different handlers based on data type
+              if (jsonData.message) {
+                handlers.onError?.(jsonData.message)
               }
-
-              try {
-                const jsonData = JSON.parse(data)
-
-                // 根据前面的 event 类型分发到不同的处理器
-                if (jsonData.message) {
-                  handlers.onError?.(jsonData.message)
-                }
-                else if (jsonData.searchQuery) {
-                  handlers.onSearchQuery?.(jsonData)
-                }
-                else if (jsonData.searchResults) {
-                  handlers.onSearchResults?.(jsonData)
-                }
-                else if (jsonData.m) {
-                  handlers.onDelta?.(jsonData.m)
-                }
-                else {
-                  handlers.onMessage?.(jsonData)
-                }
+              else if (jsonData.searchQuery) {
+                handlers.onSearchQuery?.(jsonData)
+              }
+              else if (jsonData.searchResults) {
+                handlers.onSearchResults?.(jsonData)
               }
-              catch (e) {
-                console.error('Failed to parse SSE data:', data, e)
+              else if (jsonData.m) {
+                handlers.onDelta?.(jsonData.m)
               }
+              else {
+                handlers.onMessage?.(jsonData)
+              }
+            }
+            catch (e) {
+              console.error('Failed to parse SSE data:', data, e)
             }
           }
-
-          readStream()
-        }).catch(reject)
-      }
-
-      readStream()
-    }).catch(reject)
+        },
+        onError: (error: Error) => {
+          handlers.onError?.(error.message)
+          reject(error)
+        },
+        onComplete: () => {
+          handlers.onEnd?.()
+          resolve()
+        },
+      },
+    )
   })
 }
 
diff --git a/src/utils/request/fetchService.ts b/src/utils/request/fetchService.ts
@@ -0,0 +1,147 @@
+import { useAuthStore } from '@/store'
+
+export interface FetchRequestConfig {
+  url: string
+  method?: string
+  headers?: Record<string, string>
+  body?: any
+  signal?: AbortSignal
+}
+
+export interface FetchResponse<T = any> {
+  data: T
+  status: number
+  statusText: string
+  headers: Headers
+}
+
+export interface SSEStreamOptions {
+  onChunk?: (chunk: string) => void
+  onError?: (error: Error) => void
+  onComplete?: () => void
+}
+
+class FetchService {
+  private baseURL: string
+  private defaultHeaders: Record<string, string>
+
+  constructor() {
+    this.baseURL = import.meta.env.VITE_GLOB_API_URL || ''
+    this.defaultHeaders = {
+      'Content-Type': 'application/json',
+    }
+  }
+
+  // Request interceptor - automatically add authentication headers and other configurations
+  private requestInterceptor(config: FetchRequestConfig): FetchRequestConfig {
+    const token = useAuthStore().token
+    const headers = { ...this.defaultHeaders, ...config.headers }
+
+    if (token) {
+      headers.Authorization = `Bearer ${token}`
+    }
+
+    return {
+      ...config,
+      headers,
+    }
+  }
+
+  // Response interceptor - handle error status
+  private async responseInterceptor(response: Response): Promise<Response> {
+    if (!response.ok) {
+      throw new Error(`HTTP error! status: ${response.status}`)
+    }
+    return response
+  }
+
+  // POST request
+  async post<T = any>(config: FetchRequestConfig): Promise<FetchResponse<T>> {
+    const processedConfig = this.requestInterceptor(config)
+    const url = `${this.baseURL}${processedConfig.url}`
+
+    const response = await fetch(url, {
+      method: 'POST',
+      headers: processedConfig.headers,
+      body: typeof processedConfig.body === 'object'
+        ? JSON.stringify(processedConfig.body)
+        : processedConfig.body,
+      signal: processedConfig.signal,
+    })
+
+    const processedResponse = await this.responseInterceptor(response)
+    const data = await processedResponse.json()
+
+    return {
+      data,
+      status: processedResponse.status,
+      statusText: processedResponse.statusText,
+      headers: processedResponse.headers,
+    }
+  }
+
+  // SSE streaming request
+  async postStream(config: FetchRequestConfig, options: SSEStreamOptions): Promise<void> {
+    const processedConfig = this.requestInterceptor(config)
+    const url = `${this.baseURL}${processedConfig.url}`
+
+    try {
+      const response = await fetch(url, {
+        method: 'POST',
+        headers: processedConfig.headers,
+        body: typeof processedConfig.body === 'object'
+          ? JSON.stringify(processedConfig.body)
+          : processedConfig.body,
+        signal: processedConfig.signal,
+      })
+
+      await this.responseInterceptor(response)
+
+      if (!response.body) {
+        throw new Error('ReadableStream not supported')
+      }
+
+      const reader = response.body.getReader()
+      const decoder = new TextDecoder()
+      let buffer = ''
+
+      try {
+        while (true) {
+          const { done, value } = await reader.read()
+
+          if (done) {
+            options.onComplete?.()
+            break
+          }
+
+          // Decode the chunk and add to buffer
+          buffer += decoder.decode(value, { stream: true })
+
+          // Process complete lines
+          const lines = buffer.split('\n')
+          // Keep the last potentially incomplete line
+          buffer = lines.pop() || ''
+
+          for (const line of lines) {
+            if (line.trim()) {
+              options.onChunk?.(line)
+            }
+          }
+        }
+      }
+      catch (error) {
+        options.onError?.(error as Error)
+        throw error
+      }
+    }
+    catch (error) {
+      options.onError?.(error as Error)
+      throw error
+    }
+  }
+}
+
+// Create singleton instance
+const fetchService = new FetchService()
+
+export default fetchService
diff --git a/src/views/chat/index.vue b/src/views/chat/index.vue
@@ -369,6 +369,10 @@ async function onRegenerate(index: number) {
     let lastText = ''
     let accumulatedReasoning = ''
     const fetchChatAPIOnce = async () => {
+      let searchQuery: string
+      let searchResults: Chat.SearchResult[]
+      let searchUsageTime: number
+
       await fetchChatAPIProcessSSE({
         roomId: currentChatRoom.value!.roomId,
         uuid: chatUuid || Date.now(),
@@ -377,6 +381,13 @@ async function onRegenerate(index: number) {
         options,
         signal: controller.signal,
       }, {
+        onSearchQuery: (data) => {
+          searchQuery = data.searchQuery
+        },
+        onSearchResults: (data) => {
+          searchResults = data.searchResults
+          searchUsageTime = data.searchUsageTime
+        },
         onDelta: async (delta) => {
           // 处理增量数据
           if (delta.text) {
@@ -391,6 +402,9 @@ async function onRegenerate(index: number) {
             index,
             {
               dateTime: new Date().toLocaleString(),
+              searchQuery,
+              searchResults,
+              searchUsageTime,
               reasoning: accumulatedReasoning,
               text: lastText,
               inversion: false,
@@ -406,6 +420,13 @@ async function onRegenerate(index: number) {
         },
         onMessage: async (data) => {
           // Handle complete message data (compatibility mode)
+          if (data.searchQuery)
+            searchQuery = data.searchQuery
+          if (data.searchResults)
+            searchResults = data.searchResults
+          if (data.searchUsageTime)
+            searchUsageTime = data.searchUsageTime
+          // Handle complete message data (compatibility mode)
           const usage = (data.detail && data.detail.usage)
             ? {
                 completion_tokens: data.detail.usage.completion_tokens || null,
@@ -419,6 +440,9 @@ async function onRegenerate(index: number) {
             index,
             {
               dateTime: new Date().toLocaleString(),
+              searchQuery,
+              searchResults,
+              searchUsageTime,
               reasoning: data?.reasoning,
               finish_reason: data?.finish_reason,
               text: data.text ?? '',
