vllm_mlx.engine_core

Engine Core for vllm-mlx continuous batching.

View the complete module source at #L1-L794.

API details

Each callable below includes its exact signature, type annotations, inputs, defaults, return contract, documented exceptions, implementation source, and parsed docstring sections when the source provides them.

vllm_mlx.engine_core

Engine Core for vllm-mlx continuous batching.

This module provides the EngineCore class that coordinates: - Model loading and management - Request scheduling via Scheduler - Async request processing - Output streaming

The design follows vLLM's engine architecture adapted for MLX.

vllm_mlx.engine_core.logger module-attribute

logger = logging.getLogger(__name__)

vllm_mlx.engine_core.EngineConfig dataclass

EngineConfig(model_name: str = '', scheduler_config: Optional[SchedulerConfig] = None, step_interval: float = 0.001, stream_interval: int = 1, gpu_memory_utilization: float = 0.9)

Configuration for the engine.

vllm_mlx.engine_core.EngineConfig.model_name class-attribute instance-attribute

model_name: str = ''

vllm_mlx.engine_core.EngineConfig.scheduler_config class-attribute instance-attribute

scheduler_config: Optional[SchedulerConfig] = None

vllm_mlx.engine_core.EngineConfig.step_interval class-attribute instance-attribute

step_interval: float = 0.001

vllm_mlx.engine_core.EngineConfig.stream_interval class-attribute instance-attribute

stream_interval: int = 1

vllm_mlx.engine_core.EngineConfig.gpu_memory_utilization class-attribute instance-attribute

gpu_memory_utilization: float = 0.9

vllm_mlx.engine_core.EngineCore

EngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None, engine_id: Optional[str] = None, force_model_ownership: bool = True)

Core engine for vllm-mlx inference with continuous batching.

This engine runs the generation loop and manages request lifecycle. It provides both sync and async interfaces for request handling.

Initialize the engine.

Parameters:

• model (Any) –

  The MLX model

• tokenizer (Any) –

  The tokenizer

• config (Optional[EngineConfig], default: None ) –

  Engine configuration

• engine_id (Optional[str], default: None ) –

  Optional unique ID for this engine (auto-generated if None)

• force_model_ownership (bool, default: True ) –

  If True (default), forcibly take model ownership from any existing engine. If False, raises ModelOwnershipError if model is in use.

Source code in vllm_mlx/engine_core.py

def __init__(
    self,
    model: Any,
    tokenizer: Any,
    config: Optional[EngineConfig] = None,
    engine_id: Optional[str] = None,
    force_model_ownership: bool = True,
):
    """
    Initialize the engine.

    Args:
        model: The MLX model
        tokenizer: The tokenizer
        config: Engine configuration
        engine_id: Optional unique ID for this engine (auto-generated if None)
        force_model_ownership: If True (default), forcibly take model ownership
                               from any existing engine. If False, raises
                               ModelOwnershipError if model is in use.
    """
    self.model = model
    self.tokenizer = tokenizer
    self.config = config or EngineConfig()
    self._engine_id = engine_id or str(uuid.uuid4())
    self._owns_model = False
    self._closed = False

    # Acquire model ownership
    registry = get_registry()
    registry.acquire(
        model=model,
        engine=self,
        engine_id=self._engine_id,
        force=force_model_ownership,
    )
    self._owns_model = True

    # Create scheduler
    scheduler_config = self.config.scheduler_config or SchedulerConfig()
    self.scheduler = Scheduler(
        model=model,
        tokenizer=tokenizer,
        config=scheduler_config,
    )

    # Output collectors for low-latency streaming (vLLM pattern)
    self._output_collectors: Dict[str, RequestOutputCollector] = {}
    self._stream_states: Dict[str, RequestStreamState] = {}
    self._finished_events: Dict[str, asyncio.Event] = {}

    # Engine state
    self._running = False
    self._task: Optional[asyncio.Task] = None
    self._start_time: Optional[float] = None
    self._steps_executed = 0

    logger.debug(f"Engine {self._engine_id} initialized")

vllm_mlx.engine_core.EngineCore.model instance-attribute

model = model

vllm_mlx.engine_core.EngineCore.tokenizer instance-attribute

tokenizer = tokenizer

vllm_mlx.engine_core.EngineCore.config instance-attribute

config = config or EngineConfig()

vllm_mlx.engine_core.EngineCore._engine_id instance-attribute

_engine_id = engine_id or str(uuid.uuid4())

vllm_mlx.engine_core.EngineCore._closed instance-attribute

_closed = False

vllm_mlx.engine_core.EngineCore._owns_model instance-attribute

_owns_model = True

vllm_mlx.engine_core.EngineCore.scheduler instance-attribute

scheduler = Scheduler(model=model, tokenizer=tokenizer, config=scheduler_config)

vllm_mlx.engine_core.EngineCore._output_collectors instance-attribute

_output_collectors: Dict[str, RequestOutputCollector] = {}

vllm_mlx.engine_core.EngineCore._stream_states instance-attribute

_stream_states: Dict[str, RequestStreamState] = {}

vllm_mlx.engine_core.EngineCore._finished_events instance-attribute

_finished_events: Dict[str, Event] = {}

vllm_mlx.engine_core.EngineCore._running instance-attribute

_running = False

vllm_mlx.engine_core.EngineCore._task instance-attribute

_task: Optional[Task] = None

vllm_mlx.engine_core.EngineCore._start_time instance-attribute

_start_time: Optional[float] = None

vllm_mlx.engine_core.EngineCore._steps_executed instance-attribute

_steps_executed = 0

vllm_mlx.engine_core.EngineCore.engine_id property

engine_id: str

Get the engine ID.

vllm_mlx.engine_core.EngineCore.start async

start() -> None

Start the engine loop.

Source code in vllm_mlx/engine_core.py

async def start(self) -> None:
    """Start the engine loop."""
    if self._running:
        return

    self._running = True
    self._start_time = time.time()
    self._task = asyncio.create_task(self._engine_loop())
    logger.info("Engine started")

vllm_mlx.engine_core.EngineCore.stop async

stop() -> None

Stop the engine loop.

Source code in vllm_mlx/engine_core.py

async def stop(self) -> None:
    """Stop the engine loop."""
    self._running = False
    if self._task:
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass
        self._task = None
    # Safety net: close batch generator if _engine_loop didn't get a
    # chance to clean up (e.g. it was never started).  The call is
    # idempotent — _close_batch_generator checks for None.
    self.scheduler._close_batch_generator()
    logger.info("Engine stopped")

vllm_mlx.engine_core.EngineCore.is_running

is_running() -> bool

Check if engine is running.

Source code in vllm_mlx/engine_core.py

def is_running(self) -> bool:
    """Check if engine is running."""
    return self._running

vllm_mlx.engine_core.EngineCore._engine_loop async

_engine_loop() -> None

Main engine loop.

scheduler.step runs on one dedicated worker thread. MLX streams are thread-local, so we rebind generation streams inside that worker.

Source code in vllm_mlx/engine_core.py

