一、开发目标
使用离线模型进行应用开发的时候，绕不开的话题就是线程并发的问题。最基础最简单的场景：如何让模型在推理过程中能够实现并行推理的同时，降低单张显卡内存的占用量？
最直接的方案就是：数据并行+张量并行。当然还有专家并行，但本文章只聊数据并行和张量并行。
数据并行：通俗来讲，就是将大模型的权重参数分发到多张显卡上，在推理时针对一条Prompt同时进行推理，然后进行输出合并，适用于模型规模不大的场景，能够提高应用吞吐率。
张量并行：将单个模型中的张量运算分布到多张显卡上同时执行。适用于模型规模大的场景，不能提高吞吐率，但能节省显存。
综合运用上述两种并行策略，可实现多显卡并行推理的同时大大节省单张显卡显存。
二、环境准备
SGLang只能运行在Linux系统下，本人使用Winows11作为开发环境的操作系统，所以必须安装WSL2，WSL2的安装方法可百度一下，WSL2中安装Ubuntu-22.04.5，该版本的Ubuntu系统自带Python3.10。
安装好基础环境之后，创建Python运行虚拟环境
conda create -y -p D:\MyChat\ENV python=3.10conda指令需要安装Anaconda，具体安装过程可以参照我之前的文章。
在虚拟环境中安装torch，注：SGLang最新版本目前使用的是torch2.6，我使用的SGLang版本是0.4.3.post2，所以我的torch是2.5.1版本，对应的cuda是12.4。
pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124最后就是安装SGLang包（需要翻墙，安装过程很慢），在安装SGLang包的过程中有几个包特别大，通过命令安装一直不成功，所以我提前手动下载了那几个特别大的包，手动安装完成后，再通过命令安装的SGLang
pip install --upgrade pip #flashinfer_python-0.2.2+cu124torch2.5-cp38-abi3-linux_x86_64.whl就是比较大的包之一 #手动下载后存放在root文件夹中,pip命令执行时根据命令指向文件夹，请自行调整whl文件位置 pip install ../whl/flashinfer_python-0.2.2+cu124torch2.5-cp38-abi3-linux_x86_64.whl #vllm-0.7.2-cp38-abi3-manylinux1_x86_64.whl就是比较大的包之一 #手动下载后存放在root文件夹中,pip命令执行时根据命令指向文件夹，请自行调整whl文件位置 pip install ../whl/vllm-0.7.2-cp38-abi3-manylinux1_x86_64.whl pip install sglang==0.4.3.post2 #为了后续对外提供模型推理接口，安装的web框架 uvicorn==0.34.0 fastapi==0.115.11 #为了能够开启异步线程而安装 sanic==24.12.0注：在安装这些环境的过程中，各个包的版本特别重要，需要仔细，否则会有很多坑
三、SGLang关键参数说明
SGLang框架的使用方式有很多种场景
一、直接调用huggingface模型，通过控制台命令直接启动（带参数）模型。
二、另一种就是离线模型，将模型下载到本地，通过SGLang框架加载模型，完成推理。
对于开发而言，我选择的是第二种
# 张量并行参数，表示模型计算过程中使用多少张显卡（每张显卡只承载部分模型，适合体量大的模型） tp_size:int=1 # 数据并行，表示模型被分发到多少张显卡上进行并行推理（每张显卡都要承载完整模型，适合体量小的模型） dp_size:int=1 # 用于设置 KV Cache 内存池 的最大内存占用比例,如果设置0.9则表示在计算过程中KV Cache 最多可以使用 90% 的 GPU 内存，过高容易导致内存溢出，过低会影响计算性能 # 通常在并行推理中不宜设置过高，需要为每一条推理预留出内存空间 mem_fraction_static:float=0.7 # 用于将预填充阶段的输入 prompt 分块（chunk） 处理,通过分块处理，减少预填充阶段的内存占用和计算量，从而加速推理过程 # 如果 chunked_prefill_size=128，模型会将输入 prompt 分成每块 128 个 token 进行处理,，特别适合于长prompt，设置过大会导致性能降低，设置过小会导致内存不足 chunked_prefill_size:int=None # 是否启用Torch加速，对FP8模型不生效 enable_torch_compile:bool=False # 如要使用lora模型，则提供lora模型的模型保存路径，如果 lora_paths=["adapter1", "adapter2"]，则批次中的第一个元素会使用 adapter1，第二个元素会使用 adapter2 lora_paths: Optional[List[str]] = None # 每个推理批次中最多使用多少个lora模型 max_loras_per_batch: int = 8 # 是否禁止使用cuda_graph，如何使用Lora则必须将此参数设置为True disable_cuda_graph:bool=False # 是否禁止使用radix_cache，如何使用Lora则必须将此参数设置为True disable_radix_cache:bool=False在我的开发中只简单使用了上述部分参数，SGLang提供的参数非常多，具体的可参见官方文档
四、源码
在文章中展示的源码部分主要解决一个问题：在实现批量推理的基础上，以异步流的形式，向多个客户端返回模型推理结果
所以源码结构主要分为：接口、处理类、模型加载、公共帮助类，四个文件
接口
import asyncio from typing import List from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect import uvicorn from fastapi.responses import StreamingResponse import sys import os sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../'))) from Core.Chat_Wrapper import Chat_Wrapper from Init.Model_Init import Model_Init from Core.Batch_Queue_Handle import Batch_Queue_Handle from Core.Common import valid_prompt Api=FastAPI() # 单次流式生成 @Api.post("/stream/chat") async def Stream_Async_Chat(prompt:List[dict]): message= valid_prompt(prompt) if message is not None: return message wrapper=Chat_Wrapper() return StreamingResponse(wrapper.Async_Stream_Generation(prompt=prompt), media_type="text/plain") # 批次流式生成生成（获取生成内容） @Api.websocket("/ws/chat/") async def websocket_endpoint(websocket: WebSocket): """ WebSocket 接口，支持流式返回 Token，提高响应速度。 """ await websocket.accept() client_id = id(websocket) # 以 WebSocket 对象 ID 作为唯一标识 Batch_Queue_Handle.instance().clients[client_id] = websocket try: while True: data = await websocket.receive_text() await Batch_Queue_Handle.instance().queue.put({"client_id": client_id, "prompt": data}) # 任务加入队列 except WebSocketDisconnect: del Batch_Queue_Handle.instance().clients[client_id] # 连接断开时删除 @Api.on_event("startup") async def lifespan(): instance=Batch_Queue_Handle.instance() asyncio.create_task(instance.batch_process()) # 启动后台任务 if __name__ == "__main__": # 初始化加载模型 Model_Init.instance() uvicorn.run(Api, host="172.26.115.148", port=8080)接口中包含两个方法，一个是以SSE方式进行异步流式输出的接口、一个是以WebSocket方式异步批量推理后流式输出的接口。从线程角度看，两个接口都能实现并发推理，只是当线程请求真正调用模型进行推理时，推理的方式不一样。
非批量消息处理类
from typing import List, Optional, Union from sglang.utils import trim_overlap import sys import os sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../'))) from Init.Model_Init import Model_Init from Core.Common import build_prompt_from_history class Chat_Wrapper(): # 停止词 stop:Optional[Union[str, List[str]]] = None # 采样温度，控制生成文本的多样性，温度越高多样性越强 temperature:float=1.0 # token概率选择，参数值越小，token可选择范围越小（eg.0.1 只会选择推理概率超过90%的token）参数值=1时，表示不进行限制 top_p: float = 1.0 # 限制模型只从概率最高的K个token中选择拟生成的token top_k: int = -1 # 根据拟生成的token进行重复性控制，-2~2之间，负数表示鼓励重复，正数表示鼓励不重复 frequency_penalty: float = 0.0 # 根据已经生成的token进行重复性控制，-2~2之间，负数表示鼓励重复，正数表示鼓励不重复 presence_penalty: float = 0.0 # 对已经出现在提示或生成文本中的token进行重复性控制。值在0到2之间，小于1的值鼓励重复，大于1的值鼓励多样性 repetition_penalty: float = 1.0 # 强制模型生成至少min_new_tokens个token，直到遇到停止词或EOS token min_new_tokens: int = 0 # 控制在生成过程中是否在特殊token之间添加空格，例如在EOS token中 spaces_between_special_tokens: bool = True # 指定每个请求生成的输出候选文本的数量，例如每次输入提示词，模型都输出n条对应的输出 n: int = 1 # 控制是否从生成的文本中修剪停止词或EOS token。如果设置为True，生成的文本会保留停止词和EOS token；如果设置为False，则会自动删除它们 no_stop_trim: bool = True # 控制在生成过程中是否跳过特殊token（如<EOS>、<BOS>等）。如果设置为True，生成的文本中不会包含这些特殊token skip_special_tokens: bool = False max_new_tokens:int=12800 # lora微调模型地址 lora_path:Optional[List[Optional[str]]] = None def __init__(self,**kwargs): self.params = {"max_new_tokens":self.max_new_tokens} if "stop" in kwargs: self.params["stop"]=kwargs["stop"] if "temperature" in kwargs: self.params["temperature"] = float(kwargs["temperature"]) if "top_p" in kwargs: self.params["top_p"] = float(kwargs["top_p"]) if "top_k" in kwargs: self.params["top_k"] = int(kwargs["top_k"]) if "frequency_penalty" in kwargs: self.params["frequency_penalty"] = float(kwargs["frequency_penalty"]) if "presence_penalty" in kwargs: self.params["presence_penalty"] = float(kwargs["presence_penalty"]) if "repetition_penalty" in kwargs: self.params["repetition_penalty"] = float(kwargs["repetition_penalty"]) if "min_new_tokens" in kwargs: self.params["min_new_tokens"] = int(kwargs["min_new_tokens"]) if "spaces_between_special_tokens" in kwargs: self.params["spaces_between_special_tokens"] = bool(kwargs["spaces_between_special_tokens"]) if "n" in kwargs: self.params["n"] = int(kwargs["n"]) if "no_stop_trim" in kwargs: self.params["no_stop_trim"] = bool(kwargs["no_stop_trim"]) if "skip_special_tokens" in kwargs: self.params["skip_special_tokens"] = bool(kwargs["skip_special_tokens"]) if "max_new_tokens" in kwargs: self.params["max_new_tokens"] = bool(kwargs["max_new_tokens"]) if "lora_path" in kwargs: self.lora_path = kwargs["lora_path"] self.history: List[dict] = [{ "role": "system", "content": "你是一个人工智能助手，你为用户解决各种问题" }] self.llm=Model_Init.instance().llm def _build_prompt(self,prompt:Union[str,List[dict]]): context:str="" container=[] if isinstance(prompt,str): container=self.history else: container=prompt context=build_prompt_from_history(container) return context def Generation(self,prompt:Union[str,List[dict]])->str: assert(self.llm is not None),"模型引擎未能加载成功" # 如果prompt不包含历史数据，则将prompt组装到对象实例内部的history if isinstance(prompt,str): self.history.append({"role": "user", "content": prompt}) # 将history对象转换成字符串 context=self._build_prompt(prompt) output = self.llm.generate(prompt=context, sampling_params=self.params,lora_path=self.lora_path) if isinstance(prompt,str): self.history.append({"role": "assistant", "content": output["text"]}) return output["text"] async def Async_Generation(self,prompt:Union[str,List[dict]])->str: assert(self.llm is not None),"模型引擎未能加载成功" # 如果prompt不包含历史数据，则将prompt组装到对象实例内部的history if isinstance(prompt,str): self.history.append({"role": "user", "content": prompt}) context=self._build_prompt(prompt) output = await self.llm.async_generate(prompt=context, sampling_params=self.params,lora_path=self.lora_path) if isinstance(prompt,str): self.history.append({"role": "assistant", "content": output["text"]}) return output["text"] def Stream_Generation(self,prompt:Union[str,List[dict]]): assert(self.llm is not None),"模型引擎未能加载成功" # 如果prompt不包含历史数据，则将prompt组装到对象实例内部的history if isinstance(prompt,str): self.history.append({"role": "user", "content": prompt}) context=self._build_prompt(prompt) final_text = "" for chunk in self.llm.generate(prompt=context, sampling_params=self.params,lora_path=self.lora_path,stream=True): chunk_text = chunk["text"] cleaned_chunk = trim_overlap(final_text, chunk_text) if isinstance(prompt,str): if final_text=="": self.history.append({"role": "assistant", "content": cleaned_chunk}) else: self.history[-1] = {"role": "assistant","content": f"{self.history[-1]['content']}{cleaned_chunk}"} final_text += cleaned_chunk yield cleaned_chunk async def Async_Stream_Generation(self,prompt:Union[str,List[dict]]): final_text = "" if isinstance(prompt,str): self.history.append({"role": "user", "content": prompt}) context=self._build_prompt(prompt) generator = await self.llm.async_generate(prompt=context, sampling_params=self.params,lora_path=self.lora_path,stream=True) async for chunk in generator: chunk_text = chunk["text"] cleaned_chunk = trim_overlap(final_text, chunk_text) if isinstance(prompt,str): if final_text=="": self.history.append({"role": "assistant", "content": cleaned_chunk}) else: self.history[-1] = {"role": "assistant","content": f"{self.history[-1]['content']}{cleaned_chunk}"} final_text += cleaned_chunk yield cleaned_chunk上述消息处理类包含一个同步推理和一个异步推理，同时内置了对话上下文对象，该对象可以忽略，正常情况下应该由客户端构造该对象。
批量消息处理类
import asyncio from threading import RLock from typing import List from Core.Chat_Wrapper_Batch import Chat_Wrapper_Batch from Core.Common import build_prompt_from_prompts class Batch_Queue_Handle(): single_lock = RLock() def __init__(self,max_batch:int=3): assert(max_batch<=3),"单批次处理推理任务最多不超过3条" self.max_batch=max_batch self.queue = asyncio.Queue() self.clients={} async def Generation(self,messages): wrapper=Chat_Wrapper_Batch() task_ids=[] prompts=[item["prompt"] for item in messages] async for index,chunk in wrapper.Async_Stream_Generation(prompts=prompts): client_id=messages[index]["client_id"] await self.clients[client_id].send_text(chunk) async def batch_process(self): while True: await asyncio.sleep(3) # 每 3 秒批量处理一次 messages = [] #[{"client_id": client_id, "prompt": prompt}] prompt:"system：.....\n user: ......\n assistant: .....\n user: ......"] while not self.queue.empty() and len(messages) < self.max_batch: messages.append(await self.queue.get()) # 从队列中取出任务 if messages: await self.Generation(messages) @classmethod def instance(cls, *args, **kwargs): if not hasattr(Batch_Queue_Handle, "_instance"): with Batch_Queue_Handle.single_lock: if not hasattr(Batch_Queue_Handle, "_instance"): Batch_Queue_Handle._instance = cls(*args, **kwargs) return Batch_Queue_Handle._instance该类主要是为websocket编写，将客户端请求放入一个队列中，每3秒钟，调用Chat_Wrapper_Batch类的Async_Stream_Generation方法执行一次批量推理。
Chat_Wrapper_Batch类：
import re from typing import List, Optional, Union import sys import os sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../'))) from Init.Model_Init import Model_Init from sglang.utils import trim_overlap from Core.Common import build_prompt_from_histories,build_prompt_from_prompts class Chat_Wrapper_Batch(): # 停止词 stop:Optional[Union[str, List[str]]] = None # 采样温度，控制生成文本的多样性，温度越高多样性越强 temperature:float=1.0 # token概率选择，参数值越小，token可选择范围越小（eg.0.1 只会选择推理概率超过90%的token）参数值=1时，表示不进行限制 top_p: float = 1.0 # 限制模型只从概率最高的K个token中选择拟生成的token top_k: int = -1 # 根据拟生成的token进行重复性控制，-2~2之间，负数表示鼓励重复，正数表示鼓励不重复 frequency_penalty: float = 0.0 # 根据已经生成的token进行重复性控制，-2~2之间，负数表示鼓励重复，正数表示鼓励不重复 presence_penalty: float = 0.0 # 对已经出现在提示或生成文本中的token进行重复性控制。值在0到2之间，小于1的值鼓励重复，大于1的值鼓励多样性 repetition_penalty: float = 1.0 # 强制模型生成至少min_new_tokens个token，直到遇到停止词或EOS token min_new_tokens: int = 0 # 控制在生成过程中是否在特殊token之间添加空格，例如在EOS token中 spaces_between_special_tokens: bool = True # 指定每个请求生成的输出候选文本的数量，例如每次输入提示词，模型都输出n条对应的输出 n: int = 1 # 控制是否从生成的文本中修剪停止词或EOS token。如果设置为True，生成的文本会保留停止词和EOS token；如果设置为False，则会自动删除它们 no_stop_trim: bool = True # 控制在生成过程中是否跳过特殊token（如<EOS>、<BOS>等）。如果设置为True，生成的文本中不会包含这些特殊token skip_special_tokens: bool = False max_new_tokens:int=500 # lora微调模型地址 lora_path:Optional[List[Optional[str]]] = None def __init__(self,**kwargs): self.params = {"max_new_tokens":self.max_new_tokens} if "stop" in kwargs: self.params["stop"]=kwargs["stop"] if "temperature" in kwargs: self.params["temperature"] = float(kwargs["temperature"]) if "top_p" in kwargs: self.params["top_p"] = float(kwargs["top_p"]) if "top_k" in kwargs: self.params["top_k"] = int(kwargs["top_k"]) if "frequency_penalty" in kwargs: self.params["frequency_penalty"] = float(kwargs["frequency_penalty"]) if "presence_penalty" in kwargs: self.params["presence_penalty"] = float(kwargs["presence_penalty"]) if "repetition_penalty" in kwargs: self.params["repetition_penalty"] = float(kwargs["repetition_penalty"]) if "min_new_tokens" in kwargs: self.params["min_new_tokens"] = int(kwargs["min_new_tokens"]) if "spaces_between_special_tokens" in kwargs: self.params["spaces_between_special_tokens"] = bool(kwargs["spaces_between_special_tokens"]) if "n" in kwargs: self.params["n"] = int(kwargs["n"]) if "no_stop_trim" in kwargs: self.params["no_stop_trim"] = bool(kwargs["no_stop_trim"]) if "skip_special_tokens" in kwargs: self.params["skip_special_tokens"] = bool(kwargs["skip_special_tokens"]) if "max_new_tokens" in kwargs: self.params["max_new_tokens"] = bool(kwargs["max_new_tokens"]) if "lora_path" in kwargs: self.lora_path = kwargs["lora_path"] self.llm=Model_Init.instance().llm def _build_prompt(self,histories): batch_prompt=build_prompt_from_histories(histories=histories) return batch_prompt def _re_build_history(self,context:str): pattern = re.compile(r"(?P<role>user|assistant|system)\s*[:：]\s*(?P<content>.+)") matches = pattern.findall(context) dialogue = [{"role": role, "content": content} for role, content in matches] return dialogue def Generation(self,histories:List[List[dict]]): assert(self.llm is not None),"模型引擎未能加载成功" result:List[List[dict]]=[] batch_prompt=self._build_prompt(histories) outputs = self.llm.generate(prompt=batch_prompt, sampling_params=self.params,lora_path=self.lora_path) for prompt, output in zip(batch_prompt, outputs): context=f"{prompt}\n assistant: {output['text']}" dialogue=self._re_build_history(context=context) result.append(dialogue) return result async def Async_Generation(self,histories:List[List[dict]]): assert(self.llm is not None),"模型引擎未能加载成功" result:List[List[dict]]=[] batch_prompt=self._build_prompt(histories) outputs =await self.llm.async_generate(prompt=batch_prompt, sampling_params=self.params,lora_path=self.lora_path) for prompt, output in zip(batch_prompt, outputs): context=f"{prompt}\n assistant: {output['text']}" dialogue=self._re_build_history(context=context) result.append(dialogue) return result async def Async_Stream_Generation(self,histories:List[List[dict]]=None,prompts:List[str]=None): assert(self.llm is not None),"模型引擎未能加载成功" assert(histories is not None or prompts is not None),"输入数据不能为空" if histories is not None and len(histories)>0: batch_prompt=self._build_prompt(histories) else: batch_prompt=build_prompt_from_prompts(prompts) generator =await self.llm.async_generate(prompt=batch_prompt, sampling_params=self.params,lora_path=self.lora_path,stream=True) final_text = "" async for output in generator: chunk_text = output["text"] cleaned_chunk = trim_overlap(final_text, chunk_text) final_text += cleaned_chunk yield int(output["index"]),cleaned_chunk模型加载类
from typing import List, Optional import sglang as sgl from threading import RLock class Model_Init(): single_lock = RLock() model_path:str="/home/DeepSeek_8B" #"/app/model" # 张量并行参数，表示模型计算过程中使用多少张显卡（每张显卡只承载部分模型，适合体量大的模型） tp_size:int=1 # 数据并行，表示模型被分发到多少张显卡上进行并行推理（每张显卡都要承载完整模型，适合体量小的模型） dp_size:int=1 # 用于设置 KV Cache 内存池 的最大内存占用比例,如果设置0.9则表示在计算过程中KV Cache 最多可以使用 90% 的 GPU 内存，过高容易导致内存溢出，过低会影响计算性能 # 通常在并行推理中不宜设置过高，需要为每一条推理预留出内存空间 mem_fraction_static:float=0.7 # 用于将预填充阶段的输入 prompt 分块（chunk） 处理,通过分块处理，减少预填充阶段的内存占用和计算量，从而加速推理过程 # 如果 chunked_prefill_size=128，模型会将输入 prompt 分成每块 128 个 token 进行处理,，特别适合于长prompt，设置过大会导致性能降低，设置过小会导致内存不足 chunked_prefill_size:int=None # 是否启用Torch加速，对FP8模型不生效 enable_torch_compile:bool=False # 如要使用lora模型，则提供lora模型的模型保存路径，如果 lora_paths=["adapter1", "adapter2"]，则批次中的第一个元素会使用 adapter1，第二个元素会使用 adapter2 lora_paths: Optional[List[str]] = None # 每个推理批次中最多使用多少个lora模型 max_loras_per_batch: int = 8 # 是否禁止使用cuda_graph，如何使用Lora则必须将此参数设置为True disable_cuda_graph:bool=False # 是否禁止使用radix_cache，如何使用Lora则必须将此参数设置为True disable_radix_cache:bool=False def __init__(self,**kwargs): self.llm=None if "lora_paths" in kwargs: self.lora_paths = kwargs["lora_paths"] if "max_loras_per_batch" in kwargs: self.max_loras_per_batch = kwargs["max_loras_per_batch"] if self.lora_paths is not None and len(self.lora_paths)>0: self.disable_cuda_graph=True self.disable_radix_cache=True try: llm = sgl.Engine(model_path=self.model_path, tp_size=self.tp_size, dp_size=self.dp_size, mem_fraction_static=self.mem_fraction_static, chunked_prefill_size=self.chunked_prefill_size, enable_torch_compile=self.enable_torch_compile, lora_paths=self.lora_paths, max_loras_per_batch=self.max_loras_per_batch, disable_cuda_graph=self.disable_cuda_graph, disable_radix_cache=self.disable_radix_cache ) self.llm=llm except Exception as exc: exception = exc @property def LLM_Engine(self): if self.llm is not None: return self.llm return None def Close(self): if self.llm is not None: self.llm.shutdown() self.llm=None @classmethod def instance(cls, *args, **kwargs): if not hasattr(Model_Init, "_instance"): with Model_Init.single_lock: if not hasattr(Model_Init, "_instance"): Model_Init._instance = cls(*args, **kwargs) return Model_Init._instance公共帮助类
from typing import List from sglang import assistant, system, user import re def build_prompt_from_history(history:List[dict]): context:str="" for turn in history: role = turn["role"] content = turn["content"] if role=="system": context+=f"system:{content}\n" elif role=="user": context+=f"user:{content}\n" else: context+=f"assistant:{content}\n" return context def build_prompt_from_histories(histories:List[List[dict]]): batch_prompt:List[str]=[] for history in histories: context:str="" for turn in history: role = turn["role"] content = turn["content"] if role=="system": context+=f"system:{content}\n" elif role=="user": context+=f"user:{content}\n" else: context+=f"assistant:{content}\n" batch_prompt.append(context) return batch_prompt def build_prompt_from_prompt(prompt:str): pattern = re.compile(r"(?P<role>user|assistant|system)\s*[:：]\s*(?P<content>.+)") matches = pattern.findall(prompt) dialogue = [{"role": role, "content": content} for role, content in matches] context=build_prompt_from_history(dialogue) return context def build_prompt_from_prompts(prompts:List[str]): assert(prompts is not None and len(prompts)>0),"传入参数为空" result_prompts=[] for prompt in prompts: context=build_prompt_from_prompt(prompt=prompt) result_prompts.append(context) return result_prompts def valid_prompt(propmt:List[dict]): valid_roles = {"system", "user", "assistant"} required_keys = {"role", "content"} if propmt is None or len(propmt)==0: return "输入内容不能为空" for item in propmt: if not required_keys.issubset(item.keys()): return "输入内容中必须包含role和content属性" if item["role"] not in valid_roles: return "输入内容中的role只能是system|user|assistant中的一种" if not isinstance(item["role"], str) or not isinstance(item["content"], str): return "输入内容中的role和content的值必须是字符串" return None主要用于构造和解析对话上下文，SGLang中的对话上下文，按照system,user,assistant三种身份进行设置，兼容了OpenAI标准。
五、效果图
通过使用APIFOX，对WebSocket接口进行测试，开启两个线程，同时连接Websocket，并同时发送消，对比消息接收时间，可以发现，确实是在并行推理。
