vllm.renderers.inkling_encoding ¶

def _flatten_text_content(content: Any) -> str:
    """Flatten a tool-response content (string or text parts) to text."""
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    parts: list[str] = []
    for kind, text in _iter_render_parts(content):
        if kind != "text":
            raise ValueError(
                "Inkling tool response content must be text, "
                f"got a part of kind {kind!r}"
            )
        parts.append(text)
    return "".join(parts)

def _iter_render_parts(content: Any) -> Iterator[tuple[str, str]]:
    """Yield (kind, text) per content part: kind in {text, image, audio}."""
    if content is None:
        return
    if isinstance(content, str):
        if content:
            yield ("text", content)
        return
    if not isinstance(content, Sequence) or isinstance(content, (bytes, bytearray)):
        raise TypeError("message content must be a string or a sequence of parts")
    for part in content:
        if isinstance(part, str):
            yield ("text", part)
            continue
        if not isinstance(part, Mapping):
            raise TypeError(f"content part must be mapping, got {type(part).__name__}")
        ptype = part.get("type")
        if ptype in (None, "text", "input_text"):
            text = part.get("text", "")
            yield ("text", text if isinstance(text, str) else "")
        elif ptype in _IMAGE_PART_TYPES:
            yield ("image", "")
        elif ptype in _AUDIO_PART_TYPES:
            yield ("audio", "")
        else:
            raise ValueError(f"unsupported content part type: {ptype!r}")

def render_inkling_messages(
    messages: Sequence[Mapping[str, Any]],
    tokenizer: InklingTextTokenizer,
    *,
    add_generation_prompt: bool = True,
    tools: Sequence[Mapping[str, Any]] | None = None,
    reasoning_effort: float | None = None,
) -> list[int]:
    """Render chat messages to Inkling input ids.

    PURE renderer: emits Inkling framing plus bare media markers; media
    encoding and placeholder expansion happen later in the MM processor.
    ``add_generation_prompt`` appends the assistant turn opener so the
    model continues into the response.
    """
    input_ids: list[int] = []
    tool_call_id_to_name: dict[str, str] = {}

    # Request-level tools plus per-developer-message tools (Rust renderer
    # semantics) are declared in a single leading system block.
    all_tools = list(tools or [])
    for message in messages:
        if message.get("role") == "developer":
            all_tools.extend(message.get("tools") or [])

    if all_tools:
        _append_message(
            input_ids,
            tokenizer,
            "system",
            "xml",
            _tool_declare_json(all_tools),
            author_name="tool_declare",
        )

    for message in messages:
        role = _expect_role(message)
        if role not in {"system", "developer"} and reasoning_effort is not None:
            _append_reasoning_effort(input_ids, tokenizer, reasoning_effort)
            reasoning_effort = None

        if role == "tool":
            tool_name = message.get("name") or tool_call_id_to_name.get(
                str(message.get("tool_call_id") or ""), ""
            )
            _append_message(
                input_ids,
                tokenizer,
                "tool",
                "text",
                _flatten_text_content(message.get("content")),
                author_name=str(tool_name),
            )
            continue

        if role == "assistant":
            reasoning_content = message.get("reasoning")
            if reasoning_content is None:
                reasoning_content = message.get("reasoning_content")
            if reasoning_content:
                if not isinstance(reasoning_content, str):
                    raise TypeError(
                        "assistant reasoning_content must be a string for "
                        "Inkling rendering"
                    )
                _append_message(
                    input_ids,
                    tokenizer,
                    "assistant",
                    "thinking",
                    reasoning_content,
                )

        for kind, text in _iter_render_parts(message.get("content")):
            _append_message(input_ids, tokenizer, role, kind, text)

        if role == "assistant":
            for tool_call in message.get("tool_calls") or []:
                name, args = _tool_call_name_and_args(tool_call)
                tool_call_id = _as_mapping(tool_call).get("id")
                if tool_call_id:
                    tool_call_id_to_name[str(tool_call_id)] = name
                _append_message(
                    input_ids,
                    tokenizer,
                    "assistant",
                    "invoke_tool_json",
                    _tool_call_json(name, args),
                    author_name=name,
                )

            input_ids.append(tokenizer.encode_special(CONTENT_MODEL_END_SAMPLING))

    if reasoning_effort is not None:
        _append_reasoning_effort(input_ids, tokenizer, reasoning_effort)

    if add_generation_prompt:
        input_ids.append(tokenizer.encode_special(MESSAGE_MODEL))
    return input_ids