async def _engine_loop(self) -> None:
    """Main engine loop.

    scheduler.step runs on one dedicated worker thread. MLX streams are
    thread-local, so we rebind generation streams inside that worker.
    """

    loop = asyncio.get_running_loop()
    worker = ThreadPoolExecutor(max_workers=1, thread_name_prefix="engine-core")
    worker_stream_bound = False
    model_thread_stream_bound = False
    use_worker_thread = True
    stream_thread_fallback_used = False

    def _bind_worker_streams_once() -> None:
        nonlocal worker_stream_bound
        if not worker_stream_bound:
            bind_generation_streams()
            worker_stream_bound = True

    def _bind_model_streams_once() -> None:
        nonlocal model_thread_stream_bound
        if not model_thread_stream_bound:
            bind_generation_streams()
            model_thread_stream_bound = True

    def _step_on_worker():
        _bind_worker_streams_once()
        output = self.scheduler.step()
        self._steps_executed += 1

        if self._steps_executed % _memory_check_interval == 0:
            try:
                active_mem = mx.get_active_memory()
                if active_mem > _memory_pressure_threshold:
                    mx.clear_cache()
                    logger.warning(
                        f"[Memory pressure] {active_mem / 1e9:.1f}GB > "
                        f"{_memory_pressure_threshold / 1e9:.0f}GB threshold, "
                        f"forced cache clear"
                    )
            except Exception:
                pass

        return output

    def _step_on_model_thread():
        _bind_model_streams_once()
        output = self.scheduler.step()
        self._steps_executed += 1

        if self._steps_executed % _memory_check_interval == 0:
            try:
                active_mem = mx.get_active_memory()
                if active_mem > _memory_pressure_threshold:
                    mx.clear_cache()
                    logger.warning(
                        f"[Memory pressure] {active_mem / 1e9:.1f}GB > "
                        f"{_memory_pressure_threshold / 1e9:.0f}GB threshold, "
                        f"forced cache clear"
                    )
            except Exception:
                pass

        return output

    def _recover_stream_thread_error_on_worker() -> None:
        _bind_worker_streams_once()
        self.scheduler._recover_from_cache_error()
        self.scheduler._reschedule_running_requests()

    def _clear_cache_on_worker() -> None:
        _bind_worker_streams_once()
        mx.clear_cache()

    def _close_batch_generator_on_worker() -> None:
        _bind_worker_streams_once()
        self.scheduler._close_batch_generator()

    step_interval = self.config.step_interval
    stream_interval = self.config.stream_interval
    use_simple_streaming = stream_interval == 1

    # Emergency memory pressure threshold — dynamic based on gpu_memory_utilization
    _gpu_mem_util = self.config.gpu_memory_utilization
    try:
        _device_mem = mx.device_info().get("memory_size", 200 * 1024 * 1024 * 1024)
        _memory_pressure_threshold = int(
            _device_mem * min(_gpu_mem_util + 0.05, 0.99)
        )
    except Exception:
        _memory_pressure_threshold = 200 * 1024 * 1024 * 1024
    _memory_check_interval = 64

    try:
        while self._running:
            try:
                if self.scheduler.has_requests():
                    if use_worker_thread:
                        try:
                            output = await loop.run_in_executor(
                                worker, _step_on_worker
                            )
                        except Exception as e:
                            if (
                                _is_stream_thread_error(e)
                                and not stream_thread_fallback_used
                            ):
                                await loop.run_in_executor(
                                    worker, _recover_stream_thread_error_on_worker
                                )
                                use_worker_thread = False
                                stream_thread_fallback_used = True
                                _bind_model_streams_once()
                                logger.warning(
                                    "Detected MLX stream/thread mismatch on worker "
                                    "step; switched this engine to model-thread stepping"
                                )
                                continue
                            raise
                    else:
                        output = _step_on_model_thread()
                    # Yield to event loop after each step.
                    await asyncio.sleep(0)

                    # Fast path: distribute outputs to collectors
                    outputs = output.outputs
                    if outputs:
                        collectors = self._output_collectors
                        states = self._stream_states
                        events = self._finished_events

                        for req_output in outputs:
                            rid = req_output.request_id
                            collector = collectors.get(rid)

                            if collector is not None:
                                # Optimized: skip stream_interval check when interval=1
                                if use_simple_streaming:
                                    collector.put(req_output)
                                else:
                                    state = states.get(rid)
                                    if state and state.should_send(
                                        req_output.completion_tokens,
                                        req_output.finished,
                                    ):
                                        collector.put(req_output)
                                        state.mark_sent(
                                            req_output.completion_tokens
                                        )

                            if req_output.finished:
                                event = events.get(rid)
                                if event:
                                    event.set()

                        # Free Metal buffers after distributing finished outputs
                        if output.finished_request_ids:
                            if use_worker_thread:
                                await loop.run_in_executor(
                                    worker, _clear_cache_on_worker
                                )
                            else:
                                mx.clear_cache()

                        # Always yield to prevent event loop starvation.
                        # Without this, orphaned requests (client disconnected but
                        # request still in scheduler) block the entire event loop,
                        # making the server unresponsive to all HTTP requests.
                        await asyncio.sleep(0)
                else:
                    # No work, yield control
                    await asyncio.sleep(step_interval)

            except asyncio.CancelledError:
                raise
            except Exception as e:
                import traceback

                logger.error(f"Engine loop error: {e}\n{traceback.format_exc()}")
                await asyncio.sleep(0.1)
    finally:
        try:
            if use_worker_thread:
                await loop.run_in_executor(worker, _close_batch_generator_on_worker)
            else:
                self.scheduler._close_batch_generator()
        finally:
            worker.shutdown(wait=True)

vllm_mlx.engine_core.EngineCore.add_request async

add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, images: Optional[List[Any]] = None, videos: Optional[List[Any]] = None, prefix_boundary: int = 0) -> str

Add a request for processing.

Parameters:

• prompt (Union[str, List[int]]) –

  Input prompt (string or token IDs)

• sampling_params (Optional[SamplingParams], default: None ) –

  Generation parameters

• request_id (Optional[str], default: None ) –

  Optional custom request ID

• images (Optional[List[Any]], default: None ) –

  Optional images for multimodal

• videos (Optional[List[Any]], default: None ) –

  Optional videos for multimodal

• prefix_boundary (int, default: 0 ) –

  Token count for shared prefix (for cache)

Returns:

• str –

  The request ID

Source code in vllm_mlx/engine_core.py

async def add_request(
    self,
    prompt: Union[str, List[int]],
    sampling_params: Optional[SamplingParams] = None,
    request_id: Optional[str] = None,
    images: Optional[List[Any]] = None,
    videos: Optional[List[Any]] = None,
    prefix_boundary: int = 0,
) -> str:
    """
    Add a request for processing.

    Args:
        prompt: Input prompt (string or token IDs)
        sampling_params: Generation parameters
        request_id: Optional custom request ID
        images: Optional images for multimodal
        videos: Optional videos for multimodal
        prefix_boundary: Token count for shared prefix (for cache)

    Returns:
        The request ID
    """
    if request_id is None:
        request_id = str(uuid.uuid4())

    if sampling_params is None:
        sampling_params = SamplingParams()

    request = Request(
        request_id=request_id,
        prompt=prompt,
        sampling_params=sampling_params,
        images=images,
        videos=videos,
        prefix_boundary=prefix_boundary,
    )

    # Setup output collector with stream_interval from config
    self._output_collectors[request_id] = RequestOutputCollector(aggregate=True)
    self._stream_states[request_id] = RequestStreamState(
        stream_interval=self.config.stream_interval
    )
    self._finished_events[request_id] = asyncio.Event()

    # Add to scheduler
    self.scheduler.add_request(request)

    return request_id

vllm_mlx.engine_core.EngineCore.abort_request async

abort_request(request_id: str) -> bool

Abort a request.

Source code in vllm_mlx/engine_core.py

async def abort_request(self, request_id: str) -> bool:
    """Abort a request."""
    result = self.scheduler.abort_request(request_id)
    self._cleanup_request(request_id)
    return result

vllm_mlx.engine_core.EngineCore._cleanup_request

_cleanup_request(request_id: str) -> None

Clean up request tracking.

Source code in vllm_mlx/engine_core.py

def _cleanup_request(self, request_id: str) -> None:
    """Clean up request tracking."""
    collector = self._output_collectors.pop(request_id, None)
    if collector:
        collector.clear()
    self._stream_states.pop(request_id, None)
    self._finished_events.pop(request_id, None)
    self.scheduler.remove_finished_request(request_id)

vllm_mlx.engine_core.EngineCore.stream_outputs async

stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]

Stream outputs for a request with low-latency non-blocking pattern.

Uses the vLLM pattern: get_nowait() or await get() This avoids unnecessary task switches when output is available.

Parameters:

• request_id (str) –

  The request ID

• timeout (Optional[float], default: None ) –

  Optional timeout in seconds

Yields:

• AsyncIterator[RequestOutput] –

  RequestOutput objects as tokens are generated

Source code in vllm_mlx/engine_core.py

