vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker ¶

VoxCPM2 AR talker — PagedAttention pipeline with per-request state.

Architecture

MiniCPM4PagedForVoxCPM2 (base_lm, 28 layers, PagedAttention + fp32 RoPE) → FSQ → MiniCPM4PagedResidualLM (8 layers, PagedAttention, no RoPE) → LocDiT (CFM solver) → AudioVAE → 48kHz waveform

logger `module-attribute` ¶

logger = init_logger(__name__)

VoxCPM2TalkerForConditionalGeneration ¶

Bases: Module

config `instance-attribute` ¶

config = hf_config

has_postprocess `instance-attribute` ¶

has_postprocess = True

has_preprocess `instance-attribute` ¶

has_preprocess = True

have_multimodal_outputs `instance-attribute` ¶

have_multimodal_outputs = True

hf_to_vllm_mapper `class-attribute` `instance-attribute` ¶

hf_to_vllm_mapper = WeightsMapper(
    orig_to_new_prefix={"base_lm.": "model."}
)

make_empty_intermediate_tensors `instance-attribute` ¶

make_empty_intermediate_tensors = (
    make_empty_intermediate_tensors
)

model `instance-attribute` ¶

model = MiniCPM4PagedForVoxCPM2(
    vllm_config=vllm_config,
    prefix=maybe_prefix(prefix, "model"),
)

residual_model `instance-attribute` ¶

residual_model = MiniCPM4PagedResidualLM(
    vllm_config=vllm_config,
    prefix=maybe_prefix(prefix, "residual_model"),
)

tts `property` ¶

tts: Module

vllm_config `instance-attribute` ¶

vllm_config = vllm_config

compute_logits ¶

compute_logits(
    hidden_states: Tensor | OmniOutput,
    sampling_metadata: Any = None,
) -> Tensor | None

embed_input_ids ¶

embed_input_ids(input_ids: Tensor, **_: Any) -> Tensor

forward ¶

forward(
    input_ids: Tensor,
    positions: Tensor,
    intermediate_tensors: IntermediateTensors | None = None,
    inputs_embeds: Tensor | None = None,
    **kwargs: Any,
) -> Tensor | IntermediateTensors

load_weights ¶

load_weights(
    weights: Iterable[tuple[str, Tensor]],
) -> set[str]

make_omni_output ¶

make_omni_output(
    model_outputs: Tensor | OmniOutput, **kwargs: Any
) -> OmniOutput

on_requests_finished ¶

on_requests_finished(
    finished_req_ids: set[str] | list[str],
) -> None

postprocess ¶

postprocess(
    hidden_states: Tensor, **info: Any
) -> dict[str, Any]

preprocess ¶

preprocess(
    input_ids: Tensor,
    input_embeds: Tensor | None,
    **info_dict: Any,
) -> tuple[Tensor, Tensor, dict[str, Any]]

build_cjk_split_map ¶

build_cjk_split_map(tokenizer: Any) -> dict[int, list[int]]

Build {multichar_cjk_token_id: [single_char_ids]} from tokenizer vocab.

build_voxcpm2_prompt ¶

build_voxcpm2_prompt(
    hf_config: Any,
    tokenizer: Any,
    split_map: dict[int, list[int]],
    text: str,
    ref_audio: Any | None = None,
    ref_sr: int | None = None,
    ref_text: str | None = None,
    voice_profile: dict[str, Any] | None = None,
) -> dict[str, Any]

Build a VoxCPM2 prefill prompt whose prompt_token_ids length matches the talker-side prefill length.

Used by both online serving (serving_speech._build_voxcpm2_prompt) and the offline example, so the talker-side length assertion never fires.

is_cjk_char ¶

is_cjk_char(c: str) -> bool

Check if a character is a CJK ideograph.

split_multichar_chinese ¶

split_multichar_chinese(
    token_ids: list[int], split_map: dict[int, list[int]]
) -> list[int]

Replace multichar Chinese token IDs with single-char IDs (idempotent).

vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker ¶

logger module-attribute ¶