feat: think tag parse (#3805) (#3808)

* feat: think tag parse

* remove some model config

* feat: parse think tag test

docSite/content/zh-cn/docs/development/upgrading/490.md

+---
+title: 'V4.9.0(进行中)'
+description: 'FastGPT V4.9.0 更新说明'
+icon: 'upgrade'
+draft: false
+toc: true
+weight: 803
+---
+
+
+## 完整更新内容
+
+1. 新增 - AI 对话节点解析 <think></think> 标签内容，便于各类模型进行思考链输出。
+2. 修复 - 思考链流输出时，有时与正文顺序偏差。
\ No newline at end of file

packages/global/core/ai/type.d.ts

 import openai from 'openai';
 import type {
   ChatCompletionMessageToolCall,
-  ChatCompletionChunk,
   ChatCompletionMessageParam as SdkChatCompletionMessageParam,
   ChatCompletionToolMessageParam,
   ChatCompletionContentPart as SdkChatCompletionContentPart,
   ChatCompletionUserMessageParam as SdkChatCompletionUserMessageParam,
   ChatCompletionToolMessageParam as SdkChatCompletionToolMessageParam,
-  ChatCompletionAssistantMessageParam as SdkChatCompletionAssistantMessageParam,
-  ChatCompletionContentPartText
+  ChatCompletionAssistantMessageParam as SdkChatCompletionAssistantMessageParam
 } from 'openai/resources';
 import { ChatMessageTypeEnum } from './constants';
 import { WorkflowInteractiveResponseType } from '../workflow/template/system/interactive/type';
@@ -71,7 +69,8 @@ export type ChatCompletionMessageFunctionCall =
   };
 
 // Stream response
-export type StreamChatType = Stream<ChatCompletionChunk>;
+export type StreamChatType = Stream<openai.Chat.Completions.ChatCompletionChunk>;
+export type UnStreamChatType = openai.Chat.Completions.ChatCompletion;
 
 export default openai;
 export * from 'openai';

packages/global/core/workflow/runtime/utils.ts

@@ -10,6 +10,7 @@ import { FlowNodeOutputItemType, ReferenceValueType } from '../type/io';
 import { ChatItemType, NodeOutputItemType } from '../../../core/chat/type';
 import { ChatItemValueTypeEnum, ChatRoleEnum } from '../../../core/chat/constants';
 import { replaceVariable, valToStr } from '../../../common/string/tools';
+import { ChatCompletionChunk } from 'openai/resources';
 
 export const getMaxHistoryLimitFromNodes = (nodes: StoreNodeItemType[]): number => {
   let limit = 10;
@@ -419,3 +420,137 @@ export function rewriteNodeOutputByHistories(
     };
   });
 }
