Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/llm_datadist_v1/configs.py: 90%
256 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-27 10:03 +0800
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# -----------------------------------------------------------------------------------------------------------
4# Copyright (c) 2025 Huawei Technologies Co., Ltd.
5# This program is free software, you can redistribute it and/or modify it under the terms and conditions of
6# CANN Open Software License Agreement Version 2.0 (the "License").
7# Please refer to the License for details. You may not use this file except in compliance with the License.
8# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
9# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
10# See LICENSE in the root of the software repository for the full text of the License.
11# -----------------------------------------------------------------------------------------------------------
13import json
14import socket
15from enum import IntEnum
16from typing import List, Tuple, Union
18from .status import raise_if_false
19from .utils.utils import (
20 check_dict,
21 check_int32,
22 check_isinstance,
23 check_type,
24 check_uint16,
25 check_uint32,
26 check_uint64,
27)
29_INVALID_ID = 2**64 - 1
32class LLMRole(IntEnum):
33 PROMPT = 1
34 DECODER = 2
35 MIX = 3
38def trans_str_ip(ip):
39 if check_type(ip, str):
40 try:
41 ip_bytes = socket.inet_aton(ip)
42 return int.from_bytes(ip_bytes, byteorder="little")
43 except Exception:
44 raise RuntimeError(f"Can not parse ip str:{ip}")
45 return ip
48class LLMClusterInfo(object):
49 def __init__(self):
50 self._remote_cluster_id = None
51 self._remote_role_type = None
52 self._local_ip_info_list: List[Tuple[int, int]] = []
53 self._remote_ip_info_list: List[Tuple[int, int]] = []
55 def _check_inputs(self, ip, port):
56 check_isinstance("ip", ip, [str, int])
57 check_uint16("port", port)
58 return trans_str_ip(ip)
60 @property
61 def remote_role_type(self):
62 return self._remote_role_type
64 @property
65 def remote_cluster_id(self):
66 return self._remote_cluster_id
68 @property
69 def local_ip_info_list(self):
70 return self._local_ip_info_list
72 @property
73 def remote_ip_info_list(self):
74 return self._remote_ip_info_list
76 @remote_role_type.setter
77 def remote_role_type(self, remote_role_type: Union[LLMRole, int]):
78 check_isinstance("remote_role_type", remote_role_type, [LLMRole, int])
79 self._remote_role_type = remote_role_type
81 @remote_cluster_id.setter
82 def remote_cluster_id(self, remote_cluster_id):
83 check_uint64("remote_cluster_id", remote_cluster_id)
84 self._remote_cluster_id = remote_cluster_id
86 def append_local_ip_info(self, ip: Union[str, int], port: int):
87 """
88 添加本地IP信息
89 Args:
90 ip: IP
91 port: 端口
92 """
93 ip = self._check_inputs(ip, port)
94 self._local_ip_info_list.append((ip, port))
96 def append_remote_ip_info(self, ip: Union[str, int], port: int):
97 """
98 添加对端IP信息
99 Args:
100 ip: IP
101 port: 端口
102 """
103 ip = self._check_inputs(ip, port)
104 self._remote_ip_info_list.append((ip, port))
107class LlmConfig(object):
108 def __init__(self):
109 self._options = {}
110 self._listen_ip_info = ""
111 self._device_id = None
112 self._sync_kv_timeout = None
113 self._deploy_res_path = ""
114 self._ge_options = {}
115 self._enable_switch_role = False
117 # below is offline
118 self._cluster_info = ""
119 self._output_max_size = ""
120 self._mem_utilization = 0.95
121 self._buf_pool_cfg = ""
122 self._mem_pool_cfg = ""
123 self._host_mem_pool_cfg = ""
124 self._enable_cache_manager = None
125 self._enable_remote_cache_accessible = None
126 self._rdma_traffic_class = None
127 self._rdma_service_level = None
128 self._local_comm_res = None
130 def generate_options(self):
131 """
132 生成LLM Engine配置项
133 Returns:
134 配置项dict
135 """
136 return self.gen_options()
138 def gen_options(self):
139 if self.ge_options:
140 self._options.update(self.ge_options)
141 if self.listen_ip_info:
142 self._options["llm.listenIpInfo"] = str(self.listen_ip_info)
143 if self.device_id is not None:
144 if check_type(self.device_id, int):
145 self._options["ge.exec.deviceId"] = str(self.device_id)
146 self._options["ge.session_device_id"] = str(self.device_id)
147 else:
148 self._options["ge.session_device_id"] = str(self.device_id[0])
149 self._options["ge.exec.deviceId"] = ";".join([str(dev) for dev in self.device_id])
150 if self.sync_kv_timeout is not None:
151 self._options["llm.SyncKvCacheWaitTime"] = str(self.sync_kv_timeout)
152 if self.deploy_res_path:
153 self._options["llm.deployResPath"] = str(self.deploy_res_path)
154 if self.buf_pool_cfg:
155 self._options["llm.BufPoolCfg"] = str(self.buf_pool_cfg)
156 if self._mem_pool_cfg:
157 self._options["llm.MemPoolConfig"] = str(self._mem_pool_cfg)
158 if self._host_mem_pool_cfg:
159 self._options["llm.HostMemPoolConfig"] = str(self._host_mem_pool_cfg)
160 if self._enable_cache_manager is not None:
161 self._options["llm.EnableCacheManager"] = "1" if self._enable_cache_manager else "0"
162 if self._enable_remote_cache_accessible is not None:
163 self._options["llm.EnableRemoteCacheAccessible"] = "1" if self._enable_remote_cache_accessible else "0"
165 # below is offline
166 if self._cluster_info:
167 self._options["llm.ClusterInfo"] = str(self.cluster_info)
168 if self._output_max_size:
169 self._options["llm.OutputMaxSize"] = str(self.output_max_size)
170 if self._enable_switch_role:
171 self._options["llm.EnableSwitchRole"] = "1"
172 if self._mem_utilization is not None:
173 self._options["llm.MemoryUtilization"] = str(self.mem_utilization)
174 if self.rdma_traffic_class is not None:
175 self._options["llm.RdmaTrafficClass"] = str(self.rdma_traffic_class)
176 if self.rdma_service_level is not None:
177 self._options["llm.RdmaServiceLevel"] = str(self.rdma_service_level)
178 if self._local_comm_res is not None:
179 self._options["llm.LocalCommRes"] = str(self.local_comm_res)
180 return self.options
182 @property
183 def ge_options(self):
184 return self._ge_options
186 @ge_options.setter
187 def ge_options(self, ge_options):
188 check_isinstance("ge_options", ge_options, dict)
189 check_dict("ge_options", ge_options, str, str)
190 self._ge_options = ge_options
192 @property
193 def device_id(self):
194 return self._device_id
196 @device_id.setter
197 def device_id(self, device_id):
198 check_isinstance("device_id", device_id, [list, tuple, int])
199 if check_type(device_id, list) or check_type(device_id, tuple):
200 check_isinstance("device_id", device_id, [list, tuple], int)
201 [raise_if_false(dev_id >= 0, "device_id should be greater than or equal to zero.") for dev_id in device_id]
202 [check_int32("device_id", dev_id) for dev_id in device_id]
203 else:
204 check_isinstance("device_id", device_id, int)
205 raise_if_false(device_id >= 0, "device_id should be greater than or equal to zero.")
206 check_int32("device_id", device_id)
207 self._device_id = device_id
209 @property
210 def listen_ip_info(self):
211 return self._listen_ip_info
213 @listen_ip_info.setter
214 def listen_ip_info(self, listen_ip_info):
215 check_isinstance("listen_ip_info", listen_ip_info, str)
216 self._listen_ip_info = listen_ip_info
218 @property
219 def deploy_res_path(self):
220 return self._deploy_res_path
222 @deploy_res_path.setter
223 def deploy_res_path(self, deploy_res_path):
224 check_isinstance("deploy_res_path", deploy_res_path, str)
225 self._deploy_res_path = deploy_res_path
227 @property
228 def buf_pool_cfg(self):
229 return self._buf_pool_cfg
231 @buf_pool_cfg.setter
232 def buf_pool_cfg(self, buf_pool_cfg):
233 check_isinstance("buf_pool_cfg", buf_pool_cfg, str)
234 self._buf_pool_cfg = buf_pool_cfg
236 @property
237 def output_max_size(self):
238 return self._output_max_size
240 @output_max_size.setter
241 def output_max_size(self, output_max_size):
242 check_isinstance("output_max_size", output_max_size, int)
243 self._output_max_size = output_max_size
245 @property
246 def mem_utilization(self):
247 return self._mem_utilization
249 @mem_utilization.setter
250 def mem_utilization(self, mem_utilization):
251 check_isinstance("mem_utilization", mem_utilization, float)
252 raise_if_false(
253 ((mem_utilization >= 0.0) and (mem_utilization <= 1.0)),
254 f"mem_utilization must be in range [0,1], current:{mem_utilization}",
255 )
256 self._mem_utilization = mem_utilization
258 @property
259 def options(self):
260 return self._options
262 @property
263 def cluster_info(self):
264 return self._cluster_info
266 @property
267 def sync_kv_timeout(self):
268 return self._sync_kv_timeout
270 @cluster_info.setter
271 def cluster_info(self, cluster_info):
272 check_isinstance("cluster_info", cluster_info, str)
273 cluster_info_dict = json.loads(cluster_info)
274 if "listen_ip_info" in cluster_info_dict:
275 for ip_info in cluster_info_dict["listen_ip_info"]:
276 ip_info["ip"] = trans_str_ip(ip_info["ip"])
277 self._cluster_info = json.dumps(cluster_info_dict)
279 @sync_kv_timeout.setter
280 def sync_kv_timeout(self, sync_kv_timeout):
281 check_isinstance("sync_kv_timeout", sync_kv_timeout, [int, str])
282 if check_type(sync_kv_timeout, str):
283 raise_if_false(sync_kv_timeout.isdigit(), "sync_kv_timeout must be digit.")
284 raise_if_false(int(sync_kv_timeout) > 0, "sync_kv_timeout should be greater than zero.")
285 check_int32("sync_kv_timeout", int(sync_kv_timeout))
286 self._sync_kv_timeout = sync_kv_timeout
288 @property
289 def enable_switch_role(self):
290 return self._enable_switch_role
292 @enable_switch_role.setter
293 def enable_switch_role(self, enable_switch_role: bool):
294 check_isinstance("enable_switch_role", enable_switch_role, [bool])
295 self._enable_switch_role = enable_switch_role
297 @property
298 def enable_cache_manager(self):
299 return False if self._enable_cache_manager is None else self._enable_cache_manager
301 @enable_cache_manager.setter
302 def enable_cache_manager(self, enable_cache_manager: bool):
303 check_isinstance("enable_cache_manager", enable_cache_manager, [bool])
304 self._enable_cache_manager = enable_cache_manager
306 @property
307 def enable_remote_cache_accessible(self):
308 return False if self._enable_remote_cache_accessible is None else self._enable_remote_cache_accessible
310 @enable_remote_cache_accessible.setter
311 def enable_remote_cache_accessible(self, enable_remote_cache_accessible: bool):
312 check_isinstance("enable_remote_cache_accessible", enable_remote_cache_accessible, [bool])
313 self._enable_remote_cache_accessible = enable_remote_cache_accessible
315 @property
316 def mem_pool_cfg(self) -> str:
317 return self._mem_pool_cfg
319 @mem_pool_cfg.setter
320 def mem_pool_cfg(self, mem_pool_cfg: str):
321 check_isinstance("mem_pool_cfg", mem_pool_cfg, str)
322 self._mem_pool_cfg = mem_pool_cfg
324 @property
325 def host_mem_pool_cfg(self) -> str:
326 return self._host_mem_pool_cfg
328 @host_mem_pool_cfg.setter
329 def host_mem_pool_cfg(self, host_mem_pool_cfg: str):
330 check_isinstance("host_mem_pool_cfg", host_mem_pool_cfg, str)
331 self._host_mem_pool_cfg = host_mem_pool_cfg
333 @property
334 def rdma_traffic_class(self) -> str:
335 return self._rdma_traffic_class
337 @rdma_traffic_class.setter
338 def rdma_traffic_class(self, rdma_traffic_class: int):
339 check_uint32("rdma_traffic_class", rdma_traffic_class)
340 self._rdma_traffic_class = rdma_traffic_class
342 @property
343 def rdma_service_level(self) -> str:
344 return self._rdma_service_level
346 @rdma_service_level.setter
347 def rdma_service_level(self, rdma_service_level: int):
348 check_uint32("rdma_service_level", rdma_service_level)
349 self._rdma_service_level = rdma_service_level
351 @property
352 def local_comm_res(self):
353 return "" if self._local_comm_res is None else self._local_comm_res
355 @local_comm_res.setter
356 def local_comm_res(self, local_comm_res):
357 check_isinstance("local_comm_res", local_comm_res, str)
358 self._local_comm_res = local_comm_res