Skip to content

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__)

VoxCPM2PostprocessInput

Bases: TypedDict

request_id instance-attribute 

request_id: str

VoxCPM2PreprocessInput

Bases: TypedDict

additional_information instance-attribute 

additional_information: dict[str, Any]

prompt_audio instance-attribute 

prompt_audio: object

prompt_text instance-attribute 

prompt_text: str | list[str] | None

ref_audio instance-attribute 

ref_audio: object

reference_audio instance-attribute 

reference_audio: object

request_id instance-attribute 

request_id: str

text_token_ids instance-attribute 

text_token_ids: list[list[int]]

voice_created_at instance-attribute 

voice_created_at: int | str | None

voice_name instance-attribute 

voice_name: str | list[str] | None

voice_profile instance-attribute 

voice_profile: dict[str, Any] | list[dict[str, Any]] | None

VoxCPM2TalkerForConditionalGeneration

Bases: Module

config instance-attribute 

config = vllm_config.model_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 = (
    self.model.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

get_runner_assisted_full_attention_metadata_request 

get_runner_assisted_full_attention_metadata_request(
    *,
    req_ids: Sequence[str],
    num_reqs: int,
    num_scheduled_tokens: Sequence[int],
    num_computed_tokens: Sequence[int],
    max_num_scheduled_tokens: int,
) -> RunnerAssistedFullAttentionMetadataRequest | None

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: Unpack[VoxCPM2PostprocessInput],
) -> dict[str, Any]

preprocess 

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

set_runner_assisted_full_attention_metadata_context 

set_runner_assisted_full_attention_metadata_context(
    *, enabled: bool, num_reqs: int = 0
) -> None

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: _VoxCPM2PromptConfigLike,
    tokenizer: Any,
    split_map: dict[int, list[int]],
    text: str,
    ref_audio: Sequence[float] | Tensor | 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).