vllm_mlx.output_collector

Output collector for streaming with low-latency optimizations.

View the complete module source at #L1-L212.

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.output_collector

Output collector for streaming with low-latency optimizations.

This module implements the RequestOutputCollector pattern from vLLM, providing non-blocking output collection with intelligent aggregation.

vllm_mlx.output_collector.RequestOutputCollector

RequestOutputCollector(aggregate: bool = True)

Per-request output collector with smart buffering.

This class implements the vLLM pattern for efficient streaming: - Non-blocking get_nowait() to avoid unnecessary task switches - Output aggregation when producer is faster than consumer - Event-based signaling for efficient waiting - Tracking of active consumers for yield optimization

Usage

collector = RequestOutputCollector()

Producer side (engine loop)

collector.put(output)

Consumer side (streaming generator)

output = collector.get_nowait() or await collector.get()

Initialize the collector.

Parameters:

• aggregate (bool, default: True ) –

  If True, merge outputs when producer gets ahead. This prevents buffer explosion under load.

Source code in vllm_mlx/output_collector.py

def __init__(self, aggregate: bool = True):
    """
    Initialize the collector.

    Args:
        aggregate: If True, merge outputs when producer gets ahead.
                   This prevents buffer explosion under load.
    """
    self.output: Optional[RequestOutput] = None
    self.ready = asyncio.Event()
    self.aggregate = aggregate
    self._is_waiting = False

vllm_mlx.output_collector.RequestOutputCollector._waiting_consumers class-attribute instance-attribute

_waiting_consumers: int = 0

vllm_mlx.output_collector.RequestOutputCollector._waiting_lock class-attribute instance-attribute

_waiting_lock: Lock = threading.Lock()

vllm_mlx.output_collector.RequestOutputCollector.output instance-attribute

output: Optional[RequestOutput] = None

vllm_mlx.output_collector.RequestOutputCollector.ready instance-attribute

ready = asyncio.Event()

vllm_mlx.output_collector.RequestOutputCollector.aggregate instance-attribute

aggregate = aggregate

vllm_mlx.output_collector.RequestOutputCollector._is_waiting instance-attribute

_is_waiting = False

vllm_mlx.output_collector.RequestOutputCollector.put

put(output: RequestOutput) -> None

Put an output into the collector (non-blocking).

If aggregation is enabled and an output already exists, the new output is merged with the existing one.

Parameters:

• output (RequestOutput) –

  The RequestOutput to store

Source code in vllm_mlx/output_collector.py

def put(self, output: RequestOutput) -> None:
    """
    Put an output into the collector (non-blocking).

    If aggregation is enabled and an output already exists,
    the new output is merged with the existing one.

    Args:
        output: The RequestOutput to store
    """
    if self.output is None:
        self.output = output
    elif self.aggregate:
        # Merge: combine tokens when producer is ahead
        self.output = self._merge_outputs(self.output, output)
    else:
        # Replace: just use the new output
        self.output = output
    self.ready.set()

vllm_mlx.output_collector.RequestOutputCollector.get_nowait

get_nowait() -> Optional[RequestOutput]

Get output without blocking.

This avoids task switching when output is available, reducing latency under load.

Returns:

• Optional[RequestOutput] –

  The output if available, None otherwise

Source code in vllm_mlx/output_collector.py

def get_nowait(self) -> Optional[RequestOutput]:
    """
    Get output without blocking.

    This avoids task switching when output is available,
    reducing latency under load.

    Returns:
        The output if available, None otherwise
    """
    output = self.output
    if output is not None:
        self.output = None
        self.ready.clear()
    return output

vllm_mlx.output_collector.RequestOutputCollector.get async

get() -> RequestOutput

Get output, blocking only if none available.

This method blocks until an output is available. For low-latency streaming, prefer: output = collector.get_nowait() or await collector.get()

Returns:

• RequestOutput –

  The RequestOutput

Source code in vllm_mlx/output_collector.py

