Collecting Trajectories from Your Agents

By the end of this lesson you will be able to say what an agent log has to contain before it can be trained on, decide which runs are worth imitating and which are traps, get the credentials and machine names out of a transcript before anyone reads it, convert a run into the message format a fine-tuning trainer expects, and split the result so that the tasks you measure with never appear in the tasks you trained on.

None of this is glamorous. All of it is the part that decides whether the fine-tune in the next lesson means anything.

An agent run is already a dataset, almost

Part 24’s loop writes one JSON line per turn. Part 26’s harness writes one file per task attempt plus a results file with the outcome. Part 25’s coding agents each write their own session format. Between them you have, after a week of ordinary use, a few hundred recordings of a model doing your work.

What you do not automatically have is a recording that can be trained on. For that, each episode needs five things:

• the task, as the user turn that started it;
• the tool list the model was shown, as JSON schemas, because the tool descriptions are part of the prompt and training on a different list from the one you serve teaches the wrong mapping;
• every assistant turn in full, including the text it wrote and the calls it made with their arguments;
• every tool result, as the model saw it;
• the outcome, so that a filter can tell a run that worked from one that did not.

✕Common mistake

The commonest gap is the third. A log built for debugging records what a tool was called with and what came back, because that is what a person reading a failed run needs. It leaves out what the model said between the calls, because a person can infer it. A trainer cannot: the assistant turns are precisely the tokens the loss is computed on, and an episode without them teaches a sequence of tool calls attached to nothing.

Part 24’s minimal-agent.py records the assistant’s text alongside the tool names, so its transcripts convert faithfully. Part 26’s agent-eval.py records the tool steps, so an episode rebuilt from it is a plausible run rather than the one that happened. The collector below marks those episodes lossy and the filter drops them by default. If you want to train on your own scaffold’s output, the fix is in the scaffold: log the assistant messages.

RunnableAll tracks

collect-trajectories.py

"""Turn agent run logs into one JSON-lines file of complete trajectories.

Purpose: the first stage of Part 27. Reads the logs the course's own agents already
    write - Part 24's minimal-agent.py transcripts and Part 26's agent-eval.py results -
    and normalises them into one episode per line: the tool list, the whole message
    sequence including tool calls and tool results, and the outcome. Everything
    downstream in this part reads that one shape, so a reader whose agent logs something
    else has one adapter to write rather than four.
Platform: all (standard library only; no model, no accelerator, no network)
Minimum memory: 8 GB nominally, and far less in practice: this is text in memory
Assumes: Python 3.10 or newer. For --from part-24, a lab notebook containing records
    written by minimal-agent.py --labbook, whose "transcript" paths still resolve. For
    --from part-26, the results file agent-eval.py wrote with --out, whose
    "trajectory_file" paths still resolve. --tools takes an OpenAI-style tool list, which
    is what Part 24's Toolbox.schemas() returns.

Usage: python3 collect-trajectories.py --from part-24 --labbook labbook.md \\
           --tools tools.json --out raw/trajectories.jsonl
       python3 collect-trajectories.py --from part-26 --results agent-results.json \\
           --tools tools.json --out raw/trajectories.jsonl
       python3 collect-trajectories.py --from jsonl --input other-agent.jsonl \\
           --out raw/trajectories.jsonl
       python3 collect-trajectories.py --make-example raw/example-trajectories.jsonl

The episode shape, which is the contract for the rest of this part:

  {"id": "...", "task_id": "...", "task": "...", "source": "part-24/minimal-agent",
   "model": "local/agent", "scaffold": "minimal-agent",
   "tools": [ {"type": "function", "function": {"name": ..., "description": ...,
                                                "parameters": { JSON schema }}} ],
   "messages": [ {"role": "system"|"user"|"assistant"|"tool", ...} ],
   "outcome": {"passed": true, "stopped": "finished", "steps": 4, "turns": 3,
               "tokens": 2841, "seconds": 22.4},
   "lossy": false, "collected_at": "2026-09-09T09:00:00Z"}

An assistant message that calls tools carries "tool_calls", each one
{"type": "function", "function": {"name": ..., "arguments": { ... }}} with the arguments
as an object rather than a string: that is the shape the Transformers chat templates and
TRL's SFT trainer expect. trajectories-to-sft.py converts to the string form where a
tool chain wants one.
"""
from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


# --------------------------------------------------------------------------------------
# Reading
# --------------------------------------------------------------------------------------

def read_jsonl(path: Path) -> list[dict[str, Any]]:
    """Every JSON object on its own line. Lines that are not objects are skipped, so a
    notebook that also contains prose is a valid input."""
    rows: list[dict[str, Any]] = []
    if not path.is_file():
        sys.exit(f"{path} does not exist")
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = line.strip()
        if not line.startswith("{"):
            continue
        try:
            value = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(value, dict):
            rows.append(value)
    return rows


def parse_arguments(raw: Any) -> dict[str, Any]:
    """Tool-call arguments as an object, whatever the log stored them as."""
    if isinstance(raw, dict):
        return raw
    if isinstance(raw, str) and raw.strip():
        try:
            value = json.loads(raw)
        except json.JSONDecodeError:
            return {"_unparsed": raw}
        return value if isinstance(value, dict) else {"_value": value}
    return {}


def now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


# --------------------------------------------------------------------------------------
# Part 24: minimal-agent.py transcripts
# --------------------------------------------------------------------------------------

def episode_from_part24(run: dict[str, Any], result: dict[str, Any], tools: list[dict],
                        system_prompt: str | None, keep_reasoning: bool,
                        finish_tool: str) -> dict[str, Any] | None:
    """One episode from one entry in a Part 24 lab-notebook record.

    The transcript holds the assistant's text and the arguments of every executed call.
    The call to the finish tool is the one that never appears as an executed call,
    because the loop stops on it, so it is rebuilt from the recorded answer. Rebuilding
    it matters: a model trained on trajectories that end in plain text learns to stop
    calling finish, and the harness that checks for finish then scores it as a failure.
    """
    transcript_path = Path(str(result.get("transcript", "")))
    if not transcript_path.is_file():
        print(f"  skipped {result.get('id')}: transcript {transcript_path} is gone")
        return None
    rows = read_jsonl(transcript_path)
    if not rows:
        print(f"  skipped {result.get('id')}: transcript is empty")
        return None

    messages: list[dict[str, Any]] = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": result.get("task", "")})

    # Group the transcript by turn, in file order. Each turn is one assistant row
    # followed by the rows of the tools that turn executed.
    turns: list[list[dict[str, Any]]] = []
    for row in rows:
        if "assistant" in row or not turns:
            turns.append([row])
        else:
            turns[-1].append(row)

    answer = result.get("answer") or ""
    for group in turns:
        head = group[0]
        assistant = head.get("assistant") or {}
        tool_rows = [r for r in group if r.get("tool")]
        names = list(assistant.get("tool_calls") or [])
        if not names:
            content = assistant.get("content")
            if content:
                messages.append({"role": "assistant", "content": content})
            continue

        # Match each announced call to the next executed row with that name. The one
        # that has no row is the finish call.
        remaining = list(tool_rows)
        calls: list[dict[str, Any]] = []
        matched: list[dict[str, Any] | None] = []
        for name in names:
            row = next((r for r in remaining if r.get("tool") == name), None)
            if row is not None:
                remaining.remove(row)
                calls.append({"type": "function", "function": {
                    "name": name, "arguments": parse_arguments(row.get("arguments"))}})
                matched.append(row)
            elif name == finish_tool:
                calls.append({"type": "function", "function": {
                    "name": finish_tool, "arguments": {"answer": answer}}})
                matched.append(None)
            else:
                # Announced, never executed, and not the finish tool: the loop refused
                # it. Dropping it silently would train the model on a call it did not
                # make, so the whole episode is dropped instead.
                print(f"  skipped {result.get('id')}: call to {name!r} has no recorded result")
                return None

        message: dict[str, Any] = {"role": "assistant",
                                   "content": assistant.get("content"),
                                   "tool_calls": calls}
        if keep_reasoning and assistant.get("reasoning"):
            message["reasoning_content"] = assistant["reasoning"]
        messages.append(message)
        for name, row in zip(names, matched):
            if row is None:
                continue
            messages.append({"role": "tool", "name": name,
                             "content": str(row.get("result", ""))})

    return {
        "id": f"{run.get('model', 'model')}-{result.get('id', 'task')}-{transcript_path.stem}",
        "task_id": result.get("id"),
        "task": result.get("task", ""),
        "source": "part-24/minimal-agent",
        "model": run.get("model"),
        "scaffold": "minimal-agent",
        "tools": tools,
        "messages": messages,
        "outcome": {
            "passed": result.get("passed"),
            "stopped": result.get("stopped"),
            "steps": result.get("turns"),
            "turns": result.get("turns"),
            "tokens": result.get("tokens"),
            "seconds": result.get("seconds"),
            "tool_errors": sum(1 for r in rows if r.get("is_error")),
        },
        "lossy": False,
        "collected_at": now(),
    }


