feat: API工具支持文件上传

src/backend/bisheng/linsight/worker.py

@@ -80,7 +80,7 @@ async def qsize(self):
     async def put(self, data, timeout=None):
         await self.__db.arpush(self.key, data, expiration=timeout)  # Add a new element to the far right of the queue
 
-    async def get_wait(self, timeout=None):
+    async def get_wait(self, timeout=1):
         # Returns the first element of the queue, if empty, wait until an element is queued (the timeout threshold istimeout, if isNonehas been waiting)
         item = await self.__db.ablpop(self.key, timeout=timeout)
         return item
@@ -155,7 +155,7 @@ async def async_run(self):
             try:
                 session_version_id = await self.queue.get_wait()
                 if session_version_id is None:
-                    logger.info("No session_version_id found in queue, waiting...")
+                    logger.debug("No session_version_id found in queue, waiting...")
                     self.semaphore.release()
                     continue
 

src/backend/bisheng/tool/domain/services/openapi.py

@@ -70,7 +70,7 @@ def parse_paths(self) -> list[dict]:
                     continue
 
                 if 'requestBody' in method_info:
-                    for _, content in method_info['requestBody']['content'].items():
+                    for content_type, content in method_info['requestBody']['content'].items():
                         if '$ref' in content['schema']:
                             schema_ref = content['schema']['$ref']
                             schema_name = schema_ref.split('/')[-1]
@@ -80,16 +80,22 @@ def parse_paths(self) -> list[dict]:
 
                         if 'properties' in schema:
                             for param_name, param_info in schema['properties'].items():
+                                param_schema = {
+                                    'type': param_info.get('type', 'string'),
+                                    'title': param_info.get('title', param_name),
+                                    'properties': param_info.get('properties', {})
+                                }
+                                if param_info.get('format'):
+                                    param_schema['format'] = param_info.get('format')
+                                if param_info.get('items'):
+                                    param_schema['items'] = param_info.get('items')
                                 param = {
                                     'name': param_name,
                                     'description': param_info.get('description', ''),
-                                    'in': 'body',
+                                    'in': 'formData' if content_type == 'multipart/form-data' else 'body',
+                                    'content_type': content_type,
                                     'required': param_name in schema.get('required', []),
-                                    'schema': {
-                                        'type': param_info.get('type', 'string'),
-                                        'title': param_info.get('title', param_name),
-                                        'properties': param_info.get('properties', {})
-                                    },
+                                    'schema': param_schema,
                                 }
                                 one_api_info['parameters'].append(param)
                 else:

src/backend/bisheng/workflow/nodes/input/input.py

@@ -66,6 +66,8 @@ def __init__(self, *args, **kwargs):
 
         self.node_params = new_node_params
         self._image_ext = ['png', 'jpg', 'jpeg', 'bmp']
+        self._audio_ext = ['mp3', 'wav', 'm4a', 'aac', 'ogg', 'flac']
+        self._video_ext = ['mp4', 'avi', 'mov', 'wmv', 'flv', 'mkv']
 
         self._vector_client = None
         self._es_client = None
@@ -163,6 +165,10 @@ def _parse_upload_file_variables(self, key_info: Dict, key_value: Dict) -> Dict:
         if file_parse_mode == ParseModeEnum.KEEP_RAW:
             if key_info.get("file_type") in ["image", "all"]:
                 ret[key_info['image_file']] = key_value.get(key_info['image_file'], [])
+            if key_info.get("file_type") in ["audio", "all"]:
+                ret['audio_files'] = key_value.get('audio_files', [])
+            if key_info.get("file_type") in ["video", "all"]:
+                ret['video_files'] = key_value.get('video_files', [])
             ret[key_info['file_path']] = key_value.get(key_info['file_path'], [])
         elif file_parse_mode == ParseModeEnum.EXTRACT_TEXT:
             ret[key_info['file_content']] = key_value.get(key_info['file_content'], "")
@@ -282,13 +288,17 @@ def parse_upload_file(self, key: str, key_info: dict, value: List[str]) -> dict
         all_file_content = ''
         original_file_path = []
         image_files_path = []
+        audio_files_path = []
+        video_files_path = []
         if not value:
             logger.warning(f"{self.id}.{key} value is None")
             return {
                 key_info['key']: all_metadata,
                 key_info['file_content']: all_file_content,
                 key_info['file_path']: original_file_path,
-                key_info['image_file']: image_files_path
+                key_info['image_file']: image_files_path,
+                'audio_files': audio_files_path,
+                'video_files': video_files_path,
             }
 
         file_parse_mode = key_info.get('file_parse_mode', ParseModeEnum.INGEST_TO_KNOWLEDGE_BASE)
