mistral_common.protocol.instruct.messages

@classmethod
def from_openai(cls, openai_message: dict[str, Any]) -> "AssistantMessage":
    r"""Create an AssistantMessage from OpenAI format.

    Handles conversion of OpenAI's `reasoning`/`reasoning_content` fields to
    Mistral's ThinkChunk format.

    Args:
        openai_message: Dictionary matching OpenAI's assistant message schema.
            Can contain "content", `tool_calls`, "reasoning", or `reasoning_content`.

    Returns:
        AssistantMessage instance with thinking content converted to ThinkChunk
        and `tool_calls` parsed from OpenAI format.

    Raises:
        InvalidAssistantMessageException: If message has both thinking chunks
            in `content` and a top-level `reasoning`/`reasoning_content` field.
        ValueError: If both `reasoning` and `reasoning_content` are present but unequal.
    """
    openai_tool_calls = openai_message.get("tool_calls", None)
    if openai_tool_calls is None:
        tools_calls: list[ToolCall] | None = None
    elif isinstance(openai_tool_calls, list):
        tools_calls = []
        for openai_tool_call in openai_tool_calls or []:
            tools_calls.append(ToolCall.from_openai(openai_tool_call))
    else:
        raise ValueError(f"tool_calls must be a list, got {type(openai_tool_calls)}")
    content = cls._content_from_openai(openai_message.get("content"))

    reasoning_content: str | None = openai_message.get("reasoning_content")
    reasoning: str | None = openai_message.get("reasoning")

    match reasoning_content, reasoning:
        case None, None:
            openai_thinking = None
        case None, _:
            openai_thinking = reasoning
        case _, None:
            openai_thinking = reasoning_content
        case _, _:
            if reasoning_content != reasoning:
                raise ValueError("`reasoning_content` and `reasoning` should be equal.")
            openai_thinking = reasoning

    if openai_thinking is not None:
        has_thinking_chunks = isinstance(content, list) and any(isinstance(chunk, ThinkChunk) for chunk in content)
        if has_thinking_chunks:
            raise InvalidAssistantMessageException(
                "Message cannot have both thinking chunks in `content` and a top-level"
                " `reasoning` or `reasoning_content` field."
            )

        reasoning_chunk = ThinkChunk(thinking=openai_thinking, closed=True)
        if isinstance(content, str):
            content = [reasoning_chunk, TextChunk(text=content)]
        elif content is None:
            content = [reasoning_chunk]
        else:
            content.insert(0, reasoning_chunk)

    return cls.model_validate(
        {
            "role": openai_message["role"],
            "content": content,
            "tool_calls": tools_calls,
        }
    )

def to_openai(
    self,
    reasoning_field_format: ReasoningFieldFormat | None = None,
) -> dict[str, Any]:
    r"""Convert this assistant message to OpenAI format.

    Handles conversion of thinking chunks to various OpenAI-compatible formats.

    Args:
        reasoning_field_format: Format for serializing thinking content:
            - `None`: Defaults to `thinking_chunks` but emits FutureWarning if
              content contains ThinkChunk (will change to reasoning in 1.13.0)
            - `thinking_chunks`: Keep thinking as inline chunks
            - reasoning: Use flat "reasoning" field (vLLM convention)
            - `reasoning_content`: Use flat `reasoning_content` field (SGLang)

    Returns:
        Dictionary with "role" set to "assistant", and `content`/`tool_calls` serialized.

    Raises:
        InvalidAssistantMessageException: If ThinkChunks are not leading (must
            appear before any other content chunks).
        RuntimeError: If content chunks are not all ThinkChunk or TextChunk
            when using `reasoning`/`reasoning_content` formats.
        ValueError: If `reasoning_field_format` is not supported.
    """
    out_dict: dict[str, Any] = {
        "role": self.role,
    }
    if self.tool_calls is not None:
        out_dict["tool_calls"] = [tool_call.to_openai() for tool_call in self.tool_calls]

    if self.content is None:
        return out_dict

    if isinstance(self.content, str):
        out_dict["content"] = self.content
        return out_dict

    last_think_idx: int = -1
    for i, chunk in enumerate(self.content):
        if isinstance(chunk, ThinkChunk):
            if (i - last_think_idx) > 1:
                raise InvalidAssistantMessageException(
                    "ThinkChunks must be leading: all ThinkChunks must appear before any other content chunk."
                )
            last_think_idx = i

    if reasoning_field_format is None and last_think_idx >= 0:
        warnings.warn(
            "`convert_thinking_format` defaults to 'thinking_chunks' but will change to 'reasoning' "
            "in 1.13.0. Pass `reasoning_field_format` explicitly to silence this warning.",
            FutureWarning,
            stacklevel=2,
        )

    match reasoning_field_format:
        case None | ReasoningFieldFormat.thinking_chunks:
            out_dict["content"] = self._content_to_openai(self.content)
        case ReasoningFieldFormat.reasoning | ReasoningFieldFormat.reasoning_content:
            think_chunks, content_chunks = self.content[: last_think_idx + 1], self.content[last_think_idx + 1 :]
            if not _are_think_chunks(think_chunks) or not _are_text_chunks(content_chunks):
                raise RuntimeError("Impossible, only think or content chunks should have been present.")
            if len(think_chunks) > 0:
                out_dict[reasoning_field_format.value] = "\n".join(tc.thinking for tc in think_chunks)

            if len(content_chunks) == 1:
                out_dict["content"] = content_chunks[0].text
            elif content_chunks:
                out_dict["content"] = self._content_to_openai(content_chunks)
        case _:
            raise ValueError(f"{reasoning_field_format=} is not supported.")

    return out_dict

