#!/usr/bin/env python3 """Add conservative DeepSeek pre-labels to wearer-centric EOT/INT candidates. The model receives the candidate source, structured participant timing and the transcript excerpt. It cannot change any source timestamp. Its output is stored as an auditable suggestion beside the untouched rule candidate and must still be accepted, relabeled, excluded, or sent to second review by a human. """ from __future__ import annotations import argparse from collections import Counter import hashlib import json import os from pathlib import Path import random import re import time from typing import Any, Sequence import urllib.error import urllib.request PROMPT_VERSION = "easycom-wearer-centric-eot-int-prelabel-v2" DEFAULT_MODEL = "deepseek-v4-flash" DEFAULT_BASE_URL = "https://api.deepseek.com" VERDICTS = {"POSITIVE", "HARD_NEGATIVE", "NOT_APPLICABLE", "EXCLUDE", "UNSURE"} EVENT_TYPES = {"EOT_RESPONSE", "INT_YIELD"} HARD_NEGATIVE_TYPES = { "EOT_MIDTURN_PAUSE", "EOT_OTHER_TO_OTHER", "EOT_WEARER_BACKCHANNEL", "INT_OTHER_BACKCHANNEL", "INT_NON_TAKEOVER_OVERLAP", "INT_WEARER_HOLDS_FLOOR", "OTHER_HARD_NEGATIVE", } SYSTEM_PROMPT = """You conservatively pre-label candidate regions for a scientific study of wearer-relative conversational floor control. Evidence rules: 1. Use candidate source, structured participant timing, per-speaker activity, and transcript together. Timing evidence has priority over a speculative transcript interpretation. 2. Never alter or propose an event timestamp. Do not infer prosody, gaze, intention, or acoustic cues that were not supplied. 3. A wearer onset is not automatically a floor take; a wearer offset is not automatically a yield; an other onset is not automatically a takeover. A pause is not automatically EOT, overlap is not automatically interruption, and words such as "yeah" are not automatically backchannels. Check who speaks sustainably afterward and who resumes. 4. Use UNSURE when the evidence does not establish conversational function. Never force a label for coverage. Top-level verdicts: - POSITIVE: the system should commit. - HARD_NEGATIVE: a valid, interpretable region where the system should not commit. - NOT_APPLICABLE: a real speech transition outside the committed EOT/INT task, such as a wearer starting after long silence or naturally ending into silence. - EXCLUDE: the region cannot be scored reliably because of annotation corruption, a recording boundary, missing context, a non-speech artifact, timestamp corruption, or another unresolvable defect. - UNSURE: evidence is insufficient and a second human review is needed. POSITIVE event_type must be exactly EOT_RESPONSE or INT_YIELD. EOT_RESPONSE means a non-wearer genuinely releases the floor and the wearer then takes it with a sustained response. INT_YIELD means a non-wearer genuinely takes over during wearer speech and the wearer relinquishes the floor. HARD_NEGATIVE subtype must be exactly one of EOT_MIDTURN_PAUSE, EOT_OTHER_TO_OTHER, EOT_WEARER_BACKCHANNEL, INT_OTHER_BACKCHANNEL, INT_NON_TAKEOVER_OVERLAP, INT_WEARER_HOLDS_FLOOR, or OTHER_HARD_NEGATIVE. Return exactly one JSON object with this shape and no markdown: {"items":[{"uid":"...","verdict":"UNSURE","event_type":null,"subtype":null,"confidence":0.25,"reason":"brief evidence-based sentence","needs_human_review":true}]} Return every supplied uid exactly once. Copy uid byte-for-byte and never add a backslash before an underscore. confidence must be a number from 0 to 1.""" def _normalize_returned_uid(value: Any) -> Any: """Undo DeepSeek's occasional Markdown escaping without weakening identity checks.""" if not isinstance(value, str): return value return re.sub(r"\\+_", "_", value) def _participant_uid_suffix(row: dict[str, Any], ordinal: int) -> str: previous = row.get("previous_other_participant_ids") or [] incoming = row.get("incoming_other_participant_ids") or [] previous_key = "-".join(f"p{x}" for x in previous) or "none" incoming_key = "-".join(f"p{x}" for x in incoming) or "none" return f"prev-{previous_key}__in-{incoming_key}__row-{ordinal}" def read_candidates(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] with path.open("r", encoding="utf-8") as handle: for line_number, line in enumerate(handle, 1): if not line.strip(): continue row = json.loads(line) if not isinstance(row, dict) or not isinstance(row.get("uid"), str): raise ValueError(f"Expected candidate object with uid at {path}:{line_number}") if not (row.get("candidate_source") or row.get("task")): raise ValueError(f"Missing candidate source/task at {path}:{line_number}") rows.append(row) if not rows: raise ValueError(f"No candidates in {path}") by_uid: dict[str, list[int]] = {} for index, row in enumerate(rows): by_uid.setdefault(row["uid"], []).append(index) for original_uid, indices in by_uid.items(): if len(indices) == 1: continue for ordinal, index in enumerate(indices, 1): row = dict(rows[index]) row["uid_disambiguated_from"] = original_uid row["uid"] = f"{original_uid}/{_participant_uid_suffix(row, ordinal)}" row["event_id"] = row["uid"] rows[index] = row if len({row["uid"] for row in rows}) != len(rows): raise ValueError(f"Could not deterministically disambiguate duplicate candidate UIDs in {path}") return rows def candidate_for_prompt(row: dict[str, Any]) -> dict[str, Any]: fields = ( "uid", "candidate_source", "rule_candidate_type", "task", "anchor_type", "anchor_dense_seconds", "event_time", "region_start_seconds", "region_end_seconds", "temporal_evidence", "speaker_timeline", "context", ) payload = {field: row.get(field) for field in fields if row.get(field) is not None} # Older v1 files remain reviewable and can still be pre-labeled. if "candidate_source" not in payload: payload["candidate_source"] = row.get("anchor_type", row.get("event_type", "UNKNOWN")) if "event_time" not in payload: payload["event_time"] = row.get("anchor_dense_seconds") return payload def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str: instruction = ( "Primary pass: classify conservatively." if pass_index == 0 else "Independent skeptical pass: seek alternative explanations and choose uncertain when semantics are unsupported." ) return ( instruction + "\nCandidate JSON:\n" + json.dumps([candidate_for_prompt(row) for row in rows], ensure_ascii=False, separators=(",", ":")) ) def _transport_rows(rows: Sequence[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, str]]: """Use simple batch-local IDs so the model never has to copy path-like UIDs.""" transported: list[dict[str, Any]] = [] original_by_transport: dict[str, str] = {} for index, row in enumerate(rows): transport_uid = f"item-{index:04d}" original_by_transport[transport_uid] = row["uid"] transported.append({**row, "uid": transport_uid}) return transported, original_by_transport def validate_decisions(payload: Any, expected_uids: Sequence[str]) -> list[dict[str, Any]]: if not isinstance(payload, dict) or not isinstance(payload.get("items"), list): raise ValueError("Response must be an object with an items list") expected = set(expected_uids) decisions: dict[str, dict[str, Any]] = {} for item in payload["items"]: if not isinstance(item, dict): raise ValueError("Every decision must be an object") raw_uid = item.get("uid") uid = _normalize_returned_uid(raw_uid) verdict = item.get("verdict") event_type = item.get("event_type") subtype = item.get("subtype") confidence = item.get("confidence") reason = item.get("reason") needs_human_review = item.get("needs_human_review") if uid not in expected or uid in decisions: raise ValueError(f"Unexpected or duplicate uid: {raw_uid!r}") if verdict not in VERDICTS: raise ValueError(f"Invalid verdict for {uid}: {verdict!r}") if verdict == "POSITIVE" and event_type not in EVENT_TYPES: raise ValueError(f"POSITIVE requires EOT_RESPONSE or INT_YIELD for {uid}") if verdict != "POSITIVE" and event_type is not None: raise ValueError(f"Only POSITIVE may have event_type for {uid}") if verdict == "HARD_NEGATIVE" and subtype not in HARD_NEGATIVE_TYPES: raise ValueError(f"HARD_NEGATIVE requires a valid subtype for {uid}") if verdict != "HARD_NEGATIVE" and subtype is not None: raise ValueError(f"Only HARD_NEGATIVE may have subtype for {uid}") if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1: raise ValueError(f"confidence must be in [0, 1] for {uid}") if not isinstance(reason, str) or not reason.strip(): raise ValueError(f"Missing reason for {uid}") if not isinstance(needs_human_review, bool): raise ValueError(f"needs_human_review must be boolean for {uid}") decisions[str(uid)] = { "uid": str(uid), "verdict": str(verdict), "event_type": event_type, "subtype": subtype, "confidence": float(confidence), "reason": reason.strip(), "needs_human_review": needs_human_review, } if set(decisions) != expected: raise ValueError(f"Response uid mismatch; missing={sorted(expected - set(decisions))}") return [decisions[uid] for uid in expected_uids] def _post( *, rows: Sequence[dict[str, Any]], pass_index: int, api_key: str, model: str, base_url: str, timeout_seconds: float, max_retries: int, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: transported, original_by_transport = _transport_rows(rows) body = { "model": model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": build_user_prompt(transported, pass_index)}, ], "response_format": {"type": "json_object"}, "thinking": {"type": "disabled"}, "temperature": 0, "max_tokens": 2048, "stream": False, } request = urllib.request.Request( base_url.rstrip("/") + "/chat/completions", data=json.dumps(body, ensure_ascii=False).encode("utf-8"), headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, method="POST", ) last_error: Exception | None = None for attempt in range(max_retries + 1): try: with urllib.request.urlopen(request, timeout=timeout_seconds) as response: envelope = json.loads(response.read().decode("utf-8")) content = envelope["choices"][0]["message"]["content"] parsed = json.loads(content) decisions = validate_decisions(parsed, list(original_by_transport)) for decision in decisions: decision["uid"] = original_by_transport[decision["uid"]] return decisions, envelope.get("usage", {}) except ( urllib.error.HTTPError, urllib.error.URLError, TimeoutError, KeyError, TypeError, json.JSONDecodeError, ValueError, ) as error: last_error = error if attempt >= max_retries: break time.sleep(min(30.0, 2.0**attempt)) raise RuntimeError(f"DeepSeek request failed after {max_retries + 1} attempts: {last_error}") def add_suggestions( rows: Sequence[dict[str, Any]], *, api_key: str, model: str, base_url: str, passes: int, batch_size: int = 8, timeout_seconds: float, max_retries: int, ) -> tuple[list[dict[str, Any]], Counter[str]]: by_uid: dict[str, list[dict[str, Any]]] = {row["uid"]: [] for row in rows} prompt_hashes: dict[str, list[str]] = {row["uid"]: [] for row in rows} usage: Counter[str] = Counter() if batch_size <= 0: raise ValueError("batch_size must be positive") for pass_index in range(passes): for start in range(0, len(rows), batch_size): batch = rows[start:start + batch_size] transported, _ = _transport_rows(batch) prompt_hash = hashlib.sha256( (SYSTEM_PROMPT + "\n" + build_user_prompt(transported, pass_index)).encode("utf-8") ).hexdigest() decisions, pass_usage = _post( rows=batch, pass_index=pass_index, api_key=api_key, model=model, base_url=base_url, timeout_seconds=timeout_seconds, max_retries=max_retries, ) for decision in decisions: by_uid[decision["uid"]].append(decision) prompt_hashes[decision["uid"]].append(prompt_hash) for key, value in pass_usage.items(): if isinstance(value, int): usage[key] += value output: list[dict[str, Any]] = [] for source in rows: decisions = by_uid[source["uid"]] labels = {(d["verdict"], d["event_type"], d["subtype"]) for d in decisions} unanimous = len(labels) == 1 if unanimous: verdict, event_type, subtype = next(iter(labels)) confidence = min(float(item["confidence"]) for item in decisions) reason = " | ".join(item["reason"] for item in decisions) needs_human_review = any(item["needs_human_review"] for item in decisions) else: verdict, event_type, subtype = "UNSURE", None, None confidence = min(float(item["confidence"]) for item in decisions) reason = "The independent DeepSeek passes disagreed." needs_human_review = True enriched = dict(source) enriched["llm_suggestion"] = { "verdict": verdict, "event_type": event_type, "subtype": subtype, "confidence": confidence, "reason": reason, "needs_human_review": needs_human_review, "unanimous": unanimous, } enriched.update( ai_verdict=verdict, ai_event_type=event_type, ai_subtype=subtype, ai_confidence=confidence, ai_reason=reason, human_status="UNREVIEWED", human_verdict=None, human_event_type=None, human_subtype=None, exclude_reason=None, ) enriched["llm_adjudication"] = { "model": model, "prompt_version": PROMPT_VERSION, "passes": passes, "request_prompt_sha256": prompt_hashes[source["uid"]], "decisions": decisions, "human_review_required": True, } output.append(enriched) return output, usage def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--model", default=os.environ.get("DEEPSEEK_MODEL", DEFAULT_MODEL)) parser.add_argument("--base-url", default=os.environ.get("DEEPSEEK_BASE_URL", DEFAULT_BASE_URL)) parser.add_argument("--passes", type=int, default=2) parser.add_argument("--batch-size", type=int, default=8) parser.add_argument("--max-events", type=int) parser.add_argument("--selection-seed", type=int, default=2027) parser.add_argument("--timeout-seconds", type=float, default=120.0) parser.add_argument("--max-retries", type=int, default=3) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--allow-external-transcript-upload", action="store_true") parser.add_argument("--overwrite", action="store_true") return parser.parse_args() def main() -> None: args = parse_args() if args.passes <= 0 or args.batch_size <= 0 or args.max_retries < 0: raise ValueError("passes/batch-size must be positive and max-retries non-negative") rows = read_candidates(args.input.expanduser().resolve()) if args.max_events is not None: if args.max_events < 0: raise ValueError("--max-events must be non-negative") rng = random.Random(args.selection_seed) rows = rng.sample(rows, min(args.max_events, len(rows))) rows.sort(key=lambda row: row["uid"]) print(f"Candidates={len(rows)} model={args.model} passes={args.passes}") if args.dry_run: print("DRY RUN: no API request will be made.\n") print(build_user_prompt(rows, 0)) return if not args.allow_external_transcript_upload: raise PermissionError( "Refusing to upload transcript excerpts; inspect --dry-run, then pass " "--allow-external-transcript-upload when authorized." ) api_key = os.environ.get("DEEPSEEK_API_KEY") if not api_key: raise RuntimeError("Set DEEPSEEK_API_KEY in the environment") output = args.output.expanduser().resolve() if output.exists() and not args.overwrite: raise FileExistsError(f"Refusing to overwrite {output}; pass --overwrite") enriched, usage = add_suggestions( rows, api_key=api_key, model=args.model, base_url=args.base_url, passes=args.passes, batch_size=args.batch_size, timeout_seconds=args.timeout_seconds, max_retries=args.max_retries, ) output.parent.mkdir(parents=True, exist_ok=True) with output.open("w", encoding="utf-8") as handle: for row in enriched: handle.write(json.dumps(row, ensure_ascii=False) + "\n") print(f"Saved {len(enriched)} suggested candidates to {output}") print("Usage:", dict(usage)) if __name__ == "__main__": main()