@@ -319,6 +329,10 @@ def parse_upload_file(self, key: str, key_info: dict, value: List[str]) -> dict
             original_file_path.append(one_file_url)
             if file_ext in self._image_ext:
                 image_files_path.append(one_file_url)
+            elif file_ext in self._audio_ext:
+                audio_files_path.append(one_file_url)
+            elif file_ext in self._video_ext:
+                video_files_path.append(one_file_url)
 
             if file_parse_mode == ParseModeEnum.KEEP_RAW:
                 continue
@@ -359,5 +373,7 @@ def parse_upload_file(self, key: str, key_info: dict, value: List[str]) -> dict
             key_info['key']: all_metadata,
             key_info['file_content']: all_file_content,
             key_info['file_path']: original_file_path,
-            key_info['image_file']: image_files_path
+            key_info['image_file']: image_files_path,
+            'audio_files': audio_files_path,
+            'video_files': video_files_path,
         }

src/backend/bisheng/workflow/nodes/tool/tool.py

@@ -61,7 +61,10 @@ def parse_tool_input(self) -> dict:
 
         return ret
 
-    def parse_template_msg(self, msg: str):
+    def parse_template_msg(self, msg: Any):
+        if not isinstance(msg, str):
+            return msg
+
         msg_template = PromptTemplateParser(template=msg)
         variables = msg_template.extract()
         if len(variables) > 0:

src/backend/bisheng_langchain/gpts/tools/api_tools/openapi.py

@@ -1,5 +1,8 @@
+import base64
+from io import BytesIO
 from typing import Any, Optional
 
+from aiohttp import FormData
 from langchain_core.tools import BaseTool
 from loguru import logger
 
@@ -21,6 +24,25 @@ def get_real_path(self, path_params: dict | None):
     def get_request_method(self):
         return self.params['method'].lower()
 
+    @staticmethod
+    def is_file_param(param_info: dict) -> bool:
+        schema = param_info.get('schema', {})
+        return param_info.get('content_type') == 'multipart/form-data' and schema.get('format') == 'binary'
+
+    @staticmethod
+    def build_file_tuple(value: Any):
+        if isinstance(value, dict):
+            filename = value.get('name') or value.get('filename') or 'upload.bin'
+            content_type = value.get('content_type') or value.get('type') or 'application/octet-stream'
+            content = value.get('content', '')
+            if value.get('encoding') == 'base64':
+                content_bytes = base64.b64decode(content)
+            else:
+                content_bytes = str(content).encode()
+            return filename, BytesIO(content_bytes), content_type
+
+        return 'upload.txt', BytesIO(str(value).encode()), 'text/plain'
+
     def get_params_json(self, **kwargs):
         params_define = {}
         for one in self.params['parameters']:
@@ -29,13 +51,19 @@ def get_params_json(self, **kwargs):
         path_params = {}
         params = {}
         json_data = {}
+        form_data = {}
+        files = {}
         for k, v in kwargs.items():
             if params_define.get(k):
                 if params_define[k]['in'] == 'query':
                     field_type = params_define[k]['schema']['type']
                     params[k] = convert_openapi_field_value(v, field_type)
                 elif params_define[k]['in'] == 'path':
                     path_params[k] = v
+                elif self.is_file_param(params_define[k]):
+                    files[k] = self.build_file_tuple(v)
+                elif params_define[k].get('content_type') == 'multipart/form-data':
+                    form_data[k] = v
                 else:
                     field_type = params_define[k]['schema']['type']
                     json_data[k] = convert_openapi_field_value(v, field_type)
@@ -54,7 +82,24 @@ def get_params_json(self, **kwargs):
                                      None) or self.params.get('parameter_name')
             if parameter_name:
                 params.update({parameter_name: self.api_key})