@classmethod
def from_openai(cls, openai_message: dict[str, Any]) -> "BaseMessage":
    r"""Create a message instance from OpenAI format.

    Must be implemented by concrete subclasses.

    Args:
        openai_message: Dictionary matching OpenAI's message schema.

    Returns:
        Message instance of the appropriate subclass.

    Raises:
        NotImplementedError: Always, as this is an abstract method.
    """
    raise NotImplementedError(f"from_openai method not implemented for {cls.__name__}.")

def to_openai(self) -> dict[str, Any]:
    r"""Convert this message to OpenAI format.

    Must be implemented by concrete subclasses.

    Returns:
        Dictionary matching OpenAI's message schema.

    Raises:
        NotImplementedError: Always, as this is an abstract method.
    """
    raise NotImplementedError(f"to_openai method not implemented for {type(self).__name__}")

@classmethod
def from_openai(cls, openai_message: dict[str, Any]) -> "SystemMessage":
    r"""Create a SystemMessage from OpenAI format.

    Args:
        openai_message: Dictionary with "role" and "content" keys.

    Returns:
        SystemMessage instance with content deserialized from OpenAI format.
    """
    return cls.model_validate(
        {"role": openai_message["role"], "content": cls._content_from_openai(openai_message["content"])}
    )

def to_openai(self) -> dict[str, Any]:
    r"""Convert this system message to OpenAI format.

    Returns:
        Dictionary with "role" set to "system" and "content" serialized.
    """
    return {"role": self.role, "content": self._content_to_openai(self.content)}

@classmethod
def from_openai(cls, openai_message: dict[str, Any]) -> "ToolMessage":
    r"""Create a ToolMessage from OpenAI format.

    Args:
        openai_message: Dictionary with "role", `tool_call_id`, "content",
            and optionally "name" keys.

    Returns:
        ToolMessage instance with content and `tool_call_id` parsed from OpenAI format.
    """
    content = cls._content_from_openai(openai_message["content"])
    tool_message = cls.model_validate(
        {
            "role": openai_message["role"],
            "tool_call_id": openai_message["tool_call_id"],
            "content": content,
            "name": openai_message.get("name"),
        }
    )
    return tool_message

def to_openai(self) -> dict[str, Any]:
    r"""Convert this tool message to OpenAI format.

    Returns:
        Dictionary with "role" set to "tool", `tool_call_id`, and "content".

    Raises:
        AssertionError: If `tool_call_id` is `None` (required for tool messages).
    """
    assert self.tool_call_id is not None, "tool_call_id must be provided for tool messages."
    return {
        "role": self.role,
        "tool_call_id": self.tool_call_id,
        "content": self._content_to_openai(self.content),
    }

@classmethod
def from_openai(cls, openai_message: dict[str, Any]) -> "UserMessage":
    r"""Create a UserMessage from OpenAI format.

    Args:
        openai_message: Dictionary with "role" and "content" keys.

    Returns:
        UserMessage instance with content deserialized from OpenAI format.
    """
    return cls.model_validate(
        {"role": openai_message["role"], "content": cls._content_from_openai(openai_message["content"])}
    )

def to_openai(self) -> dict[str, Any]:
    r"""Convert this user message to OpenAI format.

    Returns:
        Dictionary with "role" set to "user" and "content" serialized.
    """
    return {"role": self.role, "content": self._content_to_openai(self.content)}