async def stream_outputs(
    self,
    request_id: str,
    timeout: Optional[float] = None,
) -> AsyncIterator[RequestOutput]:
    """
    Stream outputs for a request with low-latency non-blocking pattern.

    Uses the vLLM pattern: get_nowait() or await get()
    This avoids unnecessary task switches when output is available.

    Args:
        request_id: The request ID
        timeout: Optional timeout in seconds

    Yields:
        RequestOutput objects as tokens are generated
    """
    import time as _time

    _t0 = _time.monotonic()
    _token_count = 0

    collector = self._output_collectors.get(request_id)
    if collector is None:
        logger.warning(
            f"[stream_outputs] {request_id[:12]} no collector found, returning immediately"
        )
        return

    logger.info(f"[stream_outputs] {request_id[:12]} START waiting for tokens")

    finished_normally = False
    try:
        while True:
            try:
                if timeout:
                    output = collector.get_nowait()
                    if output is None:
                        output = await asyncio.wait_for(
                            collector.get(), timeout=timeout
                        )
                else:
                    output = collector.get_nowait() or await collector.get()

                _token_count += 1
                if _token_count == 1:
                    logger.info(
                        f"[stream_outputs] {request_id[:12]} first token after "
                        f"{_time.monotonic() - _t0:.1f}s"
                    )

                if output.finished:
                    finished_normally = True
                    logger.info(
                        f"[stream_outputs] {request_id[:12]} finished normally, "
                        f"{_token_count} tokens in {_time.monotonic() - _t0:.1f}s"
                    )
                    yield output
                    break

                yield output

            except asyncio.TimeoutError:
                logger.warning(
                    f"[stream_outputs] {request_id[:12]} TIMEOUT after "
                    f"{_token_count} tokens, {_time.monotonic() - _t0:.1f}s"
                )
                break

    except (GeneratorExit, asyncio.CancelledError) as exc:
        logger.info(
            f"[stream_outputs] {request_id[:12]} {type(exc).__name__} after "
            f"{_token_count} tokens, {_time.monotonic() - _t0:.1f}s"
        )

    finally:
        if not finished_normally:
            logger.info(
                f"[stream_outputs] {request_id[:12]} ABORTING orphaned request "
                f"({_token_count} tokens generated in {_time.monotonic() - _t0:.1f}s)"
            )
            aborted = self.scheduler.abort_request(request_id)
            logger.info(
                f"[stream_outputs] {request_id[:12]} abort_request returned {aborted}"
            )
        self._cleanup_request(request_id)
        logger.info(f"[stream_outputs] {request_id[:12]} cleanup done")

vllm_mlx.engine_core.EngineCore.generate async

generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> RequestOutput

Generate a complete response (non-streaming).

This method is optimized to avoid streaming overhead when you only need the final result.

Parameters:

• prompt (Union[str, List[int]]) –

  Input prompt

• sampling_params (Optional[SamplingParams], default: None ) –

  Generation parameters

• request_id (Optional[str], default: None ) –

  Optional request ID

Returns:

• RequestOutput –

  Final RequestOutput with complete text

Source code in vllm_mlx/engine_core.py

async def generate(
    self,
    prompt: Union[str, List[int]],
    sampling_params: Optional[SamplingParams] = None,
    request_id: Optional[str] = None,
    **kwargs,
) -> RequestOutput:
    """
    Generate a complete response (non-streaming).

    This method is optimized to avoid streaming overhead when
    you only need the final result.

    Args:
        prompt: Input prompt
        sampling_params: Generation parameters
        request_id: Optional request ID

    Returns:
        Final RequestOutput with complete text
    """
    request_id = await self.add_request(
        prompt=prompt,
        sampling_params=sampling_params,
        request_id=request_id,
        **kwargs,
    )

    # Wait for completion using event instead of streaming
    # This avoids the waiting_consumer tracking overhead
    event = self._finished_events.get(request_id)
    if event is None:
        raise RuntimeError(f"No event for request {request_id}")

    try:
        # Wait for the request to finish
        await event.wait()

        # Get the final output from collector
        collector = self._output_collectors.get(request_id)
        if collector is None:
            raise RuntimeError(f"No collector for request {request_id}")

        # Drain all outputs and get the last one
        final_output = None
        while True:
            output = collector.get_nowait()
            if output is None:
                break
            final_output = output

        if final_output is None:
            raise RuntimeError(f"No output for request {request_id}")

        return final_output

    except (asyncio.CancelledError, GeneratorExit):
        logger.info(f"[generate] {request_id[:12]} CANCELLED, aborting request")
        self.scheduler.abort_request(request_id)
        raise

    finally:
        self._cleanup_request(request_id)

vllm_mlx.engine_core.EngineCore.generate_batch_sync

generate_batch_sync(prompts: List[Union[str, List[int]]], sampling_params: Optional[SamplingParams] = None) -> List[RequestOutput]

Generate responses synchronously for maximum throughput.

This bypasses the async engine loop entirely, running the scheduler directly for optimal batching performance. Use this when you don't need streaming and want maximum throughput.

Parameters:

• prompts (List[Union[str, List[int]]]) –

  List of input prompts

• sampling_params (Optional[SamplingParams], default: None ) –

  Generation parameters (same for all)

Returns:

• List[RequestOutput] –

  List of RequestOutput in same order as prompts

Source code in vllm_mlx/engine_core.py

def generate_batch_sync(
    self,
    prompts: List[Union[str, List[int]]],
    sampling_params: Optional[SamplingParams] = None,
) -> List[RequestOutput]:
    """
    Generate responses synchronously for maximum throughput.

    This bypasses the async engine loop entirely, running the scheduler
    directly for optimal batching performance. Use this when you don't
    need streaming and want maximum throughput.

    Args:
        prompts: List of input prompts
        sampling_params: Generation parameters (same for all)

    Returns:
        List of RequestOutput in same order as prompts
    """
    from .request import Request
    import uuid as uuid_module

    if sampling_params is None:
        sampling_params = SamplingParams()

    # Add all requests to scheduler
    request_ids = []
    for prompt in prompts:
        request_id = str(uuid_module.uuid4())
        request = Request(
            request_id=request_id,
            prompt=prompt,
            sampling_params=sampling_params,
        )
        self.scheduler.add_request(request)
        request_ids.append(request_id)

    # Bind MLX generation streams to the calling thread so that
    # scheduler.step() can evaluate KV cache state without hitting
    # "There is no Stream(gpu, N) in current thread" errors.
    bind_generation_streams()

    # Process until all done - direct scheduler access, no async overhead
    results: Dict[str, RequestOutput] = {}
    while self.scheduler.has_requests():
        output = self.scheduler.step()
        for req_output in output.outputs:
            if req_output.finished:
                results[req_output.request_id] = req_output

    # Cleanup
    for rid in request_ids:
        self.scheduler.remove_finished_request(rid)

    # Return in original order
    return [results[rid] for rid in request_ids]

vllm_mlx.engine_core.EngineCore.get_stats

get_stats() -> Dict[str, Any]

Get engine statistics.

Source code in vllm_mlx/engine_core.py

def get_stats(self) -> Dict[str, Any]:
    """Get engine statistics."""
    scheduler_stats = self.scheduler.get_stats()
    uptime = time.time() - self._start_time if self._start_time else 0

    return {
        "running": self._running,
        "uptime_seconds": uptime,
        "steps_executed": self._steps_executed,
        "active_requests": len(self._output_collectors),
        "stream_interval": self.config.stream_interval,
        "requests": self.scheduler.get_running_requests_info(),
        **scheduler_stats,
    }

vllm_mlx.engine_core.EngineCore.get_cache_stats

get_cache_stats() -> Optional[Dict[str, Any]]

Get prefix cache statistics.

Source code in vllm_mlx/engine_core.py

def get_cache_stats(self) -> Optional[Dict[str, Any]]:
    """Get prefix cache statistics."""
    return self.scheduler.get_cache_stats()

vllm_mlx.engine_core.EngineCore.save_cache_to_disk

save_cache_to_disk(cache_dir: str) -> bool

Save prefix cache to disk.

Source code in vllm_mlx/engine_core.py

def save_cache_to_disk(self, cache_dir: str) -> bool:
    """Save prefix cache to disk."""
    return self.scheduler.save_cache_to_disk(cache_dir)

vllm_mlx.engine_core.EngineCore.load_cache_from_disk

load_cache_from_disk(cache_dir: str) -> int

Load prefix cache from disk.

Source code in vllm_mlx/engine_core.py

def load_cache_from_disk(self, cache_dir: str) -> int:
    """Load prefix cache from disk."""
    return self.scheduler.load_cache_from_disk(cache_dir)

vllm_mlx.engine_core.EngineCore.clear_runtime_caches

clear_runtime_caches() -> Dict[str, Any] | None

Clear scheduler-managed runtime caches.

Source code in vllm_mlx/engine_core.py

def clear_runtime_caches(self) -> Dict[str, Any] | None:
    """Clear scheduler-managed runtime caches."""
    return self.scheduler.clear_runtime_caches()

vllm_mlx.engine_core.EngineCore.clear_prefix_cache

clear_prefix_cache() -> None

Clear the prefix cache (delegates to scheduler).

Source code in vllm_mlx/engine_core.py

def clear_prefix_cache(self) -> None:
    """Clear the prefix cache (delegates to scheduler)."""
    if hasattr(self.scheduler, "clear_prefix_cache"):
        self.scheduler.clear_prefix_cache()

vllm_mlx.engine_core.EngineCore._release_model

_release_model() -> None

Release model ownership.

Source code in vllm_mlx/engine_core.py

