vllm.renderers.cohere ¶

def _build_render_config(
    conversation: list[ConversationMessage],
    chat_template_kwargs: dict[str, Any],
    chat_template: str | None = None,
) -> tuple[str, dict[str, Any]]:
    """Build the ``render_cmd3`` / ``render_cmd4`` config dict.

    ``chat_template`` is the standard vLLM per-request template override
    (from ``ChatCompletionRequest.chat_template`` / the server-side
    ``--chat-template`` flag). When non-``None`` it is forwarded to melody
    as ``template_jinja`` so a single vLLM-native path drives template
    source through the trust guard.

    Returns ``(format, config_dict)`` where ``format`` is either ``"cmd3"``
    or ``"cmd4"``.
    """
    fmt = chat_template_kwargs.get("cohere_format", _DEFAULT_FORMAT)
    if fmt not in _VALID_FORMATS:
        raise ValueError(
            f"Invalid cohere_format={fmt!r}; expected one of {_VALID_FORMATS}"
        )

    # Reject Cohere-only inlets for raw Jinja source. These keys are
    # valid at Cohere's own API surface but not at vLLM's: callers must
    # use the standard vLLM ``chat_template`` request field so the
    # ``--trust-request-chat-template`` guard applies uniformly.
    for cohere_only_key in ("template_jinja", "template"):
        if chat_template_kwargs.get(cohere_only_key) is not None:
            raise ValueError(
                f"chat_template_kwargs.{cohere_only_key!r} is not accepted "
                "in vLLM (it is a Cohere-only field); pass the template "
                "body via the standard 'chat_template' request field "
                "instead (requires --trust-request-chat-template)."
            )

    config: dict[str, Any] = {
        "messages": _conversation_to_melody_messages(
            conversation,
            chat_template_kwargs.get(MESSAGES_CITATIONS_KEY),
        ),
    }

    # ``template_id`` is a selector for one of melody's built-in template
    # variants (not raw source), so it is safe to accept from the client.
    if template_id := chat_template_kwargs.get("template_id"):
        config["template_id"] = template_id
    # Raw template source flows exclusively through the standard vLLM
    # ``chat_template`` field so it is subject to the
    # ``--trust-request-chat-template`` guard in OnlineRenderer.
    if chat_template:
        config["template_jinja"] = chat_template
    # Only support Jinja with vllm
    config["use_jinja"] = True

    # Documents
    documents = chat_template_kwargs.get("documents") or []
    if documents:
        config["documents"] = [_document_to_melody(d) for d in documents]

    # Tools - prefer explicit ``available_tools``, fall back to OpenAI ``tools``
    tools = (
        chat_template_kwargs.get("available_tools")
        or chat_template_kwargs.get("tools")
        or []
    )
    if tools:
        config["available_tools"] = [_tool_to_melody(t) for t in tools]

    # Reasoning toggle (cmd3 + cmd4)
    if (rt := chat_template_kwargs.get("reasoning_type")) is not None:
        config["reasoning_type"] = str(rt)
    elif "thinking" in chat_template_kwargs:
        # Cohere v2 ``thinking: {type: enabled|disabled}`` shorthand
        thinking = chat_template_kwargs["thinking"]
        t = thinking.get("type") if isinstance(thinking, dict) else thinking
        if t in ("enabled", "disabled"):
            config["reasoning_type"] = t

    if (di := chat_template_kwargs.get("dev_instruction")) is not None:
        config["dev_instruction"] = str(di)

    # JSON / structured outputs
    if (rf := chat_template_kwargs.get("response_format")) is not None:
        rf = rf.model_dump() if hasattr(rf, "model_dump") else dict(rf)
        rf_type = rf.get("type")
        if rf_type == "json_object":
            config["json_mode"] = True
        elif rf_type in ("json_schema", "json"):
            schema = rf.get("schema") or rf.get("json_schema")
            if isinstance(schema, dict) and "schema" in schema:
                schema = schema["schema"]
            if schema is not None:
                config["json_schema"] = (
                    schema if isinstance(schema, str) else json.dumps(schema)
                )
    if (js := chat_template_kwargs.get("json_schema")) is not None:
        config["json_schema"] = js if isinstance(js, str) else json.dumps(js)
    if "json_mode" in chat_template_kwargs:
        config["json_mode"] = bool(chat_template_kwargs["json_mode"])

    if fmt == "cmd3":
        if (sm := chat_template_kwargs.get("safety_mode")) is not None:
            config["safety_mode"] = str(sm).lower()
        # citation_quality: ``on`` / ``off``
        cq = chat_template_kwargs.get("citation_quality")
        if cq is None and (co := chat_template_kwargs.get("citation_options")):
            mode = co.get("mode") if isinstance(co, dict) else None
            if mode is not None:
                cq = "on" if str(mode).lower() != "off" else "off"
        if cq is not None:
            config["citation_quality"] = str(cq).lower()
        if "skip_preamble" in chat_template_kwargs:
            config["skip_preamble"] = bool(chat_template_kwargs["skip_preamble"])
    else:  # cmd4
        # cmd4 uses ``grounding`` rather than safety_mode/citation_quality.
        # melody's cmd4 only accepts ``unknown`` / ``enabled`` / ``disabled``,
        # so the Cohere v2 ``citation_options.mode`` values
        # (``FAST`` / ``ACCURATE`` / ``OFF``) have to be normalized -- a
        # raw lowercased passthrough would raise from
        # ``render_cmd4`` (cmd4 has no fast/accurate distinction at the
        # prompt-template layer; both request grounding-on).
        if (gr := chat_template_kwargs.get("grounding")) is not None:
            config["grounding"] = _normalize_cmd4_grounding(gr)
        elif co := chat_template_kwargs.get("citation_options"):
            mode = co.get("mode") if isinstance(co, dict) else None
            if mode is not None:
                config["grounding"] = _normalize_cmd4_grounding(mode)
        if (pi := chat_template_kwargs.get("platform_instruction")) is not None:
            config["platform_instruction"] = str(pi)

    # Anything we didn't explicitly interpret above is forwarded to melody
    # as a Jinja template variable. This matches vLLM's documented contract
    # for ``chat_template_kwargs`` ("kwargs accessible by the template")
    # and lets callers write e.g. ``{"reasoning_effort": "low"}`` directly
    # without a nested ``additional_template_fields`` wrapper.
    extra = {
        k: v
        for k, v in chat_template_kwargs.items()
        if k not in _RENDERER_CONSUMED_KEYS
    }
    if extra:
        config["additional_template_fields"] = extra

    return fmt, config