+
+// Parse <think></think> tags to think and answer - unstream response
+export const parseReasoningContent = (text: string): [string, string] => {
+  const regex = /<think>([\s\S]*?)<\/think>/;
+  const match = text.match(regex);
+
+  if (!match) {
+    return ['', text];
+  }
+
+  const thinkContent = match[1].trim();
+
+  // Add answer (remaining text after think tag)
+  const answerContent = text.slice(match.index! + match[0].length);
+
+  return [thinkContent, answerContent];
+};
+
+// Parse <think></think> tags to think and answer - stream response
+export const parseReasoningStreamContent = () => {
+  let isInThinkTag: boolean | undefined;
+
+  const startTag = '<think>';
+  let startTagBuffer = '';
+
+  const endTag = '</think>';
+  let endTagBuffer = '';
+
+  /* 
+    parseReasoning - 只控制是否主动解析 <think></think>，如果接口已经解析了，仍然会返回 think 内容。
+  */
+  const parsePart = (
+    part: {
+      choices: {
+        delta: {
+          content?: string;
+          reasoning_content?: string;
+        };
+      }[];
+    },
+    parseReasoning = false
+  ): [string, string] => {
+    const content = part.choices?.[0]?.delta?.content || '';
+
+    // @ts-ignore
+    const reasoningContent = part.choices?.[0]?.delta?.reasoning_content || '';
+    if (reasoningContent || !parseReasoning) {
+      isInThinkTag = false;
+      return [reasoningContent, content];
+    }
+
+    if (!content) {
+      return ['', ''];
+    }
+
+    // 如果不在 think 标签中，或者有 reasoningContent(接口已解析），则返回 reasoningContent 和 content
+    if (isInThinkTag === false) {
+      return ['', content];
+    }
+
+    // 检测是否为 think 标签开头的数据
+    if (isInThinkTag === undefined) {
+      // Parse content think and answer
+      startTagBuffer += content;
+      // 太少内容时候，暂时不解析
+      if (startTagBuffer.length < startTag.length) {
+        return ['', ''];
+      }
+
+      if (startTagBuffer.startsWith(startTag)) {
+        isInThinkTag = true;
+        return [startTagBuffer.slice(startTag.length), ''];
+      }
+
+      // 如果未命中 think 标签，则认为不在 think 标签中，返回 buffer 内容作为 content
+      isInThinkTag = false;
+      return ['', startTagBuffer];
+    }
+
+    // 确认是 think 标签内容，开始返回 think 内容，并实时检测 </think>
+    /* 
+      检测 </think> 方案。
+      存储所有疑似 </think> 的内容，直到检测到完整的 </think> 标签或超出 </think> 长度。
+      content 返回值包含以下几种情况:
+        abc - 完全未命中尾标签
+        abc<th - 命中一部分尾标签
+        abc</think> - 完全命中尾标签
+        abc</think>abc - 完全命中尾标签
+        </think>abc - 完全命中尾标签
+        k>abc - 命中一部分尾标签
+    */
+    // endTagBuffer 专门用来记录疑似尾标签的内容
+    if (endTagBuffer) {
+      endTagBuffer += content;
+      if (endTagBuffer.includes(endTag)) {
+        isInThinkTag = false;
+        const answer = endTagBuffer.slice(endTag.length);
+        return ['', answer];
+      } else if (endTagBuffer.length >= endTag.length) {
+        // 缓存内容超出尾标签长度，且仍未命中 </think>，则认为本次猜测 </think> 失败，仍处于 think 阶段。
+        const tmp = endTagBuffer;
+        endTagBuffer = '';
+        return [tmp, ''];
+      }
+      return ['', ''];
+    } else if (content.includes(endTag)) {
+      // 返回内容，完整命中</think>，直接结束
+      isInThinkTag = false;
+      const [think, answer] = content.split(endTag);
+      return [think, answer];
+    } else {
+      // 无 buffer，且未命中 </think>，开始疑似 </think> 检测。
+      for (let i = 1; i < endTag.length; i++) {
+        const partialEndTag = endTag.slice(0, i);
+        // 命中一部分尾标签
+        if (content.endsWith(partialEndTag)) {
+          const think = content.slice(0, -partialEndTag.length);
+          endTagBuffer += partialEndTag;
+          return [think, ''];
+        }
+      }
+    }
+
+    // 完全未命中尾标签，还是 think 阶段。
+    return [content, ''];
+  };
+
+  const getStartTagBuffer = () => startTagBuffer;
+
+  return {
+    parsePart,
+    getStartTagBuffer
+  };
+};

packages/service/core/ai/config.ts

 import OpenAI from '@fastgpt/global/core/ai';
 import {
   ChatCompletionCreateParamsNonStreaming,
-  ChatCompletionCreateParamsStreaming
+  ChatCompletionCreateParamsStreaming,
+  StreamChatType,
+  UnStreamChatType
 } from '@fastgpt/global/core/ai/type';
 import { getErrText } from '@fastgpt/global/common/error/utils';
 import { addLog } from '../../common/system/log';
@@ -38,29 +40,30 @@ export const getAxiosConfig = (props?: { userKey?: OpenaiAccountType }) => {
   };
 };
 