def _release_model(self) -> None:
    """Release model ownership."""
    if self._owns_model and not self._closed:
        registry = get_registry()
        registry.release(self.model, self._engine_id)
        self._owns_model = False
        logger.debug(f"Engine {self._engine_id} released model ownership")

vllm_mlx.engine_core.EngineCore.close

close() -> None

Explicitly close the engine and release resources.

This should be called when done using the engine, especially if you plan to create another engine with the same model.

Source code in vllm_mlx/engine_core.py

def close(self) -> None:
    """
    Explicitly close the engine and release resources.

    This should be called when done using the engine, especially
    if you plan to create another engine with the same model.
    """
    if self._closed:
        return

    # Release model ownership BEFORE setting _closed
    # (_release_model checks not self._closed)
    if self._owns_model:
        registry = get_registry()
        registry.release(self.model, self._engine_id)
        self._owns_model = False
        logger.debug(f"Engine {self._engine_id} released model ownership")

    self._closed = True

    # Reset scheduler to clear BatchGenerator and all caches
    self.scheduler.deep_reset()

    # Clear output collectors
    for collector in self._output_collectors.values():
        collector.clear()
    self._output_collectors.clear()
    self._stream_states.clear()
    self._finished_events.clear()

    logger.debug(f"Engine {self._engine_id} closed")

vllm_mlx.engine_core.EngineCore.__del__

__del__()

Cleanup on destruction.

Source code in vllm_mlx/engine_core.py

def __del__(self):
    """Cleanup on destruction."""
    try:
        self._release_model()
    except Exception:
        # Ignore errors during garbage collection
        pass

vllm_mlx.engine_core.AsyncEngineCore

AsyncEngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None)

Async context manager wrapper for EngineCore.

Usage

async with AsyncEngineCore(model, tokenizer) as engine: request_id = await engine.add_request("Hello") async for output in engine.stream_outputs(request_id): print(output.new_text)

Source code in vllm_mlx/engine_core.py

def __init__(
    self,
    model: Any,
    tokenizer: Any,
    config: Optional[EngineConfig] = None,
):
    self.engine = EngineCore(model, tokenizer, config)

vllm_mlx.engine_core.AsyncEngineCore.engine instance-attribute

engine = EngineCore(model, tokenizer, config)

vllm_mlx.engine_core.AsyncEngineCore.__aenter__ async

__aenter__() -> AsyncEngineCore

Source code in vllm_mlx/engine_core.py

async def __aenter__(self) -> "AsyncEngineCore":
    await self.engine.start()
    return self

vllm_mlx.engine_core.AsyncEngineCore.__aexit__ async

__aexit__(*args) -> None

Source code in vllm_mlx/engine_core.py

async def __aexit__(self, *args) -> None:
    await self.engine.stop()

vllm_mlx.engine_core.AsyncEngineCore.start

start() -> None

Start engine (creates task in current loop).

Source code in vllm_mlx/engine_core.py

def start(self) -> None:
    """Start engine (creates task in current loop)."""
    self._start_task = asyncio.create_task(self.engine.start())

vllm_mlx.engine_core.AsyncEngineCore.stop async

stop() -> None

Stop the engine.

Source code in vllm_mlx/engine_core.py

async def stop(self) -> None:
    """Stop the engine."""
    await self.engine.stop()

vllm_mlx.engine_core.AsyncEngineCore.add_request async

add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> str

Add a request.

Source code in vllm_mlx/engine_core.py

async def add_request(
    self,
    prompt: Union[str, List[int]],
    sampling_params: Optional[SamplingParams] = None,
    request_id: Optional[str] = None,
    **kwargs,
) -> str:
    """Add a request."""
    return await self.engine.add_request(
        prompt=prompt,
        sampling_params=sampling_params,
        request_id=request_id,
        **kwargs,
    )

vllm_mlx.engine_core.AsyncEngineCore.abort_request async

abort_request(request_id: str) -> bool

Abort a request.

Source code in vllm_mlx/engine_core.py

async def abort_request(self, request_id: str) -> bool:
    """Abort a request."""
    return await self.engine.abort_request(request_id)

vllm_mlx.engine_core.AsyncEngineCore.stream_outputs async

stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]

Stream outputs.

Source code in vllm_mlx/engine_core.py

async def stream_outputs(
    self,
    request_id: str,
    timeout: Optional[float] = None,
) -> AsyncIterator[RequestOutput]:
    """Stream outputs."""
    async for output in self.engine.stream_outputs(request_id, timeout):
        yield output

vllm_mlx.engine_core.AsyncEngineCore.generate async

generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, **kwargs) -> RequestOutput

Generate complete response.

Source code in vllm_mlx/engine_core.py

async def generate(
    self,
    prompt: Union[str, List[int]],
    sampling_params: Optional[SamplingParams] = None,
    **kwargs,
) -> RequestOutput:
    """Generate complete response."""
    return await self.engine.generate(
        prompt=prompt,
        sampling_params=sampling_params,
        **kwargs,
    )

vllm_mlx.engine_core.AsyncEngineCore.get_stats

get_stats() -> Dict[str, Any]

Get engine stats.

Source code in vllm_mlx/engine_core.py

def get_stats(self) -> Dict[str, Any]:
    """Get engine stats."""
    return self.engine.get_stats()

vllm_mlx.engine_core.AsyncEngineCore.get_cache_stats

get_cache_stats() -> Optional[Dict[str, Any]]

Get prefix cache statistics.

Source code in vllm_mlx/engine_core.py

def get_cache_stats(self) -> Optional[Dict[str, Any]]:
    """Get prefix cache statistics."""
    return self.engine.get_cache_stats()

vllm_mlx.engine_core.AsyncEngineCore.save_cache_to_disk

save_cache_to_disk(cache_dir: str) -> bool

Save prefix cache to disk.

Source code in vllm_mlx/engine_core.py

def save_cache_to_disk(self, cache_dir: str) -> bool:
    """Save prefix cache to disk."""
    return self.engine.save_cache_to_disk(cache_dir)

vllm_mlx.engine_core.AsyncEngineCore.load_cache_from_disk

load_cache_from_disk(cache_dir: str) -> int

Load prefix cache from disk.

Source code in vllm_mlx/engine_core.py

def load_cache_from_disk(self, cache_dir: str) -> int:
    """Load prefix cache from disk."""
    return self.engine.load_cache_from_disk(cache_dir)

vllm_mlx.engine_core.AsyncEngineCore.clear_runtime_caches

clear_runtime_caches() -> Dict[str, Any] | None

Clear scheduler-managed runtime caches.

Source code in vllm_mlx/engine_core.py

def clear_runtime_caches(self) -> Dict[str, Any] | None:
    """Clear scheduler-managed runtime caches."""
    return self.engine.clear_runtime_caches()

vllm_mlx.engine_core._is_stream_thread_error

_is_stream_thread_error(error: Exception) -> bool

True when MLX reports stream ownership mismatch across threads.

Source code in vllm_mlx/engine_core.py

def _is_stream_thread_error(error: Exception) -> bool:
    """True when MLX reports stream ownership mismatch across threads."""
    message = str(error)
    return "no Stream(" in message or "no Stream(gpu" in message

Complete contract reference

Expand any definition for its exact inputs, annotations, defaults, return contract, directly raised exceptions, source-grounded behavior, and immutable line link. This section includes private and nested definitions that ordinary API generators omit.

vllm_mlx.engine_core._is_stream_thread_error · function

vllm_mlx.engine_core._is_stream_thread_error(error: Exception) -> bool

True when MLX reports stream ownership mismatch across threads.

Parameters

Name	Type	Required	Default	Description
error	Exception	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: 'no Stream(' in message or 'no Stream(gpu' in message

Exceptions and behavior

Function _is_stream_thread_error calls str; returns 'no Stream(' in message or 'no Stream(gpu' in message. No direct raise statement appears in this definition.

View source #L33-L36.

vllm_mlx.engine_core.EngineConfig · class

vllm_mlx.engine_core.EngineConfig(model_name: str = '', scheduler_config: Optional[SchedulerConfig] = None, step_interval: float = 0.001, stream_interval: int = 1, gpu_memory_utilization: float = 0.9)

Configuration for the engine.

Parameters

Name	Type	Required	Default	Description
model_name	str	no	''	Optional constructor field; defaults to ''.
scheduler_config	Optional[SchedulerConfig]	no	None	Optional constructor field; defaults to None.
step_interval	float	no	0.001	Optional constructor field; defaults to 0.001.
stream_interval	int	no	1	Optional constructor field; defaults to 1.
gpu_memory_utilization	float	no	0.9	Optional constructor field; defaults to 0.9.

Returns

• Constructs: vllm_mlx.engine_core.EngineConfig

