Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583#!/usr/bin/env python3"""Shared thinking-renderer validation for Machine training gates."""from __future__ import annotations
import jsonimport refrom pathlib import Pathfrom typing import Any
# Observed model-native thinking channels. Add new entries from dated renderer# receipts; never authorize by model-name substring alone.THINKING_CHANNEL_MARKERS: list[dict[str, Any]] = [ { "id": "tml_content_thinking", "open": "<|content_thinking|>", "close": None, # open-marker channel; text must appear after the marker }, { "id": "qwen_think", "open": "<think>", "close": "</think>", },]
def flatten_values(x): if hasattr(x, "detach"): x = x.detach().cpu() if hasattr(x, "tolist"): x = x.tolist() if isinstance(x, (list, tuple)): out = [] for item in x: out.extend(flatten_values(item)) return out try: return [float(x)] except Exception: return []
def extract_thinking_texts(messages: list[dict[str, Any]]) -> list[str]: texts: list[str] = [] for message in messages: if message.get("role") != "assistant": continue content = message.get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict) and part.get("type") == "thinking": thinking = part.get("thinking") if isinstance(thinking, str) and thinking.strip(): texts.append(thinking.strip()) return texts
def messages_have_thinking(messages: list[dict[str, Any]]) -> bool: return bool(extract_thinking_texts(messages))
def assistant_tool_names(messages: list[dict[str, Any]]) -> list[str]: names: list[str] = [] for message in messages: if message.get("role") != "assistant": continue tool_calls = message.get("tool_calls") if not isinstance(tool_calls, list): continue for call in tool_calls: if not isinstance(call, dict): continue fn = call.get("function") if isinstance(fn, dict) and isinstance(fn.get("name"), str) and fn["name"].strip(): names.append(fn["name"].strip()) return names
def decode_model_input(tokenizer, model_input) -> str: ids = model_input.to_ints() if hasattr(ids, "tolist"): ids = ids.tolist() else: ids = list(ids) return tokenizer.decode(ids)
def normalize_renderer_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: """Convert OpenAI-style tool_call dicts into cookbook ToolCall objects.""" from tinker_cookbook.renderers import ToolCall
normalized: list[dict[str, Any]] = [] for message in messages: if not isinstance(message, dict): normalized.append(message) continue row = dict(message) calls = row.get("tool_calls") if isinstance(calls, list): row["tool_calls"] = [ ToolCall.model_validate(call) if isinstance(call, dict) else call for call in calls ] normalized.append(row) return normalized
def detect_thinking_channels(rendered: str) -> list[str]: found: list[str] = [] for channel in THINKING_CHANNEL_MARKERS: open_marker = channel["open"] close_marker = channel["close"] if close_marker is None: if open_marker in rendered: found.append(channel["id"]) else: pattern = re.escape(open_marker) + r"[\s\S]*?" + re.escape(close_marker) if re.search(pattern, rendered): found.append(channel["id"]) return found
def thinking_text_in_channel(rendered: str, text: str, channel_ids: list[str]) -> bool: snippet = text[: min(40, len(text))] if not snippet: return False for channel in THINKING_CHANNEL_MARKERS: if channel["id"] not in channel_ids: continue open_marker = channel["open"] close_marker = channel["close"] if close_marker is None: idx = rendered.find(open_marker) if idx < 0: continue # Require the reasoning snippet after the open marker (not only somewhere in the prompt). if snippet in rendered[idx + len(open_marker):]: return True else: pattern = re.escape(open_marker) + r"([\s\S]*?)" + re.escape(close_marker) for match in re.finditer(pattern, rendered): if snippet in match.group(1): return True return False
def probe_rendered_shape(messages: list[dict[str, Any]], renderer, tokenizer) -> dict[str, Any]: """Probe one assistant-target prefix under LAST_ASSISTANT_MESSAGE.""" from tinker_cookbook.renderers import TrainOnWhat from assistant_targets import last_assistant_message
model_input, weights = renderer.build_supervised_example( normalize_renderer_messages(messages), train_on_what=TrainOnWhat.LAST_ASSISTANT_MESSAGE, ) flat_weights = flatten_values(weights) rendered = decode_model_input(tokenizer, model_input) last = last_assistant_message(messages) last_only = [last] if last else [] thinking_texts = extract_thinking_texts(last_only) tool_names = assistant_tool_names(last_only) channel_ids = detect_thinking_channels(rendered) has_content_thinking = "tml_content_thinking" in channel_ids has_qwen_think = "qwen_think" in channel_ids has_thinking_channel = bool(channel_ids) has_content_text = "<|content_text|>" in rendered or bool(re.search(r"</think>", rendered)) flattened = False if thinking_texts: if not has_thinking_channel: flattened = True else: for text in thinking_texts: if not thinking_text_in_channel(rendered, text, channel_ids): flattened = True break tool_preserved = True for name in tool_names: if name not in rendered: tool_preserved = False break weighted = sum(1 for value in flat_weights if value > 0) return { "totalTokens": len(flat_weights), "weightedTokens": weighted, "rendered": rendered, "hasContentThinking": has_content_thinking, "hasQwenThink": has_qwen_think, "hasThinkingChannel": has_thinking_channel, "thinkingChannels": channel_ids, "hasContentText": has_content_text, "flattenedThinking": flattened, "toolNames": tool_names, "toolPreserved": tool_preserved, "thinkingCount": len(thinking_texts), "trainOnWhat": "LAST_ASSISTANT_MESSAGE", }
def validate_assistant_target_thinking(messages: list[dict[str, Any]], renderer, tokenizer) -> dict[str, Any]: """Validate one expanded prefix whose LAST assistant may carry ThinkingParts.""" from assistant_targets import last_assistant_message
last = last_assistant_message(messages) if not last or not messages_have_thinking([last]): try: probe = probe_rendered_shape(messages, renderer, tokenizer) except Exception as e: return { "ok": False, "hasThinking": False, "error": f"{type(e).__name__}: {e}", "reason": "renderer execution failed for assistant target", } if probe.get("weightedTokens", 0) <= 0: return { "ok": False, "hasThinking": False, "reason": "assistant target produced no positively weighted tokens", **probe, } return {"ok": True, "hasThinking": False, **probe}
try: probe = probe_rendered_shape(messages, renderer, tokenizer) except Exception as e: return { "ok": False, "hasThinking": True, "error": f"{type(e).__name__}: {e}", "reason": "renderer execution failed for structured thinking trace", } if probe["flattenedThinking"]: return { "ok": False, "hasThinking": True, "reason": "structured thinking was flattened or stripped by the renderer", **probe, } if probe["toolNames"] and not probe["toolPreserved"]: return { "ok": False, "hasThinking": True, "reason": "tool calls were not preserved during rendering", **probe, } if probe["thinkingCount"] > 0 and not probe["hasThinkingChannel"]: return { "ok": False, "hasThinking": True, "reason": "thinking parts did not render to a thinking channel", **probe, } if probe["thinkingCount"] > 0 and probe.get("weightedTokens", 0) <= 0: return { "ok": False, "hasThinking": True, "reason": "thinking trace produced no positively weighted tokens", **probe, } # Tool-call-first thinking targets must keep thinking and the tool call in one weighted target. if probe["toolNames"] and probe["thinkingCount"] > 0: if not probe["toolPreserved"] or not probe["hasThinkingChannel"]: return { "ok": False, "hasThinking": True, "reason": "thinking and tool call were not both present in the positively weighted target", **probe, } return {"ok": True, "hasThinking": True, **probe}
def validate_messages_thinking_renderer(messages: list[dict[str, Any]], renderer, tokenizer) -> dict[str, Any]: """Expand a conversation and validate each assistant-target prefix under LAST mode.""" from assistant_targets import MalformedConversationError, expand_assistant_targets, last_assistant_message
if not messages_have_thinking(messages): return {"ok": True, "hasThinking": False} try: targets = expand_assistant_targets(messages) except MalformedConversationError as e: return { "ok": False, "hasThinking": True, "reason": f"malformed conversation ordering: {e}", }
target_receipts: list[dict[str, Any]] = [] thinking_ok = False for target_index, target in enumerate(targets): last = last_assistant_message(target) last_has_thinking = bool(last and messages_have_thinking([last])) result = validate_assistant_target_thinking(target, renderer, tokenizer) receipt = { "targetIndex": target_index, "ok": result.get("ok"), "hasThinking": result.get("hasThinking"), "totalTokens": result.get("totalTokens"), "weightedTokens": result.get("weightedTokens"), "hasContentThinking": result.get("hasContentThinking"), "hasQwenThink": result.get("hasQwenThink"), "hasThinkingChannel": result.get("hasThinkingChannel"), "thinkingChannels": result.get("thinkingChannels"), "toolPreserved": result.get("toolPreserved"), "toolNames": result.get("toolNames"), "reason": result.get("reason"), } # Later targets should retain earlier tool results as context in the rendered prefix. if target_index > 0: prior_tools = [m for m in target[:-1] if isinstance(m, dict) and m.get("role") == "tool"] tool_context_present = True rendered = result.get("rendered") or "" for tool_msg in prior_tools: content = tool_msg.get("content") snippet = "" if isinstance(content, str): snippet = content.strip()[:40] elif isinstance(content, list): for part in content: if isinstance(part, dict) and isinstance(part.get("text"), str): snippet = part["text"].strip()[:40] break if snippet and snippet not in rendered: tool_context_present = False break receipt["toolContextPreserved"] = tool_context_present if prior_tools and not tool_context_present: result = { **result, "ok": False, "reason": "prior tool results were not preserved as context for a later assistant target", } receipt["ok"] = False receipt["reason"] = result["reason"] target_receipts.append(receipt) if last_has_thinking: thinking_ok = thinking_ok or bool(result.get("ok") and result.get("hasThinking")) if not result.get("ok"): return { "ok": False, "hasThinking": True, "reason": result.get("reason"), "error": result.get("error"), "targetReceipts": target_receipts, "targetCount": len(targets), } elif not result.get("ok"): return { "ok": False, "hasThinking": True, "reason": result.get("reason") or "non-thinking assistant target failed under LAST_ASSISTANT_MESSAGE", "error": result.get("error"), "targetReceipts": target_receipts, "targetCount": len(targets), }
if not thinking_ok: return { "ok": False, "hasThinking": True, "reason": "no ThinkingPart was validated as the LAST assistant message of an expanded target", "targetReceipts": target_receipts, "targetCount": len(targets), }
# Aggregate primary receipt from the first thinking target for dataset summaries. primary = next((r for r in target_receipts if r.get("hasThinking")), target_receipts[0]) return { "ok": True, "hasThinking": True, "totalTokens": primary.get("totalTokens"), "weightedTokens": primary.get("weightedTokens"), "hasContentThinking": primary.get("hasContentThinking"), "hasQwenThink": primary.get("hasQwenThink"), "hasThinkingChannel": primary.get("hasThinkingChannel"), "thinkingChannels": primary.get("thinkingChannels"), "hasContentText": True, "toolPreserved": primary.get("toolPreserved"), "toolNames": primary.get("toolNames"), "thinkingCount": 1, "targetReceipts": target_receipts, "targetCount": len(targets), "trainOnWhat": "LAST_ASSISTANT_MESSAGE", }
def scan_dataset_thinking(path: str) -> dict[str, Any]: """Scan the entire JSONL for good rows and ThinkingParts before importing optional deps.""" thinking_rows: list[dict[str, Any]] = [] valid_rows = 0 source_line = 0 with open(path, encoding="utf-8") as handle: for raw in handle: source_line += 1 line = raw.strip() if not line: continue try: row = json.loads(line) except Exception: continue if not isinstance(row, dict) or row.get("tag", "good") != "good": continue messages = row.get("messages") if not isinstance(messages, list) or not messages: continue valid_rows += 1 if messages_have_thinking(messages): thinking_rows.append({"line": source_line, "messages": messages}) return { "validRows": valid_rows, "totalThinkingRows": len(thinking_rows), "hasThinkingDataset": len(thinking_rows) > 0, "thinkingRows": thinking_rows, }
def validate_dataset_thinking_renderer( file_path: str, model_name: str, max_examples: int | None = None,) -> dict[str, Any]: """ Validate structured thinking against the exact recommended renderer for model_name.
max_examples is retained for API compatibility with older callers. Structured-thinking compatibility always inspects every thinking row; exceeding a positive cap blocks. Prefer max_examples=None (validate all) at the training boundary. """ scan = scan_dataset_thinking(file_path) if not scan["hasThinkingDataset"]: return { "ok": True, "stage": "done", "blocked": False, "hasThinkingDataset": False, "model": model_name, "validRows": scan["validRows"], "totalThinkingRows": 0, "checkedThinkingRows": 0, }
thinking_rows: list[dict[str, Any]] = scan["thinkingRows"] total_thinking = scan["totalThinkingRows"]
# Optional safety cap: if set and exceeded, fail closed rather than silently sample. if max_examples is not None and max_examples > 0 and total_thinking > max_examples: return { "ok": False, "stage": "cap", "blocked": True, "hasThinkingDataset": True, "model": model_name, "validRows": scan["validRows"], "totalThinkingRows": total_thinking, "checkedThinkingRows": 0, "reason": ( f"thinking dataset has {total_thinking} thinking rows; " f"cap is {max_examples}. Validate all thinking rows at the training boundary." ), }
try: from tinker_cookbook import model_info from tinker_cookbook.renderers import get_renderer from tinker_cookbook.tokenizer_utils import get_tokenizer except Exception as e: return { "ok": False, "stage": "import", "blocked": True, "hasThinkingDataset": True, "model": model_name, "validRows": scan["validRows"], "totalThinkingRows": total_thinking, "checkedThinkingRows": 0, "error": f"{type(e).__name__}: {e}", "reason": "thinking dataset requires tinker-cookbook and renderer dependencies", }
try: renderer_name = model_info.get_recommended_renderer_name(model_name) tokenizer = get_tokenizer(model_name) renderer = get_renderer(renderer_name, tokenizer) except Exception as e: return { "ok": False, "stage": "renderer", "blocked": True, "hasThinkingDataset": True, "model": model_name, "validRows": scan["validRows"], "totalThinkingRows": total_thinking, "checkedThinkingRows": 0, "error": f"{type(e).__name__}: {e}", "reason": f"could not load renderer for model {model_name}", }
failures: list[dict[str, Any]] = [] checked = 0 receipts: list[dict[str, Any]] = [] for row in thinking_rows: checked += 1 result = validate_messages_thinking_renderer(row["messages"], renderer, tokenizer) receipt = { "line": row["line"], "totalTokens": result.get("totalTokens"), "weightedTokens": result.get("weightedTokens"), "hasContentThinking": result.get("hasContentThinking"), "hasQwenThink": result.get("hasQwenThink"), "hasThinkingChannel": result.get("hasThinkingChannel"), "thinkingChannels": result.get("thinkingChannels"), "hasContentText": result.get("hasContentText"), "toolPreserved": result.get("toolPreserved"), "targetCount": result.get("targetCount"), "targetReceipts": result.get("targetReceipts"), "trainOnWhat": result.get("trainOnWhat"), } receipts.append(receipt) if not result.get("ok"): failures.append({ "line": row["line"], "reason": result.get("reason"), "error": result.get("error"), })
if failures: return { "ok": False, "stage": "done", "blocked": True, "hasThinkingDataset": True, "model": model_name, "renderer": renderer_name, "validRows": scan["validRows"], "totalThinkingRows": total_thinking, "checkedThinkingRows": checked, "failures": failures[:20], "receipts": receipts[:5], "reason": failures[0]["reason"], }
return { "ok": True, "stage": "done", "blocked": False, "hasThinkingDataset": True, "model": model_name, "renderer": renderer_name, "validRows": scan["validRows"], "totalThinkingRows": total_thinking, "checkedThinkingRows": checked, "receipts": receipts[:5], }
def main(argv: list[str]) -> int: if len(argv) < 3: print(json.dumps({ "ok": False, "stage": "usage", "error": "usage: thinking_renderer_validation.py FILE MODEL [MAX_EXAMPLES]", })) return 0 file_path = argv[1] model_name = argv[2] # Prefer validating all thinking rows. Pass 0 or omit for all; positive caps fail closed if exceeded. max_examples: int | None = None if len(argv) > 3: raw = int(argv[3]) max_examples = None if raw <= 0 else raw if not Path(file_path).exists(): print(json.dumps({"ok": False, "stage": "file", "error": f"file not found: {file_path}"})) return 0 print(json.dumps(validate_dataset_thinking_renderer(file_path, model_name, max_examples))) return 0
if __name__ == "__main__": import sys
raise SystemExit(main(sys.argv))