def _content_blocks(content: Any) -> list[dict[str, Any]]:
    """Convert OpenAI ``ConversationMessage.content`` into melody content blocks.

    The chat_utils ``content_format="openai"`` produces a list of
    ``{type: text, text: ...}`` dicts already; we pass those through with
    minimal coercion. Plain string content is wrapped in a single text block.
    Image / multimodal placeholder dicts (``{"type": "image"}``) are
    forwarded as-is so that templates can reference image placeholders that
    the upstream tokenizer expands separately.
    """
    if content is None:
        return []
    if isinstance(content, str):
        return [{"type": MelodyContentType.TEXT, "text": content}]
    blocks: list[dict[str, Any]] = []
    for part in content:
        if isinstance(part, str):
            blocks.append({"type": MelodyContentType.TEXT, "text": part})
            continue
        if not isinstance(part, dict):
            raise TypeError(f"Unexpected content part: {part!r}")
        part_type = part.get("type", MelodyContentType.TEXT)
        if part_type in ("text", "input_text", "output_text", "refusal"):
            blocks.append(
                {"type": MelodyContentType.TEXT, "text": part.get("text", "")}
            )
        elif part_type == MelodyContentType.THINKING:
            blocks.append(
                {
                    "type": MelodyContentType.THINKING,
                    "thinking": part.get("thinking", ""),
                }
            )
        elif part_type == MelodyContentType.IMAGE:
            blocks.append(
                {
                    "type": MelodyContentType.IMAGE,
                    "image": {
                        "template_placeholder": part.get(
                            "template_placeholder", "<image>"
                        ),
                    },
                }
            )
        elif part_type == MelodyContentType.DOCUMENT:
            doc = part.get("document")
            if isinstance(doc, dict):
                blocks.append({"type": MelodyContentType.DOCUMENT, "document": doc})
            else:
                # Fall back to wrapping arbitrary string as text.
                blocks.append({"type": MelodyContentType.TEXT, "text": json.dumps(doc)})
        elif part_type == "tool_reference":
            # Tool references are rendered by name; emit as text so the
            # renderer downstream is content-format agnostic.
            blocks.append(
                {
                    "type": MelodyContentType.TEXT,
                    "text": part.get("name") or part.get("text", ""),
                }
            )
        else:
            # Unknown block type: render as text fallback.
            text = part.get("text") or part.get(part_type) or ""
            text_str = text if isinstance(text, str) else json.dumps(text)
            blocks.append({"type": MelodyContentType.TEXT, "text": text_str})
    return blocks

