Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / data / online / format.py: 76%
209 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-11-15 07:18 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Shared fixed-format conversion for messages, ShareGPT, and interleaved media.
17Each converter accepts an explicit source contract. They do not infer arbitrary
18source schemas or select training supervision. Explicit format converters own
19message field mapping. Model transforms own chat templates, media encoding,
20labels, and training constraints.
21"""
23import copy
24import os
25import re
26from collections.abc import Mapping, Sequence
27from dataclasses import dataclass
28from typing import Any
30from hyper_parallel.data.constants import ONLINE_SOURCE_PATH_KEY
33_SHAREGPT_ROLE_MAP = {
34 "system": "system",
35 "human": "user",
36 "user": "user",
37 "gpt": "assistant",
38 "assistant": "assistant",
39 "chatgpt": "assistant",
40 "model": "assistant",
41 "bot": "assistant",
42}
45def normalize_conversation(messages: Any) -> list[dict[str, Any]]:
46 """Normalize message fields without rendering templates or selecting supervision.
48 Args:
49 messages: OpenAI role/content or ShareGPT from/value message sequence.
51 Returns:
52 New message dictionaries with normalized roles and content. OpenAI
53 extension fields and ShareGPT loss_mask are preserved. Media content
54 is left untouched; callers own media expansion and encoding.
56 Raises:
57 ValueError: If message containers, required fields or ShareGPT roles are invalid.
58 """
59 if not isinstance(messages, Sequence) or isinstance(messages, (str, bytes)):
60 raise ValueError("Conversation messages must be a sequence of mappings")
62 normalized_messages = []
63 for message in messages:
64 if not isinstance(message, Mapping):
65 raise ValueError("Every conversation message must be a mapping")
67 if "role" in message:
68 tool_only_assistant = message["role"] == "assistant" and bool(message.get("tool_calls"))
69 if "content" not in message and not tool_only_assistant:
70 raise ValueError("A role/content message must contain field 'content'")
71 normalized_message = dict(message)
72 if tool_only_assistant:
73 normalized_message.setdefault("content", "")
74 elif "from" in message:
75 if "value" not in message:
76 raise ValueError("A ShareGPT from/value message must contain field 'value'")
77 source_role = message["from"]
78 try:
79 role = _SHAREGPT_ROLE_MAP[source_role]
80 except (KeyError, TypeError) as exc:
81 raise ValueError(f"Unsupported ShareGPT role: {source_role!r}") from exc
82 normalized_message = {
83 "role": role,
84 "content": message["value"],
85 }
86 if "loss_mask" in message:
87 normalized_message["loss_mask"] = message["loss_mask"]
88 else:
89 raise ValueError("A conversation message must use role/content or from/value fields")
91 normalized_messages.append(normalized_message)
93 return normalized_messages
96def resolve_media_path(media_path: str, source_directory: str) -> str:
97 """Anchor a local media reference while preserving absolute paths and URLs.
99 Args:
100 media_path: Local path or HTTP, HTTPS, or data URL.
101 source_directory: Directory containing the source record file.
103 Returns:
104 Original remote/absolute reference or normalized local path.
106 Note:
107 This path helper is shared by all record styles; it does not convert fields.
109 Example:
110 resolve_media_path("cat.jpg", "/data") -> "/data/cat.jpg"
111 resolve_media_path("https://example.com/cat.jpg", "/data")
112 -> "https://example.com/cat.jpg"
113 """
114 if os.path.isabs(media_path) or media_path.startswith(("http://", "https://", "data:")):
115 return media_path
117 resolved_path = os.path.normpath(os.path.join(source_directory, media_path))
118 return resolved_path
121def resolve_chat_media_paths(sample: Mapping[str, Any]) -> list[dict[str, Any]]:
122 """Copy native messages and anchor media fields to their source file.
124 Args:
125 sample: Raw messages record with optional source-file metadata.
127 Returns:
128 Independent messages preserving roles, order, template fields and media.
129 Without source metadata, relative message references are left unchanged.
131 Raises:
132 ValueError: If messages is empty or contains a non-mapping message.
134 Note:
135 Accepts OpenAI image_url blocks, HF typed media blocks, or native model
136 messages. Each block retains its input style; only its path is resolved.
137 ShareGPT from/value and Alpaca instruction/output are not parsed here.
139 Example:
140 HF-style input (source file /data/train.jsonl):
141 {"messages": [{"role": "user", "content": [
142 {"type": "image", "path": "cat.jpg"}]}]}
143 HF-style output:
144 [{"role": "user", "content": [
145 {"type": "image", "path": "/data/cat.jpg"}]}]
146 Field names stay unchanged; a final assistant reply is not required.
147 """
148 source_messages = sample.get("messages")
149 if not isinstance(source_messages, list):
150 raise ValueError("Omni sample messages must be a list")
152 if not source_messages:
153 raise ValueError("Omni sample must contain non-empty messages")
155 messages = copy.deepcopy(source_messages)
156 source_path = sample.get(ONLINE_SOURCE_PATH_KEY)
157 source_directory = os.path.dirname(os.path.abspath(source_path)) if source_path else None
158 for message in messages:
159 if not isinstance(message, dict):
160 raise ValueError("Omni sample messages must contain mapping records")
162 content = message.get("content")
163 if not isinstance(content, list) or source_directory is None:
164 continue
165 for block in content:
166 _resolve_media_block_paths(block, source_directory)
167 return messages
170def _resolve_media_block_paths(block: Any, source_directory: str) -> None:
171 """Resolve paths in place, preserving OpenAI, HF, or native media fields."""
172 if not isinstance(block, dict):
173 return
175 block_type = block.get("type")
176 if block_type == "image_url":
177 image_url = block.get("image_url")
178 if isinstance(image_url, str):
179 block["image_url"] = resolve_media_path(image_url, source_directory)
180 elif isinstance(image_url, dict) and isinstance(image_url.get("url"), str):
181 image_url["url"] = resolve_media_path(image_url["url"], source_directory)
182 return
184 if block_type not in ("image", "video", "audio"):
185 return
187 for field in (block_type, "path", "url"):
188 value = block.get(field)
189 if isinstance(value, str):
190 block[field] = resolve_media_path(value, source_directory)
191 elif field == "video" and isinstance(value, list):
192 block[field] = [
193 resolve_media_path(frame, source_directory) if isinstance(frame, str) else frame
194 for frame in value
195 ]
198def normalize_chat_media_fields(sample: Mapping[str, Any]) -> list[dict[str, Any]]:
199 """Normalize fixed OpenAI/HF media fields without imposing a training mode.
201 Args:
202 sample: Record containing OpenAI/HF-style messages and optional source metadata.
204 Returns:
205 HF-style messages with image_url converted to image/url blocks and string
206 image/path references expressed as url. Non-string media objects and
207 video frame lists retain their processor-supported representation.
209 Raises:
210 ValueError: If messages, content blocks, or image_url references are invalid.
212 Note:
213 Input: OpenAI Chat Completions or HF-style messages. Output: HF-style
214 typed media blocks. Shared by DeepSeek and AutoProcessor adapters.
215 Pretraining, SFT, and inference constraints belong to the caller.
216 This interface does not parse ShareGPT from/value or Alpaca records.
218 Example:
219 OpenAI Chat Completions-style input:
220 {"messages": [
221 {"role": "user", "content": [
222 {"type": "image_url", "image_url": {"url": "https://example.com/cat.jpg"}},
223 {"type": "text", "text": "Describe it."}]},
224 {"role": "assistant", "content": "A cat."}]}
225 HF-style output:
226 [{"role": "user", "content": [
227 {"type": "image", "url": "https://example.com/cat.jpg"},
228 {"type": "text", "text": "Describe it."}]},
229 {"role": "assistant", "content": "A cat."}]
230 HF local-file input (source file /data/train.jsonl):
231 {"type": "image", "path": "cat.jpg"}
232 HF-style media block output:
233 {"type": "image", "url": "/data/cat.jpg"}
234 HF image payloads such as {"type": "image", "image": pil_image} are kept.
235 Text-only messages keep their content; no chat template is rendered.
236 """
237 messages = resolve_chat_media_paths(sample)
238 for message in messages:
239 content = message.get("content")
240 if not isinstance(content, list):
241 continue
242 for block in content:
243 if not isinstance(block, dict):
244 raise ValueError("Multimodal content blocks must be mapping records")
246 block_type = block.get("type")
247 if block_type == "image_url":
248 image_url = block.pop("image_url", None)
249 reference = image_url.get("url") if isinstance(image_url, dict) else image_url
250 if not isinstance(reference, str) or not reference:
251 raise ValueError("image_url content requires a nonempty URL or path")
253 block["type"] = "image"
254 block["url"] = reference
255 elif block_type in ("image", "video", "audio") and "url" not in block:
256 reference = block.get(block_type, block.get("path"))
257 if isinstance(reference, str):
258 block["url"] = reference
259 block.pop(block_type, None)
260 block.pop("path", None)
261 return messages
264def _read_sharegpt_media(sample: Mapping[str, Any]) -> dict[str, list[str]]:
265 """Read and validate ShareGPT image and video reference columns."""
266 media = {"image": sample.get("images", []), "video": sample.get("videos", [])}
267 for media_type, references in media.items():
268 if not isinstance(references, list):
269 raise ValueError(f"ShareGPT {media_type} references must be a list")
271 for reference in references:
272 if not isinstance(reference, str) or not reference.strip():
273 raise ValueError(f"ShareGPT {media_type} references must be nonempty strings")
275 return media
278def convert_sharegpt_to_messages(sample: Mapping[str, Any]) -> list[dict[str, Any]]:
279 """Convert fixed ShareGPT conversations and media columns to HF-style messages.
281 Args:
282 sample: conversations with from/value, optional images/videos string lists,
283 and optional source-file metadata. Media placeholders consume entries
284 from their corresponding lists in conversation order.
286 Returns:
287 Messages preserving text and media order, with resolved media references.
289 Raises:
290 ValueError: If roles, values, media lists, or placeholder counts are invalid.
292 Example:
293 ShareGPT input (source file /data/train.jsonl):
294 {"conversations": [{"from": "human", "value": "<image>Describe it."},
295 {"from": "gpt", "value": "A cat."}], "images": ["cat.jpg"]}
296 HF-style output:
297 [{"role": "user", "content": [{"type": "image", "url": "/data/cat.jpg"},
298 {"type": "text", "text": "Describe it."}]},
299 {"role": "assistant", "content": "A cat."}]
300 No turns are discarded and no training target is selected here.
301 """
302 conversations = sample.get("conversations")
303 if not isinstance(conversations, list) or not conversations:
304 raise ValueError("ShareGPT requires nonempty conversations with from/value")
306 media = _read_sharegpt_media(sample)
307 consumed = {"image": 0, "video": 0}
308 messages = []
309 for turn in conversations:
310 if not isinstance(turn, dict) or not isinstance(turn.get("from"), str):
311 raise ValueError("ShareGPT conversations require mapping turns with a from role")
312 if turn["from"] not in ("human", "gpt", "system") or not isinstance(turn.get("value"), str):
313 raise ValueError("ShareGPT requires human/gpt/system roles and string values")
315 for message in normalize_conversation(conversations):
316 content = []
317 for part in re.split(r"(<image>|<video>)", message["content"]):
318 if part in ("<image>", "<video>"):
319 media_type = part[1:-1]
320 index = consumed[media_type]
321 if index >= len(media[media_type]):
322 raise ValueError(f"ShareGPT has more <{media_type}> placeholders than media references")
324 content.append({"type": media_type, "url": media[media_type][index]})
325 consumed[media_type] += 1
326 elif part:
327 content.append({"type": "text", "text": part})
328 if any(block["type"] != "text" for block in content):
329 message["content"] = content
330 messages.append(message)
331 for media_type, references in media.items():
332 if consumed[media_type] != len(references):
333 raise ValueError(f"ShareGPT has unused {media_type} references without matching placeholders")
335 converted_sample = dict(sample)
336 converted_sample["messages"] = messages
337 normalized_messages = normalize_chat_media_fields(converted_sample)
338 return normalized_messages
341def convert_interleaved_to_content(sample: Mapping[str, Any]) -> list[dict[str, Any]]:
342 """Convert aligned text/image lists to ordered HF-style content blocks.
344 Args:
345 sample: Equal-length texts/images lists with exactly one non-None entry
346 per position. Images are local paths or URLs. Source metadata is optional.
348 Returns:
349 Ordered text/image blocks without invented conversation roles or labels.
351 Raises:
352 ValueError: If lists, aligned entries, or image references are invalid.
354 Example:
355 Interleaved input (source file /data/train.jsonl):
356 {"texts": ["Look: ", None, "A cat."], "images": [None, "cat.jpg", None]}
357 HF-style content output:
358 [{"type": "text", "text": "Look: "}, {"type": "image", "url": "/data/cat.jpg"},
359 {"type": "text", "text": "A cat."}]
360 The caller owns pretraining/SFT encoding and supervision.
361 """
362 texts, images = sample.get("texts"), sample.get("images")
363 if not isinstance(texts, list) or not isinstance(images, list):
364 raise ValueError("Interleaved texts/images must be lists")
366 if not texts or len(texts) != len(images):
367 raise ValueError("Interleaved texts/images must be nonempty lists of equal length")
369 source_path = sample.get(ONLINE_SOURCE_PATH_KEY)
370 source_directory = os.path.dirname(os.path.abspath(source_path)) if source_path else None
371 content = []
372 for text, image in zip(texts, images):
373 if (text is None) == (image is None):
374 raise ValueError("Interleaved positions require exactly one text or image entry")
376 if text is not None:
377 if not isinstance(text, str):
378 raise ValueError("Interleaved text entries must be strings")
380 content.append({"type": "text", "text": text})
381 else:
382 if not isinstance(image, str) or not image.strip():
383 raise ValueError("Interleaved image references must be nonempty strings")
385 image_reference = image
386 if source_directory is not None:
387 image_reference = resolve_media_path(image, source_directory)
388 content.append({"type": "image", "url": image_reference})
389 return content
392@dataclass
393class InstructionMessages:
394 """Convert explicitly mapped instruction fields into standard messages."""
396 column_mapping: Mapping[str, str]
397 input_separator: str = "\n\n"
399 def __post_init__(self) -> None:
400 """Validate logical-to-source column mappings and the input separator."""
401 if not isinstance(self.column_mapping, Mapping):
402 raise ValueError("column_mapping must be a mapping")
403 for required_key in ("instruction", "output"):
404 source_key = self.column_mapping.get(required_key)
405 if not isinstance(source_key, str) or not source_key:
406 raise ValueError(f"column_mapping must define a non-empty {required_key!r} source field")
407 input_key = self.column_mapping.get("input")
408 if input_key is not None and (not isinstance(input_key, str) or not input_key):
409 raise ValueError("column_mapping 'input' source field must be a non-empty string")
410 if not isinstance(self.input_separator, str):
411 raise ValueError("input_separator must be a string")
413 def _get_required_text(self, sample: Mapping[str, Any], logical_key: str) -> str:
414 """Return one required string field selected by the column mapping."""
415 source_key = self.column_mapping[logical_key]
416 try:
417 value = sample[source_key]
418 except KeyError as exc:
419 raise ValueError(f"Instruction sample does not contain field {source_key!r}") from exc
420 if not isinstance(value, str):
421 raise ValueError(f"Instruction field {source_key!r} must be a string")
422 if not value:
423 raise ValueError(f"Instruction field {source_key!r} must not be empty")
424 return value
426 def __call__(self, sample: Mapping[str, Any]) -> list[dict[str, Any]]:
427 """Read instruction fields and return user/assistant messages."""
428 instruction = self._get_required_text(sample, "instruction")
429 output = self._get_required_text(sample, "output")
431 input_text = ""
432 input_key = self.column_mapping.get("input")
433 if input_key is not None:
434 input_text = sample.get(input_key, "")
435 if not isinstance(input_text, str):
436 raise ValueError(f"Instruction field {input_key!r} must be a string")
438 user_content = instruction
439 if input_text:
440 user_content = f"{instruction}{self.input_separator}{input_text}"
442 messages = [
443 {"role": "user", "content": user_content},
444 {"role": "assistant", "content": output},
445 ]
446 return messages