Exceptions and behavior

Class EngineConfig declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L40-L47.

vllm_mlx.engine_core.EngineCore · class

vllm_mlx.engine_core.EngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None, engine_id: Optional[str] = None, force_model_ownership: bool = True)

Core engine for vllm-mlx inference with continuous batching.

Parameters

Name	Type	Required	Default	Description
model	Any	yes	none	The MLX model
tokenizer	Any	yes	none	The tokenizer
config	Optional[EngineConfig]	no	None	Engine configuration
engine_id	Optional[str]	no	None	Optional unique ID for this engine (auto-generated if None)
force_model_ownership	bool	no	True	If True (default), forcibly take model ownership from any existing engine. If False, raises ModelOwnershipError if model is in use.

Returns

• Constructs: vllm_mlx.engine_core.EngineCore

Exceptions and behavior

Class EngineCore declares 21 direct member(s). No direct raise statement appears in this definition.

View source #L50-L698.

vllm_mlx.engine_core.EngineCore.__init__ · method

vllm_mlx.engine_core.EngineCore.__init__(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None, engine_id: Optional[str] = None, force_model_ownership: bool = True) -> not annotated

Initialize the engine.

Parameters

Name	Type	Required	Default	Description
model	Any	yes	none	The MLX model
tokenizer	Any	yes	none	The tokenizer
config	Optional[EngineConfig]	no	None	Engine configuration
engine_id	Optional[str]	no	None	Optional unique ID for this engine (auto-generated if None)
force_model_ownership	bool	no	True	If True (default), forcibly take model ownership from any existing engine. If False, raises ModelOwnershipError if model is in use.

Returns

• Type: not annotated

Exceptions and behavior

Method EngineCore.__init__ updates self.model, self.tokenizer, self.config, self._engine_id; calls EngineConfig, str, uuid.uuid4, get_registry. No direct raise statement appears in this definition.

View source #L58-L114.

vllm_mlx.engine_core.EngineCore.start · method

async vllm_mlx.engine_core.EngineCore.start() -> None

Start the engine loop.

Parameters

This callable has no explicit inputs.

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method EngineCore.start updates self._running, self._start_time, self._task; calls time.time, asyncio.create_task, self._engine_loop, logger.info; returns None. No direct raise statement appears in this definition.

View source #L116-L124.

vllm_mlx.engine_core.EngineCore.stop · method

async vllm_mlx.engine_core.EngineCore.stop() -> None

Stop the engine loop.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method EngineCore.stop updates self._running, self._task; calls self._task.cancel, self.scheduler._close_batch_generator, logger.info; awaits asynchronous work. No direct raise statement appears in this definition.

View source #L126-L140.

vllm_mlx.engine_core.EngineCore.is_running · method

vllm_mlx.engine_core.EngineCore.is_running() -> bool

Check if engine is running.

Parameters

This callable has no explicit inputs.

Returns

• Type: bool
• Direct return expressions: self._running

Exceptions and behavior

Method EngineCore.is_running returns self._running. No direct raise statement appears in this definition.

View source #L142-L144.

vllm_mlx.engine_core.EngineCore._engine_loop · method

async vllm_mlx.engine_core.EngineCore._engine_loop() -> None

Main engine loop.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method EngineCore._engine_loop calls asyncio.get_running_loop, ThreadPoolExecutor, mx.device_info().get, mx.device_info; awaits asynchronous work. No direct raise statement appears in this definition.

View source #L146-L334.

vllm_mlx.engine_core.EngineCore._engine_loop._bind_worker_streams_once · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._bind_worker_streams_once() -> None

Nested Function EngineCore._engine_loop._bind_worker_streams_once calls bind_generation_streams.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Nested Function EngineCore._engine_loop._bind_worker_streams_once calls bind_generation_streams. No direct raise statement appears in this definition.

View source #L160-L164.

vllm_mlx.engine_core.EngineCore._engine_loop._bind_model_streams_once · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._bind_model_streams_once() -> None

Nested Function EngineCore._engine_loop._bind_model_streams_once calls bind_generation_streams.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Nested Function EngineCore._engine_loop._bind_model_streams_once calls bind_generation_streams. No direct raise statement appears in this definition.

View source #L166-L170.

vllm_mlx.engine_core.EngineCore._engine_loop._step_on_worker · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._step_on_worker() -> not annotated

Nested Function EngineCore._engine_loop._step_on_worker updates self._steps_executed; calls _bind_worker_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output.

Parameters

This callable has no explicit inputs.

Returns

• Type: not annotated
• Direct return expressions: output

Exceptions and behavior

Nested Function EngineCore._engine_loop._step_on_worker updates self._steps_executed; calls _bind_worker_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output. No direct raise statement appears in this definition.

View source #L172-L190.

vllm_mlx.engine_core.EngineCore._engine_loop._step_on_model_thread · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._step_on_model_thread() -> not annotated

Nested Function EngineCore._engine_loop._step_on_model_thread updates self._steps_executed; calls _bind_model_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output.

Parameters

This callable has no explicit inputs.

Returns

• Type: not annotated
• Direct return expressions: output

Exceptions and behavior

Nested Function EngineCore._engine_loop._step_on_model_thread updates self._steps_executed; calls _bind_model_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output. No direct raise statement appears in this definition.

View source #L192-L210.

vllm_mlx.engine_core.EngineCore._engine_loop._recover_stream_thread_error_on_worker · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._recover_stream_thread_error_on_worker() -> None

Nested Function EngineCore._engine_loop._recover_stream_thread_error_on_worker calls _bind_worker_streams_once, self.scheduler._recover_from_cache_error, self.scheduler._reschedule_running_requests.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Nested Function EngineCore._engine_loop._recover_stream_thread_error_on_worker calls _bind_worker_streams_once, self.scheduler._recover_from_cache_error, self.scheduler._reschedule_running_requests. No direct raise statement appears in this definition.

View source #L212-L215.

vllm_mlx.engine_core.EngineCore._engine_loop._clear_cache_on_worker · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._clear_cache_on_worker() -> None

Nested Function EngineCore._engine_loop._clear_cache_on_worker calls _bind_worker_streams_once, mx.clear_cache.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Nested Function EngineCore._engine_loop._clear_cache_on_worker calls _bind_worker_streams_once, mx.clear_cache. No direct raise statement appears in this definition.

View source #L217-L219.

vllm_mlx.engine_core.EngineCore._engine_loop._close_batch_generator_on_worker · nested function

vllm_mlx.engine_core.EngineCore._engine_loop._close_batch_generator_on_worker() -> None

Nested Function EngineCore._engine_loop._close_batch_generator_on_worker calls _bind_worker_streams_once, self.scheduler._close_batch_generator.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Nested Function EngineCore._engine_loop._close_batch_generator_on_worker calls _bind_worker_streams_once, self.scheduler._close_batch_generator. No direct raise statement appears in this definition.

View source #L221-L223.

vllm_mlx.engine_core.EngineCore.add_request · method

async vllm_mlx.engine_core.EngineCore.add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, images: Optional[List[Any]] = None, videos: Optional[List[Any]] = None, prefix_boundary: int = 0) -> str

Add a request for processing.

Parameters

Name	Type	Required	Default	Description
prompt	Union[str, List[int]]	yes	none	Input prompt (string or token IDs)
sampling_params	Optional[SamplingParams]	no	None	Generation parameters
request_id	Optional[str]	no	None	Optional custom request ID
images	Optional[List[Any]]	no	None	Optional images for multimodal
videos	Optional[List[Any]]	no	None	Optional videos for multimodal
prefix_boundary	int	no	0	Token count for shared prefix (for cache)

Returns

• Type: str
• Direct return expressions: request_id

Exceptions and behavior

Method EngineCore.add_request calls str, uuid.uuid4, SamplingParams, Request; returns request_id. No direct raise statement appears in this definition.

View source #L336-L384.

vllm_mlx.engine_core.EngineCore.abort_request · method

async vllm_mlx.engine_core.EngineCore.abort_request(request_id: str) -> bool

Abort a request.

Parameters

Name	Type	Required	Default	Description
request_id	str	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: result

Exceptions and behavior

Method EngineCore.abort_request calls self.scheduler.abort_request, self._cleanup_request; returns result. No direct raise statement appears in this definition.

View source #L386-L390.

vllm_mlx.engine_core.EngineCore._cleanup_request · method

vllm_mlx.engine_core.EngineCore._cleanup_request(request_id: str) -> None

Clean up request tracking.

Parameters

Name	Type	Required	Default	Description
request_id	str	yes	none	Required positional or keyword input.