def _conversation_to_melody_messages(
    conversation: list[ConversationMessage],
    messages_citations: dict[int, list[dict[str, Any]]] | None = None,
) -> list[dict[str, Any]]:
    """Convert vLLM ``ConversationMessage``s into melody's message dict shape.

    ``messages_citations`` is an optional index-keyed lookup of
    per-message citations from the request, populated by
    ``CohereServingChatV2`` under the
    :data:`MESSAGES_CITATIONS_KEY` chat_template_kwargs entry. Values
    must already be in melody's ``FilterCitation`` dict shape (see
    ``CohereServingChatV2._sdk_citation_to_melody``). Index is the
    position in the *input* ``conversation`` list, which currently maps
    1-to-1 with ``CohereChatV2Request.messages`` for text-only Cohere
    requests. This mapping breaks silently if ``parse_chat_messages``
    ever expands a single input message into multiple entries (e.g.
    multi-modal splitting), which is not the case today.
    """
    out: list[dict[str, Any]] = []
    for i, msg in enumerate(conversation):
        role = _role_to_melody(msg.get("role", "user"))
        content_blocks = _content_blocks(msg.get("content"))

        # Treat reasoning content as a thinking block on assistant turns
        # so multi-turn reasoning is preserved in the rendered prompt.
        #
        # Cohere models output thought as either a ``thinking``
        # block (reasoning models) or a ``tool_plan`` field (older non-
        # reasoning Command models), but vLLM's ConversationMessage has a
        # single unified ``reasoning`` field that drops that difference.
        # The cmd3 / cmd4 jinja templates render ``thinking`` blocks in both
        # the case of tool calls and regular thinking blocks, so this is fine
        # from the renderer's perspective.
        reasoning = msg.get("reasoning") or msg.get("reasoning_content")
        if role == "chatbot" and reasoning:
            content_blocks.insert(
                0,
                {"type": MelodyContentType.THINKING, "thinking": reasoning},
            )

        tool_calls = [_normalize_tool_call(tc) for tc in (msg.get("tool_calls") or [])]

        out_msg: dict[str, Any] = {
            "role": role,
            "content": content_blocks,
            "tool_calls": tool_calls,
        }
        tool_call_id = msg.get("tool_call_id")
        if tool_call_id:
            out_msg["tool_call_id"] = tool_call_id
        if messages_citations and (cites := messages_citations.get(i)):
            out_msg["citations"] = cites
        out.append(out_msg)
    return out

