def _as_media_list(value: Any) -> List[Any]:
"""Normalize one-or-many media references to a list."""
if value is None:
return []
if isinstance(value, (list, tuple)):
return list(value)
return [value]
def _iter_batched_rows(examples: Dict[str, List[Any]]) -> List[Dict[str, Any]]:
"""Convert an HF ``map(batched=True)`` payload into row dictionaries."""
if not examples:
return []
first_key = next(iter(examples))
size = len(examples[first_key])
return [{key: value[idx] for key, value in examples.items()} for idx in range(size)]
def _build_vl_messages(row: Dict[str, Any], data_cfg: Any) -> List[Dict[str, Any]]:
"""Normalize one dataset row into a multimodal chat-template message list."""
messages_key = getattr(data_cfg, "messages_key", "messages")
image_key = getattr(data_cfg, "image_key", "image")
text_key = getattr(data_cfg, "text_key", "text")
image_queue = _as_media_list(
row.get(image_key, row.get("images", row.get("image"))),
)
video_queue = _as_media_list(row.get("videos", row.get("video")))
def _normalize_content(content: Any) -> List[Dict[str, Any]]:
if isinstance(content, str):
return [{"type": "text", "text": content}]
if not isinstance(content, list):
raise ValueError(
"VL HuggingFace rows must provide list-style message content "
f"or plain text, but got {type(content)}."
)
out: List[Dict[str, Any]] = []
for item in content:
if isinstance(item, str):
out.append({"type": "text", "text": item})
continue
if not isinstance(item, dict):
raise ValueError(
"VL message content items must be dict/str, "
f"but got {type(item)}."
)
item_type = item.get("type", "text")
if item_type == "text":
out.append({"type": "text", "text": item.get("text", item.get("content", ""))})
continue
if item_type == "image":
image = item.get("image", item.get("path", item.get("url")))
if image is None:
if not image_queue:
raise ValueError(
"VL message declared an image placeholder but no "
f"row-level media was found in '{image_key}'."
)
image = image_queue.pop(0)
out.append({"type": "image", "image": image})
continue
if item_type == "video":
video = item.get("video", item.get("path", item.get("url")))
if video is None:
if not video_queue:
raise ValueError(
"VL message declared a video placeholder but no "
"row-level video field was found."
)
video = video_queue.pop(0)
out.append({"type": "video", "video": video})
continue
raise ValueError(f"Unsupported VL content type: {item_type}")
return out
if row.get(messages_key) is not None:
messages = row[messages_key]
if not isinstance(messages, list):
raise ValueError(
f"data.messages_key='{messages_key}' must point to a list, "
f"but got {type(messages)}."
)
return [
{
"role": message.get("role", "user"),
"content": _normalize_content(message.get("content", "")),
}