async def get(self) -> RequestOutput:
    """
    Get output, blocking only if none available.

    This method blocks until an output is available.
    For low-latency streaming, prefer:
        output = collector.get_nowait() or await collector.get()

    Returns:
        The RequestOutput
    """
    # Track that we're waiting (for yield optimization)
    if not self._is_waiting:
        self._is_waiting = True
        with RequestOutputCollector._waiting_lock:
            RequestOutputCollector._waiting_consumers += 1
    try:
        while self.output is None:
            await self.ready.wait()
        output = self.get_nowait()
        # This should never be None after wait, but satisfy type checker
        assert output is not None
        return output
    finally:
        if self._is_waiting:
            self._is_waiting = False
            with RequestOutputCollector._waiting_lock:
                RequestOutputCollector._waiting_consumers -= 1

vllm_mlx.output_collector.RequestOutputCollector._merge_outputs

_merge_outputs(existing: RequestOutput, new: RequestOutput) -> RequestOutput

Merge two outputs when producer gets ahead of consumer.

This combines the token lists and text, keeping the latest status information.

Parameters:

• existing (RequestOutput) –

  The existing output in the buffer

• new (RequestOutput) –

  The new output to merge

Returns:

• RequestOutput –

  Merged RequestOutput

Source code in vllm_mlx/output_collector.py

def _merge_outputs(
    self,
    existing: RequestOutput,
    new: RequestOutput,
) -> RequestOutput:
    """
    Merge two outputs when producer gets ahead of consumer.

    This combines the token lists and text, keeping the latest
    status information.

    Args:
        existing: The existing output in the buffer
        new: The new output to merge

    Returns:
        Merged RequestOutput
    """
    # Combine new tokens
    merged_new_token_ids = existing.new_token_ids + new.new_token_ids
    merged_new_text = existing.new_text + new.new_text

    return RequestOutput(
        request_id=new.request_id,
        new_token_ids=merged_new_token_ids,
        new_text=merged_new_text,
        output_token_ids=new.output_token_ids,  # Use latest cumulative
        output_text=new.output_text,  # Use latest cumulative
        finished=new.finished,
        finish_reason=new.finish_reason,
        prompt_tokens=new.prompt_tokens,
        completion_tokens=new.completion_tokens,
    )

vllm_mlx.output_collector.RequestOutputCollector.clear

clear() -> None

Clear any pending output.

Source code in vllm_mlx/output_collector.py

def clear(self) -> None:
    """Clear any pending output."""
    self.output = None
    self.ready.clear()
    if self._is_waiting:
        self._is_waiting = False
        with RequestOutputCollector._waiting_lock:
            RequestOutputCollector._waiting_consumers -= 1

vllm_mlx.output_collector.RequestOutputCollector.has_waiting_consumers classmethod

has_waiting_consumers() -> bool

Check if any collector has waiting consumers.

Used by engine to optimize: only yield when someone is waiting.

Source code in vllm_mlx/output_collector.py

@classmethod
def has_waiting_consumers(cls) -> bool:
    """Check if any collector has waiting consumers.

    Used by engine to optimize: only yield when someone is waiting.
    """
    with cls._waiting_lock:
        return cls._waiting_consumers > 0

vllm_mlx.output_collector.RequestStreamState dataclass

RequestStreamState(stream_interval: int = 1, sent_tokens: int = 0)

Tracks streaming state for a request.

This is used to implement stream_interval batching, allowing tokens to be accumulated before sending.

vllm_mlx.output_collector.RequestStreamState.stream_interval class-attribute instance-attribute

stream_interval: int = 1

vllm_mlx.output_collector.RequestStreamState.sent_tokens class-attribute instance-attribute

sent_tokens: int = 0

vllm_mlx.output_collector.RequestStreamState.should_send

should_send(total_tokens: int, finished: bool) -> bool

Determine if output should be sent based on stream_interval.

Parameters:

• total_tokens (int) –

  Total tokens generated so far

• finished (bool) –

  Whether generation is complete

Returns:

• bool –

  True if output should be sent

Source code in vllm_mlx/output_collector.py

def should_send(self, total_tokens: int, finished: bool) -> bool:
    """
    Determine if output should be sent based on stream_interval.

    Args:
        total_tokens: Total tokens generated so far
        finished: Whether generation is complete

    Returns:
        True if output should be sent
    """
    # Always send on finish
    if finished:
        return True
    # Always send first token (for low TTFT)
    if self.sent_tokens == 0:
        return True
    # Send if we've accumulated enough tokens
    return (total_tokens - self.sent_tokens) >= self.stream_interval