Returns

• Type: None

Exceptions and behavior

Method EngineCore._cleanup_request calls self._output_collectors.pop, collector.clear, self._stream_states.pop, self._finished_events.pop. No direct raise statement appears in this definition.

View source #L392-L399.

vllm_mlx.engine_core.EngineCore.stream_outputs · method

async vllm_mlx.engine_core.EngineCore.stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]

Stream outputs for a request with low-latency non-blocking pattern.

Parameters

Name	Type	Required	Default	Description
request_id	str	yes	none	The request ID
timeout	Optional[float]	no	None	Optional timeout in seconds

Returns

• Type: AsyncIterator[RequestOutput]
• Direct return expressions: None
• Yields values incrementally.

Exceptions and behavior

Method EngineCore.stream_outputs calls _time.monotonic, self._output_collectors.get, logger.warning, logger.info; awaits asynchronous work; yields values incrementally; returns None. No direct raise statement appears in this definition.

View source #L401-L488.

vllm_mlx.engine_core.EngineCore.generate · method

async vllm_mlx.engine_core.EngineCore.generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> RequestOutput

Generate a complete response (non-streaming).

Parameters

Name	Type	Required	Default	Description
prompt	Union[str, List[int]]	yes	none	Input prompt
sampling_params	Optional[SamplingParams]	no	None	Generation parameters
request_id	Optional[str]	no	None	Optional request ID
**kwargs	not annotated	no	none	Additional variadic keyword inputs accepted by this callable.

Returns

• Type: RequestOutput
• Direct return expressions: final_output

Exceptions and behavior

Method EngineCore.generate calls self.add_request, self._finished_events.get, RuntimeError, event.wait; awaits asynchronous work; can raise RuntimeError; returns final_output. Directly raised exceptions: RuntimeError.

View source #L490-L552.

vllm_mlx.engine_core.EngineCore.generate_batch_sync · method

vllm_mlx.engine_core.EngineCore.generate_batch_sync(prompts: List[Union[str, List[int]]], sampling_params: Optional[SamplingParams] = None) -> List[RequestOutput]

Generate responses synchronously for maximum throughput.

Parameters

Name	Type	Required	Default	Description
prompts	List[Union[str, List[int]]]	yes	none	List of input prompts
sampling_params	Optional[SamplingParams]	no	None	Generation parameters (same for all)

Returns

• Type: List[RequestOutput]
• Direct return expressions: [results[rid] for rid in request_ids]

Exceptions and behavior

Method EngineCore.generate_batch_sync calls SamplingParams, str, uuid_module.uuid4, Request; returns [results[rid] for rid in request_ids]. No direct raise statement appears in this definition.

View source #L554-L609.

vllm_mlx.engine_core.EngineCore.get_stats · method

vllm_mlx.engine_core.EngineCore.get_stats() -> Dict[str, Any]

Get engine statistics.

Parameters

This callable has no explicit inputs.

Returns

