perf: system tool support sse

Author: c121914yu

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

@@ -32,6 +32,10 @@ curl --location --request POST 'https://{{host}}/api/admin/initv4101' \
 
 - 给自动同步的知识库加入新的定时任务。
 
+## 🚀 新增内容
+
+1. 系统工具支持流输出。
+   
 ## ⚙️ 优化
 
 1. 定时任务报错日志记录到对话日志。

packages/global/common/middle/tracks/constants.ts

@@ -3,5 +3,6 @@ export enum TrackEnum {
   createApp = 'createApp',
   useAppTemplate = 'useAppTemplate',
   createDataset = 'createDataset',
-  appNodes = 'appNodes'
+  appNodes = 'appNodes',
+  runSystemTool = 'runSystemTool'
 }

packages/service/common/middle/tracks/utils.ts

@@ -10,7 +10,7 @@ import { type ShortUrlParams } from '@fastgpt/global/support/marketing/type';
 
 const createTrack = ({ event, data }: { event: TrackEnum; data: Record<string, any> }) => {
   if (!global.feConfigs?.isPlus) return;
-  addLog.info('Push tracks', {
+  addLog.debug('Push tracks', {
     event,
     ...data
   });
@@ -65,5 +65,13 @@ export const pushTrack = {
         }
       });
     } catch (error) {}
+  },
+  runSystemTool: (
+    data: PushTrackCommonType & { toolId: string; result: 1 | 0; usagePoint?: number; msg?: string }
+  ) => {
+    return createTrack({
+      event: TrackEnum.runSystemTool,
+      data
+    });
   }
 };

packages/service/core/app/tool/api.ts

@@ -1,12 +1,13 @@
-import createClient, { type SystemVarType } from '@fastgpt-sdk/plugin';
+import createClient, { RunToolWithStream } from '@fastgpt-sdk/plugin';
 import { PluginSourceEnum } from '@fastgpt/global/core/app/plugin/constants';
-import { runToolStream, type StreamDataAnswerTypeEnum } from '@fastgpt-sdk/plugin';
 