-        return params, json_data, path_params
+        return params, json_data, path_params, form_data, files
+
+    def get_body_kwargs(self, json_data: dict, form_data: dict, files: dict) -> dict:
+        if files:
+            return {'data': form_data, 'files': files}
+        return {'json': json_data}
+
+    def get_async_body_kwargs(self, json_data: dict, form_data: dict, files: dict) -> dict:
+        if not files:
+            return {'json': json_data}
+
+        payload = FormData()
+        for key, value in form_data.items():
+            payload.add_field(key, str(value))
+        for key, file_info in files.items():
+            filename, file_data, content_type = file_info
+            payload.add_field(key, file_data, filename=filename, content_type=content_type)
+        return {'data': payload}
 
     def parse_args_schema(self):
         args_schema = {
@@ -114,19 +159,20 @@ def run(self, **kwargs) -> str:
         if 'proxy' in kwargs:
             extra['proxy'] = kwargs.pop('proxy')
 
-        params, json_data, path_params = self.get_params_json(**kwargs)
+        params, json_data, path_params, form_data, files = self.get_params_json(**kwargs)
         path = self.get_real_path(path_params)
         logger.info('api_call url={}', path)
         method = self.get_request_method()
+        body_kwargs = self.get_body_kwargs(json_data, form_data, files)
 
         if method == 'get':
             resp = self.client.get(path, params=params, **extra)
         elif method == 'post':
-            resp = self.client.post(path, params=params, json=json_data, **extra)
+            resp = self.client.post(path, params=params, **body_kwargs, **extra)
         elif method == 'put':
-            resp = self.client.put(path, params=params, json=json_data, **extra)
+            resp = self.client.put(path, params=params, **body_kwargs, **extra)
         elif method == 'delete':
-            resp = self.client.delete(path, params=params, json=json_data, **extra)
+            resp = self.client.delete(path, params=params, **body_kwargs, **extra)
         else:
             raise Exception(f'http method is not support: {method}')
         if resp.status_code != 200:
@@ -140,19 +186,20 @@ async def arun(self, **kwargs) -> str:
         if 'proxy' in kwargs:
             extra['proxy'] = kwargs.pop('proxy')
 
-        params, json_data, path_params = self.get_params_json(**kwargs)
+        params, json_data, path_params, form_data, files = self.get_params_json(**kwargs)
         path = self.get_real_path(path_params)
         logger.info('api_call url={} params={}', path, params)
         method = self.get_request_method()
+        body_kwargs = self.get_async_body_kwargs(json_data, form_data, files)
 
         if method == 'get':
             resp = await self.async_client.aget(path, params=params, **extra)
         elif method == 'post':
-            resp = await self.async_client.apost(path, params=params, json=json_data, **extra)
+            resp = await self.async_client.apost(path, params=params, **body_kwargs, **extra)
         elif method == 'put':
-            resp = await self.async_client.aput(path, params=params, json=json_data, **extra)
+            resp = await self.async_client.aput(path, params=params, **body_kwargs, **extra)
         elif method == 'delete':
-            resp = await self.async_client.adelete(path, params=params, json=json_data, **extra)
+            resp = await self.async_client.adelete(path, params=params, **body_kwargs, **extra)
         else:
             raise Exception(f'http method is not support: {method}')
         return resp

src/backend/bisheng_langchain/utils/requests.py

@@ -31,7 +31,7 @@ def get(self, url: str, **kwargs: Any) -> requests.Response:
                             proxies=None if not self.proxy else {'http': self.proxy, 'https': self.proxy},
                             **kwargs)
 