def _document_to_melody(doc: Any) -> dict[str, Any]:
    """Coerce a Cohere v2 document into melody's ``Document`` (dict) shape."""
    match doc:
        case str():
            return {"text": doc}
        # Cohere v2 wraps documents in {id, data: {...}}; the ``**rest``
        # capture keeps the outer ``id`` (and any other outer keys) so we
        # can lift them onto the payload without losing them.
        case {"data": dict() as data, **rest}:
            payload = dict(data)
            if (outer_id := rest.get("id")) is not None and "id" not in payload:
                payload["id"] = outer_id
            return payload
        case dict():
            return dict(doc)
        case _:
            raise TypeError(f"Unsupported document type: {type(doc).__name__}")

def _normalize_cmd4_grounding(value: Any) -> str:
    """Coerce a user-facing grounding/citation mode to melody's vocab.

    Raises ``ValueError`` rather than letting an unrecognized value
    slip through to ``render_cmd4`` and surface as a generic
    ``Invalid config: grounding`` from melody.
    """
    out = _CMD4_GROUNDING_FROM_MODE.get(value.lower())
    if out is None:
        raise ValueError(
            f"Unrecognized cmd4 grounding value: {value!r}. Expected one "
            f"of ENABLED / DISABLED / FAST / ACCURATE / OFF (from "
            f"citation_options.mode), or the direct melody values "
            f"enabled / disabled / unknown."
        )
    return out

def _normalize_tool_call(tc: dict[str, Any] | Any) -> dict[str, Any]:
    """Normalize an OpenAI tool call dict into melody's tool call shape.

    melody expects ``{id, name, parameters: <json str>}`` whereas OpenAI
    delivers ``{id, type, function: {name, arguments: <json|str>}}``.
    """
    # Pydantic objects
    if hasattr(tc, "model_dump"):
        tc = tc.model_dump()
    if not isinstance(tc, dict):
        raise TypeError(f"Unexpected tool_call value: {tc!r}")

    fn = tc.get("function") or {}
    name = fn.get("name") or tc.get("name") or ""
    args = fn.get("arguments")
    if args is None:
        args = tc.get("arguments", {})
    # melody expects a JSON-encoded string
    if not isinstance(args, str):
        args = json.dumps(args, ensure_ascii=False)
    return {
        "id": tc.get("id") or "",
        "name": name,
        "parameters": args,
    }

def _role_to_melody(role: str) -> str:
    """Map an OpenAI role to the role string melody expects.

    The cmd3 / cmd4 jinja templates only recognize ``system``, ``user``,
    ``assistant``/``chatbot``, and ``tool``: any other role is silently
    dropped by the template's role-dispatch chain (no fallback branch),
    which produces a malformed prompt without any error. We therefore
    refuse unknown roles up front rather than letting them disappear.

    Aliases:

    * ``assistant`` -> ``chatbot`` (Cohere's historical assistant role
      name used by the templates).
    * ``developer`` -> ``system`` (OpenAI's ``developer`` role is
      documented as high-priority instructions, which maps onto the
      ``system`` slot in Cohere's prompt format).
    """
    role = role.lower()
    if role == "assistant":
        return "chatbot"
    if role == "developer":
        return "system"
    if role in _MELODY_ROLES:
        return role
    raise ValueError(
        f"Unsupported message role for the cohere renderer: {role!r}. "
        "Expected one of: system, developer, user, assistant, chatbot, tool."
    )

def _tool_to_melody(tool: Any) -> dict[str, Any]:
    """Coerce a chat completions tool definition into melody's ``Tool`` shape.

    Accepts either the raw OpenAI tool wrapper ``{type:"function", function:
    {name, description, parameters}}`` or a flat ``{name, description,
    parameters}`` dict (which is what melody itself expects).
    """
    if hasattr(tool, "model_dump"):
        tool = tool.model_dump()
    if not isinstance(tool, dict):
        raise TypeError(f"Unsupported tool type: {type(tool).__name__}")
    if "function" in tool and isinstance(tool["function"], dict):
        fn = tool["function"]
    else:
        fn = tool
    return {
        "name": fn.get("name", ""),
        "description": fn.get("description", "") or "",
        "parameters": fn.get("parameters") or {},
    }