# --------------------------------------------------------------------------------------
# Part 26: agent-eval.py results
# --------------------------------------------------------------------------------------

def episode_from_part26(run: dict[str, Any], row: dict[str, Any], tools: list[dict],
                        system_prompt: str | None) -> dict[str, Any] | None:
    """One episode from one row of a Part 26 results file.

    Part 26's trajectory records what a tool was called with and what came back, which is
    what a reader needs to debug a run. It does not record the assistant's own words, so
    what is reconstructed here is a plausible trajectory rather than the one that
    happened, and every episode is marked lossy. Training on these teaches tool sequences
    and teaches nothing about what the model said between them; the honest fix is to log
    the assistant messages in the first place, which is what Part 24's agent does.
    """
    path = Path(str(row.get("trajectory_file", "")))
    steps = read_jsonl(path) if path.is_file() else []
    tool_steps = [s for s in steps if s.get("tool")]
    if not tool_steps:
        return None

    messages: list[dict[str, Any]] = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    meta = next((s.get("meta") for s in steps if isinstance(s.get("meta"), dict)), {}) or {}
    task_text = meta.get("task") or row.get("task") or ""
    messages.append({"role": "user", "content": task_text})

    for step in tool_steps:
        name = str(step["tool"])
        messages.append({"role": "assistant", "content": None, "tool_calls": [
            {"type": "function", "function": {"name": name,
                                              "arguments": parse_arguments(step.get("arguments"))}}]})
        messages.append({"role": "tool", "name": name, "content": str(step.get("result", ""))})
    messages.append({"role": "assistant", "content": row.get("answer", "")})

    checks = row.get("checks") or {}
    return {
        "id": f"{run.get('run_id', 'run')}-{row.get('id')}-r{row.get('attempt', 1)}",
        "task_id": row.get("id"),
        "task": task_text,
        "source": "part-26/agent-eval",
        "model": run.get("model"),
        "scaffold": run.get("scaffold"),
        "tools": tools,
        "messages": messages,
        "outcome": {
            "passed": checks.get("passed"),
            "stopped": row.get("stopped"),
            "steps": row.get("steps"),
            "turns": None,
            "tokens": row.get("tokens"),
            "seconds": row.get("seconds"),
            "tool_errors": sum(1 for s in tool_steps if s.get("is_error")),
        },
        "lossy": True,
        "collected_at": now(),
    }


# --------------------------------------------------------------------------------------
# Tools
# --------------------------------------------------------------------------------------

