compat vllm>=0.8 (#3574)

Author: Jintao-Huang

docs/source/Customization/自定义数据集.md

@@ -161,7 +161,7 @@ Agent格式请参考[Agent文档](../Instruction/智能体的支持.md)。
     "ms_dataset_id": "xxx/xxx"
   },
   {
-    "dataset_path": "<dataset_path>"
+    "dataset_path": "<dataset_dir/dataset_path>"
   },
   {
     "ms_dataset_id": "<dataset_id>",

docs/source_en/Customization/Custom-dataset.md

@@ -169,7 +169,7 @@ You can refer to the ms-swift built-in [dataset_info.json](https://github.com/mo
     "ms_dataset_id": "xxx/xxx"
   },
   {
-    "dataset_path": "<dataset_path>"
+    "dataset_path": "<dataset_dir/dataset_path>"
   },
   {
     "ms_dataset_id": "<dataset_id>",

swift/llm/dataset/loader.py

@@ -93,7 +93,7 @@ def get_dataset_meta(self, use_hf: bool):
         if dataset_type == 'path':
             dataset_meta = dataset_meta_mapping.get((dataset_type, self.dataset.lower()))
         else:
-            dataset_type = {True: 'hf', False: 'ms'}[use_hf]
+            dataset_type = 'repo' if os.path.isdir(self.dataset) else {True: 'hf', False: 'ms'}[use_hf]
             dataset_meta = dataset_meta_mapping.get((dataset_type, self.dataset.lower()))
         return dataset_meta or self._get_matched_dataset_meta(dataset_meta_mapping) or DatasetMeta()
 
@@ -105,7 +105,8 @@ def _get_dataset_meta_mapping() -> Dict[Tuple[str, str], DatasetMeta]:
         _dataset_meta_mapping = {}
         for dataset_meta in DATASET_MAPPING.values():
             if dataset_meta.dataset_path is not None:
-                _dataset_meta_mapping[('path', dataset_meta.dataset_path.lower())] = dataset_meta
+                dataset_type = 'repo' if os.path.isdir(dataset_meta.dataset_path) else 'path'
+                _dataset_meta_mapping[(dataset_type, dataset_meta.dataset_path.lower())] = dataset_meta
             if dataset_meta.ms_dataset_id is not None:
                 _dataset_meta_mapping[('ms', dataset_meta.ms_dataset_id.lower())] = dataset_meta
             if dataset_meta.hf_dataset_id is not None:

swift/llm/infer/deploy.py

@@ -57,8 +57,7 @@ def _compute_infer_stats(self):
     def lifespan(self, app: FastAPI):
         args = self.args
         if args.log_interval > 0:
-            thread = Thread(target=lambda: asyncio.run(self._log_stats_hook()))
-            thread.daemon = True
+            thread = Thread(target=lambda: asyncio.run(self._log_stats_hook()), daemon=True)
             thread.start()
         try:
             yield

swift/llm/infer/infer_engine/pt_engine.py

@@ -92,8 +92,7 @@ def _post_init(self):
 
     def _start_infer_worker(self):
         if self._task_thread is None:
-            self._task_thread = Thread(target=self._infer_worker)
-            self._task_thread.daemon = True
+            self._task_thread = Thread(target=self._infer_worker, daemon=True)
             self._task_thread.start()
 
     def _fetch_infer_requests(self):

swift/llm/infer/infer_engine/vllm_engine.py

@@ -379,7 +379,13 @@ async def _infer_full_async(
         return self._create_chat_completion_response(result, template, generation_config, request_id)
 
     def _batch_infer_stream(self, *args, **kwargs):
-        self.engine.engine.model_executor.parallel_worker_tasks = None
+        if hasattr(self.engine, 'engine'):
+            self.engine.engine.model_executor.parallel_worker_tasks = None
+        elif hasattr(self.engine, 'engine_core'):
+            # vllm>=0.8
+            self.engine.engine_core.outputs_queue = None
+            self.engine.engine_core.queue_task = None
+            self.engine.output_handler = None
         return super()._batch_infer_stream(*args, **kwargs)
 
     def infer(

swift/utils/io_utils.py

@@ -76,8 +76,7 @@ def _append(self, obj: Union[Dict, List[Dict]], gather_obj: bool = False):
     def append(self, obj: Union[Dict, List[Dict]], gather_obj: bool = False):
         if self.enable_async:
             if self._thread is None:
-                self._thread = Thread(target=self._append_worker)
-                self._thread.daemon = True
+                self._thread = Thread(target=self._append_worker, daemon=True)
                 self._thread.start()
             self._queue.put({'obj': obj, 'gather_obj': gather_obj})
         else: