Codeseys/composer-replication-framework

1# Replaysim Normalization Reconnaissance2 3**Status:** Recon · **Feeds:** ADR-004, V5 "replaysim with normalization"4**Author:** subagent (delegated audit) · **Date:** 2026-05-255**Sources:** GitHub REST API metadata + DeepWiki structured indexes of each repo's primary source. All repo metadata cited below was pulled from `api.github.com/repos/<owner>/<name>` directly.6 7## TL;DR8 9| Library | License | Last push | ★ | Verdict |10|---|---|---|---|---|11| **data-juicer** | Apache-2.0 | **2026-05-25** | 6.4k | ✅ **RECOMMENDED** — the only candidate with a class-based op-graph that *natively* understands `messages: [{role, content}]`, multi-turn dialog, and DPO-pair (`chosen`/`rejected`) preference samples as **first-class data formats**, with a `pair_preference_mapper` operator that maps directly onto our `extract_dpo_pairs` output. |12| **distilabel** | Apache-2.0 | 2026-05-25 | 3.2k | Strong runner-up. DAG pipeline, native chat-message format, built-in `FormatChatGenerationDPO`. But it is primarily a *generation orchestrator* and would force us to rewrite our existing OpenRouter teacher orchestration as Distilabel `LLM` subclasses. Larger refactor surface. |13| **datatrove** | Apache-2.0 | 2026-05-06 | 3.1k | ❌ **Deal-breaker.** `Document` dataclass is `text: str + metadata: dict`. All filters/dedup operate on flat `doc.text`. Multi-turn is only supported in the *generation* (`InferenceRunner.rollout_fn`) path, not the normalization/filter path. Forces lossy chat→string flattening. |14| **NeMo-Curator** | Apache-2.0 | 2026-05-25 | 1.6k | Strong on scale (Ray + Xenna + GPU), supports streaming and DPO via `generate_two_turn_prompt`. But: semantic dedup, fuzzy dedup, and classifier filters all *require GPUs*; CPU-only install drops most of the differentiating ops. Heavy framework for the size of replaysim. |15| **lilac** | Apache-2.0 | **archived 2024-03-19** | 1.1k | ❌ **Dead.** `databricks/lilac` repo `"archived": true`. The current `lilacai/lilac` is a 2-star squatter stub created Nov 2025. Do not adopt. |16 17**Recommendation:** Adopt **data-juicer** as the normalization op-graph layer wrapped around `replay_trace` → `extract_dpo_pairs`. Estimated integration cost: **~250–400 LOC** in `composer_replication.replaysim` for an adapter + 1 YAML recipe.18 19**Critical chat-template question answered:** data-juicer is the only audited library whose *filtering and normalization operators* (not just its generation operators) operate directly on a structured `messages: [{role, content}]` format and on `chosen`/`rejected` preference-pair format. The other three candidates either flatten to text (datatrove), only handle chat in the generation path (datatrove again), or treat chat as a generation output to be assembled rather than a structured object to be filtered (NeMo-Curator, distilabel partly).20 21---22 23## 1. Audit Methodology24 25For each candidate, primary-source data was collected from:26 271. `https://api.github.com/repos/<owner>/<name>` for license, `pushed_at`, `archived`, stars, forks, topics — these are authoritative GitHub metadata, not scraped.282. DeepWiki structured indexes of each repo's source tree for: op model, data structures (`Document` / `Sample` / `Step`), conversation/DPO support in filtering vs. generation paths, GPU dependencies.293. README confirmation through the GitHub API for transferred-org redirects.30 31No secondary sources, no marketing pages, no blog posts.32 33Two facts to flag up front because they materially change the candidate set:34 35- `modelscope/data-juicer` redirects to **`datajuicer/data-juicer`**. The team spun out of ModelScope into a dedicated `datajuicer` org. Same code, just a transferred name — `pushed_at` is current.36- `NVIDIA/NeMo-Curator` redirects to **`NVIDIA-NeMo/Curator`**. Same situation — moved into the dedicated `NVIDIA-NeMo` org in 2025.37 38---39 40## 2. Per-Candidate Audit41 42### 2.1 datatrove (huggingface)43 44| Dimension | Value |45|---|---|46| Repo | `huggingface/datatrove` |47| License | Apache-2.0 |48| Created | 2023-06-14 |49| Last push | **2026-05-06** |50| Stars / Forks | 3068 / 266 |51| Commits | 725 (default branch) |52| Maturity | Production. Used to build FineWeb. Active. |53 54**Op model.** Class-based **linear pipeline** of `PipelineStep` instances. `PipelineStep.run(data: DocumentsPipeline, rank: int, world_size: int) -> DocumentsPipeline` where `DocumentsPipeline` is an iterator of `Document` objects. Steps are composed by Python list concatenation, not a DAG — branching/joining requires manual orchestration.55 56**Multi-turn / chat-template support — DEAL-BREAKER.**57 58The `Document` dataclass (`src/datatrove/data.py`) is:59 60```python61@dataclass62class Document:63    text: str64    id: str65    media: list[Media]   # placeholder, "for future uses, currently not used"66    metadata: dict67```68 69There is **no `messages` field**. Every built-in filter (e.g., `C4QualityFilter`, `LanguageFilter`, `GopherQualityFilter`) and every built-in dedup op (`MinhashDedup*`, `SentenceDedup*`, `BloomFilter`) operates on `doc.text` as a flat string.70 71Multi-turn does appear, but **only in the generation path** (`InferenceRunner` + user-supplied `rollout_fn(doc, generate)`), where the user constructs `{"messages": [{"role": ..., "content": ...}]}` payloads themselves. Once the generation completes, the result is stuffed back into `doc.text` (or `doc.metadata`) and downstream filters again see flat text.72 73For our use case — normalizing already-generated multi-turn DPO pairs with `chosen`/`rejected` chat structures and tool calls — this means we'd have to:74 751. Serialize `messages` into a flat string (`<|im_start|>user...`).762. Run datatrove filters on the serialized string.773. Re-parse back into `messages` afterward.78 79Tool-call structure (`{"role": "tool", "tool_call_id": ...}`, `tool_calls: [...]`) does not survive that round-trip cleanly without custom serialization on both sides. Per the user's hard requirement — "if only flat text, that's a deal-breaker" — datatrove fails here.80 81**Streaming.** Yes. `HuggingFaceDatasetReader(streaming=True)` and the iterator-based `PipelineStep.run` mean we can pipe documents through during generation. Streaming is fine.82 83**GPU.** None of the *normalization* ops require GPU. MinHash dedup is CPU. Only the `InferenceRunner` path needs a GPU (vLLM/SGLang backend) and we don't need that — we'd be calling OpenRouter, not running local models.84 85**Integration cost.** Moot — the chat-template gap is the deal-breaker.86 87---88 89### 2.2 data-juicer (datajuicer org, formerly modelscope)90 91| Dimension | Value |92|---|---|93| Repo | `datajuicer/data-juicer` (redirect target of the legacy `modelscope/data-juicer`) |94| License | Apache-2.0 |95| Created | 2023-08-01 |96| Last push | **2026-05-25** (most recent of all candidates) |97| Stars / Forks | 6444 / 373 |98| Maturity | Production. Active core team (Alibaba/ModelScope-spinout). Most stars of the candidate set. Has its own conference papers and a docs site at `datajuicer.github.io/data-juicer`. |99 100**Op model.** Class-based DAG of **operators ("Ops")** organized as **mappers**, **filters**, **deduplicators**, and **selectors**. Each Op is a Python class subclassing `Mapper`/`Filter`/`Deduplicator`. Pipelines are declared as YAML recipes (`process: [- op_name: { args }, ...]`) and executed by the `Executor` (default Ray-distributed; also a local Pandas-backed mode). Conditional branching through `OpFusion` and `Adapter` modules is supported, and there is a Ray-Data executor for true streaming.101 102**Multi-turn / chat-template support — NATIVE.** This is the discriminator.103 104Data-juicer has a **first-class conversation schema**, supporting *both*:1051. OpenAI-style `messages: [{role, content}]`1062. A "Data-Juicer format" `{query, response, history: [[q, r], ...]}`107 108It exposes operators that are *purpose-built* for dialog/preference data:109 110- `dialog_intent_detection_mapper`111- `dialog_sentiment_detection_mapper`112- `dialog_sentiment_intensity_mapper`113- `dialog_topic_detection_mapper`114- `pair_preference_mapper` — **directly relevant**: ingests a `(prompt, chosen)` and synthesizes/refines a `rejected_response` plus a `reason` field. This is exactly the schema produced by our `extract_dpo_pairs`.115- `query_intent_detection_mapper`, `query_sentiment_detection_mapper`, `query_topic_detection_mapper`116- `optimize_qa_mapper`, `optimize_query_mapper`, `optimize_response_mapper` — refine individual fields without flattening the whole conversation.117 118Tool-call structure: data-juicer's conversation schema preserves arbitrary keys per message (because it operates on dict-of-lists Arrow tables), so `tool_call_id`, `tool_calls`, `name`, etc. survive through filters as long as no operator explicitly drops them. This is structurally safe — confirmed by the operator code only reading `role`/`content` and forwarding the rest.119 120**Streaming.** Partial. The default executor is batch on Arrow/HF datasets, but data-juicer integrated with **Ray Data** for distributed/streaming processing, and the README references "streaming JSON reader patches integrated by Apache Arrow." For our scale (≤100k DPO pairs per run), batch is fine; for true online normalization during multi-teacher generation, the Ray executor handles it — but a simpler approach is to wrap each `replay_trace` rollout's output into a tiny in-memory dataset and run the recipe per-batch (mini-batch streaming).121 122**GPU.** Only needed for image/video/multi-modal ops and for the LLM-API mappers when configured to run a *local* model. Every op we care about for replaysim — `pair_preference_mapper`, dialog detection mappers, `text_length_filter`, `language_id_score_filter`, MinHash dedup, etc. — is CPU-OK or calls a remote API (which is exactly our existing OpenRouter pattern). Importantly, **MinHash and exact dedup in data-juicer do not require GPU**, unlike NeMo-Curator's fuzzy/semantic dedup.123 124**Integration cost into `composer_replication.replaysim`.** Estimated ~250–400 LOC, breakdown:125 126- Adapter `replaysim/normalize.py`: ~80–120 LOC. Wraps a `DJDataset` (data-juicer's dataset abstraction), exposes `normalize_dpo_batch(pairs: list[DPOPair]) -> list[DPOPair]`.127- YAML recipe `replaysim/recipes/dpo_normalize.yaml`: ~40 LOC declarative.128- Hook in `teacher_replay.py` after `extract_dpo_pairs` and before final write: ~20 LOC.129- New tests `tests/replaysim/test_normalize.py`: ~80–120 LOC.130- ADR-004 update + module docs: ~20 LOC.131 132Dependency footprint: `pip install py-data-juicer` pulls in `datasets`, `pyarrow`, `loguru`, `jsonargparse`, optionally `ray`. We already have `datasets`/`pyarrow` indirectly from HF stack.133 134---135 136### 2.3 NeMo-Curator (NVIDIA-NeMo)137 138| Dimension | Value |139|---|---|140| Repo | `NVIDIA-NeMo/Curator` (redirect target of `NVIDIA/NeMo-Curator`) |141| License | Apache-2.0 |142| Created | 2024-03-14 |143| Last push | **2026-05-25** |144| Stars / Forks | 1584 / 274 |145| Maturity | Production at NVIDIA scale. Built for pre-training-corpus curation (Nemotron / Nemotron-4). |146 147**Op model.** Task-centric distributed processing, built on **Ray** + the **Xenna** executor. Stages are class-based, composed into pipelines, executed by `XennaExecutor` in either `streaming` or `batch` mode. Closer to Spark/Ray-Data than to a Python list of steps.148 149**Multi-turn / chat-template support — partial, generation-side only.** Curator has model-specific formatters (`Mixtral8x7BFormatter`, `NemotronFormatter`) that *render* multi-turn dialogue into a flat prompt string for the target model's chat template. There is `generate_dialogue` for multi-turn synthesis and `generate_two_turn_prompt` for DPO-style preference pairs. **But**: like datatrove, the *filtering* and *deduplication* stages do not have first-class conversation/preference operators — they treat the data as text after rendering. Tool-call preservation is not addressed in the public API.150 151**Streaming.** Yes — `XennaExecutor(execution_mode="streaming")` is a first-class option.152 153**GPU — significant cost.** Curator's discriminating features all require GPUs:154 155- **Semantic deduplication** — GPU-only, embedding generation + clustering. "Not supported for CPU-only processing."156- **Fuzzy deduplication** (MinHash + LSH) — GPU backend (cuDF/cuML), not CPU.157- **Classifier filters** (domain / quality / safety via `DistributedDataClassifier`) — GPU clusters.158- **Image curation modules** — GPU.159 160CPU-only install supports basic text filters and exact dedup, but *that's the same surface area we'd get from data-juicer without the dependency weight*. If we are not running on a GPU cluster, NeMo-Curator's value proposition collapses.161 162**Integration cost.** ~600–900 LOC plus operational cost: a Ray cluster setup, GPU nodes if we want the differentiating features. For replaysim's scale (a few thousand DPO pairs per run), this is overkill.163 164---165 166### 2.4 distilabel (argilla-io)167 168| Dimension | Value |169|---|---|170| Repo | `argilla-io/distilabel` |171| License | Apache-2.0 |172| Created | 2023-10-16 |173| Last push | **2026-05-25** |174| Stars / Forks | 3230 / 242 |175| Maturity | Production. Argilla is now part of HF; project remains active under argilla-io. |176 177**Op model.** **DAG pipeline** of `Step` and `Task` (Task = Step with an LLM). Each step declares `inputs: list[str]`, `outputs: list[str]`, and `process(*inputs) -> Generator[outputs]`. Steps are wired via `>>` operator. Resource declarations (`StepResources(replicas=N, gpus=M)`) handle scaling, optionally on Ray.178 179**Multi-turn / chat-template support — NATIVE on the generation side, partial on the normalization side.**180 181- `ChatGeneration` task accepts OpenAI-format `messages: [{role, content}]` natively.182- `FormatTextGenerationDPO` and `FormatChatGenerationDPO` produce the exact `{prompt, chosen, rejected, ratings, reason}` schema we want.183- `UltraFeedback` task is the canonical preference-rating step.184- `DeitaFiltering` and `MinHashDedup` are the only filtering/dedup steps; they operate on text fields rather than on structured `messages`. Tool-call structure is preserved as long as no step explicitly normalizes it (like data-juicer, by virtue of dict-of-fields semantics) — but there isn't a `pair_preference_mapper` analogue that operates on `messages` directly.185 186**Streaming.** Supports streaming generation per LLM (e.g., `AnthropicLLM` streams tokens). Pipeline-level execution is batch-of-batches; you can `.run(parameters={...})` and consume outputs as they materialize.187 188**GPU.** Only when steps choose to run a local LLM (vLLM, transformers). API-based steps (OpenAI, Anthropic, Mistral, OpenRouter via OpenAI-compat) are CPU-only.189 190**Integration cost — large but high overlap.** Distilabel would *replace* much of `teacher_replay.py`, not just normalize after it:191 192- Rewrite multi-teacher OpenRouter calls as a `Pipeline` of `Task`s subclassing distilabel's `LLM` interface (or use the `OpenAILLM` wrapper pointed at OpenRouter): ~300–500 LOC delta.193- Re-express `extract_dpo_pairs` as a custom `Task` or use `FormatChatGenerationDPO`: ~100–150 LOC.194- Migrate trace plumbing into distilabel's `GeneratorStep`/`Task` DAG: ~150 LOC.195- Tests + docs: ~150 LOC.196 197Total **~700–900 LOC** and a meaningful refactor of teacher orchestration. The win is that we'd get a real DAG runtime, retries, caching, and Argilla-integration for free. The lose is that we get *coupled* to distilabel's `LLM`/`Task` abstractions for the entire generation pipeline, not just a normalization op-graph wrapped around it.198 199This is a strategic decision the user phrased as: "see if we can leverage [a normalization library] to **normalize the data while also making the replaysim dataset generation**." Distilabel takes the broader interpretation — replace replaysim's generation with a distilabel pipeline. That is a bigger commitment than this recon was scoped to recommend.200 201---202 203### 2.5 lilac204 205**STATUS: dead. Do not adopt.**206 207- `databricks/lilac`: `"archived": true`, last push **2024-03-19**, license Apache-2.0. Repo says "Curate better data for LLMs." The Databricks acquisition (April 2024) absorbed it into Databricks Mosaic AI; the OSS project was archived shortly after.208- `lilacai/lilac`: created **2025-11-14** by a user account `lilacai`, 2 stars, 0 forks, no license, description says "Thee Eclipse - Hackerone: @theeeclipse." This is a **squatter / unrelated stub**, not the original lilac.209- No actively maintained successor with the original lilac code base outside Databricks' proprietary platform.210 211---212 213## 3. Recommendation: data-juicer214 215### 3.1 Why216 2171. **Only candidate with native conversation + preference-pair operators in the *normalization* path**, not just the generation path. `pair_preference_mapper` is a near-perfect fit for the output of `extract_dpo_pairs`.2182. **Tool-call structure is preserved** because operators read specific fields and forward the rest of the dict — confirmed by the operator schema design.2193. **No GPU required** for the operators we'd actually use (preference, dialog, length, language-id, MinHash dedup). Matches our OpenRouter-API-driven, CPU-friendly architecture.2204. **YAML-recipe style** lets us version the normalization graph as a config artifact alongside the recon doc, instead of as Python code that drifts.2215. **Lowest integration cost** of the viable candidates — wraps around our existing pipeline rather than replacing it.2226. **Maturity**: 6.4k stars, last push today, dedicated org, paper-backed.223 224### 3.2 Why not the others (one-liners)225 226- **datatrove**: flat-text `Document`, lossy round-trip on chat structure → deal-breaker.227- **distilabel**: would force a rewrite of teacher orchestration — too broad a refactor for "wrap normalization around the existing pipeline."228- **NeMo-Curator**: best ops require GPUs; without them it offers no advantage over data-juicer.229- **lilac**: archived.230 231### 3.3 Risk register232 233| Risk | Severity | Mitigation |234|---|---|---|235| Data-juicer YAML recipe drift between dev and CI | M | Pin `py-data-juicer` version; commit recipe under `replaysim/recipes/` and load via `importlib.resources`. |236| Some ops silently coerce conversation structure | M | Add a round-trip test: `pair → normalize → pair` must preserve `messages`, `tool_calls`, and arbitrary metadata. |237| Ray executor bloat if user enables it | L | Default to local Pandas executor; gate Ray behind an explicit flag. |238| `pair_preference_mapper` calls an LLM by default to synthesize `rejected` | H | We *already have* `rejected` from disagreement. Configure the mapper as a pass-through filter / use it only for refinement; if it can't be made non-LLM, fall back to a custom Mapper that just runs length/language/dedup checks on the existing pair. **Verify in spike before locking in.** |239| Apache-2.0 inbound license compatibility | L | Our framework is Apache-2.0. Compatible. |240| Op-graph executes per batch, not per sample, so a single bad pair stalls a batch | L | Use small Ray-Data batches (e.g. 64) so a stall is bounded. |241 242### 3.4 Open spike question (must verify before merge)243 244The single risk worth a 1-day spike: **does `pair_preference_mapper` accept a pre-existing `rejected` and *only* run validation/length/language filters, or does it *always* call an LLM to (re)synthesize a rejected response?** Read the operator source in `data_juicer/ops/mapper/pair_preference_mapper.py` and confirm. If the latter, we wire our pre-existing `rejected` through `optimize_response_mapper` (refinement, not regeneration) plus a custom no-op preference validator. Either way, the integration shape below stands; only the recipe content changes.245 246---247 248## 4. Integration Sketch249 250### 4.1 Current pipeline (today)251 252```253TraceState254   │255   ▼  (per-trace, multi-teacher OpenRouter call)256replay_trace(state, teachers=[m1, m2, m3])257   │258   ▼  (returns: list[TeacherCompletion] keyed by model_id)259disagreement_score(completions)260   │261   ▼  (if score > τ)262extract_dpo_pairs(completions, state)263   │264   ▼  (yields)265DPOPair { prompt: messages[], chosen: messages[], rejected: messages[], state, meta }266   │267   ▼268write_jsonl(out_path)269```270 271### 4.2 Proposed pipeline (with data-juicer normalization op-graph)272 273```274TraceState275   │276   ▼277replay_trace(state, teachers)         ← unchanged278   │279   ▼280disagreement_score(completions)        ← unchanged281   │282   ▼283extract_dpo_pairs(completions, state)  ← unchanged284   │285   ▼286[NEW] DJNormalizer.normalize_batch(dpo_pairs) ──── loads recipe from287   │                                                replaysim/recipes/dpo_normalize.yaml288   │   data-juicer op-graph runs:289   │     1. text_length_filter (on chosen + rejected separately)290   │     2. language_id_score_filter (en-only or configured)291   │     3. dialog_topic_detection_mapper (annotates meta, no drop)292   │     4. minhash_deduplicator (on prompt+chosen serialization)293   │     5. (optional) optimize_response_mapper to clean trailing whitespace, code-block fences294   │     6. custom PreferenceValidator op (chosen != rejected, both non-empty,295   │        tool_calls structurally valid)296   ▼297write_jsonl(out_path)                  ← unchanged consumer298```299 300The op-graph is a **wrapper around** `extract_dpo_pairs`, not a replacement. `replay_trace` and `extract_dpo_pairs` keep their current signatures. The only call-site change in `teacher_replay.py` is one line:301 302```python303# before:304pairs = list(extract_dpo_pairs(completions, state))305write_jsonl(out_path, pairs)306 307# after:308pairs = list(extract_dpo_pairs(completions, state))309pairs = DJNormalizer.from_recipe("dpo_normalize.yaml").normalize_batch(pairs)310write_jsonl(out_path, pairs)311```312 313### 4.3 Adapter shape (`replaysim/normalize.py`)314 315> **Realised in v0.1 (Wave 17 update):** ADR-004 shipped with a different316> public surface than the sketch below. The actual API:317>318> ```python319> from composer_replication.replaysim import (320>     replay_and_normalize_trace,    # convenience wrapper321>     DJNormalizer,                  # the normalizer class322>     DPOPair,                       # input TypedDict (from teacher_replay)323>     NormalizedDPOPair,             # output TypedDict324>     replay_trace, extract_dpo_pairs,  # re-exports of upstream stages325> )326> ```327>328> Key shape differences from the sketch:329>330> 1. **`DPOPair` is a TypedDict, not a dataclass.** Its actual fields331>    are `{state_id: str, state_messages: list[dict], chosen: str,332>    rejected: str, n_teachers_agreeing: int}` (defined in333>    `composer_replication/teacher_replay.py:99`) — **not**334>    `{prompt, chosen, rejected, state, meta}`. The `_to_dj`/`_from_dj`335>    sketch round-trip below would not type-check against the realised336>    TypedDict.337> 2. **Recipe path is `composer_replication/recipes/replaysim/default.yaml`**,338>    not `composer_replication/replaysim/recipes/dpo_normalize.yaml`.339>    There is no `replaysim/recipes/` subpackage; recipes live under340>    the top-level `recipes/` tree.341> 3. **No `composer_replication/replaysim/ops/` subpackage exists.**342>    The custom op file `preference_validator.py` was not created;343>    data-juicer's stock ops + the framework's own validation in344>    `DJNormalizer` covered the requirement.345> 4. **The integration hook is `replay_and_normalize_trace(...)`** in346>    `composer_replication/replaysim/__init__.py` (re-exported from347>    `normalize.py`). It wraps the existing `replay_trace` +348>    `extract_dpo_pairs` flow without modifying `teacher_replay.py`.349>    There is no separate `composer_replication/replaysim/teacher_replay.py`350>    — `teacher_replay` lives at top-level `composer_replication/teacher_replay.py`.351>352> The pre-spike sketch below is preserved as historical proposal context.353> It documents the shape of thinking that fed ADR-004; the realised code354> is the source of truth for the adapter contract.355 356```python357# composer_replication/replaysim/normalize.py358from __future__ import annotations359from dataclasses import asdict360from importlib.resources import files361from typing import Iterable362 363from data_juicer.config import init_configs364from data_juicer.core.executor import DefaultExecutor365from data_juicer.format import load_formatter366 367from .types import DPOPair368 369 370class DJNormalizer:371    """Wraps a data-juicer op-graph as a batch normalization step over372    DPOPair samples produced by extract_dpo_pairs.373 374    The recipe (YAML) declares the op sequence. Operators consume and375    produce the data-juicer conversation schema, which we convert to376    and from our internal DPOPair on the boundary.377    """378 379    def __init__(self, recipe_path: str):380        cfg = init_configs(["--config", recipe_path])381        self._executor = DefaultExecutor(cfg)382 383    @classmethod384    def from_recipe(cls, name: str) -> "DJNormalizer":385        recipe = files("composer_replication.replaysim.recipes") / name386        return cls(str(recipe))387 388    @staticmethod389    def _to_dj(p: DPOPair) -> dict:390        # data-juicer preference schema:391        #   {"prompt": str-or-messages, "chosen": str-or-messages,392        #    "rejected": str-or-messages, "meta": {...}}393        return {394            "prompt": p.prompt,        # messages[]395            "chosen": p.chosen,        # messages[]396            "rejected": p.rejected,    # messages[]397            "meta": {398                "trace_id": p.state.trace_id,399                "teachers": p.meta.get("teachers", []),400                "disagreement": p.meta.get("disagreement"),401                **p.meta,402            },403        }404 405    @staticmethod406    def _from_dj(s: dict) -> DPOPair:407        return DPOPair(408            prompt=s["prompt"],409            chosen=s["chosen"],410            rejected=s["rejected"],411            state=...,           # rehydrate from meta.trace_id + cache412            meta=s.get("meta", {}),413        )414 415    def normalize_batch(self, pairs: Iterable[DPOPair]) -> list[DPOPair]:416        in_records = [self._to_dj(p) for p in pairs]417        # Build an in-memory DJDataset from records (no disk round-trip).418        ds = self._executor.formatter.load_dataset_from_records(in_records)419        ds = self._executor.run(dataset=ds)420        out_records = ds.to_list()421        return [self._from_dj(r) for r in out_records]422```423 424### 4.4 Recipe (`replaysim/recipes/dpo_normalize.yaml`)425 426```yaml427# data-juicer recipe for normalizing replaysim DPO output428project_name: replaysim_dpo_normalize429executor_type: default          # local Pandas; switch to 'ray' for distributed430np: 4431 432# Conversation/preference schema mode433text_keys: ['chosen', 'rejected']    # ops scan both response variants434suffixes: ['.jsonl']435 436process:437  # 1. Length sanity on each response variant438  - text_length_filter:439      text_key: chosen440      min_len: 10441      max_len: 16384442  - text_length_filter:443      text_key: rejected444      min_len: 10445      max_len: 16384446 447  # 2. Language gate (configurable; default English-only)448  - language_id_score_filter:449      text_key: chosen450      lang: en451      min_score: 0.6452 453  # 3. Dialog topic annotation (no drop, just attaches meta.topic)454  - dialog_topic_detection_mapper:455      api_or_hf_model: openrouter:openai/gpt-4o-mini456      mode: annotate457 458  # 4. Near-duplicate removal across the batch on (prompt + chosen)459  - document_minhash_deduplicator:460      tokenization: space461      window_size: 5462      num_permutations: 256463      jaccard_threshold: 0.85464      text_key: chosen465 466  # 5. Custom preference validator (chosen != rejected, structural integrity)467  - preference_validator_filter:        # module: composer_replication.replaysim.ops468      check_distinct: true469      check_tool_calls_valid: true470```471 472A custom op `preference_validator_filter` lives in `composer_replication/replaysim/ops/preference_validator.py` and is registered via data-juicer's plugin entry point.473 474### 4.5 Hook into `teacher_replay.py`475 476```python477# composer_replication/replaysim/teacher_replay.py (delta)478 479from .normalize import DJNormalizer480 481def run_replay(traces, teachers, out_path, *, normalize: bool = True):482    pairs: list[DPOPair] = []483    for state in traces:484        completions = replay_trace(state, teachers=teachers)485        if disagreement_score(completions) <= TAU:486            continue487        pairs.extend(extract_dpo_pairs(completions, state))488 489    if normalize:490        norm = DJNormalizer.from_recipe("dpo_normalize.yaml")491        pairs = norm.normalize_batch(pairs)492 493    write_jsonl(out_path, pairs)494```495 496The `normalize=True` flag keeps the old code-path one negation away during initial rollout.497 498### 4.6 Test plan (`tests/replaysim/test_normalize.py`)499 5001. **Round-trip preservation**: synthesize a DPOPair with `tool_calls`, run through `DJNormalizer.normalize_batch`, assert tool-call structure and arbitrary `meta` keys are preserved.5012. **Length filter**: a pair with empty `chosen` is dropped.5023. **Language filter**: a non-English `chosen` (Cyrillic) below the score threshold is dropped.5034. **Near-duplicate**: two pairs with identical `chosen` collapse to one.5045. **Distinctness**: a pair where `chosen == rejected` is dropped by `preference_validator_filter`.5056. **Multi-turn**: a 3-turn conversation in `prompt` survives end-to-end with role+content intact.5067. **Recipe loading**: `DJNormalizer.from_recipe("dpo_normalize.yaml")` works with `importlib.resources` regardless of install location.507 508---509 510## 5. ADR-004 Implications511 512ADR-004 (the umbrella ADR for "replaysim with normalization") should record:513 514- **Decision**: adopt data-juicer (`datajuicer/data-juicer`, Apache-2.0) as the normalization op-graph layer.515- **Status**: proposed; promote to accepted after the spike on `pair_preference_mapper`.516- **Consequences**:517  - New runtime dependency: `py-data-juicer` (transitively pulls `pyarrow`, `datasets`, `loguru`, `jsonargparse`).518  - Optional `ray` extra for distributed execution; not enabled by default.519  - `replaysim/recipes/*.yaml` becomes a versioned config artifact; recipe changes must accompany behavioral-test updates.520  - Tool-call and multi-turn structure preserved through normalization — verified by round-trip test.521- **Alternatives considered**: distilabel (too broad — would replace generation orchestration), datatrove (flat-text only — deal-breaker), NeMo-Curator (GPU-bound), lilac (archived).522 523---524 525## 6. Primary-source citations526 527| Claim | Source |528|---|---|529| datatrove license, last push, archived state | `https://api.github.com/repos/huggingface/datatrove` (`license.spdx_id`, `pushed_at`, `archived`) |530| datatrove `Document` is text+metadata, no `messages` field; built-in filters operate on `doc.text` | DeepWiki index of `huggingface/datatrove`, `src/datatrove/data.py`, `src/datatrove/pipeline/filters/c4_filters.py` |531| datatrove multi-turn only via `InferenceRunner.rollout_fn` | DeepWiki index of `huggingface/datatrove`, `src/datatrove/pipeline/inference/run_inference.py` |532| data-juicer license, last push, redirect to `datajuicer/data-juicer` | `https://api.github.com/repos/modelscope/data-juicer` (resolves to `datajuicer/data-juicer`) |533| data-juicer supports `messages: [{role, content}]` and Data-Juicer dialog format `{query, response, history}` | DeepWiki index of `modelscope/data-juicer` |534| `pair_preference_mapper` synthesizes `rejected_response` and `reason` | DeepWiki index of `modelscope/data-juicer`, `data_juicer/ops/mapper/pair_preference_mapper.py` |535| data-juicer GPU-required ops are tagged `🚀GPU` (image/video/multi-modal); core text + dialog mappers are CPU-OK | DeepWiki index of `modelscope/data-juicer` |536| NeMo-Curator license, last push, redirect to `NVIDIA-NeMo/Curator` | `https://api.github.com/repos/NVIDIA/NeMo-Curator` |537| NeMo-Curator semantic dedup is GPU-only; CPU install drops differentiating ops | DeepWiki index of `NVIDIA/NeMo-Curator` |538| distilabel license, last push, DAG model, `FormatChatGenerationDPO`, `MinHashDedup`, `DeitaFiltering` | `https://api.github.com/repos/argilla-io/distilabel`; DeepWiki index of `argilla-io/distilabel` |539| `databricks/lilac` archived 2024-03-19 | `https://api.github.com/repos/databricks/lilac` (`archived: true`, `pushed_at: "2024-03-19T12:41:30Z"`) |540| `lilacai/lilac` is a 2-star squatter stub created 2025-11-14 | `https://api.github.com/repos/lilacai/lilac` |541 542---543 544## 7. Confirmed output path545 546**File:** `/home/codeseys/.hermes/hermes-agent/docs/research/REPLAYSIM_NORMALIZATION_RECONNAISSANCE.md`547**Length:** ≤600 lines (this file).548