def infer_tools(episodes: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Build a tool list from the calls that were actually made.

    This exists so that the pipeline can be run before the real schemas are to hand, and
    it is not what you should train on. The description of a tool is part of the prompt
    the model reads at serving time; inventing an empty one here teaches the model to
    map an empty description onto a call, and at serving time it sees the real
    description instead. Every inferred entry says so in its own description.
    """
    seen: dict[str, set[str]] = {}
    for episode in episodes:
        for message in episode["messages"]:
            for call in message.get("tool_calls") or []:
                function = call.get("function", {})
                name = function.get("name")
                if not name:
                    continue
                keys = seen.setdefault(name, set())
                keys.update(k for k in (function.get("arguments") or {}) if isinstance(k, str))
    tools = []
    for name in sorted(seen):
        properties = {key: {"type": "string"} for key in sorted(seen[name])}
        tools.append({"type": "function", "function": {
            "name": name,
            "description": "Inferred from observed calls; replace with the tool's real description.",
            "parameters": {"type": "object", "properties": properties,
                           "required": sorted(properties)},
        }})
    return tools


# --------------------------------------------------------------------------------------
# The worked example
# --------------------------------------------------------------------------------------

EXAMPLE_TOOLS = [
    {"type": "function", "function": {
        "name": "list_files",
        "description": "List the file names in a directory inside the workspace.",
        "parameters": {"type": "object",
                       "properties": {"path": {"type": "string",
                                               "description": "Directory relative to the workspace root."}},
                       "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "read_file",
        "description": "Read a text file inside the workspace and return its contents.",
        "parameters": {"type": "object",
                       "properties": {"path": {"type": "string",
                                               "description": "File relative to the workspace root."}},
                       "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "finish",
        "description": "Give the final answer and end the run.",
        "parameters": {"type": "object",
                       "properties": {"answer": {"type": "string",
                                                 "description": "The answer, in plain prose."}},
                       "required": ["answer"]}}},
]

EXAMPLE_SYSTEM = ("You are a careful assistant working inside a sandboxed workspace. "
                  "Use the tools to find things out, take one step at a time, and call "
                  "finish exactly once when you have the answer.")


# The example transcripts have to contain something worth redacting, and this file is
# published, so nothing here may be a real secret or a real address. These four values
# are the safe stand-ins: an address in the loopback range, a documentation e-mail
# domain, a home path with a placeholder user name, and a token that is a prefix plus a
# run of one letter, assembled here so that no line of this file looks like a
# credential. Real transcripts contain the real things, which is the whole point of the
# scrubber.
SAFE_ENDPOINT = "http://127.0.0.1:4000/v1"
SAFE_EMAIL = "ops@example.com"
SAFE_HOME_PATH = "/home/user/deploy/logs"
SAFE_FAKE_TOKEN = "hf_" + "y" * 34


def example_episodes() -> list[dict[str, Any]]:
    """Four short episodes: two that succeeded, one that succeeded the long way round,
    and one that failed. They exercise every filter in trajectories-to-sft.py and carry
    the kind of leak scrub-trajectories.py is looking for."""
    def episode(task_id: str, task: str, calls: list[tuple[str, dict, str]], answer: str,
                passed: bool, stopped: str, tokens: int, seconds: float,
                tool_errors: int = 0) -> dict[str, Any]:
        messages: list[dict[str, Any]] = [
            {"role": "system", "content": EXAMPLE_SYSTEM},
            {"role": "user", "content": task},
        ]
        for name, arguments, result in calls:
            messages.append({"role": "assistant", "content": None, "tool_calls": [
                {"type": "function", "function": {"name": name, "arguments": arguments}}]})
            messages.append({"role": "tool", "name": name, "content": result})
        messages.append({"role": "assistant", "content": None, "tool_calls": [
            {"type": "function", "function": {"name": "finish",
                                              "arguments": {"answer": answer}}}]})
        return {
            "id": f"example-{task_id}", "task_id": task_id, "task": task,
            "source": "part-27/example", "model": "local/agent", "scaffold": "minimal-agent",
            "tools": EXAMPLE_TOOLS, "messages": messages,
            "outcome": {"passed": passed, "stopped": stopped, "steps": len(calls) + 1,
                        "turns": len(calls) + 1, "tokens": tokens, "seconds": seconds,
                        "tool_errors": tool_errors},
            "lossy": False, "collected_at": now(),
        }

    return [
        episode("e-index-file", "Which file in the workspace describes the release process?",
                [("list_files", {"path": "."}, "notes.md\nrelease.md\nrunbook.md"),
                 ("read_file", {"path": "release.md"},
                  "# Release\nTag the commit, then run the publish job.")],
                "release.md describes the release process.", True, "finished", 1840, 14.2),
        episode("e-owner", "Who is listed as the owner of the deploy job, and where do its logs go?",
                [("read_file", {"path": "runbook.md"},
                  f"Owner: platform team (contact {SAFE_EMAIL})\n"
                  f"Gateway: {SAFE_ENDPOINT} with token {SAFE_FAKE_TOKEN}\n"
                  f"Logs: {SAFE_HOME_PATH}, kept for 28 days")],
                "The platform team owns it and the logs are kept for 28 days.",
                True, "finished", 1210, 9.8),
        episode("e-long-way", "How many markdown files does the workspace contain?",
                [("list_files", {"path": "."}, "notes.md\nrelease.md\nrunbook.md"),
                 ("read_file", {"path": "notes.md"}, "Scratch notes."),
                 ("read_file", {"path": "release.md"}, "# Release"),
                 ("read_file", {"path": "runbook.md"}, "# Runbook"),
                 ("list_files", {"path": "."}, "notes.md\nrelease.md\nrunbook.md")],
                "Three.", True, "finished", 4900, 51.0),
        episode("e-missing", "What does the deployment checklist say about database migrations?",
                [("read_file", {"path": "checklist.md"},
                  "error: no such file inside the workspace: checklist.md")],
                "There is no deployment checklist in the workspace.",
                False, "finished", 980, 7.1, tool_errors=1),
    ]


# --------------------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--from", dest="source", choices=["part-24", "part-26", "jsonl"],
                        help="which log shape to read")
    parser.add_argument("--labbook", default=None,
                        help="part-24: the notebook minimal-agent.py appended to")
    parser.add_argument("--lab-name", default="part-24/minimal-agent",
                        help="part-24: the lab field to select in the notebook")
    parser.add_argument("--results", default=None,
                        help="part-26: the results file agent-eval.py wrote with --out")
    parser.add_argument("--input", default=None,
                        help="jsonl: a file already in this part's episode shape")
    parser.add_argument("--tools", default=None,
                        help="JSON file holding the OpenAI-style tool list the agent was given")
    parser.add_argument("--infer-tools", action="store_true",
                        help="build a tool list from the observed calls; read the warning it prints")
    parser.add_argument("--system-prompt-file", default=None,
                        help="text file holding the agent's system prompt, included in every episode")
    parser.add_argument("--keep-reasoning", action="store_true",
                        help="keep the assistant's reasoning text as reasoning_content")
    parser.add_argument("--finish-tool", default="finish",
                        help="name of the tool that ends a run; its call is rebuilt from the answer")
    parser.add_argument("--out", default="raw/trajectories.jsonl")
    parser.add_argument("--make-example", default=None, metavar="PATH",
                        help="write four worked episodes to PATH and exit")
    args = parser.parse_args()

    if args.make_example:
        out = Path(args.make_example)
        out.parent.mkdir(parents=True, exist_ok=True)
        episodes = example_episodes()
        out.write_text("".join(json.dumps(e) + "\n" for e in episodes), encoding="utf-8")
        print(f"wrote {len(episodes)} example episode(s) to {out}")
        print("Two of them carry the kind of thing scrub-trajectories.py is looking for.")
        return

    if not args.source:
        parser.error("give --from part-24, --from part-26 or --from jsonl, or --make-example")

    tools: list[dict[str, Any]] = []
    if args.tools:
        loaded = json.loads(Path(args.tools).read_text(encoding="utf-8"))
        tools = loaded["tools"] if isinstance(loaded, dict) and "tools" in loaded else loaded
        if not isinstance(tools, list):
            sys.exit(f"{args.tools} must hold a list of tool definitions")
    elif not args.infer_tools and args.source != "jsonl":
        sys.exit("pass --tools with the agent's real tool list, or --infer-tools and read "
                 "the warning. Part 24's Toolbox.schemas() returns exactly this list.")

    system_prompt = None
    if args.system_prompt_file:
        system_prompt = Path(args.system_prompt_file).read_text(encoding="utf-8").strip()

    episodes: list[dict[str, Any]] = []

    if args.source == "part-24":
        if not args.labbook:
            parser.error("--from part-24 needs --labbook")
        runs = [r for r in read_jsonl(Path(args.labbook)) if r.get("lab") == args.lab_name]
        if not runs:
            sys.exit(f"no records for lab {args.lab_name!r} in {args.labbook}")
        for run in runs:
            for result in run.get("results", []):
                episode = episode_from_part24(run, result, tools, system_prompt,
                                              args.keep_reasoning, args.finish_tool)
                if episode:
                    episodes.append(episode)

    elif args.source == "part-26":
        if not args.results:
            parser.error("--from part-26 needs --results")
        payload = json.loads(Path(args.results).read_text(encoding="utf-8"))
        run = payload.get("run", {})
        for row in payload.get("results", []):
            episode = episode_from_part26(run, row, tools, system_prompt)
            if episode:
                episodes.append(episode)
        if episodes:
            print("note: Part 26's trajectories record tool calls and results but not the "
                  "assistant's own words, so every episode here is marked lossy.")

    else:
        if not args.input:
            parser.error("--from jsonl needs --input")
        for row in read_jsonl(Path(args.input)):
            if "messages" not in row:
                continue
            row.setdefault("tools", tools)
            row.setdefault("lossy", False)
            row.setdefault("collected_at", now())
            episodes.append(row)

    if not episodes:
        sys.exit("no episodes were built; check the paths inside the log you passed")

    if args.infer_tools and not args.tools:
        tools = infer_tools(episodes)
        for episode in episodes:
            episode["tools"] = tools
        print(f"WARNING: inferred {len(tools)} tool definition(s) from the calls that were "
              "made. The descriptions are placeholders and they will be part of the "
              "training prompt. Replace them with the real schemas before you train "
              "anything you intend to serve.")

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("".join(json.dumps(e) + "\n" for e in episodes), encoding="utf-8")

    passed = sum(1 for e in episodes if e["outcome"].get("passed") is True)
    lossy = sum(1 for e in episodes if e.get("lossy"))
    tool_calls = sum(len(m.get("tool_calls") or []) for e in episodes for m in e["messages"])
    print(f"episodes:      {len(episodes)}")
    print(f"outcome passed:{passed:>4}")
    print(f"lossy:         {lossy:>4}")
    print(f"tool calls:    {tool_calls:>4}")
    print(f"tools:         {len(tools)}")
    print(f"written to {out}")
    print("\nNext: scrub-trajectories.py, before you read them and before you train on them.")


if __name__ == "__main__":
    main()

Download collect-trajectories.py522 lines

The script normalises all three shapes into one line per episode. Run it with --make-example first: it writes four worked episodes so you can see the shape and run the rest of the pipeline before you have collected anything of your own.

Filtering by outcome, and why that is not enough

The obvious filter is the right one to start with: keep the episodes whose outcome was a pass. This is what the agent-training literature does at scale. SWE-Gym, an environment of 2,438 Python task instances published in December 2024, trains software-engineering agents on trajectories sampled from it, and the paper reports gains of up to 19 percentage points absolute on SWE-Bench Verified and Lite for the fine-tuned agents; ToolLLM, from July 2023, builds its instruction data by having a strong model search for solution paths and keeping the ones that reach an answer. In both cases the selection rule is the same: the run ended in the right place, so imitate it.

Rejection sampling of this kind is cheap and it works. It is also not sufficient, for three reasons that show up immediately on a few hundred of your own runs.

A run that ended well may have got there badly. An agent that reads eleven files, repeats the same search three times and then answers correctly is a successful run and a bad example. Train on enough of them and the model learns that thoroughness means volume. The filter needs a step budget and a check for repeated identical calls, not just the outcome flag.

A run that failed may contain the behaviour you want. An episode where the model called the right tool with the right arguments and then mis-summarised the result is a good tool-use example and a bad answer example. Throwing it away is the safe default; keeping it means training on partial trajectories, which is a bigger change than it sounds and belongs after you have the simple version working.

The outcome check is only as good as the task file. Part 26’s checks are strings, regular expressions, tool names and step counts. A task whose expected answer is a single common word will be scored as passed by an agent that guessed. Read the tasks your filter is trusting before you trust it.

∴Research note

There is a fourth reason, and it is the one that catches people out. Trajectories sampled from the model you are about to train are, by definition, samples from the model’s current behaviour. Training on them sharpens what it already does; it does not teach it anything new. That is a real effect and a useful one, and it is also why the honest expectation for this part is a firmer, more consistent version of the model you have, not a cleverer one.

Secrets, before anyone reads the transcripts

A tool result is whatever the tool printed. If the agent read a configuration file, the transcript has the configuration file in it. If it ran a command that failed, the transcript has the error, which often contains an absolute path with your user name in it. If it read a runbook, the transcript may have an address, a hostname or an e-mail.

This matters twice. It matters because you may want to publish the adapter, or the dataset, or ask someone else to look at a run. And it matters because a model trained on a transcript containing a token will happily reproduce that token later, in a completely unrelated context, which is a much harder problem to notice than a leaked file.

RunnableAll tracks

scrub-trajectories.py

"""Redact credentials, addresses, e-mail and home paths from agent trajectories.

Purpose: the stage between collecting trajectories and looking at them. An agent
    transcript is a recording of your machine talking about itself: the tool results
    contain whatever was in the files it read and whatever the commands it ran printed,
    which is where tokens, hostnames, addresses, e-mail and absolute paths live. This
    script walks every string in every episode, replaces what it recognises with a
    marker, drops the episodes whose leak cannot be safely redacted, and writes a report
    saying what it found and where. The pattern set is the one this course's own
    release check runs over its published files, extended with the two shapes that
    appear in transcripts rather than in source: bearer headers and connection strings.
Platform: all (standard library only; no model, no accelerator, no network)
Minimum memory: 8 GB nominally, and far less in practice: this is text in memory
Assumes: Python 3.10 or newer. The input is the JSON-lines episode file that
    collect-trajectories.py wrote. agentlog.py sits next to this file.

Usage: python3 scrub-trajectories.py --in raw/trajectories.jsonl \\
           --out clean/trajectories.jsonl --report scrub-report.json
       python3 scrub-trajectories.py --in raw/trajectories.jsonl --out clean/trajectories.jsonl \\
           --extra "client=\\bNorthwind\\b" --extra "internal-host=\\b[a-z]+\\.corp\\.invalid\\b"
       python3 scrub-trajectories.py --in clean/trajectories.jsonl --check
       python3 scrub-trajectories.py --list-patterns

A regular expression finds the things that have a shape. It does not find the things
that only you know are sensitive: a client's name, an unreleased product, the fact that
a particular repository exists. --extra is for those, and reading a sample of the output
is not optional. --check re-runs the pattern set over a file and exits non-zero if
anything still matches, which is the step to put in front of anything you publish.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any, Iterable

import agentlog

# Each entry is (id, compiled pattern, replacement, allow pattern or None). A match on a
# line the allow pattern also matches is left alone: those are the cases where the shape
# is present but the value is not private, such as the loopback address or a
# documentation e-mail domain.
BUILTIN: list[tuple[str, str, str, str | None]] = [
    ("private-key",
     r"-----BEGIN (?:RSA|OPENSSH|EC|DSA|PGP) PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----",
     "[redacted:private-key]", None),
    ("ssh-public-key",
     r"\b(?:ssh-(?:rsa|ed25519|dss)|ecdsa-sha2-nistp\d+) AAAA[0-9A-Za-z+/=]{20,}",
     "[redacted:ssh-public-key]", None),
    ("aws-key", r"\bAKIA[0-9A-Z]{16}\b", "[redacted:aws-key]", None),
    ("github-token", r"\bgh[pousr]_[A-Za-z0-9]{30,}\b", "[redacted:github-token]", None),
    ("slack-token", r"\bxox[abpr]-[A-Za-z0-9-]{10,}", "[redacted:slack-token]", None),
    ("api-secret", r"\bsk-(?:live|test|proj|ant)-[A-Za-z0-9_-]{16,}", "[redacted:api-secret]", None),
    ("huggingface-token", r"\bhf_[A-Za-z0-9]{30,}\b", "[redacted:huggingface-token]", None),
    ("ngc-key", r"\bnvapi-[A-Za-z0-9_-]{30,}\b", "[redacted:ngc-key]", None),
    ("bearer-header", r"(?i)\b(authorization\s*:\s*bearer)\s+\S+", r"\1 [redacted:bearer]", None),
    ("assigned-secret",
     r"(?i)\b(api[_-]?key|secret|token|passw(?:or)?d)(\s*[:=]\s*)(?:\"[^\"]{6,}\"|'[^']{6,}'|\S{6,})",
     r"\1\2[redacted:secret]",
     r"(?i)(?:placeholder|example|your[_-]|xxx|<[^>]+>|redacted)"),
    ("connection-string",
     r"\b[a-z][a-z0-9+.-]*://[^\s/@]+:[^\s/@]+@[^\s\"']+",
     "[redacted:connection-string]", None),
    # No exemption for documentation domains here, unlike the release check. A published
    # page may say example.com; a training set has no reason to carry any address at all,
    # and a rule with holes in it is a rule people stop reading.
    ("email", r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b", "[redacted:email]", None),
    ("ip-address",
     r"\b(?:(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)\b",
     "[redacted:address]", r"(?:127\.0\.0\.1|0\.0\.0\.0)"),
    ("mac-address", r"\b(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\b", "[redacted:mac]", None),
    ("home-path",
     r"(?:/home/[a-z][a-z0-9_-]*|/Users/[A-Za-z][A-Za-z0-9_-]*|[A-Z]:\\Users\\[A-Za-z][A-Za-z0-9_-]*)",
     "[redacted:home]", None),
    ("root-login", r"\broot@[a-z0-9.-]+", "[redacted:root-login]", None),
]

# Leaks a marker cannot fix. A private key is not made safe by having its middle removed,
# because the fact that it was in this transcript is itself the finding, and the episode
# is worth nothing as training data anyway.
DEFAULT_DROP_ON = ("private-key",)


class Pattern:
    def __init__(self, name: str, regex: str, replacement: str, allow: str | None) -> None:
        self.name = name
        self.regex = re.compile(regex)
        self.replacement = replacement
        self.allow = re.compile(allow) if allow else None

    def apply(self, text: str) -> tuple[str, int, list[str]]:
        """Redact every match that the allow pattern does not exempt."""
        hits: list[str] = []

        def substitute(match: re.Match[str]) -> str:
            found = match.group(0)
            if self.allow and self.allow.search(found):
                return found
            hits.append(found[:12] + "…" if len(found) > 12 else found)
            return match.expand(self.replacement)

        return self.regex.sub(substitute, text), len(hits), hits


def build_patterns(extra: Iterable[str], redact_loopback: bool) -> list[Pattern]:
    patterns: list[Pattern] = []
    for name, regex, replacement, allow in BUILTIN:
        if name == "ip-address" and redact_loopback:
            allow = None
        patterns.append(Pattern(name, regex, replacement, allow))
    for item in extra:
        if "=" not in item:
            sys.exit(f"--extra needs NAME=REGEX, got {item!r}")
        name, regex = item.split("=", 1)
        try:
            patterns.append(Pattern(name.strip(), regex, f"[redacted:{name.strip()}]", None))
        except re.error as exc:
            sys.exit(f"--extra {name}: {exc}")
    return patterns


def scrub_value(value: Any, patterns: list[Pattern], counts: dict[str, int],
                samples: dict[str, list[str]], found: set[str]) -> Any:
    """Walk the whole object. Strings are redacted; everything else is rebuilt as it was."""
    if isinstance(value, str):
        text = value
        for pattern in patterns:
            text, n, hits = pattern.apply(text)
            if n:
                counts[pattern.name] = counts.get(pattern.name, 0) + n
                found.add(pattern.name)
                bucket = samples.setdefault(pattern.name, [])
                for hit in hits:
                    if len(bucket) < 5:
                        bucket.append(hit)
        return text
    if isinstance(value, list):
        return [scrub_value(v, patterns, counts, samples, found) for v in value]
    if isinstance(value, dict):
        return {k: scrub_value(v, patterns, counts, samples, found) for k, v in value.items()}
    return value


def read_episodes(path: Path) -> list[dict[str, Any]]:
    if not path.is_file():
        sys.exit(f"{path} does not exist; run collect-trajectories.py first")
    episodes = []
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = line.strip()
        if not line.startswith("{"):
            continue
        episodes.append(json.loads(line))
    if not episodes:
        sys.exit(f"{path} holds no episodes")
    return episodes


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--in", dest="input", default="raw/trajectories.jsonl")
    parser.add_argument("--out", default=None, help="where the scrubbed episodes go")
    parser.add_argument("--report", default=None, help="write the counts and samples here")
    parser.add_argument("--extra", action="append", default=[], metavar="NAME=REGEX",
                        help="one more pattern of your own; repeatable")
    parser.add_argument("--drop-on", action="append", default=None, metavar="NAME",
                        help=f"drop the whole episode on this pattern; default {DEFAULT_DROP_ON}")
    parser.add_argument("--redact-loopback", action="store_true",
                        help="also redact 127.0.0.1 and 0.0.0.0, which are kept by default")
    parser.add_argument("--check", action="store_true",
                        help="report what still matches and exit non-zero if anything does")
    parser.add_argument("--list-patterns", action="store_true")
    parser.add_argument("--labbook", default=None)
    parser.add_argument("--notes", default=None)
    args = parser.parse_args()

    if args.list_patterns:
        print(f"{'name':<20} replacement")
        for name, _regex, replacement, allow in BUILTIN:
            note = "  (allow: " + allow + ")" if allow else ""
            print(f"{name:<20} {replacement}{note}")
        print("\nAdd your own with --extra NAME=REGEX. Client names, internal hostnames and "
              "project code names have no shape a regular expression can find on its own.")
        return

    patterns = build_patterns(args.extra, args.redact_loopback)
    drop_on = set(args.drop_on if args.drop_on is not None else DEFAULT_DROP_ON)
    episodes = read_episodes(Path(args.input))

    counts: dict[str, int] = {}
    samples: dict[str, list[str]] = {}
    kept: list[dict[str, Any]] = []
    dropped: list[dict[str, str]] = []

    for episode in episodes:
        found: set[str] = set()
        scrubbed = scrub_value(episode, patterns, counts, samples, found)
        blocking = sorted(found & drop_on)
        if blocking:
            dropped.append({"id": str(episode.get("id")), "patterns": ", ".join(blocking)})
            continue
        if found:
            scrubbed["scrubbed"] = sorted(found)
        kept.append(scrubbed)

    if args.check:
        total = sum(counts.values())
        if total:
            print(f"{total} match(es) still present in {args.input}:")
            for name in sorted(counts):
                print(f"  {name:<20} {counts[name]:>5}   e.g. {samples.get(name, [''])[0]}")
            sys.exit(1)
        print(f"{args.input}: nothing matched. That is a floor, not a proof: read a sample.")
        return

    if not args.out:
        parser.error("give --out, or --check to test a file that is already scrubbed")

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("".join(json.dumps(e) + "\n" for e in kept), encoding="utf-8")

    report = {
        "input": args.input,
        "output": args.out,
        "episodes_in": len(episodes),
        "episodes_kept": len(kept),
        "episodes_dropped": dropped,
        "redactions_by_pattern": dict(sorted(counts.items())),
        "samples": {k: v for k, v in sorted(samples.items())},
        "extra_patterns": args.extra,
        "drop_on": sorted(drop_on),
        "loopback_redacted": args.redact_loopback,
    }
    if args.report:
        Path(args.report).write_text(json.dumps(report, indent=2), encoding="utf-8")

    print(f"episodes in:      {len(episodes)}")
    print(f"episodes kept:    {len(kept)}")
    print(f"episodes dropped: {len(dropped)}")
    if counts:
        print("\nredactions by pattern")
        for name in sorted(counts):
            print(f"  {name:<20} {counts[name]:>5}   e.g. {samples[name][0]}")
    else:
        print("\nNothing matched. Either the transcripts are clean or your agent's tools "
              "never printed anything private. Read ten of them and decide which.")
    for row in dropped:
        print(f"  dropped {row['id']}: {row['patterns']}")
    print(f"\nwritten to {out}")
    if args.report:
        print(f"report written to {args.report}")
    print("Now read a sample by hand. The patterns find shapes; they do not find the "
          "things only you know are sensitive.")

    if args.labbook:
        record = agentlog.record(
            labbook=args.labbook,
            lab="part-27/scrub-trajectories",
            model=None,
            dataset={"path": args.input, "sha256": agentlog.file_sha256(args.input),
                     "episodes": len(episodes)},
            data_lineage=agentlog.lineage(trajectories=args.input, scrub_report=args.report),
            hyperparameters={"extra_patterns": args.extra, "drop_on": sorted(drop_on),
                             "loopback_redacted": args.redact_loopback},
            seed=None, losses=None,
            scores={"episodes_kept": len(kept), "episodes_dropped": len(dropped),
                    "redactions": sum(counts.values())},
            config_path=__file__, notes=args.notes,
        )
        print(f"recorded run {record['run_id']} in {args.labbook}")


if __name__ == "__main__":
    main()

Download scrub-trajectories.py276 lines

The pattern set is this course’s own release check, which scans every file before publication for private keys, cloud access keys, provider tokens, e-mail addresses, addresses, home directory paths and root login targets. Two shapes are added that appear in transcripts rather than in source: an Authorization header carrying a bearer token, and a connection string with a password in the authority part. Episodes carrying a private key are dropped rather than redacted, because the fact that one was in the transcript is itself a finding and the episode is worth nothing as training data anyway.

RunnableAll tracks

scrub, then verify the output against the same patterns

python3 scrub-trajectories.py --in raw/trajectories.jsonl \
    --out clean/trajectories.jsonl --report scrub-report.json
python3 scrub-trajectories.py --in clean/trajectories.jsonl --check

Output — what you should see

episodes in:      4
episodes kept:    4
episodes dropped: 0

redactions by pattern
  email                    1   e.g. ops@example.…
  home-path                1   e.g. /home/user
  huggingface-token        1   e.g. hf_yyyyyyyyy…

⛨Security

A pattern set finds things that have a shape. It does not find a client’s name, an unreleased product, the fact that a particular repository exists, or a sentence in a design document that should not leave the building. --extra NAME=REGEX is where your own strings go, and reading a sample of the output by hand is not optional. The --check pass is a floor, not a proof.

Formatting tool calls for training

Every model family has its own wire format for a tool call. Qwen and the models that follow it use the Hermes convention: the tool schemas go inside <tools> tags in the system prompt and each call comes back inside <tool_call> tags as a JSON object with a name and an arguments field. The gpt-oss family uses harmony channels. Others use a pythonic form.

You do not write any of them by hand, and you should not. The format is produced by the model’s chat template from a structured message list, and the template travels with the tokeniser. The Transformers documentation on tool use, read on 2026-09-09, is explicit about the structure: tools are passed to apply_chat_template as a tools argument, either as JSON schemas or as Python functions whose Google-style docstrings are parsed into schemas by get_json_schema; a call goes in the tool_calls key of an assistant message as an object with a type of function and a function holding a name and an arguments object; and the result comes back as a message with the tool role. The documentation warns that although this looks like the OpenAI API, that API sends arguments as a JSON string and Transformers expects an object, which is exactly the kind of mismatch that produces a fine-tune that trains without error and calls nothing.

The dataset guide for TRL 1.12.0 · verified 2026-09-08, read the same day, adds the training-time half: a supervised fine-tuning dataset with tool calls carries an extra column named tools holding the list of available tools as codified JSON schemas, and the chat template uses it to build the system prompt. A complete row is a messages list and a tools list.

From a run to a training row

1. 1CollectOne JSON line per episode: task, tool schemas, every message, outcome.
2. 2ScrubRedact what has a shape, drop what cannot be redacted, add your own patterns.
3. 3FilterOutcome, step budget, repeated calls, calls to tools that were never declared.
4. 4Deduplicate and decontaminateExact and near-duplicates out; anything overlapping the evaluation suite out.
5. 5Split and writeTRL's conversational layout with a tools column, and mlx-lm's for Track M.

The first three stages are about honesty and the last two are about format. Every stage writes a report, because a pipeline whose intermediate steps you cannot inspect is one you will not be able to debug in a month.

The two layouts differ in one field, and the difference is documented rather than invented. mlx-lm’s fine-tuning guide, read on 2026-09-09, shows a tools-format example whose tool_calls entries carry arguments as a JSON string; the Transformers path wants an object. The converter writes both, so the same episodes train on every track without anyone editing a file.

RunnableAll tracks

trajectories-to-sft.py

"""Filter, deduplicate and decontaminate agent trajectories, and write a training set.

Purpose: the stage that decides what the fine-tune will actually learn. Reads the
    scrubbed episodes, keeps the ones whose outcome and shape are worth imitating,
    removes exact and near-duplicates, drops anything that overlaps the evaluation
    suite, splits by task so that no task appears in both halves, and writes the two
    layouts this part trains from: TRL's conversational format with a `tools` column,
    and mlx-lm's tools format for Track M. Every rejection is counted by reason and a
    sample of each reason is written out, because a filter you cannot inspect is a
    filter you cannot trust.
Platform: all (standard library only; no model, no accelerator, no network)
Minimum memory: 8 GB nominally, and far less in practice: this is text in memory
Assumes: Python 3.10 or newer. The input is the JSON-lines episode file
    scrub-trajectories.py wrote. --tasks takes the evaluation suite you will measure
    with, in the shape of Part 26's agent-tasks.json: an object with a "tasks" list
    whose items carry "id" and "task". agentlog.py sits next to this file.

Usage: python3 trajectories-to-sft.py --in clean/trajectories.jsonl --out-dir . \\
           --tasks agent-tasks.json --report filter-report.json --labbook labbook.md
       python3 trajectories-to-sft.py --in clean/trajectories.jsonl --out-dir . \\
           --include-failures --max-steps 12 --near-duplicate 0.7
       python3 trajectories-to-sft.py --in clean/trajectories.jsonl --print-example

Written under --out-dir:
  data/{train,valid}.jsonl        {"messages": [...], "tools": [...]}, arguments as objects
  data-mlx/{train,valid}.jsonl    the same episodes with tool-call arguments as strings
  filter-report.json              counts by reason, plus a sample of each

The two layouts differ in one field on purpose. The Transformers chat templates and
TRL's SFT trainer expect a tool call's `arguments` to be an object; mlx-lm's documented
tools example carries them as a JSON string. Writing both here means the same episodes
train on every track without anyone editing a file by hand.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import random
import re
import sys
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

import agentlog

WORD_RE = re.compile(r"[a-z0-9]+")


# --------------------------------------------------------------------------------------
# Text comparison, the same containment measure Part 13's decontaminate.py uses
# --------------------------------------------------------------------------------------

def words(text: str) -> list[str]:
    return WORD_RE.findall(text.lower())


def ngrams(tokens: list[str], n: int) -> set[tuple[str, ...]]:
    if len(tokens) < n:
        return {tuple(tokens)} if tokens else set()
    return {tuple(tokens[i:i + n]) for i in range(len(tokens) - n + 1)}


def containment(a: set[tuple[str, ...]], b: set[tuple[str, ...]]) -> float:
    """How much of a is also in b. Containment rather than Jaccard, because a short
    evaluation task buried inside a long training task is the case that matters."""
    return (len(a & b) / len(a)) if a else 0.0


def episode_text(episode: dict[str, Any]) -> str:
    """Everything the model would see and write, flattened, for hashing and comparison."""
    parts: list[str] = [str(episode.get("task", ""))]
    for message in episode.get("messages", []):
        if message.get("content"):
            parts.append(str(message["content"]))
        for call in message.get("tool_calls") or []:
            function = call.get("function", {})
            parts.append(str(function.get("name", "")))
            parts.append(json.dumps(function.get("arguments", {}), sort_keys=True))
    return "\n".join(parts)


def episode_hash(episode: dict[str, Any]) -> str:
    return hashlib.sha256(episode_text(episode).encode("utf-8")).hexdigest()


# --------------------------------------------------------------------------------------
# Filters
# --------------------------------------------------------------------------------------

def tool_calls_of(episode: dict[str, Any]) -> list[tuple[str, str]]:
    """(name, canonical arguments) for every call in the episode, in order."""
    out: list[tuple[str, str]] = []
    for message in episode.get("messages", []):
        for call in message.get("tool_calls") or []:
            function = call.get("function", {})
            out.append((str(function.get("name", "")),
                        json.dumps(function.get("arguments", {}), sort_keys=True)))
    return out


def shape_reason(episode: dict[str, Any], args: argparse.Namespace) -> str | None:
    """Why this episode is not worth imitating, or None if it is."""
    outcome = episode.get("outcome") or {}
    if not args.include_failures and outcome.get("passed") is not True:
        return "outcome-not-passed"
    if args.drop_lossy and episode.get("lossy"):
        return "lossy-reconstruction"
    if not episode.get("tools"):
        return "no-tool-list"
    calls = tool_calls_of(episode)
    if len(calls) < args.min_tool_calls:
        return "too-few-tool-calls"
    steps = outcome.get("steps")
    if args.max_steps and isinstance(steps, int) and steps > args.max_steps:
        return "over-step-budget"
    errors = outcome.get("tool_errors")
    if isinstance(errors, int) and errors > args.max_tool_errors:
        return "too-many-tool-errors"
    if args.drop_repeated_calls and len(calls) != len(set(calls)):
        return "repeated-identical-call"
    names = {name for name, _ in calls}
    declared = {t.get("function", {}).get("name") for t in episode["tools"]}
    if not names <= declared:
        return "call-to-undeclared-tool"
    if len(episode.get("messages", [])) > args.max_messages:
        return "too-many-messages"
    return None


# --------------------------------------------------------------------------------------
# Output shapes
# --------------------------------------------------------------------------------------

def trl_row(episode: dict[str, Any], keep_system: bool) -> dict[str, Any]:
    """TRL conversational language modelling, with the tools column the SFT trainer reads."""
    messages = [m for m in episode["messages"] if keep_system or m.get("role") != "system"]
    cleaned: list[dict[str, Any]] = []
    for message in messages:
        row: dict[str, Any] = {"role": message["role"]}
        if message.get("content") is not None:
            row["content"] = message["content"]
        if message.get("tool_calls"):
            row["tool_calls"] = [
                {"type": "function",
                 "function": {"name": c["function"]["name"],
                              "arguments": c["function"].get("arguments", {})}}
                for c in message["tool_calls"]
            ]
            row.setdefault("content", "")
        if message.get("name"):
            row["name"] = message["name"]
        cleaned.append(row)
    return {"messages": cleaned, "tools": episode["tools"]}


def mlx_row(row: dict[str, Any]) -> dict[str, Any]:
    """The same episode with tool-call arguments as JSON strings and an id per call,
    which is the shape mlx-lm's documented tools example uses."""
    messages = []
    for index, message in enumerate(row["messages"]):
        copy = dict(message)
        if copy.get("tool_calls"):
            copy["tool_calls"] = [
                {"id": f"call_{index}_{n}", "type": "function",
                 "function": {"name": c["function"]["name"],
                              "arguments": json.dumps(c["function"]["arguments"])}}
                for n, c in enumerate(message["tool_calls"])
            ]
        messages.append(copy)
    return {"messages": messages, "tools": row["tools"]}


# --------------------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--in", dest="input", default="clean/trajectories.jsonl")
    parser.add_argument("--out-dir", default=".")
    parser.add_argument("--tasks", default=None,
                        help="the evaluation suite to decontaminate against, in Part 26's shape")
    parser.add_argument("--include-failures", action="store_true",
                        help="keep episodes whose outcome was not a pass; read the lesson first")
    parser.add_argument("--keep-lossy", dest="drop_lossy", action="store_false",
                        help="keep episodes rebuilt from logs with no assistant text")
    parser.add_argument("--min-tool-calls", type=int, default=1)
    parser.add_argument("--max-steps", type=int, default=0, help="0 disables the step filter")
    parser.add_argument("--max-tool-errors", type=int, default=0)
    parser.add_argument("--drop-repeated-calls", action="store_true",
                        help="drop an episode that made the same call with the same arguments twice")
    parser.add_argument("--max-messages", type=int, default=80)
    parser.add_argument("--drop-system", dest="keep_system", action="store_false",
                        help="leave the system prompt out of the training rows")
    parser.add_argument("--near-duplicate", type=float, default=0.8,
                        help="containment above this counts as a duplicate; 0 disables")
    parser.add_argument("--contamination", type=float, default=0.6,
                        help="containment against an evaluation task above this is contamination")
    parser.add_argument("--n", type=int, default=8, help="n-gram size for both comparisons")
    parser.add_argument("--valid-fraction", type=float, default=0.15)
    parser.add_argument("--seed", type=int, default=0)
    parser.add_argument("--report", default=None)
    parser.add_argument("--print-example", action="store_true",
                        help="print one converted training row and exit")
    parser.add_argument("--labbook", default=None)
    parser.add_argument("--notes", default=None)
    args = parser.parse_args()

    path = Path(args.input)
    if not path.is_file():
        sys.exit(f"{path} does not exist; run scrub-trajectories.py first")
    episodes = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()
                if line.strip().startswith("{")]
    if not episodes:
        sys.exit(f"{path} holds no episodes")

    rejected: Counter[str] = Counter()
    examples: dict[str, list[str]] = defaultdict(list)

    def reject(episode: dict[str, Any], reason: str) -> None:
        rejected[reason] += 1
        if len(examples[reason]) < 5:
            examples[reason].append(f"{episode.get('id')}: {str(episode.get('task', ''))[:120]}")

    # 1. Shape and outcome.
    kept: list[dict[str, Any]] = []
    for episode in episodes:
        reason = shape_reason(episode, args)
        if reason:
            reject(episode, reason)
        else:
            kept.append(episode)

    # 2. Exact duplicates.
    seen_hash: set[str] = set()
    unique: list[dict[str, Any]] = []
    for episode in kept:
        digest = episode_hash(episode)
        if digest in seen_hash:
            reject(episode, "exact-duplicate")
            continue
        seen_hash.add(digest)
        unique.append(episode)

    # 3. Near-duplicates, compared against what has already been accepted.
    deduped: list[dict[str, Any]] = []
    accepted_grams: list[set[tuple[str, ...]]] = []
    if args.near_duplicate > 0:
        for episode in unique:
            grams = ngrams(words(episode_text(episode)), args.n)
            if any(containment(grams, other) >= args.near_duplicate for other in accepted_grams):
                reject(episode, "near-duplicate")
                continue
            accepted_grams.append(grams)
            deduped.append(episode)
    else:
        deduped = unique

    # 4. Decontamination against the suite this fine-tune will be measured with.
    eval_ids: set[str] = set()
    eval_grams: list[tuple[str, set[tuple[str, ...]]]] = []
    if args.tasks:
        suite = json.loads(Path(args.tasks).read_text(encoding="utf-8"))
        for task in suite.get("tasks", []):
            eval_ids.add(str(task.get("id")))
            eval_grams.append((str(task.get("id")), ngrams(words(str(task.get("task", ""))), args.n)))

    clean: list[dict[str, Any]] = []
    contaminated: list[dict[str, str]] = []
    for episode in deduped:
        if str(episode.get("task_id")) in eval_ids:
            reject(episode, "same-task-id-as-evaluation")
            contaminated.append({"id": str(episode.get("id")), "why": "task id is in the suite"})
            continue
        grams = ngrams(words(str(episode.get("task", ""))), args.n)
        hit = next((tid for tid, other in eval_grams
                    if containment(other, grams) >= args.contamination), None)
        if hit:
            reject(episode, "overlaps-evaluation-task")
            contaminated.append({"id": str(episode.get("id")),
                                 "why": f"overlaps evaluation task {hit}"})
            continue
        clean.append(episode)

    if not clean:
        sys.exit("nothing survived the filters. Read filter-report.json: the commonest cause "
                 "is that no episode passed, and the second is that every task in the "
                 "collection set is also in the evaluation suite.")

    # 5. Split by task, so that no task id appears in both halves. A model that saw the
    #    same task in training is not being evaluated on it.
    by_task: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for episode in clean:
        by_task[str(episode.get("task_id"))].append(episode)
    task_ids = sorted(by_task)
    random.Random(args.seed).shuffle(task_ids)
    want_valid = max(1, round(len(task_ids) * args.valid_fraction)) if len(task_ids) > 1 else 0
    valid_ids = set(task_ids[:want_valid])
    train_episodes = [e for tid in task_ids if tid not in valid_ids for e in by_task[tid]]
    valid_episodes = [e for tid in task_ids if tid in valid_ids for e in by_task[tid]]
    if not valid_episodes:
        print("WARNING: every episode came from one task, so there is no held-out split. "
              "Collect trajectories from more tasks before you believe any number here.")

    rows_train = [trl_row(e, args.keep_system) for e in train_episodes]
    rows_valid = [trl_row(e, args.keep_system) for e in valid_episodes]

    if args.print_example:
        if not rows_train:
            sys.exit("no training rows to print")
        print(json.dumps(rows_train[0], indent=2))
        return

    out_dir = Path(args.out_dir)
    (out_dir / "data").mkdir(parents=True, exist_ok=True)
    (out_dir / "data-mlx").mkdir(parents=True, exist_ok=True)
    for name, rows in (("train", rows_train), ("valid", rows_valid)):
        (out_dir / "data" / f"{name}.jsonl").write_text(
            "".join(json.dumps(r) + "\n" for r in rows), encoding="utf-8")
        (out_dir / "data-mlx" / f"{name}.jsonl").write_text(
            "".join(json.dumps(mlx_row(r)) + "\n" for r in rows), encoding="utf-8")

    assistant_turns = sum(1 for r in rows_train for m in r["messages"] if m["role"] == "assistant")
    calls = sum(len(m.get("tool_calls") or []) for r in rows_train for m in r["messages"])
    report = {
        "input": args.input,
        "episodes_in": len(episodes),
        "train_rows": len(rows_train),
        "valid_rows": len(rows_valid),
        "train_tasks": len(task_ids) - len(valid_ids),
        "valid_tasks": len(valid_ids),
        "assistant_turns_train": assistant_turns,
        "tool_calls_train": calls,
        "rejected_by_reason": dict(sorted(rejected.items())),
        "rejected_examples": {k: v for k, v in sorted(examples.items())},
        "contaminated": contaminated,
        "settings": {"include_failures": args.include_failures, "max_steps": args.max_steps,
                     "max_tool_errors": args.max_tool_errors,
                     "drop_repeated_calls": args.drop_repeated_calls,
                     "near_duplicate": args.near_duplicate, "contamination": args.contamination,
                     "n": args.n, "valid_fraction": args.valid_fraction, "seed": args.seed,
                     "tasks": args.tasks},
    }
    report_path = Path(args.report) if args.report else out_dir / "filter-report.json"
    report_path.write_text(json.dumps(report, indent=2), encoding="utf-8")

    print(f"episodes in:      {len(episodes)}")
    print(f"train rows:       {len(rows_train)}   from {report['train_tasks']} task(s)")
    print(f"valid rows:       {len(rows_valid)}   from {report['valid_tasks']} task(s)")
    print(f"assistant turns:  {assistant_turns}")
    print(f"tool calls:       {calls}")
    if rejected:
        print("\nrejected by reason")
        for reason, count in sorted(rejected.items()):
            print(f"  {reason:<28} {count:>5}")
    print(f"\nreport written to {report_path}")
    print("Read five rejected examples and five accepted ones before you train. The filter "
          "decides what the model imitates and it is easier to fix here than afterwards.")

    if args.labbook:
        record = agentlog.record(
            labbook=args.labbook,
            lab="part-27/trajectories-to-sft",
            model=None,
            dataset={"path": str(out_dir / "data" / "train.jsonl"),
                     "sha256": agentlog.file_sha256(out_dir / "data" / "train.jsonl"),
                     "train_examples": len(rows_train), "validation_examples": len(rows_valid)},
            data_lineage=agentlog.lineage(trajectories=args.input, filter_report=str(report_path),
                                          evaluation_suite=args.tasks),
            hyperparameters=report["settings"],
            seed=args.seed, losses=None,
            scores={"train_rows": len(rows_train), "valid_rows": len(rows_valid),
                    "rejected": sum(rejected.values())},
            config_path=__file__, notes=args.notes,
        )
        print(f"recorded run {record['run_id']} in {args.labbook}")


if __name__ == "__main__":
    main()

Download trajectories-to-sft.py382 lines

✦Tip

Before you train, render one row through the model’s own chat template and read it. Part 27’s training script has an --inspect-only flag that does exactly this and warns when the tool list or a call does not appear in the rendered text. A template that silently ignores the tools column produces a training prompt that will never occur at serving time, and the only symptom is a fine-tune that does not help.

Deduplication and held-out tasks

Two problems share one solution here.

The first is repetition. If you ran the same task five times to see the variance, you have five near-identical episodes, and a training set where one task appears five times and another once has told the model which one matters. The converter removes exact duplicates by hashing the flattened message sequence, and near-duplicates by word n-gram containment above a threshold you set, which is the same measure Part 13’s decontaminate.py uses.

The second is contamination, and it is worse because it is invisible. The suite you are going to measure with is Part 26’s agent-tasks.json. If any of those tasks, or a paraphrase of one, is in the training set, the improvement you measure afterwards is memorisation. The converter checks for both the same task identifier and a high containment against each evaluation task, and drops what it finds.

The structural fix, which matters more than the check, is to collect from a different set of tasks than you measure with. This part ships one: a collection set over the same three sample documents as Part 26’s suite, asking sixteen different questions on purpose. Use it to get the pipeline running, then replace it with tasks over your own repository, which is the only version that will improve anything you care about.

The split is by task rather than by episode, for the same reason. Two attempts at the same task are not independent, and putting one in training and one in validation gives you a validation loss that looks better than the model is.

Preserve causal order and provenance in the dataset

A trajectory should distinguish user instructions, model proposals, tool invocations, tool results and independent outcome checks. Preserve call identifiers and ordering so an answer cannot appear to use a tool result that arrived later. Record the model, prompt template, tool version and initial environment revision.

Scrub secrets before distributing or training on transcripts, while keeping a restricted original only if your retention policy permits it. Automated patterns catch some secrets but not every sensitive fact; inspect samples and use synthetic tasks for the first collection. Keep a report of removed or transformed fields so the training representation remains interpretable.

Split by source task, repository or incident before producing variants. A successful run can still contain unnecessary calls or an unsafe attempt that the sandbox blocked. Decide whether those turns belong in demonstrations rather than accepting the entire transcript because the final test passed. Outcome filtering and trajectory-quality filtering answer different questions, and both matter when teaching another model how to act.

Recap

• A trainable episode needs the task, the tool schemas, every assistant turn in full, every tool result and the outcome. A debugging log usually has four of the five.
• Filtering on the outcome is the right first rule and not a sufficient one: a successful run can still be a bad example, and the outcome check is only as good as the task file behind it.
• Transcripts contain whatever your tools printed. Scrub them with a pattern set, add your own patterns for the things that have no shape, verify the output, and read a sample.
• Tool calls are produced by the chat template from structured messages, not written by hand. Training needs a tools column of JSON schemas alongside the messages, and the arguments are an object on the Transformers path and a string on the mlx-lm one.
• Deduplicate, decontaminate against the suite you will be judged on, and split by task rather than by episode.

Check your understanding

Question 1. Your scaffold logs, for each step, the tool name, its arguments and the result, plus the final answer. What is missing before these runs can be used for supervised fine-tuning?

• Nothing; that is a complete trajectory
• The assistant turns: what the model wrote between the calls, which is where the loss lands
• The wall-clock time per step
• The quantisation the model was served at

Show the answer and why

Answer: The assistant turns: what the model wrote between the calls, which is where the loss lands

The loss in supervised fine-tuning is computed on the assistant tokens. A log of calls and results describes what happened to the environment, not what the model produced, so an episode rebuilt from it attaches a tool sequence to text the model never wrote. Fix it in the scaffold rather than in the converter.

Question 2. Which of these should be dropped from a training set of successful trajectories?

• A run that answered correctly after making the same search three times
• A run that answered correctly in two steps
• A run whose task text closely paraphrases a task in your evaluation suite
• A run that called a tool which was not in the declared tool list

Show the answer and why

Answer: A run that answered correctly after making the same search three times, A run whose task text closely paraphrases a task in your evaluation suite, A run that called a tool which was not in the declared tool list

The repeated-search run teaches padding; the paraphrase contaminates the measurement you are about to take; the undeclared call cannot be reproduced at serving time because the model will never be shown that tool. Only the two-step run is what you want more of.

Question 3. The scrubber reports no matches on your transcripts. What follows?

• The transcripts are safe to publish
• The pattern set found nothing with a recognisable shape, which is a floor rather than a proof
• The scrubber is broken
• The transcripts contain no tool results

Show the answer and why

Answer: The pattern set found nothing with a recognisable shape, which is a floor rather than a proof

A regular expression finds tokens, addresses, e-mail and home paths because those have shapes. A client name, a project code name or an unreleased feature has no shape. Add them with an extra pattern and read a sample by hand before you publish anything.

Question 4. True or false: a tool call in a training row should be written in the model family’s wire format, such as the Hermes tool_call tags.

• True
• False

Show the answer and why

Answer: False

False. The wire format is what the chat template produces from the structured message. You store the call in the tool_calls key of an assistant message with a name and an arguments object, and pass the tool schemas in the tools column; the template renders whichever format that model family uses. Writing the tags yourself pins the data to one family and breaks the moment the template changes.

Question 5. Why does the converter split train and validation by task identifier rather than by episode?

• To make the split reproducible
• Because two attempts at the same task are not independent, so a split by episode reports a validation loss better than the model is
• Because TRL requires it
• To keep the files the same size

Show the answer and why

Answer: Because two attempts at the same task are not independent, so a split by episode reports a validation loss better than the model is

Repeated attempts at one task share its wording, its documents and often most of its trajectory. Putting one attempt in each half means the validation set is measuring recall of something the model has seen, which is the same error as evaluation contamination in miniature.

Check my answersClear