-type CompletionsBodyType =
-  | ChatCompletionCreateParamsNonStreaming
-  | ChatCompletionCreateParamsStreaming;
-type InferResponseType<T extends CompletionsBodyType> =
-  T extends ChatCompletionCreateParamsStreaming
-    ? OpenAI.Chat.Completions.ChatCompletionChunk
-    : OpenAI.Chat.Completions.ChatCompletion;
-
-export const createChatCompletion = async <T extends CompletionsBodyType>({
+export const createChatCompletion = async ({
   body,
   userKey,
   timeout,
   options
 }: {
-  body: T;
+  body: ChatCompletionCreateParamsNonStreaming | ChatCompletionCreateParamsStreaming;
   userKey?: OpenaiAccountType;
   timeout?: number;
   options?: OpenAI.RequestOptions;
-}): Promise<{
-  response: InferResponseType<T>;
-  isStreamResponse: boolean;
-  getEmptyResponseTip: () => string;
-}> => {
+}): Promise<
+  {
+    getEmptyResponseTip: () => string;
+  } & (
+    | {
+        response: StreamChatType;
+        isStreamResponse: true;
+      }
+    | {
+        response: UnStreamChatType;
+        isStreamResponse: false;
+      }
+  )
+> => {
   try {
     const modelConstantsData = getLLMModel(body.model);
 
@@ -96,9 +99,17 @@ export const createChatCompletion = async <T extends CompletionsBodyType>({
       return i18nT('chat:LLM_model_response_empty');
     };
 
+    if (isStreamResponse) {
+      return {
+        response,
+        isStreamResponse: true,
+        getEmptyResponseTip
+      };
+    }
+
     return {
-      response: response as InferResponseType<T>,
-      isStreamResponse,
+      response,
+      isStreamResponse: false,
       getEmptyResponseTip
     };
   } catch (error) {

packages/service/core/ai/utils.ts

@@ -37,25 +37,26 @@ export const computedTemperature = ({
   return temperature;
 };
 
-type CompletionsBodyType = (
+type CompletionsBodyType =
   | ChatCompletionCreateParamsNonStreaming
-  | ChatCompletionCreateParamsStreaming
-) & {
-  response_format?: any;
-  json_schema?: string;
-  stop?: string;
-};
+  | ChatCompletionCreateParamsStreaming;
 type InferCompletionsBody<T> = T extends { stream: true }
   ? ChatCompletionCreateParamsStreaming
-  : ChatCompletionCreateParamsNonStreaming;
+  : T extends { stream: false }
+    ? ChatCompletionCreateParamsNonStreaming
+    : ChatCompletionCreateParamsNonStreaming | ChatCompletionCreateParamsStreaming;
 
 export const llmCompletionsBodyFormat = <T extends CompletionsBodyType>(
-  body: T,
+  body: T & {
+    response_format?: any;
+    json_schema?: string;
+    stop?: string;
+  },
   model: string | LLMModelItemType
 ): InferCompletionsBody<T> => {
   const modelData = typeof model === 'string' ? getLLMModel(model) : model;
   if (!modelData) {
-    return body as InferCompletionsBody<T>;
+    return body as unknown as InferCompletionsBody<T>;
   }
 
   const response_format = body.response_format;
@@ -91,9 +92,7 @@ export const llmCompletionsBodyFormat = <T extends CompletionsBodyType>(
     });
   }
 
-  // console.log(requestBody);
-
-  return requestBody as InferCompletionsBody<T>;
+  return requestBody as unknown as InferCompletionsBody<T>;
 };
 
 export const llmStreamResponseToText = async (response: StreamChatType) => {

packages/service/core/workflow/dispatch/chat/oneapi.ts

@@ -3,13 +3,13 @@ import { filterGPTMessageByMaxContext, loadRequestMessages } from '../../../chat
 import type { ChatItemType, UserChatItemValueItemType } from '@fastgpt/global/core/chat/type.d';
 import { ChatRoleEnum } from '@fastgpt/global/core/chat/constants';
 import { SseResponseEventEnum } from '@fastgpt/global/core/workflow/runtime/constants';
-import { textAdaptGptResponse } from '@fastgpt/global/core/workflow/runtime/utils';
+import {
+  parseReasoningContent,
+  parseReasoningStreamContent,
+  textAdaptGptResponse
+} from '@fastgpt/global/core/workflow/runtime/utils';
 import { createChatCompletion } from '../../../ai/config';
-import type {
-  ChatCompletion,
-  ChatCompletionMessageParam,
-  StreamChatType
-} from '@fastgpt/global/core/ai/type.d';
+import type { ChatCompletionMessageParam, StreamChatType } from '@fastgpt/global/core/ai/type.d';
 import { formatModelChars2Points } from '../../../../support/wallet/usage/utils';
 import type { LLMModelItemType } from '@fastgpt/global/core/ai/model.d';
 import { postTextCensor } from '../../../../common/api/requestPlusApi';
@@ -195,7 +195,13 @@ export const dispatchChatCompletion = async (props: ChatProps): Promise<ChatResp
   });
 
   const { answerText, reasoningText } = await (async () => {
-    if (res && isStreamResponse) {
+    if (isStreamResponse) {
+      if (!res) {
+        return {
+          answerText: '',
+          reasoningText: ''
+        };
+      }
       // sse response
       const { answer, reasoning } = await streamResponse({
         res,
@@ -210,34 +216,49 @@ export const dispatchChatCompletion = async (props: ChatProps): Promise<ChatResp
         reasoningText: reasoning
       };
     } else {
-      const unStreamResponse = response as ChatCompletion;
-      const answer = unStreamResponse.choices?.[0]?.message?.content || '';
-      // @ts-ignore
-      const reasoning = unStreamResponse.choices?.[0]?.message?.reasoning_content || '';
+      const { content, reasoningContent } = (() => {
+        const content = response.choices?.[0]?.message?.content || '';
+        // @ts-ignore
+        const reasoningContent: string = response.choices?.[0]?.message?.reasoning_content || '';
+
+        // API already parse reasoning content
+        if (reasoningContent || !aiChatReasoning) {
+          return {
+            content,
+            reasoningContent
+          };
+        }
+
+        const [think, answer] = parseReasoningContent(content);
+        return {
+          content: answer,
+          reasoningContent: think
+        };
+      })();
 
       // Some models do not support streaming
       if (stream) {
-        if (isResponseAnswerText && answer) {
+        if (aiChatReasoning && reasoningContent) {
           workflowStreamResponse?.({
             event: SseResponseEventEnum.fastAnswer,
             data: textAdaptGptResponse({
-              text: answer
+              reasoning_content: reasoningContent
             })
           });
         }
-        if (aiChatReasoning && reasoning) {
+        if (isResponseAnswerText && content) {
           workflowStreamResponse?.({
             event: SseResponseEventEnum.fastAnswer,
             data: textAdaptGptResponse({
-              reasoning_content: reasoning
+              text: content
             })
           });
         }
       }
 
       return {
-        answerText: answer,
-        reasoningText: reasoning
+        answerText: content,
+        reasoningText: reasoningContent
       };
     }
   })();
@@ -267,7 +288,7 @@ export const dispatchChatCompletion = async (props: ChatProps): Promise<ChatResp
   });
 
   return {
-    answerText,
+    answerText: answerText.trim(),
     reasoningText,
     [DispatchNodeResponseKeyEnum.nodeResponse]: {
       totalPoints: externalProvider.openaiAccount?.key ? 0 : totalPoints,
@@ -500,36 +521,43 @@ async function streamResponse({
   });
   let answer = '';
   let reasoning = '';
+  const { parsePart, getStartTagBuffer } = parseReasoningStreamContent();
+
   for await (const part of stream) {
     if (res.closed) {
       stream.controller?.abort();
       break;
     }
 
-    const content = part.choices?.[0]?.delta?.content || '';
+    const [reasoningContent, content] = parsePart(part, aiChatReasoning);
     answer += content;
-    if (isResponseAnswerText && content) {
+    reasoning += reasoningContent;
+
+    if (aiChatReasoning && reasoningContent) {
       workflowStreamResponse?.({
         write,
         event: SseResponseEventEnum.answer,
         data: textAdaptGptResponse({
-          text: content
+          reasoning_content: reasoningContent
         })
       });
     }
 
-    const reasoningContent = part.choices?.[0]?.delta?.reasoning_content || '';
-    reasoning += reasoningContent;
-    if (aiChatReasoning && reasoningContent) {
+    if (isResponseAnswerText && content) {
       workflowStreamResponse?.({
         write,
         event: SseResponseEventEnum.answer,
         data: textAdaptGptResponse({
-          reasoning_content: reasoningContent
+          text: content
         })
       });
     }
   }
 
+  // if answer is empty, try to get value from startTagBuffer. (Cause: The response content is too short to exceed the minimum parse length)
+  if (answer === '') {
+    answer = getStartTagBuffer();
+  }
+
   return { answer, reasoning };
 }

packages/service/core/workflow/dispatch/index.ts

@@ -243,12 +243,17 @@ export async function dispatchWorkFlow(data: Props): Promise<DispatchFlowRespons
       chatAssistantResponse = chatAssistantResponse.concat(assistantResponses);
     } else {
       if (reasoningText) {
-        chatAssistantResponse.push({
-          type: ChatItemValueTypeEnum.reasoning,
-          reasoning: {
-            content: reasoningText
-          }
-        });
+        const isResponseReasoningText = inputs.find(
+          (item) => item.key === NodeInputKeyEnum.aiChatReasoning
+        )?.value;
+        if (isResponseReasoningText) {
+          chatAssistantResponse.push({
+            type: ChatItemValueTypeEnum.reasoning,
+            reasoning: {
+              content: reasoningText
+            }
+          });
+        }
       }
       if (answerText) {
         // save assistant text response

projects/app/src/pages/api/v1/chat/utils.test.ts

+import '@/pages/api/__mocks__/base';
+import { parseReasoningStreamContent } from '@fastgpt/global/core/workflow/runtime/utils';
+
+test('Parse reasoning stream content test', async () => {
+  const partList = [
+    {
+      data: [{ content: '你好1' }, { content: '你好2' }, { content: '你好3' }],
+      correct: { answer: '你好1你好2你好3', reasoning: '' }
+    },
+    {
+      data: [
+        { reasoning_content: '这是' },
+        { reasoning_content: '思考' },
+        { reasoning_content: '过程' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<t' },
+        { content: 'hink>' },
+        { content: '这是' },
+        { content: '思考' },
+        { content: '过程' },
+        { content: '</think>' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>' },
+        { content: '这是' },
+        { content: '思考' },
+        { content: '过程' },
+        { content: '</think>' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程' },
+        { content: '</think>' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程</' },
+        { content: 'think>' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程</think>' },
+        { content: '你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程</think>你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程</th' },
+        { content: '假的' },
+        { content: '你好2' },
+        { content: '你好3' },
+        { content: '过程</think>你好1' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '你好1你好2你好3', reasoning: '这是思考过程</th假的你好2你好3过程' }
+    },
+    {
+      data: [
+        { content: '<think>这是' },
+        { content: '思考' },
+        { content: '过程</th' },
+        { content: '假的' },
+        { content: '你好2' },
+        { content: '你好3' }
+      ],
+      correct: { answer: '', reasoning: '这是思考过程</th假的你好2你好3' }
+    }
+  ];
+
+  partList.forEach((part) => {
+    const { parsePart } = parseReasoningStreamContent();
+
+    let answer = '';
+    let reasoning = '';
+    part.data.forEach((item) => {
+      const formatPart = {
+        choices: [
+          {
+            delta: {
+              role: 'assistant',
+              content: item.content,
+              reasoning_content: item.reasoning_content
+            }
+          }
+        ]
+      };
+      const [reasoningContent, content] = parsePart(formatPart, true);
+      answer += content;
+      reasoning += reasoningContent;
+    });
+
+    expect(answer).toBe(part.correct.answer);
+    expect(reasoning).toBe(part.correct.reasoning);
+  });
+});

projects/app/src/web/common/api/fetch.ts

@@ -24,7 +24,11 @@ export type StreamResponseType = {
   [DispatchNodeResponseKeyEnum.nodeResponse]: ChatHistoryItemResType[];
 };
 type ResponseQueueItemType =
-  | { event: SseResponseEventEnum.fastAnswer | SseResponseEventEnum.answer; text: string }
+  | {
+      event: SseResponseEventEnum.fastAnswer | SseResponseEventEnum.answer;
+      text?: string;
+      reasoningText?: string;
+    }
   | { event: SseResponseEventEnum.interactive; [key: string]: any }
   | {
       event:
@@ -79,7 +83,7 @@ export const streamFetch = ({
       if (abortCtrl.signal.aborted) {
         responseQueue.forEach((item) => {
           onMessage(item);
-          if (isAnswerEvent(item.event)) {
+          if (isAnswerEvent(item.event) && item.text) {
             responseText += item.text;
           }
         });
@@ -91,7 +95,7 @@ export const streamFetch = ({
         for (let i = 0; i < fetchCount; i++) {
           const item = responseQueue[i];
           onMessage(item);
-          if (isAnswerEvent(item.event)) {
+          if (isAnswerEvent(item.event) && item.text) {
             responseText += item.text;
           }
         }
@@ -180,7 +184,7 @@ export const streamFetch = ({
           // console.log(parseJson, event);
           if (event === SseResponseEventEnum.answer) {
             const reasoningText = parseJson.choices?.[0]?.delta?.reasoning_content || '';
-            onMessage({
+            pushDataToQueue({
               event,
               reasoningText
             });
@@ -194,7 +198,7 @@ export const streamFetch = ({
             }
           } else if (event === SseResponseEventEnum.fastAnswer) {
             const reasoningText = parseJson.choices?.[0]?.delta?.reasoning_content || '';
-            onMessage({
+            pushDataToQueue({
               event,
               reasoningText
             });