Coverage for cli / inference.py: 100.00%
497 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Inference endpoint management for GCO CLI.
4Provides functionality to deploy, manage, and monitor inference endpoints
5across multi-region EKS clusters via the DynamoDB-backed reconciliation
6pattern (inference_monitor).
7"""
9from __future__ import annotations
11import logging
12import secrets
13from copy import deepcopy
14from typing import TYPE_CHECKING, Any, Literal, TypedDict, TypeGuard
16from .aws_client import get_aws_client
17from .config import GCOConfig, get_config
19# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
20# Generated at (UTC): 2026-09-13T13:44:22Z
21# Generated from Git commit: c49331669c66625fecfecf44ae6ab5f95afbfcb4
22# Flowchart(s) generated from this file:
23# * ``InferenceManager.deploy`` -> ``diagrams/code_diagrams/cli/inference.InferenceManager_deploy.html``
24# (PNG: ``diagrams/code_diagrams/cli/inference.InferenceManager_deploy.png``)
25# * ``InferenceManager.canary_deploy`` -> ``diagrams/code_diagrams/cli/inference.InferenceManager_canary_deploy.html``
26# (PNG: ``diagrams/code_diagrams/cli/inference.InferenceManager_canary_deploy.png``)
27# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
28# <pyflowchart-code-diagram> END
31if TYPE_CHECKING:
32 from gco.services.inference_store import InferenceEndpointStore
34logger = logging.getLogger(__name__)
37# ---------------------------------------------------------------------------
38# Mooncake topology — optional endpoint-spec extension
39# ---------------------------------------------------------------------------
40#
41# An endpoint spec may carry an optional ``mooncake`` block describing
42# disaggregated prefill/decode (PD) serving and/or a shared KV-cache store.
43# The block is entirely additive: when it is absent the endpoint reconciles
44# exactly as it does today — one Deployment and one internal ClusterIP Service
45# behind the shared authenticated inference route.
46#
47# The definitions below describe the shape of that block (the dict written to
48# DynamoDB and read back by the per-region monitor) and the constant
49# vocabularies its enumerated fields draw from. Byte-size fields are authored
50# as base-10 integer decimal strings (see :func:`author_byte_size`) so they
51# round-trip through DynamoDB without being coerced to ``Decimal`` via a float
52# literal.
54#: Serving modes a ``mooncake`` block may declare.
55#: ``disaggregated`` splits prefill and decode; ``store`` runs a single
56#: KV-store instance; ``both`` composes the two.
57MOONCAKE_MODES: frozenset[str] = frozenset({"disaggregated", "store", "both"})
59#: KV transfer / store intents. ``rdma`` is the default high-performance
60#: intent: GCO schedules the pod on EFA and renders vLLM's point-to-point
61#: ``mooncake_protocol`` as ``efa``. ``tcp`` is the non-EFA fallback.
62MOONCAKE_TRANSFER_PROTOCOLS: frozenset[str] = frozenset({"rdma", "tcp"})
64#: KV-store offload tiers for spilling cache beyond GPU memory.
65MOONCAKE_OFFLOAD_TIERS: frozenset[str] = frozenset({"cpu", "disk", "none"})
67#: PD proxy request-scheduling strategies supported today.
68MOONCAKE_PROXY_SCHEDULING: frozenset[str] = frozenset({"round_robin"})
70#: Inclusive bounds for per-role replica counts in an XpYd topology.
71MOONCAKE_TOPOLOGY_MIN: int = 1
72MOONCAKE_TOPOLOGY_MAX: int = 1000
74#: Inclusive bounds for byte-size fields. The ceiling is the signed 64-bit
75#: maximum; authoring sizes as decimal strings in ``[MIN, MAX]`` keeps them out
76#: of float/Decimal coercion when they round-trip through DynamoDB.
77MOONCAKE_BYTE_SIZE_MIN: int = 0
78MOONCAKE_BYTE_SIZE_MAX: int = 9223372036854775807
80#: Transfer-engine defaults mirroring Mooncake's reference configuration.
81MOONCAKE_DEFAULT_BOOTSTRAP_BASE_PORT: int = 8998
82MOONCAKE_DEFAULT_NUM_WORKERS: int = 10
83MOONCAKE_DEFAULT_ABORT_REQUEST_TIMEOUT: int = 480
86class MooncakeTopology(TypedDict):
87 """An XpYd topology: ``prefill`` (X) and ``decode`` (Y) instance counts."""
89 prefill: int
90 decode: int
93class MooncakeStoreConfig(TypedDict, total=False):
94 """KV-cache store pool configuration.
96 ``global_segment_size`` and ``local_buffer_size`` are byte counts authored
97 as base-10 integer decimal strings. ``cold_tier_enabled`` opts this
98 endpoint into the asynchronous, per-region object-store cold tier; the
99 cold-tier bucket is resolved by the monitor from regional configuration and
100 is never a user-typed URI.
101 """
103 enabled: bool
104 metadata_server: str
105 master_server_address: str
106 protocol: Literal["rdma", "tcp"]
107 device_name: str
108 global_segment_size: str
109 local_buffer_size: str
110 offload: Literal["cpu", "disk", "none"]
111 cold_tier_enabled: bool
114class MooncakeTransferConfig(TypedDict, total=False):
115 """RDMA/TCP transfer-engine configuration for KV cache movement."""
117 protocol: Literal["rdma", "tcp"]
118 device_name: str
119 num_workers: int
120 bootstrap_base_port: int
121 abort_request_timeout: int
124class MooncakeProxyConfig(TypedDict, total=False):
125 """PD proxy configuration. ``admin_api_key_secret`` names the Kubernetes
126 Secret holding the proxy admin key; the key value is never carried on the
127 endpoint spec."""
129 image: str
130 scheduling: Literal["round_robin"]
131 admin_api_key_secret: str
134class MooncakeRoleAutoscaling(TypedDict, total=False):
135 """Per-role autoscaling bounds and metrics for one of prefill/decode."""
137 min_replicas: int
138 max_replicas: int
139 metrics: list[dict[str, Any]]
142class MooncakeAutoscalingConfig(TypedDict, total=False):
143 """Optional per-role pod autoscaling. When absent the topology is static."""
145 enabled: bool
146 prefill: MooncakeRoleAutoscaling
147 decode: MooncakeRoleAutoscaling
150class MooncakeSpec(TypedDict, total=False):
151 """The optional ``mooncake`` block carried on an endpoint spec dict."""
153 mode: Literal["disaggregated", "store", "both"]
154 topology: MooncakeTopology
155 store: MooncakeStoreConfig
156 transfer: MooncakeTransferConfig
157 proxy: MooncakeProxyConfig
158 autoscaling: MooncakeAutoscalingConfig
161def author_byte_size(value: int | str) -> str:
162 """Render a byte-size value as a canonical base-10 integer decimal string.
164 Mooncake store/transfer sizes (segment size, local buffer) are carried on
165 the endpoint spec as digit-only strings so they survive the DynamoDB
166 round-trip without being coerced to ``Decimal`` through a float literal.
168 Accepts a non-negative ``int`` or a string of base-10 ASCII digits and
169 returns the same whole number as ``str``. The value must fall in
170 ``[MOONCAKE_BYTE_SIZE_MIN, MOONCAKE_BYTE_SIZE_MAX]``. Signs, decimal
171 points, exponents, floats, booleans, and any non-digit text are not
172 accepted.
174 Raises:
175 ValueError: when ``value`` cannot be authored as an in-range base-10
176 integer.
177 """
178 # ``bool`` is a subclass of ``int``; reject it explicitly so ``True``/``False``
179 # never masquerade as 1/0 byte sizes.
180 if isinstance(value, bool):
181 raise ValueError(f"byte-size value must be an integer, got bool: {value!r}")
183 if isinstance(value, int):
184 size = value
185 elif isinstance(value, str):
186 text = value.strip()
187 if not text or any(ch not in "0123456789" for ch in text):
188 raise ValueError(
189 "byte-size value must be a base-10 integer string "
190 f"(ASCII digits only, no sign, point, or exponent), got {value!r}"
191 )
192 size = int(text)
193 else:
194 raise ValueError(
195 f"byte-size value must be an int or a base-10 digit string, got {type(value).__name__}"
196 )
198 if not MOONCAKE_BYTE_SIZE_MIN <= size <= MOONCAKE_BYTE_SIZE_MAX:
199 raise ValueError(
200 "byte-size value out of range "
201 f"[{MOONCAKE_BYTE_SIZE_MIN}, {MOONCAKE_BYTE_SIZE_MAX}]: {size}"
202 )
204 return str(size)
207#: Byte-size fields a ``mooncake`` store block may carry. Each is authored as a
208#: base-10 integer decimal string via :func:`author_byte_size`.
209_MOONCAKE_STORE_BYTE_SIZE_FIELDS: tuple[str, ...] = (
210 "global_segment_size",
211 "local_buffer_size",
212)
214#: Modes that run a split prefill/decode topology and therefore require a
215#: valid ``topology`` and may carry per-role autoscaling.
216_MOONCAKE_DISAGGREGATED_MODES: frozenset[str] = frozenset({"disaggregated", "both"})
219def _is_plain_int(value: Any) -> TypeGuard[int]:
220 """True when ``value`` is an ``int`` and not a ``bool``.
222 ``bool`` is a subclass of ``int``; counts and replica bounds must be real
223 integers, so ``True``/``False`` are not accepted as 1/0.
224 """
225 return isinstance(value, int) and not isinstance(value, bool)
228def _validate_role_autoscaling_bounds(role: str, role_block: dict[str, Any]) -> None:
229 """Validate one role's ``min_replicas``/``max_replicas`` bounds.
231 Raises :class:`ValueError` naming the violated bound. ``min_replicas`` must
232 be an integer ``>= 1`` and ``max_replicas`` an integer no smaller than the
233 effective minimum (which defaults to 1 when ``min_replicas`` is absent).
234 """
235 min_replicas = role_block.get("min_replicas")
236 max_replicas = role_block.get("max_replicas")
238 if min_replicas is not None:
239 if not _is_plain_int(min_replicas):
240 raise ValueError(
241 f"mooncake.autoscaling.{role}.min_replicas must be an integer, got {min_replicas!r}"
242 )
243 if min_replicas < 1:
244 raise ValueError(
245 f"mooncake.autoscaling.{role}.min_replicas must be >= 1, got {min_replicas}"
246 )
248 if max_replicas is not None:
249 if not _is_plain_int(max_replicas):
250 raise ValueError(
251 f"mooncake.autoscaling.{role}.max_replicas must be an integer, got {max_replicas!r}"
252 )
253 effective_min = min_replicas if _is_plain_int(min_replicas) else 1
254 if max_replicas < effective_min:
255 raise ValueError(
256 f"mooncake.autoscaling.{role}.max_replicas ({max_replicas}) "
257 f"must be >= min_replicas ({effective_min})"
258 )
261def validate_mooncake_spec(mooncake: dict[str, Any]) -> None:
262 """Validate a ``mooncake`` endpoint-spec block, failing fast.
264 Raises :class:`ValueError` on the first rejected field, naming the
265 offending field so the caller can correct it. The check is pure — it reads
266 nothing and writes nothing — so a caller that validates before persisting
267 leaves any previously stored spec untouched when a block is rejected.
269 The rules enforced here are:
271 * ``mode`` must be one of the supported serving modes
272 (:data:`MOONCAKE_MODES`).
273 * ``transfer`` must be a mapping when present; ``protocol`` must be one of
274 :data:`MOONCAKE_TRANSFER_PROTOCOLS` and ``device_name`` must be a string
275 (the empty string requests automatic interface detection).
276 * Store byte-size fields must author as in-range base-10 integers.
277 * ``disaggregated``/``both`` modes require integer ``topology.prefill`` and
278 ``topology.decode`` in
279 ``[MOONCAKE_TOPOLOGY_MIN, MOONCAKE_TOPOLOGY_MAX]``.
280 * ``store.cold_tier_enabled`` may be true only while ``store.enabled`` is
281 true (the cold tier extends the hot store).
282 * Autoscaling may be enabled only for ``disaggregated``/``both`` modes, and
283 each present role's ``min_replicas``/``max_replicas`` must satisfy
284 ``min_replicas >= 1`` and ``max_replicas >= min_replicas``.
285 """
286 if not isinstance(mooncake, dict):
287 raise ValueError("mooncake block must be a mapping")
289 mode = mooncake.get("mode")
290 if mode not in MOONCAKE_MODES:
291 allowed = ", ".join(sorted(MOONCAKE_MODES))
292 raise ValueError(f"mooncake.mode must be one of {{{allowed}}}, got {mode!r}")
294 store = mooncake.get("store")
295 if store is not None and not isinstance(store, dict):
296 raise ValueError("mooncake.store must be a mapping")
298 transfer = mooncake.get("transfer")
299 if transfer is not None and not isinstance(transfer, dict):
300 raise ValueError("mooncake.transfer must be a mapping")
301 if isinstance(transfer, dict):
302 protocol = transfer.get("protocol", "rdma")
303 if protocol not in MOONCAKE_TRANSFER_PROTOCOLS:
304 allowed = ", ".join(sorted(MOONCAKE_TRANSFER_PROTOCOLS))
305 raise ValueError(
306 f"mooncake.transfer.protocol must be one of {{{allowed}}}, got {protocol!r}"
307 )
308 device_name = transfer.get("device_name", "")
309 if not isinstance(device_name, str):
310 raise ValueError(f"mooncake.transfer.device_name must be a string, got {device_name!r}")
312 # Byte-size fields must author cleanly; surface the offending field name.
313 if isinstance(store, dict):
314 for field in _MOONCAKE_STORE_BYTE_SIZE_FIELDS:
315 if field in store:
316 try:
317 author_byte_size(store[field])
318 except ValueError as exc:
319 raise ValueError(f"mooncake.store.{field}: {exc}") from exc
321 # Split topologies need integer prefill/decode counts in range.
322 if mode in _MOONCAKE_DISAGGREGATED_MODES:
323 topology = mooncake.get("topology")
324 if not isinstance(topology, dict):
325 raise ValueError(
326 f"mooncake.topology is required for mode {mode!r} with integer "
327 "'prefill' and 'decode' counts"
328 )
329 for field in ("prefill", "decode"):
330 count = topology.get(field)
331 if not _is_plain_int(count):
332 raise ValueError(
333 f"mooncake.topology.{field} must be an integer in "
334 f"[{MOONCAKE_TOPOLOGY_MIN}, {MOONCAKE_TOPOLOGY_MAX}], "
335 f"got {count!r}"
336 )
337 if not MOONCAKE_TOPOLOGY_MIN <= count <= MOONCAKE_TOPOLOGY_MAX:
338 raise ValueError(
339 f"mooncake.topology.{field} out of range "
340 f"[{MOONCAKE_TOPOLOGY_MIN}, {MOONCAKE_TOPOLOGY_MAX}]: {count}"
341 )
343 # The cold tier extends the hot store; it cannot be enabled on its own.
344 if (
345 isinstance(store, dict)
346 and store.get("cold_tier_enabled") is True
347 and store.get("enabled") is not True
348 ):
349 raise ValueError(
350 "mooncake.store.cold_tier_enabled requires mooncake.store.enabled to be true"
351 )
353 autoscaling = mooncake.get("autoscaling")
354 if autoscaling is not None:
355 if not isinstance(autoscaling, dict):
356 raise ValueError("mooncake.autoscaling must be a mapping")
357 if autoscaling.get("enabled") is True and mode not in _MOONCAKE_DISAGGREGATED_MODES:
358 raise ValueError(
359 "mooncake.autoscaling.enabled requires a 'disaggregated' or "
360 f"'both' mode, got {mode!r}"
361 )
362 for role in ("prefill", "decode"):
363 role_block = autoscaling.get(role)
364 if role_block is None:
365 continue
366 if not isinstance(role_block, dict):
367 raise ValueError(f"mooncake.autoscaling.{role} must be a mapping")
368 _validate_role_autoscaling_bounds(role, role_block)
371class InferenceManager:
372 """Manages inference endpoints via the DynamoDB store."""
374 def __init__(self, config: GCOConfig | None = None):
375 self.config = config or get_config()
376 self._aws_client = get_aws_client(config)
378 def _get_store(self, region: str | None = None) -> InferenceEndpointStore:
379 """Get an InferenceEndpointStore for the global region."""
380 from gco.services.inference_store import InferenceEndpointStore
382 # Use the global region for DynamoDB (same as job store)
383 store_region = region or self.config.global_region
384 return InferenceEndpointStore(region=store_region)
386 def _build_mooncake_block(
387 self,
388 *,
389 mode: str,
390 prefill_replicas: int,
391 decode_replicas: int,
392 store: dict[str, Any] | None,
393 transfer: dict[str, Any] | None,
394 proxy: dict[str, Any] | None,
395 autoscaling: dict[str, Any] | None,
396 default_proxy_image: str | None = None,
397 ) -> dict[str, Any]:
398 """Assemble and validate an optional ``spec.mooncake`` block.
400 Composes the topology and any supplied store/transfer/proxy/autoscaling
401 sub-blocks into a single mapping, authoring store byte-size fields as
402 base-10 integer decimal strings so they round-trip through DynamoDB,
403 then validates the result. Validation is pure and runs before the
404 caller persists anything, so a rejected block leaves any previously
405 stored spec untouched. Raises :class:`ValueError` — naming the offending
406 field — when the mode is unsupported or any field is invalid.
408 The store-bearing modes (``store`` and ``both``) default the store to
409 enabled so the shared master address is wired in (the ``both``-mode
410 MultiConnector's store half depends on it), and split modes
411 (``disaggregated`` and ``both``) default the prefill-decode proxy image
412 to ``default_proxy_image`` when the caller supplies no explicit proxy
413 image.
414 """
415 block: dict[str, Any] = {"mode": mode}
417 # Split modes carry an XpYd topology; a single-instance store does not.
418 if mode in _MOONCAKE_DISAGGREGATED_MODES:
419 block["topology"] = {
420 "prefill": prefill_replicas,
421 "decode": decode_replicas,
422 }
424 # The store-bearing modes (store and both) only function with the KV
425 # store enabled: the both-mode MultiConnector's store half is wired to
426 # the shared master address, which the monitor renders only for an
427 # enabled store. So a store block is always present for those modes,
428 # defaulting enabled to True; an explicit store block still tunes
429 # offload, sizes, and the cold tier.
430 store_block = dict(store) if store is not None else None
431 if mode in ("store", "both"):
432 store_block = dict(store_block or {})
433 store_block.setdefault("enabled", True)
434 if store_block is not None:
435 # Author byte-size fields as canonical decimal strings up front so
436 # the persisted spec round-trips through DynamoDB without float or
437 # Decimal coercion. Authoring also fails fast on bad inputs.
438 for field in _MOONCAKE_STORE_BYTE_SIZE_FIELDS:
439 if field in store_block:
440 try:
441 store_block[field] = author_byte_size(store_block[field])
442 except ValueError as exc:
443 raise ValueError(f"mooncake.store.{field}: {exc}") from exc
444 block["store"] = store_block
446 if transfer is not None:
447 block["transfer"] = dict(transfer)
449 # Split modes are fronted by the prefill-decode proxy, which needs a
450 # container image. Default it to the same image the role pods serve from
451 # (the upstream vLLM image bundles the reference proxy) so a split deploy
452 # stands up without a separate proxy image; an explicit proxy image
453 # still wins.
454 proxy_block = dict(proxy) if proxy is not None else None
455 if mode in _MOONCAKE_DISAGGREGATED_MODES and default_proxy_image:
456 proxy_block = dict(proxy_block or {})
457 proxy_block.setdefault("image", default_proxy_image)
458 if proxy_block is not None:
459 block["proxy"] = proxy_block
461 if autoscaling is not None:
462 block["autoscaling"] = dict(autoscaling)
464 # Fail fast before persisting: rejects unsupported modes (naming the
465 # allowed values) and every other invalid field.
466 validate_mooncake_spec(block)
467 return block
469 def deploy(
470 self,
471 endpoint_name: str,
472 image: str | None = None,
473 target_regions: list[str] | None = None,
474 replicas: int = 1,
475 gpu_count: int = 1,
476 gpu_type: str | None = None,
477 port: int = 8000,
478 model_path: str | None = None,
479 model_source: str | None = None,
480 health_check_path: str = "/health",
481 env: dict[str, str] | None = None,
482 namespace: str = "gco-inference",
483 labels: dict[str, str] | None = None,
484 autoscaling: dict[str, Any] | None = None,
485 capacity_type: str | None = None,
486 extra_args: list[str] | None = None,
487 accelerator: str = "nvidia",
488 node_selector: dict[str, str] | None = None,
489 rewrite_image: bool = True,
490 *,
491 framework: str | None = None,
492 mooncake_mode: str | None = None,
493 prefill_replicas: int = 1,
494 decode_replicas: int = 1,
495 mooncake_store: dict[str, Any] | None = None,
496 mooncake_transfer: dict[str, Any] | None = None,
497 mooncake_proxy: dict[str, Any] | None = None,
498 mooncake_autoscaling: dict[str, Any] | None = None,
499 ) -> dict[str, Any]:
500 """
501 Deploy an inference endpoint to one or more regions.
503 The endpoint spec is written to DynamoDB. The inference_monitor
504 in each target region picks it up and creates the K8s resources.
506 Args:
507 endpoint_name: Unique name for the endpoint
508 image: Container image (e.g. vllm/vllm-openai:v0.29.0). Optional
509 when ``mooncake_mode`` is set: a disaggregated/store deploy
510 with no image falls back to the default upstream
511 Mooncake-enabled vLLM image. A plain deploy still requires an
512 image.
513 target_regions: Regions to deploy to (default: all deployed regions)
514 replicas: Number of replicas per region
515 gpu_count: GPUs per replica
516 gpu_type: GPU instance type hint for node selector
517 port: Container port
518 model_path: EFS path for model weights
519 health_check_path: Health check endpoint path
520 env: Environment variables
521 namespace: Kubernetes namespace
522 labels: Labels for the endpoint
523 rewrite_image: When True (the default), rewrite ECR URIs in
524 ``image`` to target each region's local replica. Non-ECR
525 refs (Docker Hub, GHCR, etc.) are left unchanged. When
526 False, the URI is written verbatim to every region's
527 spec — the operator is responsible for cross-region
528 pulls. Per-region rewrites are stored under a
529 ``region_overrides`` map on the spec keyed by region.
530 mooncake_mode: When set to one of ``disaggregated``, ``store``,
531 or ``both``, build and persist a ``spec.mooncake`` block for
532 disaggregated prefill/decode serving and/or a shared KV-cache
533 store. An unsupported value is rejected before anything is
534 persisted.
535 prefill_replicas: X in an XpYd topology — prefill instance count
536 for split (``disaggregated``/``both``) modes.
537 decode_replicas: Y in an XpYd topology — decode instance count for
538 split modes.
539 mooncake_store: Optional KV-store pool configuration merged into
540 ``spec.mooncake.store``. Byte-size fields are authored as
541 base-10 integer decimal strings so they round-trip through
542 DynamoDB.
543 mooncake_transfer: Optional Mooncake transfer intent and network
544 device. ``protocol`` accepts ``rdma`` (the default; scheduled
545 on EFA and rendered to vLLM as ``mooncake_protocol=efa``) or
546 ``tcp`` (no EFA placement). ``device_name`` is forwarded to
547 both the connector and mounted Mooncake configuration; an
548 empty string lets Mooncake auto-detect it.
549 mooncake_proxy: Optional PD proxy configuration merged into
550 ``spec.mooncake.proxy``.
551 mooncake_autoscaling: Optional per-role autoscaling configuration
552 merged into ``spec.mooncake.autoscaling``.
554 Returns:
555 Created endpoint record
556 """
557 if framework not in (None, "vllm", "tgi"):
558 raise ValueError("framework must be 'vllm' or 'tgi'")
559 if mooncake_mode is not None and framework == "tgi":
560 raise ValueError("Mooncake serving requires the vllm framework")
561 if mooncake_mode is not None and framework is None:
562 framework = "vllm"
564 # Build the optional mooncake block first and validate it before any
565 # persistence so a rejected block leaves any stored spec untouched. A
566 # disaggregated/store deploy without an explicit image falls back to
567 # the default upstream Mooncake-enabled vLLM image.
568 mooncake_block: dict[str, Any] | None = None
569 if mooncake_mode is not None:
570 # Resolve the image before building the block so a split mode's
571 # prefill-decode proxy can default to the same image the role pods
572 # serve from (the upstream vLLM image bundles the reference proxy).
573 if image is None:
574 from .images import default_disaggregated_image
576 image = default_disaggregated_image(config=self.config)
577 mooncake_block = self._build_mooncake_block(
578 mode=mooncake_mode,
579 prefill_replicas=prefill_replicas,
580 decode_replicas=decode_replicas,
581 store=mooncake_store,
582 transfer=mooncake_transfer,
583 proxy=mooncake_proxy,
584 autoscaling=mooncake_autoscaling,
585 default_proxy_image=image,
586 )
588 if image is None:
589 raise ValueError(
590 "an image is required (pass image, or set mooncake_mode to use "
591 "the default upstream Mooncake-enabled vLLM image)"
592 )
594 if not target_regions:
595 stacks = self._aws_client.discover_regional_stacks()
596 target_regions = list(stacks.keys())
597 if not target_regions:
598 raise ValueError("No deployed regions found. Deploy infrastructure first.")
600 # Per-region image-URI rewrites for ECR refs. Each target region
601 # gets the local replica's URI on its own spec, so the
602 # inference_monitor's pod-spec materialiser pulls in-region
603 # rather than across the WAN. Non-ECR URIs come back unchanged
604 # from the helper, so this is a no-op for Docker Hub / GHCR refs.
605 #
606 # The helper lives in ``cli._image_uri`` rather than ``cli.images``
607 # so this import doesn't create a module-level cycle:
608 # ``cli.images`` itself imports the same helper. ``cli._image_uri``
609 # is a leaf module with no project-side dependencies.
610 region_image_map: dict[str, str] = {}
611 if rewrite_image:
612 from ._image_uri import rewrite_image_uri_for_region
614 for region in target_regions:
615 region_image_map[region] = rewrite_image_uri_for_region(image, region)
617 spec = {
618 "image": image,
619 "port": port,
620 "replicas": replicas,
621 "gpu_count": gpu_count,
622 "health_check_path": health_check_path,
623 }
624 if framework:
625 spec["framework"] = framework
626 # Preserve the rewrite map on the spec so the inference_monitor
627 # service can pick the right URI per region when materialising
628 # pods. When ``rewrite_image=False`` no map is set and the flat
629 # ``image`` field is the only source.
630 if region_image_map and any(uri != image for uri in region_image_map.values()):
631 spec["region_image_uris"] = region_image_map
632 if gpu_type:
633 spec["gpu_type"] = gpu_type
634 if model_path:
635 spec["model_path"] = model_path
636 if model_source:
637 spec["model_source"] = model_source
638 if env:
639 spec["env"] = env
640 if autoscaling:
641 spec["autoscaling"] = autoscaling
642 if capacity_type:
643 spec["capacity_type"] = capacity_type
644 if extra_args:
645 spec["args"] = extra_args
646 if accelerator != "nvidia":
647 spec["accelerator"] = accelerator
648 if node_selector:
649 spec["node_selector"] = node_selector
650 if mooncake_block is not None:
651 spec["mooncake"] = mooncake_block
653 store = self._get_store()
654 result: dict[str, Any] = store.create_endpoint(
655 endpoint_name=endpoint_name,
656 spec=spec,
657 target_regions=target_regions,
658 namespace=namespace,
659 labels=labels,
660 )
661 return result
663 def list_endpoints(
664 self,
665 desired_state: str | None = None,
666 region: str | None = None,
667 ) -> list[dict[str, Any]]:
668 """List all inference endpoints."""
669 store = self._get_store()
670 result: list[dict[str, Any]] = store.list_endpoints(
671 desired_state=desired_state,
672 target_region=region,
673 )
674 return result
676 def get_endpoint(self, endpoint_name: str) -> dict[str, Any] | None:
677 """Get details of a specific endpoint."""
678 store = self._get_store()
679 result: dict[str, Any] | None = store.get_endpoint(endpoint_name)
680 return result
682 @staticmethod
683 def _load_mutable_endpoint(
684 store: InferenceEndpointStore,
685 endpoint_name: str,
686 operation: str,
687 ) -> tuple[dict[str, Any], str, str | None] | None:
688 """Read and lifecycle-fence an endpoint before an ordinary mutation."""
689 endpoint = store.get_endpoint(endpoint_name, consistent_read=True)
690 if endpoint is None:
691 return None
692 if endpoint.get("desired_state") == "deleted":
693 raise ValueError(
694 f"Endpoint '{endpoint_name}' is deleted and cannot be {operation}; "
695 "redeploy it after deletion completes."
696 )
697 lifecycle_id = endpoint.get("lifecycle_id")
698 if not isinstance(lifecycle_id, str) or not lifecycle_id:
699 migrated = store.ensure_lifecycle_metadata(endpoint)
700 if migrated is None:
701 raise ValueError(
702 f"Endpoint '{endpoint_name}' changed while initializing lifecycle identity"
703 )
704 endpoint = migrated
705 lifecycle_id = endpoint.get("lifecycle_id")
706 if not isinstance(lifecycle_id, str) or not lifecycle_id:
707 raise ValueError(f"Endpoint '{endpoint_name}' has no lifecycle identity")
708 updated_at = endpoint.get("updated_at")
709 return endpoint, lifecycle_id, updated_at if isinstance(updated_at, str) else None
711 @staticmethod
712 def _raise_write_conflict(endpoint_name: str, operation: str) -> None:
713 raise ValueError(
714 f"Endpoint '{endpoint_name}' changed while being {operation}; retry after reading status"
715 )
717 def scale(self, endpoint_name: str, replicas: int) -> dict[str, Any] | None:
718 """Scale a classic static endpoint only."""
719 store = self._get_store()
720 loaded = self._load_mutable_endpoint(store, endpoint_name, "scaled")
721 if loaded is None:
722 return None
723 endpoint, lifecycle_id, _updated_at = loaded
724 spec = endpoint.get("spec")
725 if isinstance(spec, dict) and "mooncake" in spec:
726 raise ValueError(
727 f"Endpoint '{endpoint_name}' uses Mooncake topology; use "
728 "'gco inference set-topology' instead of 'gco inference scale'."
729 )
730 autoscaling = spec.get("autoscaling") if isinstance(spec, dict) else None
731 if isinstance(autoscaling, dict) and autoscaling.get("enabled") is True:
732 raise ValueError(
733 f"Endpoint '{endpoint_name}' is autoscaled; update its min/max autoscaling "
734 "bounds or disable autoscaling before using 'gco inference scale'."
735 )
736 result: dict[str, Any] | None = store.scale_endpoint(
737 endpoint_name,
738 replicas,
739 expected_lifecycle_id=lifecycle_id,
740 )
741 if result is None:
742 self._raise_write_conflict(endpoint_name, "scaled")
743 return result
745 def set_topology(
746 self,
747 endpoint_name: str,
748 prefill: int,
749 decode: int,
750 ) -> dict[str, Any] | None:
751 """Resize a disaggregated endpoint's prefill/decode topology.
753 Updates ``spec.mooncake.topology`` to the new XpYd counts and
754 re-triggers reconciliation (via :meth:`InferenceEndpointStore.update_spec`,
755 which flips ``desired_state`` to ``deploying``) so the per-region
756 monitor adjusts the prefill and decode role replica counts.
758 Both counts must be integers in the inclusive range
759 ``[MOONCAKE_TOPOLOGY_MIN, MOONCAKE_TOPOLOGY_MAX]``. The counts are
760 validated before anything is read or written, so a rejected request
761 names the offending count and leaves the stored topology and
762 ``desired_state`` untouched.
764 Args:
765 endpoint_name: Name of the disaggregated endpoint to resize.
766 prefill: New prefill (X) instance count.
767 decode: New decode (Y) instance count.
769 Returns:
770 The updated endpoint record, or ``None`` when no endpoint with
771 ``endpoint_name`` exists.
773 Raises:
774 ValueError: when ``prefill`` or ``decode`` is not an integer in
775 ``[MOONCAKE_TOPOLOGY_MIN, MOONCAKE_TOPOLOGY_MAX]``.
776 """
777 # Validate before any read or write so a bad count names the offending
778 # field and leaves the stored topology and desired_state unchanged.
779 for field, count in (("prefill", prefill), ("decode", decode)):
780 if not _is_plain_int(count):
781 raise ValueError(
782 f"topology {field} count must be an integer in "
783 f"[{MOONCAKE_TOPOLOGY_MIN}, {MOONCAKE_TOPOLOGY_MAX}], "
784 f"got {count!r}"
785 )
786 if not MOONCAKE_TOPOLOGY_MIN <= count <= MOONCAKE_TOPOLOGY_MAX:
787 raise ValueError(
788 f"topology {field} count out of range "
789 f"[{MOONCAKE_TOPOLOGY_MIN}, {MOONCAKE_TOPOLOGY_MAX}]: {count}"
790 )
792 store = self._get_store()
793 loaded = self._load_mutable_endpoint(store, endpoint_name, "updated")
794 if loaded is None:
795 return None
796 endpoint, lifecycle_id, updated_at = loaded
798 raw_spec = endpoint.get("spec")
799 if not isinstance(raw_spec, dict):
800 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
801 spec = deepcopy(raw_spec)
802 # Preserve any existing mooncake sub-fields and replace only the
803 # topology counts.
804 mooncake = dict(spec.get("mooncake") or {})
805 mooncake["topology"] = {"prefill": prefill, "decode": decode}
806 spec["mooncake"] = mooncake
808 result: dict[str, Any] | None = store.update_spec(
809 endpoint_name,
810 spec,
811 expected_lifecycle_id=lifecycle_id,
812 expected_updated_at=updated_at,
813 )
814 if result is None:
815 self._raise_write_conflict(endpoint_name, "updated")
816 return result
818 def configure_store(
819 self,
820 endpoint_name: str,
821 store_config: dict[str, Any],
822 ) -> dict[str, Any] | None:
823 """Update an endpoint's KV-cache store configuration.
825 Merges ``store_config`` into ``spec.mooncake.store`` and re-triggers
826 reconciliation (via :meth:`InferenceEndpointStore.update_spec`, which
827 flips ``desired_state`` to ``deploying``) so the per-region monitor
828 picks up the new store settings.
830 Store byte-size fields are authored as base-10 integer decimal strings
831 (so they round-trip through DynamoDB without float/Decimal coercion)
832 and the resulting ``mooncake`` block is validated before anything is
833 written. A rejected configuration names the offending field and leaves
834 the stored spec untouched.
836 Args:
837 endpoint_name: Name of the endpoint to reconfigure.
838 store_config: KV-store pool settings merged into
839 ``spec.mooncake.store``.
841 Returns:
842 The updated endpoint record, or ``None`` when no endpoint with
843 ``endpoint_name`` exists.
845 Raises:
846 ValueError: when the resulting ``mooncake`` block is invalid (for
847 example an out-of-range byte-size field).
848 """
849 store = self._get_store()
850 loaded = self._load_mutable_endpoint(store, endpoint_name, "configured")
851 if loaded is None:
852 return None
853 endpoint, lifecycle_id, updated_at = loaded
855 raw_spec = endpoint.get("spec")
856 if not isinstance(raw_spec, dict):
857 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
858 spec = deepcopy(raw_spec)
859 # Preserve any existing mooncake sub-fields and replace only the store
860 # block, authoring byte-size fields as canonical decimal strings.
861 mooncake = dict(spec.get("mooncake") or {})
862 store_block = dict(store_config)
863 for field in _MOONCAKE_STORE_BYTE_SIZE_FIELDS:
864 if field in store_block:
865 try:
866 store_block[field] = author_byte_size(store_block[field])
867 except ValueError as exc:
868 raise ValueError(f"mooncake.store.{field}: {exc}") from exc
869 mooncake["store"] = store_block
870 spec["mooncake"] = mooncake
872 # Fail fast before persisting so a rejected block leaves the stored
873 # spec untouched.
874 validate_mooncake_spec(mooncake)
876 result: dict[str, Any] | None = store.update_spec(
877 endpoint_name,
878 spec,
879 expected_lifecycle_id=lifecycle_id,
880 expected_updated_at=updated_at,
881 )
882 if result is None:
883 self._raise_write_conflict(endpoint_name, "configured")
884 return result
886 def stop(self, endpoint_name: str) -> dict[str, Any] | None:
887 """Stop a live endpoint without crossing a deletion boundary."""
888 store = self._get_store()
889 loaded = self._load_mutable_endpoint(store, endpoint_name, "stopped")
890 if loaded is None:
891 return None
892 _endpoint, lifecycle_id, _updated_at = loaded
893 result: dict[str, Any] | None = store.update_desired_state(
894 endpoint_name,
895 "stopped",
896 expected_lifecycle_id=lifecycle_id,
897 )
898 if result is None:
899 self._raise_write_conflict(endpoint_name, "stopped")
900 return result
902 def start(self, endpoint_name: str) -> dict[str, Any] | None:
903 """Start a stopped endpoint; deleted endpoints must be redeployed."""
904 store = self._get_store()
905 result: dict[str, Any] | None = store.start_endpoint(endpoint_name)
906 return result
908 def delete(
909 self,
910 endpoint_name: str,
911 *,
912 expected_owner_label: tuple[str, str] | None = None,
913 expected_lifecycle_id: str | None = None,
914 ) -> dict[str, Any] | None:
915 """Mark an endpoint deleted with optional exact-incarnation ownership."""
916 if expected_owner_label is not None and not expected_lifecycle_id:
917 raise ValueError("Owner-conditioned deletion also requires an immutable lifecycle id")
918 store = self._get_store()
919 if expected_owner_label is None:
920 endpoint = store.get_endpoint(endpoint_name, consistent_read=True)
921 if endpoint is None:
922 return None
923 migrated = store.ensure_lifecycle_metadata(endpoint)
924 if migrated is None:
925 raise ValueError(
926 f"Endpoint '{endpoint_name}' changed while initializing deletion identity"
927 )
928 lifecycle_id = migrated.get("lifecycle_id")
929 if not isinstance(lifecycle_id, str) or not lifecycle_id:
930 raise ValueError(f"Endpoint '{endpoint_name}' has no lifecycle identity")
931 result: dict[str, Any] | None = store.update_desired_state(
932 endpoint_name,
933 "deleted",
934 expected_lifecycle_id=lifecycle_id,
935 )
936 else:
937 result = store.update_desired_state(
938 endpoint_name,
939 "deleted",
940 expected_label=expected_owner_label,
941 expected_lifecycle_id=expected_lifecycle_id,
942 )
943 return result
945 def update_image(self, endpoint_name: str, image: str) -> dict[str, Any] | None:
946 """Update the container image for an endpoint."""
947 if not isinstance(image, str) or not image.strip():
948 raise ValueError("Image must be a non-empty string")
950 store = self._get_store()
951 loaded = self._load_mutable_endpoint(store, endpoint_name, "updated")
952 if loaded is None:
953 return None
954 endpoint, lifecycle_id, updated_at = loaded
955 raw_spec = endpoint.get("spec")
956 if not isinstance(raw_spec, dict):
957 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
958 spec = deepcopy(raw_spec)
959 spec["image"] = image.strip()
960 # A direct image update is global. Stale regional rewrites would take
961 # precedence in the monitor and silently keep serving the old image.
962 spec.pop("region_image_uris", None)
963 result: dict[str, Any] | None = store.update_spec(
964 endpoint_name,
965 spec,
966 expected_lifecycle_id=lifecycle_id,
967 expected_updated_at=updated_at,
968 )
969 if result is None:
970 self._raise_write_conflict(endpoint_name, "updated")
971 return result
973 def add_region(self, endpoint_name: str, region: str) -> dict[str, Any] | None:
974 """Add a target and append it to this lifecycle's cleanup history."""
975 store = self._get_store()
976 loaded = self._load_mutable_endpoint(store, endpoint_name, "added to a Region")
977 if loaded is None:
978 return None
979 endpoint, lifecycle_id, updated_at = loaded
980 if updated_at is None:
981 raise ValueError(f"Endpoint '{endpoint_name}' has no conditional update timestamp")
982 regions = list(endpoint.get("target_regions") or [])
983 cleanup_regions = list(endpoint.get("cleanup_regions") or regions)
984 if region in regions:
985 return endpoint
986 regions.append(region)
987 if region not in cleanup_regions:
988 cleanup_regions.append(region)
989 raw_generations = endpoint.get("region_generations")
990 stored_generations = raw_generations if isinstance(raw_generations, dict) else {}
991 region_generations = {
992 cleanup_region: (
993 stored_generations[cleanup_region]
994 if isinstance(stored_generations.get(cleanup_region), str)
995 and stored_generations[cleanup_region]
996 else secrets.token_hex(32)
997 )
998 for cleanup_region in cleanup_regions
999 }
1000 # Every target membership transition gets a fresh token, even when the
1001 # Region appeared earlier in cleanup history. This invalidates a prior
1002 # terminal removal acknowledgement before resources can be recreated.
1003 region_generations[region] = secrets.token_hex(32)
1004 try:
1005 result = store.update_target_regions(
1006 endpoint_name,
1007 regions,
1008 cleanup_regions,
1009 region_generations,
1010 expected_lifecycle_id=lifecycle_id,
1011 expected_updated_at=updated_at,
1012 )
1013 except Exception as e:
1014 logger.error("Failed to add region: %s", e)
1015 return None
1016 if result is None:
1017 self._raise_write_conflict(endpoint_name, "added to a Region")
1018 return result
1020 def remove_region(self, endpoint_name: str, region: str) -> dict[str, Any] | None:
1021 """Remove a target without erasing its authoritative cleanup history."""
1022 store = self._get_store()
1023 loaded = self._load_mutable_endpoint(store, endpoint_name, "removed from a Region")
1024 if loaded is None:
1025 return None
1026 endpoint, lifecycle_id, updated_at = loaded
1027 if updated_at is None:
1028 raise ValueError(f"Endpoint '{endpoint_name}' has no conditional update timestamp")
1029 regions = list(endpoint.get("target_regions") or [])
1030 cleanup_regions = list(endpoint.get("cleanup_regions") or regions)
1031 if region not in regions:
1032 return endpoint
1033 regions.remove(region)
1034 if region not in cleanup_regions:
1035 cleanup_regions.append(region)
1036 raw_generations = endpoint.get("region_generations")
1037 stored_generations = raw_generations if isinstance(raw_generations, dict) else {}
1038 region_generations = {
1039 cleanup_region: (
1040 stored_generations[cleanup_region]
1041 if isinstance(stored_generations.get(cleanup_region), str)
1042 and stored_generations[cleanup_region]
1043 else secrets.token_hex(32)
1044 )
1045 for cleanup_region in cleanup_regions
1046 }
1047 # The cleanup pass must acknowledge this removal, not any earlier
1048 # remove/re-add cycle for the same Region.
1049 region_generations[region] = secrets.token_hex(32)
1050 try:
1051 result = store.update_target_regions(
1052 endpoint_name,
1053 regions,
1054 cleanup_regions,
1055 region_generations,
1056 expected_lifecycle_id=lifecycle_id,
1057 expected_updated_at=updated_at,
1058 )
1059 except Exception as e:
1060 logger.error("Failed to remove region: %s", e)
1061 return None
1062 if result is None:
1063 self._raise_write_conflict(endpoint_name, "removed from a Region")
1064 return result
1066 def canary_deploy(
1067 self,
1068 endpoint_name: str,
1069 image: str,
1070 weight: int = 10,
1071 replicas: int = 1,
1072 ) -> dict[str, Any] | None:
1073 """Start a canary deployment for an existing classic endpoint.
1075 Creates a canary variant with the new image receiving ``weight``%
1076 of traffic. Mooncake endpoints are excluded because their split-role
1077 topology cannot be represented by the classic canary Deployment.
1079 Args:
1080 endpoint_name: Existing endpoint to canary
1081 image: New container image for the canary
1082 weight: Percentage of traffic to route to canary (1-99)
1083 replicas: Positive number of canary replicas
1085 Returns:
1086 Updated endpoint record, or None if endpoint not found
1087 """
1088 if not isinstance(image, str) or not image.strip():
1089 raise ValueError("Canary image must be a non-empty string")
1090 if not _is_plain_int(weight) or not 1 <= weight <= 99:
1091 raise ValueError("Canary weight must be an integer between 1 and 99")
1092 if not _is_plain_int(replicas) or replicas < 1:
1093 raise ValueError("Canary replicas must be a positive integer")
1095 store = self._get_store()
1096 loaded = self._load_mutable_endpoint(store, endpoint_name, "updated")
1097 if loaded is None:
1098 return None
1099 endpoint, lifecycle_id, updated_at = loaded
1101 if endpoint.get("desired_state") not in ("running", "deploying"):
1102 raise ValueError(
1103 f"Cannot canary an endpoint in '{endpoint.get('desired_state')}' state. "
1104 "Endpoint must be running or deploying."
1105 )
1107 raw_spec = endpoint.get("spec")
1108 if not isinstance(raw_spec, dict):
1109 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
1110 if "mooncake" in raw_spec:
1111 raise ValueError("Canary deployments are not supported for Mooncake endpoints")
1113 # Never mutate the object returned by the store; callers and test
1114 # doubles may retain it as shared state.
1115 spec = deepcopy(raw_spec)
1116 spec["canary"] = {
1117 "image": image.strip(),
1118 "weight": weight,
1119 "replicas": replicas,
1120 }
1122 result: dict[str, Any] | None = store.update_spec(
1123 endpoint_name,
1124 spec,
1125 expected_lifecycle_id=lifecycle_id,
1126 expected_updated_at=updated_at,
1127 )
1128 if result is None:
1129 self._raise_write_conflict(endpoint_name, "updated")
1130 return result
1132 def promote_canary(self, endpoint_name: str) -> dict[str, Any] | None:
1133 """Promote a classic canary to primary and remove its deployment."""
1134 store = self._get_store()
1135 loaded = self._load_mutable_endpoint(store, endpoint_name, "updated")
1136 if loaded is None:
1137 return None
1138 endpoint, lifecycle_id, updated_at = loaded
1140 raw_spec = endpoint.get("spec")
1141 if not isinstance(raw_spec, dict):
1142 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
1143 if "mooncake" in raw_spec:
1144 raise ValueError("Canary promotion is not supported for Mooncake endpoints")
1146 canary = raw_spec.get("canary")
1147 if not isinstance(canary, dict):
1148 raise ValueError(f"Endpoint '{endpoint_name}' has no active canary deployment")
1149 if "image" not in canary:
1150 raise ValueError(
1151 f"Canary deployment for '{endpoint_name}' is missing the 'image' field"
1152 )
1153 canary_image = canary["image"]
1154 if not isinstance(canary_image, str) or not canary_image.strip():
1155 raise ValueError(
1156 f"Canary deployment for '{endpoint_name}' has an invalid 'image' field"
1157 )
1159 spec = deepcopy(raw_spec)
1160 spec["image"] = canary_image.strip()
1161 spec.pop("canary", None)
1162 # The canary image is explicit and global. Existing per-region primary
1163 # rewrites point at the superseded image and must not take precedence.
1164 spec.pop("region_image_uris", None)
1166 result: dict[str, Any] | None = store.update_spec(
1167 endpoint_name,
1168 spec,
1169 expected_lifecycle_id=lifecycle_id,
1170 expected_updated_at=updated_at,
1171 )
1172 if result is None:
1173 self._raise_write_conflict(endpoint_name, "updated")
1174 return result
1176 def rollback_canary(self, endpoint_name: str) -> dict[str, Any] | None:
1177 """Remove the canary deployment, keeping the primary unchanged."""
1178 store = self._get_store()
1179 loaded = self._load_mutable_endpoint(store, endpoint_name, "updated")
1180 if loaded is None:
1181 return None
1182 endpoint, lifecycle_id, updated_at = loaded
1184 raw_spec = endpoint.get("spec")
1185 if not isinstance(raw_spec, dict):
1186 raise ValueError(f"Endpoint '{endpoint_name}' has an invalid spec")
1187 if "canary" not in raw_spec:
1188 raise ValueError(f"Endpoint '{endpoint_name}' has no active canary deployment")
1190 # Rollback is deliberately allowed for a legacy invalid
1191 # Mooncake-plus-canary record so an operator can repair it.
1192 spec = deepcopy(raw_spec)
1193 spec.pop("canary", None)
1194 result: dict[str, Any] | None = store.update_spec(
1195 endpoint_name,
1196 spec,
1197 expected_lifecycle_id=lifecycle_id,
1198 expected_updated_at=updated_at,
1199 )
1200 if result is None:
1201 self._raise_write_conflict(endpoint_name, "updated")
1202 return result
1205def get_inference_manager(config: GCOConfig | None = None) -> InferenceManager:
1206 """Factory function for InferenceManager."""
1207 return InferenceManager(config)