vllm_mlx.output_collector.RequestStreamState.mark_sent

mark_sent(total_tokens: int) -> None

Update state after sending output.

Parameters:

• total_tokens (int) –

  Total tokens at time of send

Source code in vllm_mlx/output_collector.py

def mark_sent(self, total_tokens: int) -> None:
    """
    Update state after sending output.

    Args:
        total_tokens: Total tokens at time of send
    """
    self.sent_tokens = total_tokens

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.output_collector.RequestOutputCollector · class

vllm_mlx.output_collector.RequestOutputCollector(aggregate: bool = True)

Per-request output collector with smart buffering.

Parameters

Name	Type	Required	Default	Description
aggregate	bool	no	True	If True, merge outputs when producer gets ahead. This prevents buffer explosion under load.

Returns

• Constructs: vllm_mlx.output_collector.RequestOutputCollector

Exceptions and behavior

Class RequestOutputCollector declares 7 direct member(s). No direct raise statement appears in this definition.

View source #L17-L170.

vllm_mlx.output_collector.RequestOutputCollector.__init__ · method

vllm_mlx.output_collector.RequestOutputCollector.__init__(aggregate: bool = True) -> not annotated

Initialize the collector.

Parameters

Name	Type	Required	Default	Description
aggregate	bool	no	True	If True, merge outputs when producer gets ahead. This prevents buffer explosion under load.

Returns

• Type: not annotated

Exceptions and behavior

Method RequestOutputCollector.__init__ updates self.output, self.ready, self.aggregate, self._is_waiting; calls asyncio.Event. No direct raise statement appears in this definition.

View source #L42-L53.

vllm_mlx.output_collector.RequestOutputCollector.put · method

vllm_mlx.output_collector.RequestOutputCollector.put(output: RequestOutput) -> None

Put an output into the collector (non-blocking).

Parameters

Name	Type	Required	Default	Description
output	RequestOutput	yes	none	The RequestOutput to store

Returns

• Type: None

Exceptions and behavior

Method RequestOutputCollector.put updates self.output; calls self._merge_outputs, self.ready.set. No direct raise statement appears in this definition.

View source #L55-L73.

vllm_mlx.output_collector.RequestOutputCollector.get_nowait · method

vllm_mlx.output_collector.RequestOutputCollector.get_nowait() -> Optional[RequestOutput]

Get output without blocking.

Parameters

This callable has no explicit inputs.

Returns

• Type: Optional[RequestOutput]
• Direct return expressions: output

Exceptions and behavior

Method RequestOutputCollector.get_nowait updates self.output; calls self.ready.clear; returns output. No direct raise statement appears in this definition.

View source #L75-L89.

vllm_mlx.output_collector.RequestOutputCollector.get · method

async vllm_mlx.output_collector.RequestOutputCollector.get() -> RequestOutput

Get output, blocking only if none available.

Parameters

This callable has no explicit inputs.

Returns

• Type: RequestOutput
• Direct return expressions: output

Exceptions and behavior

Method RequestOutputCollector.get updates self._is_waiting; calls self.ready.wait, self.get_nowait; awaits asynchronous work; returns output. No direct raise statement appears in this definition.

View source #L91-L118.

vllm_mlx.output_collector.RequestOutputCollector._merge_outputs · method

vllm_mlx.output_collector.RequestOutputCollector._merge_outputs(existing: RequestOutput, new: RequestOutput) -> RequestOutput

Merge two outputs when producer gets ahead of consumer.

Parameters

Name	Type	Required	Default	Description
existing	RequestOutput	yes	none	The existing output in the buffer
new	RequestOutput	yes	none	The new output to merge

Returns