• Type: Dict[str, Any]
• Direct return expressions: {'running': self._running, 'uptime_seconds': uptime, 'steps_executed': self._steps_executed, 'active_requests': len(sel…

Exceptions and behavior

Method EngineCore.get_stats calls self.scheduler.get_stats, time.time, len, self.scheduler.get_running_requests_info; returns {'running': self._running, 'uptime_seconds': uptime, 'steps_executed': self._steps_executed, 'active_requests': len(sel…. No direct raise statement appears in this definition.

View source #L611-L624.

vllm_mlx.engine_core.EngineCore.get_cache_stats · method

vllm_mlx.engine_core.EngineCore.get_cache_stats() -> Optional[Dict[str, Any]]

Get prefix cache statistics.

Parameters

This callable has no explicit inputs.

Returns

• Type: Optional[Dict[str, Any]]
• Direct return expressions: self.scheduler.get_cache_stats()

Exceptions and behavior

Method EngineCore.get_cache_stats calls self.scheduler.get_cache_stats; returns self.scheduler.get_cache_stats(). No direct raise statement appears in this definition.

View source #L626-L628.

vllm_mlx.engine_core.EngineCore.save_cache_to_disk · method

vllm_mlx.engine_core.EngineCore.save_cache_to_disk(cache_dir: str) -> bool

Save prefix cache to disk.

Parameters

Name	Type	Required	Default	Description
cache_dir	str	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: self.scheduler.save_cache_to_disk(cache_dir)

Exceptions and behavior

Method EngineCore.save_cache_to_disk calls self.scheduler.save_cache_to_disk; returns self.scheduler.save_cache_to_disk(cache_dir). No direct raise statement appears in this definition.

View source #L630-L632.

vllm_mlx.engine_core.EngineCore.load_cache_from_disk · method

vllm_mlx.engine_core.EngineCore.load_cache_from_disk(cache_dir: str) -> int

Load prefix cache from disk.

Parameters

Name	Type	Required	Default	Description
cache_dir	str	yes	none	Required positional or keyword input.

Returns

• Type: int
• Direct return expressions: self.scheduler.load_cache_from_disk(cache_dir)

Exceptions and behavior

Method EngineCore.load_cache_from_disk calls self.scheduler.load_cache_from_disk; returns self.scheduler.load_cache_from_disk(cache_dir). No direct raise statement appears in this definition.

View source #L634-L636.

vllm_mlx.engine_core.EngineCore.clear_runtime_caches · method

vllm_mlx.engine_core.EngineCore.clear_runtime_caches() -> Dict[str, Any] | None

Clear scheduler-managed runtime caches.

Parameters

This callable has no explicit inputs.

Returns

• Type: Dict[str, Any] | None
• Direct return expressions: self.scheduler.clear_runtime_caches()

Exceptions and behavior

Method EngineCore.clear_runtime_caches calls self.scheduler.clear_runtime_caches; returns self.scheduler.clear_runtime_caches(). No direct raise statement appears in this definition.

View source #L638-L640.

vllm_mlx.engine_core.EngineCore.clear_prefix_cache · method

vllm_mlx.engine_core.EngineCore.clear_prefix_cache() -> None

Clear the prefix cache (delegates to scheduler).

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method EngineCore.clear_prefix_cache calls hasattr, self.scheduler.clear_prefix_cache. No direct raise statement appears in this definition.

View source #L642-L645.

vllm_mlx.engine_core.EngineCore._release_model · method

vllm_mlx.engine_core.EngineCore._release_model() -> None

Release model ownership.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method EngineCore._release_model updates self._owns_model; calls get_registry, registry.release, logger.debug. No direct raise statement appears in this definition.

View source #L647-L653.

vllm_mlx.engine_core.EngineCore.close · method

vllm_mlx.engine_core.EngineCore.close() -> None

Explicitly close the engine and release resources.

Parameters

This callable has no explicit inputs.

Returns

• Type: None
• Direct return expressions: None

Exceptions and behavior

Method EngineCore.close updates self._owns_model, self._closed; calls get_registry, registry.release, logger.debug, self.scheduler.deep_reset; returns None. No direct raise statement appears in this definition.

View source #L655-L685.

vllm_mlx.engine_core.EngineCore.__del__ · method

vllm_mlx.engine_core.EngineCore.__del__() -> not annotated

Cleanup on destruction.

Parameters

This callable has no explicit inputs.

Returns

• Type: not annotated

Exceptions and behavior

Method EngineCore.__del__ calls self._release_model. No direct raise statement appears in this definition.

View source #L687-L693.

vllm_mlx.engine_core.EngineCore.engine_id · method

vllm_mlx.engine_core.EngineCore.engine_id() -> str

Get the engine ID.

Parameters

This callable has no explicit inputs.

Returns

• Type: str
• Direct return expressions: self._engine_id

Exceptions and behavior

Method EngineCore.engine_id returns self._engine_id. No direct raise statement appears in this definition.

View source #L696-L698.

vllm_mlx.engine_core.AsyncEngineCore · class

vllm_mlx.engine_core.AsyncEngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None)

Async context manager wrapper for EngineCore.

Parameters

Name	Type	Required	Default	Description
model	Any	yes	none	Required positional or keyword input.
tokenizer	Any	yes	none	Required positional or keyword input.
config	Optional[EngineConfig]	no	None	Optional positional or keyword input; defaults to None.

Returns

• Constructs: vllm_mlx.engine_core.AsyncEngineCore

Exceptions and behavior

Class AsyncEngineCore declares 14 direct member(s). No direct raise statement appears in this definition.

View source #L701-L794.

vllm_mlx.engine_core.AsyncEngineCore.__init__ · method

vllm_mlx.engine_core.AsyncEngineCore.__init__(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None) -> not annotated

Method AsyncEngineCore.__init__ updates self.engine; calls EngineCore.

Parameters

Name	Type	Required	Default	Description
model	Any	yes	none	Required positional or keyword input.
tokenizer	Any	yes	none	Required positional or keyword input.
config	Optional[EngineConfig]	no	None	Optional positional or keyword input; defaults to None.

Returns

• Type: not annotated

Exceptions and behavior

Method AsyncEngineCore.__init__ updates self.engine; calls EngineCore. No direct raise statement appears in this definition.

View source #L712-L718.

vllm_mlx.engine_core.AsyncEngineCore.__aenter__ · method

async vllm_mlx.engine_core.AsyncEngineCore.__aenter__() -> 'AsyncEngineCore'

Method AsyncEngineCore.__aenter__ calls self.engine.start; awaits asynchronous work; returns self.

Parameters

This callable has no explicit inputs.

Returns

• Type: 'AsyncEngineCore'
• Direct return expressions: self

Exceptions and behavior

Method AsyncEngineCore.__aenter__ calls self.engine.start; awaits asynchronous work; returns self. No direct raise statement appears in this definition.

View source #L720-L722.

vllm_mlx.engine_core.AsyncEngineCore.__aexit__ · method

async vllm_mlx.engine_core.AsyncEngineCore.__aexit__(*args) -> None

Method AsyncEngineCore.__aexit__ calls self.engine.stop; awaits asynchronous work.

Parameters

Name	Type	Required	Default	Description
*args	not annotated	no	none	Additional variadic positional inputs accepted by this callable.

Returns

• Type: None

Exceptions and behavior

Method AsyncEngineCore.__aexit__ calls self.engine.stop; awaits asynchronous work. No direct raise statement appears in this definition.

View source #L724-L725.

vllm_mlx.engine_core.AsyncEngineCore.start · method

vllm_mlx.engine_core.AsyncEngineCore.start() -> None

Start engine (creates task in current loop).

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method AsyncEngineCore.start updates self._start_task; calls asyncio.create_task, self.engine.start. No direct raise statement appears in this definition.

View source #L727-L729.

vllm_mlx.engine_core.AsyncEngineCore.stop · method

async vllm_mlx.engine_core.AsyncEngineCore.stop() -> None

Stop the engine.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method AsyncEngineCore.stop calls self.engine.stop; awaits asynchronous work. No direct raise statement appears in this definition.

View source #L731-L733.

vllm_mlx.engine_core.AsyncEngineCore.add_request · method

async vllm_mlx.engine_core.AsyncEngineCore.add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> str

Add a request.

Parameters

Name	Type	Required	Default	Description
prompt	Union[str, List[int]]	yes	none	Required positional or keyword input.
sampling_params	Optional[SamplingParams]	no	None	Optional positional or keyword input; defaults to None.
request_id	Optional[str]	no	None	Optional positional or keyword input; defaults to None.
**kwargs	not annotated	no	none	Additional variadic keyword inputs accepted by this callable.

Returns

• Type: str
• Direct return expressions: await self.engine.add_request(prompt=prompt, sampling_params=sampling_params, request_id=request_id, **kwargs)

Exceptions and behavior

Method AsyncEngineCore.add_request calls self.engine.add_request; awaits asynchronous work; returns await self.engine.add_request(prompt=prompt, sampling_params=sampling_params, request_id=request_id, **kwargs). No direct raise statement appears in this definition.

View source #L735-L748.

vllm_mlx.engine_core.AsyncEngineCore.abort_request · method

async vllm_mlx.engine_core.AsyncEngineCore.abort_request(request_id: str) -> bool

Abort a request.

Parameters

Name	Type	Required	Default	Description
request_id	str	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: await self.engine.abort_request(request_id)

Exceptions and behavior

Method AsyncEngineCore.abort_request calls self.engine.abort_request; awaits asynchronous work; returns await self.engine.abort_request(request_id). No direct raise statement appears in this definition.

View source #L750-L752.

vllm_mlx.engine_core.AsyncEngineCore.stream_outputs · method

async vllm_mlx.engine_core.AsyncEngineCore.stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]

Stream outputs.

Parameters

Name	Type	Required	Default	Description
request_id	str	yes	none	Required positional or keyword input.
timeout	Optional[float]	no	None	Optional positional or keyword input; defaults to None.

Returns

• Type: AsyncIterator[RequestOutput]
• Yields values incrementally.

Exceptions and behavior

Method AsyncEngineCore.stream_outputs calls self.engine.stream_outputs; yields values incrementally. No direct raise statement appears in this definition.

View source #L754-L761.

vllm_mlx.engine_core.AsyncEngineCore.generate · method

async vllm_mlx.engine_core.AsyncEngineCore.generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, **kwargs) -> RequestOutput

Generate complete response.

Parameters

Name	Type	Required	Default	Description
prompt	Union[str, List[int]]	yes	none	Required positional or keyword input.
sampling_params	Optional[SamplingParams]	no	None	Optional positional or keyword input; defaults to None.
**kwargs	not annotated	no	none	Additional variadic keyword inputs accepted by this callable.

Returns

• Type: RequestOutput
• Direct return expressions: await self.engine.generate(prompt=prompt, sampling_params=sampling_params, **kwargs)

Exceptions and behavior

Method AsyncEngineCore.generate calls self.engine.generate; awaits asynchronous work; returns await self.engine.generate(prompt=prompt, sampling_params=sampling_params, **kwargs). No direct raise statement appears in this definition.

View source #L763-L774.

vllm_mlx.engine_core.AsyncEngineCore.get_stats · method

vllm_mlx.engine_core.AsyncEngineCore.get_stats() -> Dict[str, Any]

Get engine stats.

Parameters

This callable has no explicit inputs.

Returns

• Type: Dict[str, Any]
• Direct return expressions: self.engine.get_stats()

Exceptions and behavior

Method AsyncEngineCore.get_stats calls self.engine.get_stats; returns self.engine.get_stats(). No direct raise statement appears in this definition.

View source #L776-L778.

vllm_mlx.engine_core.AsyncEngineCore.get_cache_stats · method

vllm_mlx.engine_core.AsyncEngineCore.get_cache_stats() -> Optional[Dict[str, Any]]

Get prefix cache statistics.

Parameters

This callable has no explicit inputs.

Returns

• Type: Optional[Dict[str, Any]]
• Direct return expressions: self.engine.get_cache_stats()

Exceptions and behavior

Method AsyncEngineCore.get_cache_stats calls self.engine.get_cache_stats; returns self.engine.get_cache_stats(). No direct raise statement appears in this definition.

View source #L780-L782.

vllm_mlx.engine_core.AsyncEngineCore.save_cache_to_disk · method

vllm_mlx.engine_core.AsyncEngineCore.save_cache_to_disk(cache_dir: str) -> bool

Save prefix cache to disk.

Parameters

Name	Type	Required	Default	Description
cache_dir	str	yes	none	Required positional or keyword input.

Returns

• Type: bool
• Direct return expressions: self.engine.save_cache_to_disk(cache_dir)

Exceptions and behavior

Method AsyncEngineCore.save_cache_to_disk calls self.engine.save_cache_to_disk; returns self.engine.save_cache_to_disk(cache_dir). No direct raise statement appears in this definition.

View source #L784-L786.

vllm_mlx.engine_core.AsyncEngineCore.load_cache_from_disk · method

vllm_mlx.engine_core.AsyncEngineCore.load_cache_from_disk(cache_dir: str) -> int

Load prefix cache from disk.

Parameters

Name	Type	Required	Default	Description
cache_dir	str	yes	none	Required positional or keyword input.

Returns

• Type: int
• Direct return expressions: self.engine.load_cache_from_disk(cache_dir)

Exceptions and behavior

Method AsyncEngineCore.load_cache_from_disk calls self.engine.load_cache_from_disk; returns self.engine.load_cache_from_disk(cache_dir). No direct raise statement appears in this definition.

View source #L788-L790.

vllm_mlx.engine_core.AsyncEngineCore.clear_runtime_caches · method

vllm_mlx.engine_core.AsyncEngineCore.clear_runtime_caches() -> Dict[str, Any] | None

Clear scheduler-managed runtime caches.

Parameters

This callable has no explicit inputs.

Returns

• Type: Dict[str, Any] | None
• Direct return expressions: self.engine.clear_runtime_caches()

Exceptions and behavior

Method AsyncEngineCore.clear_runtime_caches calls self.engine.clear_runtime_caches; returns self.engine.clear_runtime_caches(). No direct raise statement appears in this definition.

View source #L792-L794.

Complete symbol map

This map also includes private definitions and nested helpers. The signature column exposes every explicit input even when an internal helper has no dedicated parameter prose.

Symbol	Kind	Signature and inputs	What it does	Source
_is_stream_thread_error	function	_is_stream_thread_error(error: Exception) -> bool	True when MLX reports stream ownership mismatch across threads.	#L33-L36
EngineConfig	class	EngineConfig(model_name: str = '', scheduler_config: Optional[SchedulerConfig] = None, step_interval: float = 0.001, stream_interval: int = 1, gpu_memory_utilization: float = 0.9)	Configuration for the engine.	#L40-L47
EngineCore	class	EngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None, engine_id: Optional[str] = None, force_model_ownership: bool = True)	Core engine for vllm-mlx inference with continuous batching.	#L50-L698
EngineCore.__init__	method	EngineCore.__init__(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None, engine_id: Optional[str] = None, force_model_ownership: bool = True) -> not annotated	Initialize the engine.	#L58-L114
EngineCore.start	method	async EngineCore.start() -> None	Start the engine loop.	#L116-L124
EngineCore.stop	method	async EngineCore.stop() -> None	Stop the engine loop.	#L126-L140
EngineCore.is_running	method	EngineCore.is_running() -> bool	Check if engine is running.	#L142-L144
EngineCore._engine_loop	method	async EngineCore._engine_loop() -> None	Main engine loop.	#L146-L334
EngineCore._engine_loop._bind_worker_streams_once	nested function	EngineCore._engine_loop._bind_worker_streams_once() -> None	Nested Function EngineCore._engine_loop._bind_worker_streams_once calls bind_generation_streams.	#L160-L164
EngineCore._engine_loop._bind_model_streams_once	nested function	EngineCore._engine_loop._bind_model_streams_once() -> None	Nested Function EngineCore._engine_loop._bind_model_streams_once calls bind_generation_streams.	#L166-L170
EngineCore._engine_loop._step_on_worker	nested function	EngineCore._engine_loop._step_on_worker() -> not annotated	Nested Function EngineCore._engine_loop._step_on_worker updates self._steps_executed; calls _bind_worker_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output.	#L172-L190
EngineCore._engine_loop._step_on_model_thread	nested function	EngineCore._engine_loop._step_on_model_thread() -> not annotated	Nested Function EngineCore._engine_loop._step_on_model_thread updates self._steps_executed; calls _bind_model_streams_once, self.scheduler.step, mx.get_active_memory, mx.clear_cache; returns output.	#L192-L210
EngineCore._engine_loop._recover_stream_thread_error_on_worker	nested function	EngineCore._engine_loop._recover_stream_thread_error_on_worker() -> None	Nested Function EngineCore._engine_loop._recover_stream_thread_error_on_worker calls _bind_worker_streams_once, self.scheduler._recover_from_cache_error, self.scheduler._reschedule_running_requests.	#L212-L215
EngineCore._engine_loop._clear_cache_on_worker	nested function	EngineCore._engine_loop._clear_cache_on_worker() -> None	Nested Function EngineCore._engine_loop._clear_cache_on_worker calls _bind_worker_streams_once, mx.clear_cache.	#L217-L219
EngineCore._engine_loop._close_batch_generator_on_worker	nested function	EngineCore._engine_loop._close_batch_generator_on_worker() -> None	Nested Function EngineCore._engine_loop._close_batch_generator_on_worker calls _bind_worker_streams_once, self.scheduler._close_batch_generator.	#L221-L223
EngineCore.add_request	method	async EngineCore.add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, images: Optional[List[Any]] = None, videos: Optional[List[Any]] = None, prefix_boundary: int = 0) -> str	Add a request for processing.	#L336-L384
EngineCore.abort_request	method	async EngineCore.abort_request(request_id: str) -> bool	Abort a request.	#L386-L390
EngineCore._cleanup_request	method	EngineCore._cleanup_request(request_id: str) -> None	Clean up request tracking.	#L392-L399
EngineCore.stream_outputs	method	async EngineCore.stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]	Stream outputs for a request with low-latency non-blocking pattern.	#L401-L488
EngineCore.generate	method	async EngineCore.generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> RequestOutput	Generate a complete response (non-streaming).	#L490-L552
EngineCore.generate_batch_sync	method	EngineCore.generate_batch_sync(prompts: List[Union[str, List[int]]], sampling_params: Optional[SamplingParams] = None) -> List[RequestOutput]	Generate responses synchronously for maximum throughput.	#L554-L609
EngineCore.get_stats	method	EngineCore.get_stats() -> Dict[str, Any]	Get engine statistics.	#L611-L624
EngineCore.get_cache_stats	method	EngineCore.get_cache_stats() -> Optional[Dict[str, Any]]	Get prefix cache statistics.	#L626-L628
EngineCore.save_cache_to_disk	method	EngineCore.save_cache_to_disk(cache_dir: str) -> bool	Save prefix cache to disk.	#L630-L632
EngineCore.load_cache_from_disk	method	EngineCore.load_cache_from_disk(cache_dir: str) -> int	Load prefix cache from disk.	#L634-L636
EngineCore.clear_runtime_caches	method	EngineCore.clear_runtime_caches() -> Dict[str, Any] \| None	Clear scheduler-managed runtime caches.	#L638-L640
EngineCore.clear_prefix_cache	method	EngineCore.clear_prefix_cache() -> None	Clear the prefix cache (delegates to scheduler).	#L642-L645
EngineCore._release_model	method	EngineCore._release_model() -> None	Release model ownership.	#L647-L653
EngineCore.close	method	EngineCore.close() -> None	Explicitly close the engine and release resources.	#L655-L685
EngineCore.__del__	method	EngineCore.__del__() -> not annotated	Cleanup on destruction.	#L687-L693
EngineCore.engine_id	method	EngineCore.engine_id() -> str	Get the engine ID.	#L696-L698
AsyncEngineCore	class	AsyncEngineCore(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None)	Async context manager wrapper for EngineCore.	#L701-L794
AsyncEngineCore.__init__	method	AsyncEngineCore.__init__(model: Any, tokenizer: Any, config: Optional[EngineConfig] = None) -> not annotated	Method AsyncEngineCore.__init__ updates self.engine; calls EngineCore.	#L712-L718
AsyncEngineCore.__aenter__	method	async AsyncEngineCore.__aenter__() -> 'AsyncEngineCore'	Method AsyncEngineCore.__aenter__ calls self.engine.start; awaits asynchronous work; returns self.	#L720-L722
AsyncEngineCore.__aexit__	method	async AsyncEngineCore.__aexit__(*args) -> None	Method AsyncEngineCore.__aexit__ calls self.engine.stop; awaits asynchronous work.	#L724-L725
AsyncEngineCore.start	method	AsyncEngineCore.start() -> None	Start engine (creates task in current loop).	#L727-L729
AsyncEngineCore.stop	method	async AsyncEngineCore.stop() -> None	Stop the engine.	#L731-L733
AsyncEngineCore.add_request	method	async AsyncEngineCore.add_request(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, request_id: Optional[str] = None, **kwargs) -> str	Add a request.	#L735-L748
AsyncEngineCore.abort_request	method	async AsyncEngineCore.abort_request(request_id: str) -> bool	Abort a request.	#L750-L752
AsyncEngineCore.stream_outputs	method	async AsyncEngineCore.stream_outputs(request_id: str, timeout: Optional[float] = None) -> AsyncIterator[RequestOutput]	Stream outputs.	#L754-L761
AsyncEngineCore.generate	method	async AsyncEngineCore.generate(prompt: Union[str, List[int]], sampling_params: Optional[SamplingParams] = None, **kwargs) -> RequestOutput	Generate complete response.	#L763-L774
AsyncEngineCore.get_stats	method	AsyncEngineCore.get_stats() -> Dict[str, Any]	Get engine stats.	#L776-L778
AsyncEngineCore.get_cache_stats	method	AsyncEngineCore.get_cache_stats() -> Optional[Dict[str, Any]]	Get prefix cache statistics.	#L780-L782
AsyncEngineCore.save_cache_to_disk	method	AsyncEngineCore.save_cache_to_disk(cache_dir: str) -> bool	Save prefix cache to disk.	#L784-L786
AsyncEngineCore.load_cache_from_disk	method	AsyncEngineCore.load_cache_from_disk(cache_dir: str) -> int	Load prefix cache from disk.	#L788-L790
AsyncEngineCore.clear_runtime_caches	method	AsyncEngineCore.clear_runtime_caches() -> Dict[str, Any] \| None	Clear scheduler-managed runtime caches.	#L792-L794