# -*- coding: utf-8 -*-
"""The DashScope formatter module (OpenAI-compatible format)."""
import base64
from typing import Any
from fnmatch import fnmatch
from abc import ABC
from pydantic import Field
from ._formatter_base import FormatterBase
from .._logging import logger
from ..message import (
Msg,
TextBlock,
ThinkingBlock,
ToolResultBlock,
URLSource,
DataBlock,
ToolCallBlock,
Base64Source,
HintBlock,
)
class _DashScopeFormatterBase(FormatterBase, ABC):
"""Base class for DashScope formatters (OpenAI-compatible format),
providing shared data block formatting logic."""
input_types: list[str] = Field(
default_factory=lambda: [
"text/plain",
"image/*",
"audio/*",
"video/*",
],
description=(
"The supported input types, aligned with the model card's "
"``input_types`` field. Media types (non ``text/plain`` / "
"``application/x-thinking`` entries) are used to filter "
"``DataBlock``\\s; ``application/x-thinking`` enables passing "
"``reasoning_content`` back to the API."
),
)
@property
def supported_input_media_types(self) -> list[str]:
"""Derive supported media types from :attr:`input_types`, excluding
``text/plain`` and ``application/x-thinking``."""
return [
t
for t in self.input_types
if t not in ("text/plain", "application/x-thinking")
]
@property
def supports_thinking_input(self) -> bool:
"""Return ``True`` if ``application/x-thinking`` is listed in
:attr:`input_types`, meaning the model accepts ``reasoning_content``
in the conversation history."""
return "application/x-thinking" in self.input_types
def _format_dashscope_data_block(
self,
block: DataBlock,
) -> dict[str, Any] | None:
"""Format a DataBlock into the OpenAI-compatible format for
DashScope API.
Supports:
- Images: ``{"type": "image_url", "image_url": {"url": ...}}``
- Videos: ``{"type": "video_url", "video_url": {"url": ...}}``
- Audio: ``{"type": "input_audio", "input_audio": {...}}``
Args:
block (`DataBlock`):
The DataBlock to format.
Returns:
`dict[str, Any] | None`:
A dictionary representing the formatted DataBlock, or ``None``
if the media type is unsupported.
"""
if not any(
fnmatch(block.source.media_type, pattern)
for pattern in self.supported_input_media_types
):
logger.warning(
"Unsupported media type %s for DashScope API. Supported "
"types: %s. This block will be skipped.",
block.source.media_type,
", ".join(self.supported_input_media_types),
)
return None
main_type = block.source.media_type.split("/")[0]
if main_type == "image":
return self._format_image_source(block.source)
if main_type == "video":
return self._format_video_source(block.source)
if main_type == "audio":
return self._format_audio_source(block.source)
logger.warning(
"Unsupported main media type %s for DashScope API. "
"This block will be skipped.",
main_type,
)
return None
@staticmethod
def _format_image_source(
source: URLSource | Base64Source,
) -> dict[str, Any]:
"""Convert an image source to OpenAI-compatible ``image_url`` format.
Local ``file://`` URLs are read from disk and converted to base64
data URIs. Remote URLs are passed through unchanged.
"""
if isinstance(source, Base64Source):
url = f"data:{source.media_type};base64,{source.data}"
elif isinstance(source, URLSource):
url_str = str(source.url)
if url_str.startswith("file://"):
local_path = url_str.removeprefix("file://")
with open(local_path, "rb") as f:
encoded = base64.b64encode(f.read()).decode("utf-8")
url = f"data:{source.media_type};base64,{encoded}"
else:
url = url_str
else:
raise ValueError(f"Unsupported image source type: {type(source)}")
return {
"type": "image_url",
"image_url": {"url": url},
}
@staticmethod
def _format_video_source(
source: URLSource | Base64Source,
) -> dict[str, Any]:
"""Convert a video source to DashScope's ``video_url`` format
(OpenAI-compatible extension).
Local ``file://`` URLs are read from disk and converted to base64
data URIs. Remote URLs are passed through unchanged.
"""
if isinstance(source, Base64Source):
url = f"data:{source.media_type};base64,{source.data}"
elif isinstance(source, URLSource):
url_str = str(source.url)
if url_str.startswith("file://"):
local_path = url_str.removeprefix("file://")
with open(local_path, "rb") as f:
encoded = base64.b64encode(f.read()).decode("utf-8")
url = f"data:{source.media_type};base64,{encoded}"
else:
url = url_str
else:
raise ValueError(f"Unsupported video source type: {type(source)}")
return {
"type": "video_url",
"video_url": {"url": url},
}
@staticmethod
def _format_audio_source(
source: URLSource | Base64Source,
) -> dict[str, Any]:
"""Convert an audio source to DashScope ``input_audio`` format.
DashScope's compatible API accepts URLs directly in the ``data``
field (unlike standard OpenAI which requires base64). Local files
are still read and base64-encoded.
"""
if isinstance(source, Base64Source):
fmt = source.media_type.split("/")[-1]
return {
"type": "input_audio",
"input_audio": {
"data": source.data,
"format": fmt,
},
}
if isinstance(source, URLSource):
url_str = str(source.url)
fmt = source.media_type.split("/")[-1]
if url_str.startswith("file://"):
local_path = url_str.removeprefix("file://")
with open(local_path, "rb") as f:
data = base64.b64encode(f.read()).decode("utf-8")
return {
"type": "input_audio",
"input_audio": {
"data": data,
"format": fmt,
},
}
else:
return {
"type": "input_audio",
"input_audio": {
"data": url_str,
"format": fmt,
},
}
raise ValueError(f"Unsupported audio source type: {type(source)}")
class DashScopeChatFormatter(_DashScopeFormatterBase):
"""The DashScope formatter class for chatbot scenario (OpenAI-compatible
format), where only a user and an agent are involved. We use the ``role``
field to identify different entities in the conversation.
This formatter outputs messages in the OpenAI Chat Completions format,
with DashScope-specific extensions for video (``video_url``) and
thinking (``reasoning_content``).
"""
# pylint: disable=too-many-branches
async def format(
self,
msgs: list[Msg],
) -> list[dict[str, Any]]:
"""Format message objects into DashScope OpenAI-compatible format.
Args:
msgs (`list[Msg]`):
The list of message objects to format.
Returns:
`list[dict[str, Any]]`:
The formatted messages as a list of dictionaries.
"""
self.assert_list_of_msgs(msgs)
formatted_msgs: list[dict] = []
i = 0
while i < len(msgs):
msg = msgs[i]
content_blocks: list[dict] = []
tool_calls = []
thinking_parts: list[str] = []
for block in msg.get_content_blocks():
if isinstance(block, TextBlock):
content_blocks.append({"type": "text", "text": block.text})
elif isinstance(block, DataBlock):
formatted_block = self._format_dashscope_data_block(
block,
)
if formatted_block:
content_blocks.append(formatted_block)
elif isinstance(block, HintBlock):
if content_blocks or tool_calls or thinking_parts:
msg_openai: dict[str, Any] = {
"role": msg.role,
"content": content_blocks or None,
}
if tool_calls:
msg_openai["tool_calls"] = tool_calls
if thinking_parts:
msg_openai["reasoning_content"] = "\n".join(
thinking_parts,
)
formatted_msgs.append(msg_openai)
content_blocks = []
tool_calls = []
thinking_parts = []
if isinstance(block.hint, str):
formatted_msgs.append(
{
"role": "user",
"content": [
{"type": "text", "text": block.hint},
],
},
)
else:
hint_parts: list[dict] = []
for sub in block.hint:
if isinstance(sub, TextBlock):
hint_parts.append(
{"type": "text", "text": sub.text},
)
elif isinstance(sub, DataBlock):
formatted_sub = (
self._format_dashscope_data_block(
sub,
)
)
if formatted_sub:
hint_parts.append(formatted_sub)
if hint_parts:
formatted_msgs.append(
{"role": "user", "content": hint_parts},
)
elif isinstance(block, ToolCallBlock):
tool_calls.append(
{
"id": block.id,
"type": "function",
"function": {
"name": block.name,
"arguments": block.input,
},
},
)
elif isinstance(block, ThinkingBlock):
if self.supports_thinking_input:
thinking_parts.append(block.thinking)
elif isinstance(block, ToolResultBlock):
if content_blocks or tool_calls or thinking_parts:
msg_flush: dict[str, Any] = {
"role": msg.role,
"content": content_blocks or None,
}
if tool_calls:
msg_flush["tool_calls"] = tool_calls
if thinking_parts:
msg_flush["reasoning_content"] = "\n".join(
thinking_parts,
)
formatted_msgs.append(msg_flush)
content_blocks = []
tool_calls = []
thinking_parts = []
(
textual_output,
multimodal_data,
) = self.convert_tool_result_to_string(block.output)
formatted_msgs.append(
{
"role": "tool",
"tool_call_id": block.id,
"content": textual_output,
"name": block.name,
},
)
if multimodal_data:
promo_content = []
for item in multimodal_data:
if isinstance(item, TextBlock):
promo_content.append(
{"type": "text", "text": item.text},
)
elif isinstance(item, DataBlock):
fmt_item = self._format_dashscope_data_block(
item,
)
if fmt_item is not None:
promo_content.append(fmt_item)
if promo_content:
formatted_msgs.append(
{
"role": "user",
"content": promo_content,
},
)
else:
logger.warning(
"Unsupported block type %s in the message, skipped.",
type(block),
)
msg_dashscope: dict[str, Any] = {
"role": msg.role,
"content": content_blocks or None,
}
if tool_calls:
msg_dashscope["tool_calls"] = tool_calls
if thinking_parts:
msg_dashscope["reasoning_content"] = "\n".join(thinking_parts)
if (
msg_dashscope["content"]
or msg_dashscope.get("tool_calls")
or msg_dashscope.get("reasoning_content")
):
formatted_msgs.append(msg_dashscope)
i += 1
return formatted_msgs
class DashScopeMultiAgentFormatter(_DashScopeFormatterBase):
"""DashScope formatter for multi-agent conversations (OpenAI-compatible
format), where more than a user and an agent are involved.
.. note:: This formatter will combine previous messages (except tool
calls/results) into a history section in the first system message with
the conversation history prompt.
.. note:: For tool calls/results, they will be presented as separate
messages as required by the API. Therefore, the tool calls/results
messages are expected to be placed at the end of the input messages.
.. tip:: Telling the assistant's name in the system prompt is very
important in multi-agent conversations. So that LLM can know who it
is playing as.
"""
conversation_history_prompt: str = Field(
description="The conversation history prompt.",
default=(
"# Conversation History\n"
"The content between tags contains "
"your conversation history\n"
),
)
async def format(self, msgs: list[Msg]) -> list[dict]:
"""Format input messages into the structure required by the DashScope
OpenAI-compatible API.
To support multi-agent conversations, this formatter processes messages
as follows:
- Prepends an instruction before the first conversation history
section.
- Combines conversation turns into a history section, where each entry
is formatted as ``{name}: {content}``.
- Wraps the conversation history with ```` and ````
tags.
Returns:
`list[dict[str, Any]]`:
A list of dictionaries formatted for the DashScope API.
"""
formatted_msgs = []
start_index = 0
if len(msgs) > 0 and msgs[0].role == "system":
formatted_msgs.append(
await self._format_system_message(msgs[0]),
)
start_index = 1
is_first_agent_message = True
async for typ, group in self._group_messages(msgs[start_index:]):
match typ:
case "tool_sequence":
formatted_msgs.extend(
await self._format_tool_sequence(group),
)
case "agent_message":
formatted_msgs.extend(
await self._format_agent_message(
group,
is_first_agent_message,
),
)
is_first_agent_message = False
return formatted_msgs
async def _format_tool_sequence(
self,
msgs: list[Msg],
) -> list[dict[str, Any]]:
"""Given a sequence of tool call/result messages, format them into
the required format for the DashScope API."""
return await DashScopeChatFormatter(
input_types=self.input_types,
).format(msgs)
async def _format_agent_message(
self,
msgs: list[Msg],
is_first: bool = True,
) -> list[dict[str, Any]]:
"""Given a sequence of messages without tool calls/results, format
them into a user message with conversation history tags."""
if is_first:
conversation_history_prompt = self.conversation_history_prompt
else:
conversation_history_prompt = ""
formatted_msgs: list[dict] = []
conversation_blocks: list = []
accumulated_text = []
media_blocks: list[dict] = []
for msg in msgs:
for block in msg.get_content_blocks():
if isinstance(block, TextBlock):
accumulated_text.append(f"{msg.name}: {block.text}")
elif isinstance(block, DataBlock):
formatted_block = self._format_dashscope_data_block(
block,
)
if formatted_block is not None:
media_blocks.append(formatted_block)
if accumulated_text:
conversation_blocks.append(
{"text": "\n".join(accumulated_text)},
)
if conversation_blocks:
if conversation_blocks[0].get("text"):
conversation_blocks[0]["text"] = (
conversation_history_prompt
+ "\n"
+ conversation_blocks[0]["text"]
)
else:
conversation_blocks.insert(
0,
{"text": conversation_history_prompt + "\n"},
)
if conversation_blocks[-1].get("text"):
conversation_blocks[-1]["text"] += "\n"
else:
conversation_blocks.append({"text": ""})
conversation_blocks_text = "\n".join(
b.get("text", "") for b in conversation_blocks
)
content_list: list[dict[str, Any]] = []
if conversation_blocks_text:
content_list.append(
{"type": "text", "text": conversation_blocks_text},
)
content_list.extend(media_blocks)
if content_list:
formatted_msgs.append({"role": "user", "content": content_list})
return formatted_msgs
@staticmethod
async def _format_system_message(
msg: Msg,
) -> dict[str, Any]:
"""Format system message for DashScope API."""
return {
"role": "system",
"content": msg.get_text_content(),
}