-    def post(self, url: str, json: Dict[str, Any], **kwargs: Any) -> requests.Response:
+    def post(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> requests.Response:
         """POST to the URL and return the text."""
         return requests.post(url,
                              json=json,
@@ -41,7 +41,7 @@ def post(self, url: str, json: Dict[str, Any], **kwargs: Any) -> requests.Respon
                              proxies=None if not self.proxy else {'http': self.proxy, 'https': self.proxy},
                              **kwargs)
 
-    def patch(self, url: str, json: Dict[str, Any], **kwargs: Any) -> requests.Response:
+    def patch(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> requests.Response:
         """PATCH the URL and return the text."""
         return requests.patch(url,
                               json=json,
@@ -51,7 +51,7 @@ def patch(self, url: str, json: Dict[str, Any], **kwargs: Any) -> requests.Respo
                               proxies=None if not self.proxy else {'http': self.proxy, 'https': self.proxy},
                               **kwargs)
 
-    def put(self, url: str, json: Dict[str, Any], **kwargs: Any) -> requests.Response:
+    def put(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> requests.Response:
         """PUT the URL and return the text."""
         return requests.put(url,
                             json=json,
@@ -103,21 +103,21 @@ async def aget(self, url: str, **kwargs: Any) -> AsyncGenerator[aiohttp.ClientRe
             yield response
 
     @asynccontextmanager
-    async def apost(self, url: str, json: Dict[str, Any],
+    async def apost(self, url: str, json: Optional[Dict[str, Any]] = None,
                     **kwargs: Any) -> AsyncGenerator[aiohttp.ClientResponse, None]:
         """POST to the URL and return the text asynchronously."""
         async with self._arequest('POST', url, json=json, auth=self.auth, **kwargs) as response:
             yield response
 
     @asynccontextmanager
-    async def apatch(self, url: str, json: Dict[str, Any],
+    async def apatch(self, url: str, json: Optional[Dict[str, Any]] = None,
                      **kwargs: Any) -> AsyncGenerator[aiohttp.ClientResponse, None]:
         """PATCH the URL and return the text asynchronously."""
         async with self._arequest('PATCH', url, json=json, auth=self.auth, **kwargs) as response:
             yield response
 
     @asynccontextmanager
-    async def aput(self, url: str, json: Dict[str, Any],
+    async def aput(self, url: str, json: Optional[Dict[str, Any]] = None,
                    **kwargs: Any) -> AsyncGenerator[aiohttp.ClientResponse, None]:
         """PUT the URL and return the text asynchronously."""
         async with self._arequest('PUT', url, json=json, auth=self.auth, **kwargs) as response:
@@ -156,15 +156,15 @@ def get(self, url: str, **kwargs: Any) -> str:
         """GET the URL and return the text."""
         return self.requests.get(url, **kwargs).text
 
-    def post(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    def post(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """POST to the URL and return the text."""
         return self.requests.post(url, json, **kwargs).text
 
-    def patch(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    def patch(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """PATCH the URL and return the text."""
         return self.requests.patch(url, json, **kwargs).text
 
-    def put(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    def put(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """PUT the URL and return the text."""
         return self.requests.put(url, json, **kwargs).text
 
@@ -177,17 +177,17 @@ async def aget(self, url: str, **kwargs: Any) -> str:
         async with self.requests.aget(url, **kwargs) as response:
             return await response.text()
 
-    async def apost(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    async def apost(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """POST to the URL and return the text asynchronously."""
         async with self.requests.apost(url, json, **kwargs) as response:
             return await response.text()
 
-    async def apatch(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    async def apatch(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """PATCH the URL and return the text asynchronously."""
         async with self.requests.apatch(url, json, **kwargs) as response:
             return await response.text()
 
-    async def aput(self, url: str, json: Dict[str, Any], **kwargs: Any) -> str:
+    async def aput(self, url: str, json: Optional[Dict[str, Any]] = None, **kwargs: Any) -> str:
         """PUT the URL and return the text asynchronously."""
         async with self.requests.aput(url, json, **kwargs) as response:
             return await response.text()

src/frontend/platform/src/pages/BuildPage/flow/FlowChat/ChatInput.tsx

@@ -22,9 +22,11 @@ import { useFileDropAndPaste } from "./useFileDropAndPaste";
 const GuideQuestionsAny = GuideQuestions as any;
 
 export const FileTypes = {
-    ALL: ['.PNG', '.JPEG', '.JPG', '.BMP', '.PDF', '.TXT', '.MD', '.HTML', '.XLS', '.XLSX', '.CSV', '.DOC', '.DOCX', '.PPT', '.PPTX'],
+    ALL: ['.PNG', '.JPEG', '.JPG', '.BMP', '.PDF', '.TXT', '.MD', '.HTML', '.XLS', '.XLSX', '.CSV', '.DOC', '.DOCX', '.PPT', '.PPTX', '.MP3', '.WAV', '.M4A', '.AAC', '.OGG', '.FLAC', '.MP4', '.AVI', '.MOV', '.WMV', '.FLV', '.MKV'],
     IMAGE: ['.PNG', '.JPEG', '.JPG', '.BMP'],
     FILE: ['.PDF', '.TXT', '.MD', '.HTML', '.XLS', '.XLSX', '.CSV', '.DOC', '.DOCX', '.PPT', '.PPTX'],
+    AUDIO: ['.MP3', '.WAV', '.M4A', '.AAC', '.OGG', '.FLAC'],
+    VIDEO: ['.MP4', '.AVI', '.MOV', '.WMV', '.FLV', '.MKV'],
 }
 
 export default function ChatInput({ autoRun, version, clear, form, wsUrl, onBeforSend, onLoad }) {

src/frontend/platform/src/pages/BuildPage/flow/FlowNode/Parameter.tsx

@@ -1,4 +1,5 @@
 import { WorkflowNode, WorkflowNodeParam } from "@/types/flow";
+import ApiFileUploadItem from "./component/ApiFileUploadItem";
 import CodeInputItem from "./component/CodeInputItem";
 import CodeOutputItem from "./component/CodeOutputItem";
 import CodePythonItem from "./component/CodePythonItem";
@@ -113,6 +114,13 @@ export default function Parameter({
                 onVarEvent={bindVarValidate}
                 i18nPrefix={i18nPrefix}
             />
+        case 'api_file_upload':
+            return <ApiFileUploadItem
+                node={node}
+                data={item}
+                onChange={handleOnNewValue}
+                onValidate={bindValidate}
+            />
         case 'output_form':
             return <OutputItem
                 nodeId={nodeId}