• Type: RequestOutput
• Direct return expressions: RequestOutput(request_id=new.request_id, new_token_ids=merged_new_token_ids, new_text=merged_new_text, output_token_ids…

Exceptions and behavior

Method RequestOutputCollector._merge_outputs calls RequestOutput; returns RequestOutput(request_id=new.request_id, new_token_ids=merged_new_token_ids, new_text=merged_new_text, output_token_ids…. No direct raise statement appears in this definition.

View source #L120-L152.

vllm_mlx.output_collector.RequestOutputCollector.clear · method

vllm_mlx.output_collector.RequestOutputCollector.clear() -> None

Clear any pending output.

Parameters

This callable has no explicit inputs.

Returns

• Type: None

Exceptions and behavior

Method RequestOutputCollector.clear updates self.output, self._is_waiting; calls self.ready.clear. No direct raise statement appears in this definition.

View source #L154-L161.

vllm_mlx.output_collector.RequestOutputCollector.has_waiting_consumers · method

vllm_mlx.output_collector.RequestOutputCollector.has_waiting_consumers() -> bool

Check if any collector has waiting consumers.

Parameters

This callable has no explicit inputs.

Returns

• Type: bool
• Direct return expressions: cls._waiting_consumers > 0

Exceptions and behavior

Method RequestOutputCollector.has_waiting_consumers returns cls._waiting_consumers > 0. No direct raise statement appears in this definition.

View source #L164-L170.

vllm_mlx.output_collector.RequestStreamState · class

vllm_mlx.output_collector.RequestStreamState(stream_interval: int = 1, sent_tokens: int = 0)

Tracks streaming state for a request.

Parameters

Name	Type	Required	Default	Description
stream_interval	int	no	1	Optional constructor field; defaults to 1.
sent_tokens	int	no	0	Optional constructor field; defaults to 0.

Returns

• Constructs: vllm_mlx.output_collector.RequestStreamState

Exceptions and behavior

Class RequestStreamState declares 2 direct member(s). No direct raise statement appears in this definition.

View source #L174-L212.

vllm_mlx.output_collector.RequestStreamState.should_send · method

vllm_mlx.output_collector.RequestStreamState.should_send(total_tokens: int, finished: bool) -> bool

Determine if output should be sent based on stream_interval.

Parameters

Name	Type	Required	Default	Description
total_tokens	int	yes	none	Total tokens generated so far
finished	bool	yes	none	Whether generation is complete

Returns

• Type: bool
• Direct return expressions: True; total_tokens - self.sent_tokens >= self.stream_interval

Exceptions and behavior

Method RequestStreamState.should_send has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L185-L203.

vllm_mlx.output_collector.RequestStreamState.mark_sent · method

vllm_mlx.output_collector.RequestStreamState.mark_sent(total_tokens: int) -> None

Update state after sending output.

Parameters

Name	Type	Required	Default	Description
total_tokens	int	yes	none	Total tokens at time of send

Returns

• Type: None

Exceptions and behavior

Method RequestStreamState.mark_sent updates self.sent_tokens. No direct raise statement appears in this definition.

View source #L205-L212.

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
RequestOutputCollector	class	RequestOutputCollector(aggregate: bool = True)	Per-request output collector with smart buffering.	#L17-L170
RequestOutputCollector.__init__	method	RequestOutputCollector.__init__(aggregate: bool = True) -> not annotated	Initialize the collector.	#L42-L53
RequestOutputCollector.put	method	RequestOutputCollector.put(output: RequestOutput) -> None	Put an output into the collector (non-blocking).	#L55-L73
RequestOutputCollector.get_nowait	method	RequestOutputCollector.get_nowait() -> Optional[RequestOutput]	Get output without blocking.	#L75-L89
RequestOutputCollector.get	method	async RequestOutputCollector.get() -> RequestOutput	Get output, blocking only if none available.	#L91-L118
RequestOutputCollector._merge_outputs	method	RequestOutputCollector._merge_outputs(existing: RequestOutput, new: RequestOutput) -> RequestOutput	Merge two outputs when producer gets ahead of consumer.	#L120-L152
RequestOutputCollector.clear	method	RequestOutputCollector.clear() -> None	Clear any pending output.	#L154-L161
RequestOutputCollector.has_waiting_consumers	method	RequestOutputCollector.has_waiting_consumers() -> bool	Check if any collector has waiting consumers.	#L164-L170
RequestStreamState	class	RequestStreamState(stream_interval: int = 1, sent_tokens: int = 0)	Tracks streaming state for a request.	#L174-L212
RequestStreamState.should_send	method	RequestStreamState.should_send(total_tokens: int, finished: bool) -> bool	Determine if output should be sent based on stream_interval.	#L185-L203
RequestStreamState.mark_sent	method	RequestStreamState.mark_sent(total_tokens: int) -> None	Update state after sending output.	#L205-L212