fix(examples): address PR #1176 review feedback

- Fix module docstring in conversation_utils.py: clarify that
  user_suffix augmentations apply to all user turns, not just the last
- Map developer role to system in normalize_messages() to preserve
  developer instructions rather than silently dropping them
- Raise ValueError instead of warning+return when combined dataset
  is empty in make_nemotron_ptv3_dataset.py
- Re-raise all exceptions in query.py LLM.generate() so datasets.map()
  halts on any failure (not only connection errors)
- Make tool role explicit in synthesize(); raise ValueError for truly
  unknown roles to catch typos/unsupported roles early
- Fix local file format detection in query.py: use "parquet" loader for
  .parquet files instead of hardcoded "json"
- Replace eval with direct argv in query.sh to eliminate shell injection

Signed-off-by: chenhany <chenhany@nvidia.com>

Author: ChenhanYu

examples/dataset/conversation_utils.py

@@ -29,8 +29,9 @@
 skeleton to the target model turn-by-turn, appending each generated response before
 sending the next user turn, so the model produces coherent multi-turn continuations.
 
-Augmentations are applied only to the *last* user message (the new prompt), not to
-earlier user turns that are already part of the established context.
+Augmentations (``user_suffix``) are applied to *all* user messages so that the
+language or style instruction is present at every turn — important for multi-turn
+synthetic generation where the model must maintain the requested style throughout.
 """
 
 import logging
@@ -190,5 +191,8 @@ def normalize_messages(example: dict[str, Any], idx: int) -> dict[str, Any]:
                 "content": m.get("content") or "",
                 "tool_call_id": m.get("tool_call_id", ""),
             })
+        elif role == "developer":
+            # Map developer-role messages to system per OpenAI schema conventions.
+            normalized.append({"role": "system", "content": m.get("content") or ""})
         # other roles (e.g. function, unknown) are dropped
     return {"messages": normalized}

examples/dataset/make_nemotron_ptv3_dataset.py

@@ -271,8 +271,7 @@ def main() -> None:
         parts_to_combine.append(non_augmentable)
 
     if not parts_to_combine:
-        logger.warning("No data to combine — all rows were filtered out. Exiting.")
-        return
+        raise ValueError("No data to combine — all rows were filtered out.")
 
     combined = concatenate_datasets(parts_to_combine)
     logger.info("Combined (pre-shuffle): %d rows", len(combined))

tools/launcher/common/query.py

@@ -78,12 +78,9 @@ def generate(self, messages, verbose=False, **chat_template_kwargs):
             new_message = {"role": "assistant", "content": new_message}
         except Exception as e:
             print(e)
-
             if "Connection error" in str(e):
                 early_termination = True
-                raise  # propagate so datasets.map() halts the shard
-
-            new_message = None
+            raise  # always propagate so datasets.map() halts the shard
 
         return new_message
 
@@ -168,10 +165,11 @@ def synthesize(data):
         elif role == "assistant":
             # Original assistant messages are not used — the model generates fresh responses.
             pass
-        else:
-            # Skip unknown roles (e.g. tool) — agentic datasets include tool turns
-            # that are not sent to the generation model.
+        elif role == "tool":
+            # Tool turns are not sent to the generation model — skip them.
             pass
+        else:
+            raise ValueError(f"Unexpected message role {role!r} in conversation.")
 
     # Restore the full reasoning trace for the last generated assistant turn.
     if enable_thinking and last_full_message is not None:
@@ -185,7 +183,9 @@ def synthesize(data):
 
 # Support both HF Hub repo IDs and local file paths (.jsonl, .json, .parquet, etc.)
 if os.path.isfile(args.data):
-    dataset = load_dataset("json", data_files=args.data, split=args.data_split)
+    ext = os.path.splitext(args.data)[1].lower()
+    fmt = "parquet" if ext == ".parquet" else "json"
+    dataset = load_dataset(fmt, data_files={"train": args.data}, split=args.data_split)
 else:
     dataset = load_dataset(args.data, split=args.data_split)
 

tools/launcher/common/vllm/query.sh

@@ -119,9 +119,8 @@ while true; do
 done
 
 pip3 install -q datasets openai 2>/dev/null || true
-cmd="python3 common/query.py http://localhost:8000/v1 ${MODEL} ${QUERY_ARGS[*]}"
-echo "Running command: $cmd"
-eval $cmd
+echo "Running: python3 common/query.py http://localhost:8000/v1 ${MODEL} ${QUERY_ARGS[*]}"
+python3 common/query.py http://localhost:8000/v1 "${MODEL}" "${QUERY_ARGS[@]}"
 echo "Main process exit"
 
 kill $SERVER_PID