-const PluginClientConfig = {
-  baseUrl: process.env.PLUGIN_BASE_URL || '',
-  token: process.env.PLUGIN_TOKEN || ''
-};
-const client = createClient(PluginClientConfig);
+const BASE_URL = process.env.PLUGIN_BASE_URL || '';
+const TOKEN = process.env.PLUGIN_TOKEN || '';
+
+const client = createClient({
+  baseUrl: BASE_URL,
+  token: TOKEN
+});
 
 export async function getSystemToolList() {
   const res = await client.tool.list();
@@ -28,18 +29,8 @@ export async function getSystemToolList() {
   return Promise.reject(res.body);
 }
 
-export async function runPluginToolStream(
-  toolId: string,
-  inputs: Record<string, any>,
-  systemVar: SystemVarType,
-  onStreamData: (type: StreamDataAnswerTypeEnum, data: string) => void
-) {
-  return await runToolStream({
-    baseUrl: PluginClientConfig.baseUrl,
-    authtoken: PluginClientConfig.token,
-    toolId,
-    inputs,
-    systemVar,
-    onStreamData
-  });
-}
+const runToolInstance = new RunToolWithStream({
+  baseUrl: BASE_URL,
+  token: TOKEN
+});
+export const runSystemTool = runToolInstance.run.bind(runToolInstance);

packages/service/core/workflow/dispatch/plugin/runTool.ts

@@ -10,13 +10,13 @@ import { NodeInputKeyEnum } from '@fastgpt/global/core/workflow/constants';
 import { MCPClient } from '../../../app/mcp';
 import { getSecretValue } from '../../../../common/secret/utils';
 import type { McpToolDataType } from '@fastgpt/global/core/app/mcpTools/type';
-import { StreamDataAnswerTypeEnum } from '@fastgpt-sdk/plugin';
-import { runPluginToolStream } from '../../../app/tool/api';
+import { runSystemTool } from '../../../app/tool/api';
 import { MongoSystemPlugin } from '../../../app/plugin/systemPluginSchema';
 import { SystemToolInputTypeEnum } from '@fastgpt/global/core/app/systemTool/constants';
 import type { StoreSecretValueType } from '@fastgpt/global/common/secret/type';
-import { getSystemPluginById, splitCombinePluginId } from '../../../app/plugin/controller';
+import { getSystemPluginById } from '../../../app/plugin/controller';
 import { textAdaptGptResponse } from '@fastgpt/global/core/workflow/runtime/utils';
+import { pushTrack } from '../../../../common/middle/tracks/utils';
 
 type SystemInputConfigType = {
   type: SystemToolInputTypeEnum;
@@ -46,10 +46,12 @@ export const dispatchRunTool = async (props: RunToolProps): Promise<RunToolRespo
     node: { name, avatar, toolConfig, version }
   } = props;
 
+  const systemToolId = toolConfig?.systemTool?.toolId;
+
   try {
     // run system tool
-    if (toolConfig?.systemTool?.toolId) {
-      const tool = await getSystemPluginById(toolConfig.systemTool!.toolId);
+    if (systemToolId) {
+      const tool = await getSystemPluginById(systemToolId);
 
       const inputConfigParams = await (async () => {
         switch (params.system_input_config?.type) {
@@ -79,10 +81,10 @@ export const dispatchRunTool = async (props: RunToolProps): Promise<RunToolRespo
       const formatToolId = tool.id.split('-')[1];
 
       const result = await (async () => {
-        const res = await runPluginToolStream(
-          formatToolId,
+        const res = await runSystemTool({
+          toolId: formatToolId,
           inputs,
-          {
+          systemVar: {
             user: {
               id: variables.userId,
               teamId: runningUserInfo.teamId,
@@ -98,31 +100,26 @@ export const dispatchRunTool = async (props: RunToolProps): Promise<RunToolRespo
             },
             time: variables.cTime
           },
-          (type, data) => {
-            if (
-              workflowStreamResponse &&
-              data &&
-              (type === StreamDataAnswerTypeEnum.Answer ||
-                type === StreamDataAnswerTypeEnum.FastAnswer)
-            ) {
+          onMessage: ({ type, content }) => {
+            if (workflowStreamResponse && content) {
               workflowStreamResponse({
-                event: type as SseResponseEventEnum,
+                event: type as unknown as SseResponseEventEnum,
                 data: textAdaptGptResponse({
-                  text: data
+                  text: content
                 })
               });
             }
           }
-        );
-        if ('error' in res) {
+        });
+        if (res.error) {
           return Promise.reject(res.error);
         }
         if (!res.output) return {};
 
         return res.output;
       })();
 
-      const usagePoints = await (async () => {
+      const usagePoints = (() => {
         if (
           params.system_input_config?.type !== SystemToolInputTypeEnum.system ||
           result[NodeOutputKeyEnum.systemError]
@@ -132,6 +129,16 @@ export const dispatchRunTool = async (props: RunToolProps): Promise<RunToolRespo
         return tool.currentCost ?? 0;
       })();
 
+      pushTrack.runSystemTool({
+        teamId: runningUserInfo.teamId,
+        tmbId: runningUserInfo.tmbId,
+        uid: runningUserInfo.tmbId,
+        toolId: tool.id,
+        result: 1,
+        usagePoint: usagePoints,
+        msg: result[NodeOutputKeyEnum.systemError]
+      });
+
       return {
         [DispatchNodeResponseKeyEnum.nodeResponse]: {
           toolRes: result,
@@ -170,6 +177,17 @@ export const dispatchRunTool = async (props: RunToolProps): Promise<RunToolRespo
       };
     }
   } catch (error) {
+    if (systemToolId) {
+      pushTrack.runSystemTool({
+        teamId: runningUserInfo.teamId,
+        tmbId: runningUserInfo.tmbId,
+        uid: runningUserInfo.tmbId,
+        toolId: systemToolId,
+        result: 0,
+        msg: getErrText(error)
+      });
+    }
+
     return {
       [DispatchNodeResponseKeyEnum.nodeResponse]: {
         moduleLogo: avatar,

packages/service/package.json

@@ -3,7 +3,7 @@
   "version": "1.0.0",
   "type": "module",
   "dependencies": {
-    "@fastgpt-sdk/plugin": "^0.1.0",
+    "@fastgpt-sdk/plugin": "^0.1.1",
     "@fastgpt/global": "workspace:*",
     "@modelcontextprotocol/sdk": "^1.12.1",
     "@node-rs/jieba": "2.0.1",