Merge remote-tracking branch 'origin/main' into wzj_pd

Author: hiworldwzj

Dockerfile

@@ -41,7 +41,7 @@ RUN pip install -r /lightllm/requirements.txt --no-cache-dir --ignore-installed
 RUN pip install --no-cache-dir nvidia-nccl-cu12==2.25.1  # for allreduce hang issues in multinode H100
 
 RUN git clone https://github.com/Dao-AILab/flash-attention.git -b v2.7.4.post1
-RUN cd flash-attention/hopper && NVCC_THREADS=16 python setup.py install
+RUN cd flash-attention/hopper && MAX_JOBS=4 NVCC_THREADS=16 python setup.py install
 
 COPY . /lightllm
 RUN pip install -e /lightllm --no-cache-dir

lightllm/models/deepseek2/model.py

@@ -74,6 +74,7 @@ def _init_some_value(self):
 
     def _init_custom(self):
         self._init_to_get_yarn_rotary()
+        dist_group_manager.new_deepep_group(self.config["n_routed_experts"])
 
     def _verify_params(self):
         return super()._verify_params()

lightllm/server/api_http.py

@@ -368,7 +368,7 @@ async def tokens(request: Request):
 
         multimodal_params_dict = request_dict.get("multimodal_params", {})
         multimodal_params = MultimodalParams(**multimodal_params_dict)
-        await multimodal_params.verify_and_preload()
+        await multimodal_params.verify_and_preload(request)
         return JSONResponse(
             {
                 "ntokens": g_objs.httpserver_manager.tokens(

lightllm/server/httpserver/manager.py

@@ -228,7 +228,7 @@ async def generate(
                 original_multimodal_params = copy.deepcopy(multimodal_params)
 
             if self.pd_mode.is_P_or_NORMAL():
-                await multimodal_params.verify_and_preload()
+                await multimodal_params.verify_and_preload(request)
 
             # 记录请求到达的相关信息
             await self._log_req_header(request_headers, group_request_id)

lightllm/server/multimodal_params.py

@@ -6,6 +6,7 @@
 from PIL import Image
 from lightllm.utils.image_utils import fetch_image
 import base64
+from fastapi import Request
 
 
 class ImageItem:
@@ -24,11 +25,12 @@ def __init__(self, **kwargs):
         self._preload_data = None
         self.extra_params = {}
 
-    async def preload(self):
+    async def preload(self, request: Request):
         try:
             if self._type == "url":
                 timeout = int(os.getenv("REQUEST_TIMEOUT", "5"))
-                img_data = await fetch_image(self._data, timeout=timeout)
+                proxy = os.getenv("REQUEST_PROXY", None)
+                img_data = await fetch_image(self._data, request, timeout=timeout, proxy=proxy)
             elif self._type == "base64":
                 img_data = base64.b64decode(self._data)
             elif self._type == "image_size":
@@ -81,9 +83,9 @@ def __init__(
         self.images = [ImageItem(**i) for i in images]
         return
 
-    async def verify_and_preload(self):
+    async def verify_and_preload(self, request: Request):
         for image in self.images:
-            await image.preload()
+            await image.preload(request)
         return
 
     def to_dict(self):

lightllm/server/router/model_infer/mode_backend/base_backend.py

@@ -192,7 +192,6 @@ def init_model(self, kvargs):
             elif self.model_type == "phi3":
                 self.model = Phi3TpPartModel(model_kvargs)
             elif self.model_type in ["deepseek_v2", "deepseek_v3"]:
-                dist_group_manager.new_deepep_group(model_cfg["n_routed_experts"])
                 self.model = Deepseek2TpPartModel(model_kvargs)
             elif self.model_type == "internvl_chat":
                 llm_model_type = model_cfg.get("llm_config").get("model_type")

lightllm/server/router/model_infer/mode_backend/dp_backend/impl.py

@@ -31,7 +31,7 @@ def init_custom(self):
         # nan 值，避免后续构建的fake请求在计算的过程中出现计算错误。
         from .pre_process import padded_prepare_prefill_inputs
 
-        kwargs, run_reqs, padded_req_num = padded_prepare_prefill_inputs([], 1, is_multimodal=False)
+        kwargs, run_reqs, padded_req_num = padded_prepare_prefill_inputs([], 1, is_multimodal=self.is_multimodal)
         self.model.forward(**kwargs)
         assert len(run_reqs) == 0 and padded_req_num == 1
         return

lightllm/utils/image_utils.py

@@ -1,7 +1,13 @@
+import time
 import base64
 import httpx
+import logging
 from PIL import Image
 from io import BytesIO
+from fastapi import Request
+from lightllm.utils.log_utils import init_logger
+
+logger = init_logger(__name__)
 
 
 def image2base64(img_str: str):
@@ -13,17 +19,24 @@ def image2base64(img_str: str):
     return base64.b64encode(buffer.getvalue()).decode("utf-8")
 
 
-async def fetch_image(url, timeout):
-    async with httpx.AsyncClient() as client:
+async def fetch_image(url, request: Request, timeout, proxy=None):
+    logger.info(f"Begin to download image from url: {url}")
+    start_time = time.time()
+    async with httpx.AsyncClient(proxy=proxy) as client:
         async with client.stream("GET", url, timeout=timeout) as response:
             response.raise_for_status()
             ans_bytes = []
-
             async for chunk in response.aiter_bytes(chunk_size=1024 * 1024):
+                if request is not None and await request.is_disconnected():
+                    await response.aclose()
+                    raise Exception("Request disconnected. User cancelled download.")
                 ans_bytes.append(chunk)
                 # 接收的数据不能大于128M
                 if len(ans_bytes) > 128:
-                    raise Exception("image data is too big")
+                    raise Exception(f"url {url} Image data is too big")
 
             content = b"".join(ans_bytes)
-            return content
+    end_time = time.time()
+    cost_time = end_time - start_time
+    logger.info(f"Download url {url} image cost time: {cost_time} seconds")
+    return content